fix: preserve role-neutral input policy across compaction and replay

This commit is contained in:
2026-10-20 04:45:50 +09:00
parent 8f759911ab
commit 3b3381900f
46 changed files with 779 additions and 315 deletions
+2 -2
View File
@@ -361,7 +361,7 @@ impl Store for FsStore {
segment_id: SegmentId, segment_id: SegmentId,
entry: &LogEntry, entry: &LogEntry,
) -> Result<(), StoreError> { ) -> Result<(), StoreError> {
entry.validate_system_inputs()?; entry.validate_input_metadata()?;
let _retention = self.retention().lock_live(session_id)?; let _retention = self.retention().lock_live(session_id)?;
let line = serde_json::to_string(entry)?; let line = serde_json::to_string(entry)?;
self.append_line(&self.log_path(session_id, segment_id), &line) self.append_line(&self.log_path(session_id, segment_id), &line)
@@ -450,7 +450,7 @@ impl Store for FsStore {
entries: &[LogEntry], entries: &[LogEntry],
) -> Result<(), StoreError> { ) -> Result<(), StoreError> {
for entry in entries { for entry in entries {
entry.validate_system_inputs()?; entry.validate_input_metadata()?;
} }
let _retention = self.retention().lock_live(session_id)?; let _retention = self.retention().lock_live(session_id)?;
let path = self.log_path(session_id, segment_id); let path = self.log_path(session_id, segment_id);
+47 -37
View File
@@ -85,21 +85,22 @@ pub struct LoggedHistoryDerivation {
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LoggedSessionHistoryMetadata { pub struct LoggedSessionHistoryMetadata {
pub entry_id: LoggedSessionHistoryEntryId, pub entry_id: LoggedSessionHistoryEntryId,
pub origin: LoggedSessionHistoryOrigin, pub origin: LoggedSessionHistoryOrigin,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub derivation: Option<LoggedHistoryDerivation>, pub derivation: Option<LoggedHistoryDerivation>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub system_input: Option<crate::SystemInputMetadata>, pub input_metadata: Option<crate::HistoryInputMetadata>,
} }
impl LoggedSessionHistoryMetadata { 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 { .. }) 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(()) Ok(())
} }
@@ -109,7 +110,7 @@ impl LoggedSessionHistoryMetadata {
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown, origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
system_input: None, input_metadata: None,
} }
} }
} }
@@ -131,34 +132,40 @@ struct HistoryEntryWire {
} }
impl TryFrom<HistoryEntryWire> for LoggedHistoryEntry { impl TryFrom<HistoryEntryWire> for LoggedHistoryEntry {
type Error = crate::SystemInputError; type Error = crate::InputMetadataError;
fn try_from(wire: HistoryEntryWire) -> Result<Self, Self::Error> { fn try_from(wire: HistoryEntryWire) -> Result<Self, Self::Error> {
let entry = Self { let entry = Self {
item: wire.item, item: wire.item,
metadata: wire.metadata, metadata: wire.metadata,
}; };
entry.validate_system_input()?; entry.validate_input_metadata()?;
Ok(entry) Ok(entry)
} }
} }
impl LoggedHistoryEntry { impl LoggedHistoryEntry {
pub fn validate_system_input(&self) -> Result<(), crate::SystemInputError> { pub fn validate_input_metadata(&self) -> Result<(), crate::InputMetadataError> {
self.metadata.validate_system_input()?; self.metadata.validate_input_metadata()?;
if self.metadata.system_input.is_some() { if let Some(input) = &self.metadata.input_metadata {
let LoggedItem::Message { match &self.item {
LoggedItem::Message {
role: crate::LoggedRole::System, role: crate::LoggedRole::System,
content, content,
} = &self.item } => {
else {
return Err(crate::SystemInputError::InconsistentRecord);
};
let [crate::LoggedContentPart::Text { text }] = content.as_slice() else { let [crate::LoggedContentPart::Text { text }] = content.as_slice() else {
return Err(crate::SystemInputError::InconsistentRecord); return Err(crate::InputMetadataError::InconsistentRecord);
}; };
if text.len() > crate::system_input::MAX_SYSTEM_INPUT_BYTES { if text.len() > crate::input_metadata::MAX_SYSTEM_INPUT_BYTES {
return Err(crate::SystemInputError::ResourceLimit); 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(()) Ok(())
@@ -183,14 +190,14 @@ struct SystemEntryWire {
} }
impl TryFrom<SystemEntryWire> for LoggedSystemHistoryEntry { impl TryFrom<SystemEntryWire> for LoggedSystemHistoryEntry {
type Error = crate::SystemInputError; type Error = crate::InputMetadataError;
fn try_from(wire: SystemEntryWire) -> Result<Self, Self::Error> { fn try_from(wire: SystemEntryWire) -> Result<Self, Self::Error> {
let entry = Self { let entry = Self {
item: wire.item, item: wire.item,
metadata: wire.metadata, metadata: wire.metadata,
}; };
entry.validate_system_input()?; entry.validate_input_metadata()?;
Ok(entry) Ok(entry)
} }
} }
@@ -201,30 +208,30 @@ impl LoggedSystemHistoryEntry {
derivation: Option<LoggedHistoryDerivation>, derivation: Option<LoggedHistoryDerivation>,
input: crate::SystemInputEnvelope, input: crate::SystemInputEnvelope,
) -> Self { ) -> Self {
let (body, system_input) = input.into_parts(); let (body, input_metadata) = input.into_parts();
Self { Self {
item: crate::SystemItem::Input { body }, item: crate::SystemItem::Input { body },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin, origin,
derivation, derivation,
system_input: Some(system_input), input_metadata: Some(input_metadata),
}, },
} }
} }
pub fn validate_system_input(&self) -> Result<(), crate::SystemInputError> { pub fn validate_input_metadata(&self) -> Result<(), crate::InputMetadataError> {
self.metadata.validate_system_input()?; self.metadata.validate_input_metadata()?;
match (&self.item, &self.metadata.system_input) { match (&self.item, &self.metadata.input_metadata) {
(crate::SystemItem::Input { body }, Some(_)) => { (crate::SystemItem::Input { body }, Some(_)) => {
if body.len() > crate::system_input::MAX_SYSTEM_INPUT_BYTES { if body.len() > crate::input_metadata::MAX_SYSTEM_INPUT_BYTES {
return Err(crate::SystemInputError::ResourceLimit); return Err(crate::InputMetadataError::ResourceLimit);
} }
} }
(crate::SystemItem::Input { .. }, None) => { (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) => {} (_, None) => {}
} }
Ok(()) Ok(())
@@ -241,9 +248,9 @@ impl LoggedSystemHistoryEntry {
if matches!(self.item, crate::SystemItem::Input { .. }) { if matches!(self.item, crate::SystemItem::Input { .. }) {
return self return self
.metadata .metadata
.system_input .input_metadata
.as_ref() .as_ref()
.and_then(crate::SystemInputMetadata::public_item) .and_then(crate::HistoryInputMetadata::public_item)
.cloned() .cloned()
.unwrap_or_else(|| self.item.public_item()); .unwrap_or_else(|| self.item.public_item());
} }
@@ -268,7 +275,7 @@ mod tests {
}], }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::HumanInput { origin: LoggedSessionHistoryOrigin::HumanInput {
account_id: "account-1".into(), account_id: "account-1".into(),
@@ -293,7 +300,7 @@ mod tests {
"item": { "kind": "input", "body": "context" }, "item": { "kind": "input", "body": "context" },
"metadata": { "metadata": {
"entry_id": "stable-entry", "origin": { "kind": "provided", "producer": "owner" }, "entry_id": "stable-entry", "origin": { "kind": "provided", "producer": "owner" },
"system_input": { "usage": { "input_metadata": { "usage": {
"recording_material": "exclude", "memory_evidence": "exclude", "recording_material": "exclude", "memory_evidence": "exclude",
"observation": "hidden", "compaction": "reference_input" "observation": "hidden", "compaction": "reference_input"
} } } }
@@ -315,10 +322,10 @@ mod tests {
missing["metadata"] missing["metadata"]
.as_object_mut() .as_object_mut()
.unwrap() .unwrap()
.remove("system_input"); .remove("input_metadata");
assert!(decode(missing)); assert!(decode(missing));
let mut absent_policy = wire.clone(); let mut absent_policy = wire.clone();
absent_policy["metadata"]["system_input"] absent_policy["metadata"]["input_metadata"]
.as_object_mut() .as_object_mut()
.unwrap() .unwrap()
.remove("usage"); .remove("usage");
@@ -327,15 +334,18 @@ mod tests {
invalid_role["item"] = invalid_role["item"] =
serde_json::to_value(LoggedItem::from(agen::Item::user_message("not system"))) serde_json::to_value(LoggedItem::from(agen::Item::user_message("not system")))
.unwrap(); .unwrap();
invalid_role["metadata"]["input_metadata"]["public"] = serde_json::json!({
"kind": "context", "body": "public system projection", "presentation": {"kind": "context"}
});
assert!(decode(invalid_role)); assert!(decode(invalid_role));
let mut oversized = wire.clone(); let mut oversized = wire.clone();
if generic { if generic {
oversized["item"]["body"] = "x" oversized["item"]["body"] = "x"
.repeat(crate::system_input::MAX_SYSTEM_INPUT_BYTES + 1) .repeat(crate::input_metadata::MAX_SYSTEM_INPUT_BYTES + 1)
.into(); .into();
} else { } else {
oversized["item"]["content"][0]["text"] = "x" 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(); .into();
} }
assert!(decode(oversized)); assert!(decode(oversized));
@@ -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 //! The enclosing history annotation owns identity, origin and derivation. None of
//! those descriptive facts selects a use policy here. Model text, independent use //! 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; const MAX_JSON_NODES: usize = 8192;
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] #[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum SystemInputError { pub enum InputMetadataError {
#[error("system input payload schema is invalid")] #[error("history input payload schema is invalid")]
InvalidSchema, InvalidSchema,
#[error("system input payload version must be nonzero")] #[error("history input payload version must be nonzero")]
InvalidVersion, InvalidVersion,
#[error("system input resource limit exceeded")] #[error("system input resource limit exceeded")]
ResourceLimit, ResourceLimit,
#[error("system input payload could not be serialized")] #[error("history input payload could not be serialized")]
Serialization, 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, 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, InvalidTypedPayload,
#[error("generic system input is missing explicit metadata")] #[error("provided input is missing explicit metadata")]
MissingMetadata, MissingMetadata,
#[error("system input metadata does not match its logical record")] #[error("input metadata does not match its logical record")]
InconsistentRecord, InconsistentRecord,
} }
@@ -70,7 +70,7 @@ struct PayloadWire {
} }
impl TryFrom<PayloadWire> for FeatureInputPayload { impl TryFrom<PayloadWire> for FeatureInputPayload {
type Error = SystemInputError; type Error = InputMetadataError;
fn try_from(wire: PayloadWire) -> Result<Self, Self::Error> { fn try_from(wire: PayloadWire) -> Result<Self, Self::Error> {
if wire.schema.is_empty() if wire.schema.is_empty()
@@ -80,10 +80,10 @@ impl TryFrom<PayloadWire> for FeatureInputPayload {
byte.is_ascii_lowercase() || byte.is_ascii_digit() || b"-_.:/".contains(&byte) byte.is_ascii_lowercase() || byte.is_ascii_digit() || b"-_.:/".contains(&byte)
}) })
{ {
return Err(SystemInputError::InvalidSchema); return Err(InputMetadataError::InvalidSchema);
} }
if wire.version == 0 { if wire.version == 0 {
return Err(SystemInputError::InvalidVersion); return Err(InputMetadataError::InvalidVersion);
} }
validate_json(&wire.value, MAX_FEATURE_PAYLOAD_BYTES)?; validate_json(&wire.value, MAX_FEATURE_PAYLOAD_BYTES)?;
Ok(Self { Ok(Self {
@@ -99,11 +99,11 @@ impl FeatureInputPayload {
schema: impl Into<String>, schema: impl Into<String>,
version: u32, version: u32,
value: &T, value: &T,
) -> Result<Self, SystemInputError> { ) -> Result<Self, InputMetadataError> {
PayloadWire { PayloadWire {
schema: schema.into(), schema: schema.into(),
version, version,
value: serde_json::to_value(value).map_err(|_| SystemInputError::Serialization)?, value: serde_json::to_value(value).map_err(|_| InputMetadataError::Serialization)?,
} }
.try_into() .try_into()
} }
@@ -122,12 +122,12 @@ impl FeatureInputPayload {
&self, &self,
schema: &str, schema: &str,
version: u32, version: u32,
) -> Result<T, SystemInputError> { ) -> Result<T, InputMetadataError> {
if self.schema != schema || self.version != version { if self.schema != schema || self.version != version {
return Err(SystemInputError::UnsupportedPayload); return Err(InputMetadataError::UnsupportedPayload);
} }
serde_json::from_value(self.value.clone()) 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. /// Public presentation is deliberately not one of these use decisions.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub struct SystemInputUsePolicy { pub struct HistoryInputUsePolicy {
pub recording_material: InputEvidenceUse, pub recording_material: InputEvidenceUse,
pub memory_evidence: InputEvidenceUse, pub memory_evidence: InputEvidenceUse,
pub observation: InputObservationUse, pub observation: InputObservationUse,
pub compaction: InputCompactionUse, 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)] #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(try_from = "MetadataWire")] #[serde(try_from = "MetadataWire")]
pub struct SystemInputMetadata { pub struct HistoryInputMetadata {
usage: SystemInputUsePolicy, usage: HistoryInputUsePolicy,
feature_payload: Option<FeatureInputPayload>, feature_payload: Option<FeatureInputPayload>,
public: Option<PublicSystemItem>, public: Option<PublicSystemItem>,
} }
@@ -179,13 +182,13 @@ pub struct SystemInputMetadata {
#[derive(Deserialize)] #[derive(Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
struct MetadataWire { struct MetadataWire {
usage: SystemInputUsePolicy, usage: HistoryInputUsePolicy,
feature_payload: Option<FeatureInputPayload>, feature_payload: Option<FeatureInputPayload>,
public: Option<PublicSystemItem>, public: Option<PublicSystemItem>,
} }
impl TryFrom<MetadataWire> for SystemInputMetadata { impl TryFrom<MetadataWire> for HistoryInputMetadata {
type Error = SystemInputError; type Error = InputMetadataError;
fn try_from(wire: MetadataWire) -> Result<Self, Self::Error> { fn try_from(wire: MetadataWire) -> Result<Self, Self::Error> {
if let Some(public) = &wire.public { if let Some(public) = &wire.public {
@@ -194,10 +197,10 @@ impl TryFrom<MetadataWire> for SystemInputMetadata {
|| public.kind.len() > MAX_SCHEMA_BYTES || public.kind.len() > MAX_SCHEMA_BYTES
|| public.kind.chars().any(char::is_control) || public.kind.chars().any(char::is_control)
{ {
return Err(SystemInputError::InvalidSchema); return Err(InputMetadataError::InvalidSchema);
} }
let value = 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)?; validate_json(&value, MAX_SYSTEM_PUBLIC_BYTES)?;
} }
Ok(Self { Ok(Self {
@@ -208,12 +211,12 @@ impl TryFrom<MetadataWire> for SystemInputMetadata {
} }
} }
impl SystemInputMetadata { impl HistoryInputMetadata {
pub fn new( pub fn new(
usage: SystemInputUsePolicy, usage: HistoryInputUsePolicy,
feature_payload: Option<FeatureInputPayload>, feature_payload: Option<FeatureInputPayload>,
public: Option<PublicSystemItem>, public: Option<PublicSystemItem>,
) -> Result<Self, SystemInputError> { ) -> Result<Self, InputMetadataError> {
MetadataWire { MetadataWire {
usage, usage,
feature_payload, feature_payload,
@@ -222,7 +225,7 @@ impl SystemInputMetadata {
.try_into() .try_into()
} }
pub fn usage(&self) -> SystemInputUsePolicy { pub fn usage(&self) -> HistoryInputUsePolicy {
self.usage self.usage
} }
@@ -258,22 +261,22 @@ impl SystemInputMetadata {
#[serde(try_from = "InputWire")] #[serde(try_from = "InputWire")]
pub struct SystemInputEnvelope { pub struct SystemInputEnvelope {
body: String, body: String,
metadata: SystemInputMetadata, metadata: HistoryInputMetadata,
} }
#[derive(Deserialize)] #[derive(Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
struct InputWire { struct InputWire {
body: String, body: String,
metadata: SystemInputMetadata, metadata: HistoryInputMetadata,
} }
impl TryFrom<InputWire> for SystemInputEnvelope { impl TryFrom<InputWire> for SystemInputEnvelope {
type Error = SystemInputError; type Error = InputMetadataError;
fn try_from(wire: InputWire) -> Result<Self, Self::Error> { fn try_from(wire: InputWire) -> Result<Self, Self::Error> {
if wire.body.len() > MAX_SYSTEM_INPUT_BYTES { if wire.body.len() > MAX_SYSTEM_INPUT_BYTES {
return Err(SystemInputError::ResourceLimit); return Err(InputMetadataError::ResourceLimit);
} }
Ok(Self { Ok(Self {
body: wire.body, body: wire.body,
@@ -285,8 +288,8 @@ impl TryFrom<InputWire> for SystemInputEnvelope {
impl SystemInputEnvelope { impl SystemInputEnvelope {
pub fn new( pub fn new(
body: impl Into<String>, body: impl Into<String>,
metadata: SystemInputMetadata, metadata: HistoryInputMetadata,
) -> Result<Self, SystemInputError> { ) -> Result<Self, InputMetadataError> {
InputWire { InputWire {
body: body.into(), body: body.into(),
metadata, metadata,
@@ -298,7 +301,7 @@ impl SystemInputEnvelope {
&self.body &self.body
} }
pub fn metadata(&self) -> &SystemInputMetadata { pub fn metadata(&self) -> &HistoryInputMetadata {
&self.metadata &self.metadata
} }
@@ -306,15 +309,15 @@ impl SystemInputEnvelope {
Item::system_message(self.body.clone()) Item::system_message(self.body.clone())
} }
pub fn into_parts(self) -> (String, SystemInputMetadata) { pub fn into_parts(self) -> (String, HistoryInputMetadata) {
(self.body, self.metadata) (self.body, self.metadata)
} }
} }
fn validate_json(value: &Value, max_bytes: usize) -> Result<(), SystemInputError> { fn validate_json(value: &Value, max_bytes: usize) -> Result<(), InputMetadataError> {
fn walk(value: &Value, depth: usize, remaining: &mut usize) -> Result<(), SystemInputError> { fn walk(value: &Value, depth: usize, remaining: &mut usize) -> Result<(), InputMetadataError> {
if depth > MAX_JSON_DEPTH || *remaining == 0 { if depth > MAX_JSON_DEPTH || *remaining == 0 {
return Err(SystemInputError::ResourceLimit); return Err(InputMetadataError::ResourceLimit);
} }
*remaining -= 1; *remaining -= 1;
match value { match value {
@@ -334,9 +337,9 @@ fn validate_json(value: &Value, max_bytes: usize) -> Result<(), SystemInputError
} }
let mut remaining = MAX_JSON_NODES; let mut remaining = MAX_JSON_NODES;
walk(value, 0, &mut remaining)?; 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 { if bytes.len() > max_bytes {
return Err(SystemInputError::ResourceLimit); return Err(InputMetadataError::ResourceLimit);
} }
Ok(()) Ok(())
} }
@@ -353,8 +356,8 @@ mod tests {
omitted: bool, omitted: bool,
} }
fn reference_policy() -> SystemInputUsePolicy { fn reference_policy() -> HistoryInputUsePolicy {
SystemInputUsePolicy { HistoryInputUsePolicy {
recording_material: InputEvidenceUse::Exclude, recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden, observation: InputObservationUse::Hidden,
@@ -375,7 +378,7 @@ mod tests {
.unwrap(); .unwrap();
let envelope = SystemInputEnvelope::new( let envelope = SystemInputEnvelope::new(
"private model context", "private model context",
SystemInputMetadata::new(reference_policy(), Some(payload), None).unwrap(), HistoryInputMetadata::new(reference_policy(), Some(payload), None).unwrap(),
) )
.unwrap(); .unwrap();
let restored: SystemInputEnvelope = let restored: SystemInputEnvelope =
@@ -387,7 +390,7 @@ mod tests {
assert_eq!(saved_payload.version(), 99); assert_eq!(saved_payload.version(), 99);
assert_eq!( assert_eq!(
saved_payload.decode::<OwnerPayload>("known-owner:context", 1), saved_payload.decode::<OwnerPayload>("known-owner:context", 1),
Err(SystemInputError::UnsupportedPayload) Err(InputMetadataError::UnsupportedPayload)
); );
assert_eq!( assert_eq!(
serde_json::to_value(&restored).unwrap(), serde_json::to_value(&restored).unwrap(),
@@ -415,7 +418,7 @@ mod tests {
.unwrap(); .unwrap();
assert_eq!( assert_eq!(
invalid.decode::<OwnerPayload>("owner:context", 1), invalid.decode::<OwnerPayload>("owner:context", 1),
Err(SystemInputError::InvalidTypedPayload) Err(InputMetadataError::InvalidTypedPayload)
); );
} }
@@ -462,14 +465,14 @@ mod tests {
] { ] {
let mut absent = valid.clone(); let mut absent = valid.clone();
absent.as_object_mut().unwrap().remove(field); absent.as_object_mut().unwrap().remove(field);
assert!(serde_json::from_value::<SystemInputUsePolicy>(absent).is_err()); assert!(serde_json::from_value::<HistoryInputUsePolicy>(absent).is_err());
let mut unknown = valid.clone(); let mut unknown = valid.clone();
unknown[field] = Value::String("future-allow".into()); unknown[field] = Value::String("future-allow".into());
assert!(serde_json::from_value::<SystemInputUsePolicy>(unknown).is_err()); assert!(serde_json::from_value::<HistoryInputUsePolicy>(unknown).is_err());
} }
let mut extra = valid; let mut extra = valid;
extra["feature_name"] = Value::String("infer-policy-from-me".into()); extra["feature_name"] = Value::String("infer-policy-from-me".into());
assert!(serde_json::from_value::<SystemInputUsePolicy>(extra).is_err()); assert!(serde_json::from_value::<HistoryInputUsePolicy>(extra).is_err());
} }
#[test] #[test]
@@ -483,7 +486,7 @@ mod tests {
body: "explicit public text".into(), body: "explicit public text".into(),
presentation: PublicSystemPresentation::Context, presentation: PublicSystemPresentation::Context,
}; };
let metadata = SystemInputMetadata::new( let metadata = HistoryInputMetadata::new(
reference_policy(), reference_policy(),
Some( Some(
FeatureInputPayload::encode( FeatureInputPayload::encode(
@@ -519,14 +522,14 @@ mod tests {
#[test] #[test]
fn use_policies_remain_independent_of_each_other_and_of_public_display() { fn use_policies_remain_independent_of_each_other_and_of_public_display() {
let usage = SystemInputUsePolicy { let usage = HistoryInputUsePolicy {
recording_material: InputEvidenceUse::Include, recording_material: InputEvidenceUse::Include,
memory_evidence: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Readable, observation: InputObservationUse::Readable,
compaction: InputCompactionUse::Retain, compaction: InputCompactionUse::Retain,
}; };
let metadata = SystemInputMetadata::new(usage, None, None).unwrap(); let metadata = HistoryInputMetadata::new(usage, None, None).unwrap();
let restored: SystemInputMetadata = let restored: HistoryInputMetadata =
serde_json::from_value(serde_json::to_value(&metadata).unwrap()).unwrap(); serde_json::from_value(serde_json::to_value(&metadata).unwrap()).unwrap();
assert_eq!(restored.usage(), usage); assert_eq!(restored.usage(), usage);
assert!(restored.public_item().is_none()); assert!(restored.public_item().is_none());
@@ -535,7 +538,7 @@ mod tests {
#[test] #[test]
fn model_body_and_public_projection_have_independent_resource_limits() { 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!( assert!(
SystemInputEnvelope::new("x".repeat(MAX_SYSTEM_INPUT_BYTES + 1), metadata).is_err() 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), body: "x".repeat(MAX_SYSTEM_PUBLIC_BYTES),
presentation: PublicSystemPresentation::Context, 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 } }); let invalid = serde_json::json!({ "body": "small", "metadata": { "public": null, "feature_payload": null } });
assert!(serde_json::from_value::<SystemInputEnvelope>(invalid).is_err()); assert!(serde_json::from_value::<SystemInputEnvelope>(invalid).is_err());
} }
+86 -30
View File
@@ -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, //! These types are intentionally private to `session-store`. Current writers,
//! replay, and public projections use [`crate::LogEntry`] exclusively; only the //! replay, and public projections use [`crate::LogEntry`] exclusively; only the
@@ -93,14 +93,14 @@ impl LegacySystemItem {
fn into_entry( fn into_entry(
self, self,
mut metadata: LoggedSessionHistoryMetadata, mut metadata: LoggedSessionHistoryMetadata,
) -> Result<LoggedSystemHistoryEntry, crate::SystemInputError> { ) -> Result<LoggedSystemHistoryEntry, crate::InputMetadataError> {
use crate::InputCompactionUse::{ReferenceInput, Retain}; use crate::InputCompactionUse::{ReferenceInput, Retain};
let item = match self { let item = match self {
Self::Core(item) => item, Self::Core(item) => item,
Self::Feature(feature) => { Self::Feature(feature) => {
metadata.validate_system_input()?; metadata.validate_input_metadata()?;
if metadata.system_input.is_some() { if metadata.input_metadata.is_some() {
return Err(crate::SystemInputError::InconsistentRecord); return Err(crate::InputMetadataError::InconsistentRecord);
} }
let (body, input_metadata) = match feature { let (body, input_metadata) = match feature {
LegacyFeatureSystemItem::JournalStartup { body, payload } => { LegacyFeatureSystemItem::JournalStartup { body, payload } => {
@@ -150,12 +150,12 @@ impl LegacySystemItem {
} }
}; };
// Identity/origin/derivation are retained, never re-inferred. // Identity/origin/derivation are retained, never re-inferred.
metadata.system_input = Some(input_metadata); metadata.input_metadata = Some(input_metadata);
SystemItem::Input { body } SystemItem::Input { body }
} }
}; };
let entry = LoggedSystemHistoryEntry { item, metadata }; let entry = LoggedSystemHistoryEntry { item, metadata };
entry.validate_system_input()?; entry.validate_input_metadata()?;
Ok(entry) Ok(entry)
} }
} }
@@ -166,8 +166,8 @@ fn migrated_feature_input<T: Serialize>(
schema: &str, schema: &str,
payload: T, payload: T,
compaction: crate::InputCompactionUse, compaction: crate::InputCompactionUse,
) -> Result<(String, crate::SystemInputMetadata), crate::SystemInputError> { ) -> Result<(String, crate::HistoryInputMetadata), crate::InputMetadataError> {
let metadata = crate::SystemInputMetadata::new( let metadata = crate::HistoryInputMetadata::new(
migration_policy(compaction), migration_policy(compaction),
Some(crate::FeatureInputPayload::encode(schema, 1, &payload)?), Some(crate::FeatureInputPayload::encode(schema, 1, &payload)?),
Some(crate::PublicSystemItem { Some(crate::PublicSystemItem {
@@ -207,7 +207,7 @@ enum LegacyAnnotatedOrCurrent {
} }
impl LegacyAnnotatedOrCurrent { impl LegacyAnnotatedOrCurrent {
fn into_current(self) -> Result<LogEntry, crate::SystemInputError> { fn into_current(self) -> Result<LogEntry, crate::InputMetadataError> {
match self { match self {
Self::Annotated(LegacyAnnotatedLogEntry::AnnotatedSystemItem { Self::Annotated(LegacyAnnotatedLogEntry::AnnotatedSystemItem {
ts, ts,
@@ -285,16 +285,19 @@ pub(crate) fn decode_entry(
segment_id: SegmentId, segment_id: SegmentId,
line_index: usize, line_index: usize,
) -> Result<LogEntry, serde_json::Error> { ) -> Result<LogEntry, serde_json::Error> {
let mut wire: serde_json::Value = serde_json::from_str(line)?;
migrate_legacy_metadata_field(&mut wire)?;
let entry = match schema_version { let entry = match schema_version {
1 => match serde_json::from_str::<LegacySessionLogEntryV1>(line)? { 1 => match serde_json::from_value::<LegacySessionLogEntryV1>(wire)? {
LegacySessionLogEntryV1::History(entry) => Entry::History(entry), LegacySessionLogEntryV1::History(entry) => Entry::History(entry),
LegacySessionLogEntryV1::Current(entry) => Entry::Current(entry), LegacySessionLogEntryV1::Current(entry) => Entry::Current(entry),
}, },
2 => match serde_json::from_str::<LegacySessionLogEntryV2>(line)? { 2 => match serde_json::from_value::<LegacySessionLogEntryV2>(wire)? {
LegacySessionLogEntryV2::History(entry) => Entry::History(entry), LegacySessionLogEntryV2::History(entry) => Entry::History(entry),
LegacySessionLogEntryV2::Current(entry) => Entry::Current(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}"), _ => unreachable!("legacy decoder called for unsupported schema {schema_version}"),
}; };
match entry { match entry {
@@ -306,6 +309,53 @@ pub(crate) fn decode_entry(
.map_err(serde::de::Error::custom) .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 /// Schema 3 used DerivedSummary for Host compaction seeds. Only the trusted
/// creation boundary may identify one; arbitrary text and later seed items do not. /// creation boundary may identify one; arbitrary text and later seed items do not.
/// Include already-upgraded entries so an interrupted multi-file migration resumes. /// 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 /// readers/restoration. Invalid or oversized payloads fail migration; the generic
/// payload/body/public limits are deliberately not relaxed for legacy data. /// payload/body/public limits are deliberately not relaxed for legacy data.
pub(crate) fn migrate_system_inputs(entry: &mut LogEntry) -> Result<(), crate::SystemInputError> { pub(crate) fn migrate_system_inputs(entry: &mut LogEntry) -> Result<(), crate::InputMetadataError> {
entry.validate_system_inputs()?; entry.validate_input_metadata()?;
match entry { match entry {
// Feature wire has already been converted by the explicit old-schema // Feature wire has already been converted by the explicit old-schema
// decoder. Ordinary LogEntry can contain only generic Input/core events. // 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::AnnotatedSegmentStart { history, .. }
| LogEntry::AnnotatedUserInput { history, .. } => { | LogEntry::AnnotatedUserInput { history, .. } => {
for entry in 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 { fn migration_policy(compaction: crate::InputCompactionUse) -> crate::HistoryInputUsePolicy {
crate::SystemInputUsePolicy { crate::HistoryInputUsePolicy {
recording_material: crate::InputEvidenceUse::Exclude, recording_material: crate::InputEvidenceUse::Exclude,
memory_evidence: crate::InputEvidenceUse::Exclude, memory_evidence: crate::InputEvidenceUse::Exclude,
observation: crate::InputObservationUse::Hidden, observation: crate::InputObservationUse::Hidden,
@@ -397,18 +447,22 @@ fn migration_policy(compaction: crate::InputCompactionUse) -> crate::SystemInput
fn migrate_flattened_system_input( fn migrate_flattened_system_input(
entry: &mut LoggedHistoryEntry, entry: &mut LoggedHistoryEntry,
) -> Result<(), crate::SystemInputError> { ) -> Result<(), crate::InputMetadataError> {
// Check before supplying legacy policy: Provided without metadata is a // Check before supplying legacy policy: Provided without metadata is a
// malformed NEW record, not permission to synthesize a producer policy. // malformed NEW record, not permission to synthesize a producer policy.
entry.validate_system_input()?; entry.validate_input_metadata()?;
if entry.metadata.system_input.is_none() if entry.metadata.input_metadata.is_none()
&& matches!( && (matches!(
entry.item, entry.item,
LoggedItem::Message { LoggedItem::Message {
role: crate::LoggedRole::System, role: crate::LoggedRole::System,
.. ..
} }
) ) || matches!(
entry.metadata.origin,
LoggedSessionHistoryOrigin::DerivedSummary
| LoggedSessionHistoryOrigin::CompactionSummary
))
{ {
let compaction = match entry.metadata.origin { let compaction = match entry.metadata.origin {
LoggedSessionHistoryOrigin::JournalContext => crate::InputCompactionUse::ReferenceInput, LoggedSessionHistoryOrigin::JournalContext => crate::InputCompactionUse::ReferenceInput,
@@ -417,14 +471,16 @@ fn migrate_flattened_system_input(
} }
_ => crate::InputCompactionUse::Retain, _ => crate::InputCompactionUse::Retain,
}; };
// No Feature payload or public projection can be recovered from text. // Legacy derived context was excluded from independent evidence. Preserve
entry.metadata.system_input = Some(crate::SystemInputMetadata::new( // 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), migration_policy(compaction),
None, None,
None, None,
)?); )?);
} }
entry.validate_system_input() entry.validate_input_metadata()
} }
enum Entry { enum Entry {
@@ -442,7 +498,7 @@ fn legacy_metadata(
identity.extend_from_slice(&(line_index as u64).to_be_bytes()); identity.extend_from_slice(&(line_index as u64).to_be_bytes());
identity.extend_from_slice(&(item_index as u64).to_be_bytes()); identity.extend_from_slice(&(item_index as u64).to_be_bytes());
LoggedSessionHistoryMetadata { LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId(format!( entry_id: LoggedSessionHistoryEntryId(format!(
"l-{}", "l-{}",
base64::Engine::encode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, identity) base64::Engine::encode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, identity)
@@ -457,7 +513,7 @@ fn canonicalize_history_entry(
segment_id: SegmentId, segment_id: SegmentId,
line_index: usize, line_index: usize,
entry: LegacyHistoryLogEntry, entry: LegacyHistoryLogEntry,
) -> Result<LogEntry, crate::SystemInputError> { ) -> Result<LogEntry, crate::InputMetadataError> {
Ok(match entry { Ok(match entry {
LegacyHistoryLogEntry::SegmentStart { LegacyHistoryLogEntry::SegmentStart {
ts, ts,
@@ -24,7 +24,7 @@ fn seed(
content: vec![crate::LoggedContentPart::Text { text: text.into() }], content: vec![crate::LoggedContentPart::Text { text: text.into() }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId("summary".into()), entry_id: LoggedSessionHistoryEntryId("summary".into()),
origin, origin,
derivation: Some(crate::LoggedHistoryDerivation { derivation: Some(crate::LoggedHistoryDerivation {
@@ -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"]}}}}
@@ -27,7 +27,7 @@ fn encode(entries: &[LogEntry]) -> Vec<u8> {
fn migrate(entries: &mut [LogEntry]) { fn migrate(entries: &mut [LogEntry]) {
for entry in entries { for entry in entries {
migrate_system_inputs(entry).unwrap(); 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.entry_id, old_metadata.entry_id);
assert_eq!(entry.metadata.origin, old_metadata.origin); assert_eq!(entry.metadata.origin, old_metadata.origin);
assert_eq!(entry.metadata.derivation, old_metadata.derivation); 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(); let public = metadata.public_item().unwrap();
assert_eq!(public.kind, old_item["kind"].as_str().unwrap()); assert_eq!(public.kind, old_item["kind"].as_str().unwrap());
assert_eq!(public.body, old_item["body"].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, history[index].metadata.derivation,
old[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_eq!(metadata.usage(), migration_policy(compaction));
assert!(metadata.public_item().is_none()); assert!(metadata.public_item().is_none());
assert!(metadata.feature_payload().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(); fs::write(&old_path, source).unwrap();
let store = WorkerSessionStore::new(root.path()).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(); let restored = store.read_all(session, segment).unwrap();
assert_eq!(encode(&restored), new_bytes); assert_eq!(encode(&restored), new_bytes);
assert_eq!(fs::read(&new_path).unwrap(), 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() { 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(); let mut oversized: Value = serde_json::from_str(V4.lines().next().unwrap()).unwrap();
oversized["entry"]["item"]["snapshot"] = 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(); let mut unknown: Value = serde_json::from_str(V4.lines().next().unwrap()).unwrap();
unknown["entry"]["item"]["kind"] = json!("unsupported_future_feature"); unknown["entry"]["item"]["kind"] = json!("unsupported_future_feature");
let mut malformed = unknown.clone(); let mut malformed = unknown.clone();
@@ -295,7 +295,7 @@ fn all_segments_preflight_keeps_old_bytes_on_decode_or_conversion_failure() {
#[test] #[test]
fn unsupported_schema_preserves_all_old_bytes() { 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 root = tempfile::tempdir().unwrap();
let (_, segment) = write_worker(root.path(), version, V4.as_bytes()); let (_, segment) = write_worker(root.path(), version, V4.as_bytes());
let path = root 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 store = FsStore::new(root.path()).unwrap();
let entries = store.read_all(session, segment).unwrap(); let entries = store.read_all(session, segment).unwrap();
for entry in entries { for entry in entries {
entry.validate_system_inputs().unwrap(); entry.validate_input_metadata().unwrap();
} }
assert_eq!( assert_eq!(
fs::read_to_string(dir.join("session-schema.json")) fs::read_to_string(dir.join("session-schema.json"))
.unwrap() .unwrap()
.trim(), .trim(),
"5" "6"
); );
let once = fs::read(&path).unwrap(); let once = fs::read(&path).unwrap();
drop(store); drop(store);
@@ -381,7 +381,7 @@ fn compaction_origin_upgrade_is_retained_only_for_pre_v4_sources() {
} }
); );
assert_eq!( assert_eq!(
metadata.system_input.as_ref().unwrap().usage().compaction, metadata.input_metadata.as_ref().unwrap().usage().compaction,
if version == 3 { if version == 3 {
InputCompactionUse::OpeningSummary InputCompactionUse::OpeningSummary
} else { } else {
@@ -423,7 +423,7 @@ fn current_v5_rejects_all_legacy_feature_kinds_without_rewriting_sources() {
assert!(serde_json::from_str::<LogEntry>(line).is_err()); assert!(serde_json::from_str::<LogEntry>(line).is_err());
let bytes = format!("{line}\n"); let bytes = format!("{line}\n");
let root = tempfile::tempdir().unwrap(); 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 let path = root
.path() .path()
.join("segments") .join("segments")
@@ -495,7 +495,7 @@ fn v1_v2_feature_wire_uses_the_same_typed_payloads_and_stable_legacy_identity()
else { else {
panic!() 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"); assert_eq!(entry.item.history_text(), "exact archived body");
let payload = entry let payload = entry
.metadata .metadata
.system_input .input_metadata
.as_ref() .as_ref()
.unwrap() .unwrap()
.feature_payload() .feature_payload()
@@ -573,7 +573,7 @@ fn frozen_behavior_counter_is_accepted_only_at_the_old_schema_boundary() {
} }
#[test] #[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(); let mut missing: Value = serde_json::from_str(V4.lines().last().unwrap()).unwrap();
missing["entry"]["metadata"] missing["entry"]["metadata"]
.as_object_mut() .as_object_mut()
@@ -581,7 +581,7 @@ fn current_v5_rejects_generic_input_without_metadata_instead_of_migrating_it() {
.remove("system_input"); .remove("system_input");
let bytes = format!("{}\n", missing); let bytes = format!("{}\n", missing);
let root = tempfile::tempdir().unwrap(); 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!(WorkerSessionStore::new(root.path()).is_err());
assert_eq!( assert_eq!(
fs::read( fs::read(
@@ -592,5 +592,112 @@ fn current_v5_rejects_generic_input_without_metadata_instead_of_migrating_it() {
.unwrap(), .unwrap(),
bytes.as_bytes() 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<Value> = 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);
}
} }
+5 -5
View File
@@ -41,6 +41,7 @@ pub(crate) fn read_without_atime_bounded(path: &Path, max_bytes: u64) -> io::Res
pub mod event_trace; pub mod event_trace;
pub mod fs_store; pub mod fs_store;
pub mod history; pub mod history;
pub mod input_metadata;
mod legacy_session_log; mod legacy_session_log;
pub mod logged_item; pub mod logged_item;
mod paste_artifact; mod paste_artifact;
@@ -51,7 +52,6 @@ pub mod retention;
pub mod segment; pub mod segment;
pub mod segment_log; pub mod segment_log;
pub mod store; pub mod store;
pub mod system_input;
pub mod system_item; pub mod system_item;
pub mod uploaded_file; pub mod uploaded_file;
pub mod worker_metadata; pub mod worker_metadata;
@@ -66,6 +66,10 @@ pub use history::{
LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, LoggedSystemHistoryEntry, LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, LoggedSystemHistoryEntry,
LoggedWorkerSubject, 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 logged_item::{LoggedContentPart, LoggedItem, LoggedRole, from_logged, to_logged};
pub use paste_artifact::PasteArtifactLimits; pub use paste_artifact::PasteArtifactLimits;
pub use public_index::{ pub use public_index::{
@@ -98,10 +102,6 @@ pub use segment::{
}; };
pub use segment_log::{LogEntry, RestoredState, SegmentOrigin, SessionExtension, collect_state}; pub use segment_log::{LogEntry, RestoredState, SegmentOrigin, SessionExtension, collect_state};
pub use store::{Store, StoreError}; pub use store::{Store, StoreError};
pub use system_input::{
FeatureInputPayload, InputCompactionUse, InputEvidenceUse, InputObservationUse,
SystemInputEnvelope, SystemInputError, SystemInputMetadata, SystemInputUsePolicy,
};
pub use system_item::{ pub use system_item::{
PromptRenderProvenance, PublicSystemItem, PublicSystemPresentation, SystemItem, SystemReminder, PromptRenderProvenance, PublicSystemItem, PublicSystemPresentation, SystemItem, SystemReminder,
SystemReminderSource, render_worker_event, SystemReminderSource, render_worker_event,
+2 -2
View File
@@ -1296,7 +1296,7 @@ mod tests {
LoggedHistoryEntry { LoggedHistoryEntry {
item, item,
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId(id.to_owned()), entry_id: LoggedSessionHistoryEntryId(id.to_owned()),
origin, origin,
derivation: None, derivation: None,
@@ -1538,7 +1538,7 @@ mod tests {
prompt_provenance: None, prompt_provenance: None,
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId("system-item".into()), entry_id: LoggedSessionHistoryEntryId("system-item".into()),
origin: LoggedSessionHistoryOrigin::BackendInstruction { origin: LoggedSessionHistoryOrigin::BackendInstruction {
operation_id: None, operation_id: None,
+5 -6
View File
@@ -371,11 +371,10 @@ impl SessionPublicMaterialReader {
let source_ids: Option<Vec<String>> = if compact { let source_ids: Option<Vec<String>> = if compact {
let summaries: Vec<_> = history let summaries: Vec<_> = history
.iter() .iter()
.filter(|e| { .filter(|entry| {
matches!( entry.metadata.input_metadata.as_ref().is_some_and(|input| {
e.metadata.origin, input.usage().compaction == crate::InputCompactionUse::OpeningSummary
LoggedSessionHistoryOrigin::CompactionSummary })
)
}) })
.collect(); .collect();
if summaries.len() != 1 { if summaries.len() != 1 {
@@ -895,7 +894,7 @@ fn canonical_body(
entry: &LoggedHistoryEntry, entry: &LoggedHistoryEntry,
purpose: PublicMaterialUse, purpose: PublicMaterialUse,
) -> Option<CanonicalBody<'_>> { ) -> Option<CanonicalBody<'_>> {
if let Some(input) = &entry.metadata.system_input { if let Some(input) = &entry.metadata.input_metadata {
let usage = input.usage(); let usage = input.usage();
let allowed = match purpose { let allowed = match purpose {
PublicMaterialUse::Recording => usage.recording_material, PublicMaterialUse::Recording => usage.recording_material,
@@ -9,7 +9,7 @@ fn entry(id: &str, item: LoggedItem) -> LoggedHistoryEntry {
LoggedHistoryEntry { LoggedHistoryEntry {
item, item,
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId(id.into()), entry_id: LoggedSessionHistoryEntryId(id.into()),
origin: LoggedSessionHistoryOrigin::LocalHumanInput, origin: LoggedSessionHistoryOrigin::LocalHumanInput,
derivation: None, derivation: None,
@@ -31,9 +31,9 @@ fn system_context(
compaction: crate::InputCompactionUse, compaction: crate::InputCompactionUse,
) -> LoggedHistoryEntry { ) -> LoggedHistoryEntry {
let mut entry = message(id, LoggedRole::System, text); let mut entry = message(id, LoggedRole::System, text);
entry.metadata.system_input = Some( entry.metadata.input_metadata = Some(
crate::SystemInputMetadata::new( crate::HistoryInputMetadata::new(
crate::SystemInputUsePolicy { crate::HistoryInputUsePolicy {
recording_material: crate::InputEvidenceUse::Exclude, recording_material: crate::InputEvidenceUse::Exclude,
memory_evidence: crate::InputEvidenceUse::Exclude, memory_evidence: crate::InputEvidenceUse::Exclude,
observation: crate::InputObservationUse::Hidden, 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(), body: "previous Journal must not become new recording material".into(),
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: Some( input_metadata: Some(
crate::SystemInputMetadata::new( crate::HistoryInputMetadata::new(
crate::SystemInputUsePolicy { crate::HistoryInputUsePolicy {
recording_material: crate::InputEvidenceUse::Exclude, recording_material: crate::InputEvidenceUse::Exclude,
memory_evidence: crate::InputEvidenceUse::Exclude, memory_evidence: crate::InputEvidenceUse::Exclude,
observation: crate::InputObservationUse::Hidden, observation: crate::InputObservationUse::Hidden,
@@ -812,9 +812,9 @@ fn schema_v3_compaction_migrates_persistently_and_strict_reader_keeps_original_i
panic!() panic!()
}; };
summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary;
summary.metadata.system_input = Some( summary.metadata.input_metadata = Some(
crate::SystemInputMetadata::new( crate::HistoryInputMetadata::new(
crate::SystemInputUsePolicy { crate::HistoryInputUsePolicy {
recording_material: crate::InputEvidenceUse::Exclude, recording_material: crate::InputEvidenceUse::Exclude,
memory_evidence: crate::InputEvidenceUse::Exclude, memory_evidence: crate::InputEvidenceUse::Exclude,
observation: crate::InputObservationUse::Hidden, observation: crate::InputObservationUse::Hidden,
@@ -878,7 +878,9 @@ fn midturn_compaction_uses_exact_source_ids_and_preserves_discarded_originals()
"never use as original", "never use as original",
crate::InputCompactionUse::OpeningSummary, crate::InputCompactionUse::OpeningSummary,
); );
summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; summary.metadata.origin = LoggedSessionHistoryOrigin::Provided {
producer: "renamed-summary-owner".into(),
};
summary.metadata.derivation = Some(crate::LoggedHistoryDerivation { summary.metadata.derivation = Some(crate::LoggedHistoryDerivation {
sources: vec![ sources: vec![
LoggedSessionHistoryEntryId("u".into()), LoggedSessionHistoryEntryId("u".into()),
@@ -1311,8 +1313,8 @@ fn compaction_without_exact_derivation_or_with_invalid_turn_is_rejected() {
#[test] #[test]
fn generic_system_material_uses_explicit_evidence_purpose_not_producer_or_body() { fn generic_system_material_uses_explicit_evidence_purpose_not_producer_or_body() {
use crate::{ use crate::{
FeatureInputPayload, InputCompactionUse, InputEvidenceUse, InputObservationUse, FeatureInputPayload, HistoryInputMetadata, HistoryInputUsePolicy, InputCompactionUse,
LoggedSystemHistoryEntry, SystemInputEnvelope, SystemInputMetadata, SystemInputUsePolicy, InputEvidenceUse, InputObservationUse, LoggedSystemHistoryEntry, SystemInputEnvelope,
}; };
for worker_layout in [false, true] { for worker_layout in [false, true] {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
@@ -1328,8 +1330,8 @@ fn generic_system_material_uses_explicit_evidence_purpose_not_producer_or_body()
None, None,
SystemInputEnvelope::new( SystemInputEnvelope::new(
format!("private model text: {body}"), format!("private model text: {body}"),
SystemInputMetadata::new( HistoryInputMetadata::new(
SystemInputUsePolicy { HistoryInputUsePolicy {
recording_material, recording_material,
memory_evidence, memory_evidence,
observation: InputObservationUse::Hidden, observation: InputObservationUse::Hidden,
+22 -13
View File
@@ -1625,9 +1625,9 @@ pub(crate) fn project_session_snapshot_for_segment(
derivation_ids(entry), derivation_ids(entry),
entry entry
.metadata .metadata
.system_input .input_metadata
.as_ref() .as_ref()
.map(crate::SystemInputMetadata::public_use_policy), .map(crate::HistoryInputMetadata::public_use_policy),
)), )),
LogEntry::RunYielded { LogEntry::RunYielded {
ts, ts,
@@ -1797,9 +1797,9 @@ fn history_entry(
SessionSnapshotEntry { SessionSnapshotEntry {
use_policy: entry use_policy: entry
.metadata .metadata
.system_input .input_metadata
.as_ref() .as_ref()
.map(crate::SystemInputMetadata::public_use_policy), .map(crate::HistoryInputMetadata::public_use_policy),
entry_id: entry.metadata.entry_id.0.clone(), entry_id: entry.metadata.entry_id.0.clone(),
timestamp, timestamp,
provenance: provenance(&entry.metadata.origin), provenance: provenance(&entry.metadata.origin),
@@ -1926,7 +1926,16 @@ pub fn session_tool_attachment_id(
} }
fn project_history_item(entry: &LoggedHistoryEntry) -> Option<SessionSnapshotEntryData> { fn project_history_item(entry: &LoggedHistoryEntry) -> Option<SessionSnapshotEntryData> {
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(); let public = input.public_projection();
return Some(SessionSnapshotEntryData::SystemItem { return Some(SessionSnapshotEntryData::SystemItem {
item_kind: public.kind.clone(), item_kind: public.kind.clone(),
@@ -2310,7 +2319,7 @@ mod tests {
content: vec![LoggedContentPart::Text { text: text.into() }], content: vec![LoggedContentPart::Text { text: text.into() }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId(entry_id.into()), entry_id: LoggedSessionHistoryEntryId(entry_id.into()),
origin, origin,
derivation: None, derivation: None,
@@ -3609,7 +3618,7 @@ mod tests {
.map(|item| LoggedHistoryEntry { .map(|item| LoggedHistoryEntry {
item, item,
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown, origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
@@ -3664,7 +3673,7 @@ mod tests {
}], }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: user_entry_id, entry_id: user_entry_id,
origin: LoggedSessionHistoryOrigin::HumanInput { origin: LoggedSessionHistoryOrigin::HumanInput {
account_id: "account-1".into(), account_id: "account-1".into(),
@@ -3734,7 +3743,7 @@ mod tests {
}], }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::FlowInstruction { origin: LoggedSessionHistoryOrigin::FlowInstruction {
selector: "builtin:coder-review".into(), selector: "builtin:coder-review".into(),
@@ -3754,7 +3763,7 @@ mod tests {
}], }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: user_entry_id.clone(), entry_id: user_entry_id.clone(),
origin: origin.clone(), origin: origin.clone(),
derivation: Some(LoggedHistoryDerivation { derivation: Some(LoggedHistoryDerivation {
@@ -3785,7 +3794,7 @@ mod tests {
fn annotated_projection_preserves_identity_and_provenance() { fn annotated_projection_preserves_identity_and_provenance() {
let session_id = crate::new_session_id(); let session_id = crate::new_session_id();
let metadata = LoggedSessionHistoryMetadata { let metadata = LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::ModelOutput { origin: LoggedSessionHistoryOrigin::ModelOutput {
worker: LoggedWorkerSubject { worker: LoggedWorkerSubject {
@@ -4348,7 +4357,7 @@ mod tests {
is_error: false, is_error: false,
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId("ancestor-image".into()), entry_id: LoggedSessionHistoryEntryId("ancestor-image".into()),
origin: LoggedSessionHistoryOrigin::LegacyUnknown, origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
@@ -4443,7 +4452,7 @@ mod tests {
is_error: false, is_error: false,
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId("tool-images".into()), entry_id: LoggedSessionHistoryEntryId("tool-images".into()),
origin: LoggedSessionHistoryOrigin::LegacyUnknown, origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
+5 -5
View File
@@ -255,18 +255,18 @@ pub struct SegmentOrigin {
impl LogEntry { impl LogEntry {
/// Validate generic System records before a Store publishes any bytes. /// 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 { match self {
Self::AnnotatedSegmentStart { history, .. } Self::AnnotatedSegmentStart { history, .. }
| Self::AnnotatedUserInput { history, .. } => { | Self::AnnotatedUserInput { history, .. } => {
for entry in history { for entry in history {
entry.validate_system_input()?; entry.validate_input_metadata()?;
} }
} }
Self::AnnotatedAssistantItem { entry, .. } Self::AnnotatedAssistantItem { entry, .. }
| Self::AnnotatedToolResult { entry, .. } | Self::AnnotatedToolResult { entry, .. }
| Self::ToolResultCorrected { entry, .. } => entry.validate_system_input()?, | Self::ToolResultCorrected { entry, .. } => entry.validate_input_metadata()?,
Self::AnnotatedSystemItem { entry, .. } => entry.validate_system_input()?, Self::AnnotatedSystemItem { entry, .. } => entry.validate_input_metadata()?,
_ => {} _ => {}
} }
Ok(()) Ok(())
@@ -554,7 +554,7 @@ mod tests {
LoggedHistoryEntry { LoggedHistoryEntry {
item: LoggedItem::from(item), item: LoggedItem::from(item),
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown, origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
+2 -2
View File
@@ -27,8 +27,8 @@ pub enum StoreError {
#[error("serialization error: {0}")] #[error("serialization error: {0}")]
Serde(#[from] serde_json::Error), Serde(#[from] serde_json::Error),
#[error("invalid system input: {0}")] #[error("invalid history input metadata: {0}")]
InvalidSystemInput(#[from] crate::SystemInputError), InvalidInputMetadata(#[from] crate::InputMetadataError),
#[error("segment not found: {0}")] #[error("segment not found: {0}")]
NotFound(SegmentId), NotFound(SegmentId),
@@ -36,7 +36,8 @@ use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use std::time::SystemTime; 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 COMPACTION_SESSION_SCHEMA_VERSION: u32 = 4;
const ANNOTATED_SESSION_SCHEMA_VERSION: u32 = 3; const ANNOTATED_SESSION_SCHEMA_VERSION: u32 = 3;
const PREVIOUS_SESSION_SCHEMA_VERSION: u32 = 2; const PREVIOUS_SESSION_SCHEMA_VERSION: u32 = 2;
@@ -225,8 +226,9 @@ impl WorkerSessionStore {
LEGACY_SESSION_SCHEMA_VERSION LEGACY_SESSION_SCHEMA_VERSION
| PREVIOUS_SESSION_SCHEMA_VERSION | PREVIOUS_SESSION_SCHEMA_VERSION
| ANNOTATED_SESSION_SCHEMA_VERSION | ANNOTATED_SESSION_SCHEMA_VERSION
| COMPACTION_SESSION_SCHEMA_VERSION => { | COMPACTION_SESSION_SCHEMA_VERSION
migrate_segment_logs_to_v5( | SYSTEM_INPUT_SESSION_SCHEMA_VERSION => {
migrate_segment_logs_to_current(
segment_log_paths(&root)?, segment_log_paths(&root)?,
manifest.session_id, manifest.session_id,
manifest.schema_version, manifest.schema_version,
@@ -753,7 +755,7 @@ impl Store for WorkerSessionStore {
segment_id: SegmentId, segment_id: SegmentId,
entry: &LogEntry, entry: &LogEntry,
) -> Result<(), StoreError> { ) -> Result<(), StoreError> {
entry.validate_system_inputs()?; entry.validate_input_metadata()?;
let _retention = self.lock_publication(session_id)?; let _retention = self.lock_publication(session_id)?;
self.ensure_session(session_id, true)?; self.ensure_session(session_id, true)?;
self.append_log_entry(&self.log_path(segment_id), entry) self.append_log_entry(&self.log_path(segment_id), entry)
@@ -813,7 +815,7 @@ impl Store for WorkerSessionStore {
entries: &[LogEntry], entries: &[LogEntry],
) -> Result<(), StoreError> { ) -> Result<(), StoreError> {
for entry in entries { for entry in entries {
entry.validate_system_inputs()?; entry.validate_input_metadata()?;
} }
let _retention = self.lock_publication(session_id)?; let _retention = self.lock_publication(session_id)?;
self.ensure_session(session_id, true)?; self.ensure_session(session_id, true)?;
@@ -1173,7 +1175,7 @@ fn segment_log_paths_in_directory(
Ok(paths) Ok(paths)
} }
fn migrate_segment_logs_to_v5( fn migrate_segment_logs_to_current(
paths: Vec<(SegmentId, PathBuf)>, paths: Vec<(SegmentId, PathBuf)>,
session_id: SessionId, session_id: SessionId,
source_schema_version: u32, 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); crate::legacy_session_log::migrate_compaction_origins(&mut plan.entries, &summaries);
for mut entry in std::mem::take(&mut plan.entries) { for mut entry in std::mem::take(&mut plan.entries) {
crate::legacy_session_log::migrate_system_inputs(&mut entry)?; 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)?; serde_json::to_writer(&mut plan.output, &entry)?;
plan.output.push(b'\n'); 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 LEGACY_SESSION_SCHEMA_VERSION
| PREVIOUS_SESSION_SCHEMA_VERSION | PREVIOUS_SESSION_SCHEMA_VERSION
| ANNOTATED_SESSION_SCHEMA_VERSION | ANNOTATED_SESSION_SCHEMA_VERSION
| COMPACTION_SESSION_SCHEMA_VERSION => { | COMPACTION_SESSION_SCHEMA_VERSION
migrate_segment_logs_to_v5( | SYSTEM_INPUT_SESSION_SCHEMA_VERSION => {
migrate_segment_logs_to_current(
segment_log_paths_in_directory(root)?, segment_log_paths_in_directory(root)?,
session_id, session_id,
version, version,
@@ -1289,7 +1292,7 @@ fn validate_canonical_segment_paths(paths: Vec<(SegmentId, PathBuf)>) -> Result<
for (_, path) in paths { for (_, path) in paths {
let entries: Vec<LogEntry> = parse_jsonl(&fs::read(&path)?)?; let entries: Vec<LogEntry> = parse_jsonl(&fs::read(&path)?)?;
for entry in entries { for entry in entries {
entry.validate_system_inputs()?; entry.validate_input_metadata()?;
} }
} }
Ok(()) Ok(())
@@ -1470,7 +1473,7 @@ mod tests {
LoggedHistoryEntry { LoggedHistoryEntry {
item: LoggedItem::from(item), item: LoggedItem::from(item),
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown, origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
@@ -2162,7 +2165,7 @@ mod tests {
assert_eq!(fs::read(&marker_path).unwrap(), old_marker); assert_eq!(fs::read(&marker_path).unwrap(), old_marker);
let updated_first: Vec<LogEntry> = parse_jsonl(&first_published).unwrap(); let updated_first: Vec<LogEntry> = parse_jsonl(&first_published).unwrap();
for entry in &updated_first { for entry in &updated_first {
entry.validate_system_inputs().unwrap(); entry.validate_input_metadata().unwrap();
} }
assert!(matches!( assert!(matches!(
&updated_first[0], &updated_first[0],
@@ -2184,7 +2187,7 @@ mod tests {
assert_eq!(identity_and_body(&first), first_identity); assert_eq!(identity_and_body(&first), first_identity);
assert_eq!(identity_and_body(&second), second_identity); assert_eq!(identity_and_body(&second), second_identity);
for entry in first.iter().chain(second.iter()) { 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); assert_eq!(fs::read(&first_path).unwrap(), first_published);
let second_published = fs::read(&second_path).unwrap(); let second_published = fs::read(&second_path).unwrap();
+8 -9
View File
@@ -11,7 +11,7 @@ fn annotated(item: Item) -> LoggedHistoryEntry {
LoggedHistoryEntry { LoggedHistoryEntry {
item: LoggedItem::from(item), item: LoggedItem::from(item),
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown, origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
@@ -428,10 +428,9 @@ fn lookup_session_of_finds_owning_session() {
#[test] #[test]
fn generic_system_append_and_fork_preserve_metadata_without_public_payload_leak() { fn generic_system_append_and_fork_preserve_metadata_without_public_payload_leak() {
use session_store::{ use session_store::{
FeatureInputPayload, InputCompactionUse, InputEvidenceUse, InputObservationUse, FeatureInputPayload, HistoryInputMetadata, HistoryInputUsePolicy, InputCompactionUse,
LoggedHistoryDerivation, LoggedSystemHistoryEntry, PublicSystemItem, InputEvidenceUse, InputObservationUse, LoggedHistoryDerivation, LoggedSystemHistoryEntry,
PublicSystemPresentation, SystemInputEnvelope, SystemInputMetadata, SystemInputUsePolicy, PublicSystemItem, PublicSystemPresentation, SystemInputEnvelope, WorkerSessionStore,
WorkerSessionStore,
}; };
for worker_layout in [false, true] { for worker_layout in [false, true] {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
@@ -455,8 +454,8 @@ fn generic_system_append_and_fork_preserve_metadata_without_public_payload_leak(
}), }),
SystemInputEnvelope::new( SystemInputEnvelope::new(
"private model body", "private model body",
SystemInputMetadata::new( HistoryInputMetadata::new(
SystemInputUsePolicy { HistoryInputUsePolicy {
recording_material: InputEvidenceUse::Exclude, recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden, observation: InputObservationUse::Hidden,
@@ -584,11 +583,11 @@ fn invalid_generic_system_input_rejects_append_create_and_external_json_without_
}; };
assert!(matches!( assert!(matches!(
store.append(sid, segment, &invalid), store.append(sid, segment, &invalid),
Err(StoreError::InvalidSystemInput(_)) Err(StoreError::InvalidInputMetadata(_))
)); ));
assert!(matches!( assert!(matches!(
store.create_segment(sid, segment, &[seed, invalid.clone()]), store.create_segment(sid, segment, &[seed, invalid.clone()]),
Err(StoreError::InvalidSystemInput(_)) Err(StoreError::InvalidInputMetadata(_))
)); ));
assert!( assert!(
serde_json::from_value::<LogEntry>(serde_json::to_value(invalid).unwrap()).is_err() serde_json::from_value::<LogEntry>(serde_json::to_value(invalid).unwrap()).is_err()
+1 -1
View File
@@ -24,7 +24,7 @@ fn annotated(items: &[Item]) -> Vec<session_store::LoggedHistoryEntry> {
.map(|item| session_store::LoggedHistoryEntry { .map(|item| session_store::LoggedHistoryEntry {
item: session_store::LoggedItem::from(item), item: session_store::LoggedItem::from(item),
metadata: session_store::LoggedSessionHistoryMetadata { metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: session_store::LoggedSessionHistoryEntryId::new(), entry_id: session_store::LoggedSessionHistoryEntryId::new(),
origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
@@ -212,7 +212,7 @@ fn message(id: &str, text: &str) -> session_store::LoggedHistoryEntry {
content: vec![session_store::LoggedContentPart::Text { text: text.into() }], content: vec![session_store::LoggedContentPart::Text { text: text.into() }],
}, },
metadata: session_store::LoggedSessionHistoryMetadata { metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: session_store::LoggedSessionHistoryEntryId(id.into()), entry_id: session_store::LoggedSessionHistoryEntryId(id.into()),
origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput, origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput,
derivation: None, derivation: None,
@@ -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::TextDelta { text } if text.contains(PARTIAL) => break,
protocol::Event::RunEnd { .. } => panic!("dialog ended before cancellation"), 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}"), 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() .iter()
.flat_map(|segment| source_store.read_all(session, *segment).unwrap()) .flat_map(|segment| source_store.read_all(session, *segment).unwrap())
.collect(); .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!( assert_eq!(
raw.iter() raw.iter()
.filter(|entry| matches!( .filter(|entry| matches!(
@@ -11,7 +11,7 @@ use agen::llm_client::{
types::Item, types::Item,
}; };
use futures::Stream; use futures::Stream;
use session_store::{CombinedStore, FsWorkerStore, LogEntry, LoggedSessionHistoryOrigin}; use session_store::{CombinedStore, FsWorkerStore, LogEntry};
use std::pin::Pin; use std::pin::Pin;
const BEFORE: &str = "FROZEN_POSITION: B partially implemented; tests not run; result unconfirmed"; 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(); let worker = prepared.worker();
assert_ne!(worker.segment_id(), source_segment); 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 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); assert_eq!(journal_items.len(), 1);
let journal_item = Item::from(journal_items[0].item.clone()); let journal_item = Item::from(journal_items[0].item.clone());
let body = journal_item.as_text().unwrap(); 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(&expected_json), "durable seed contains the full selected JSON and processing checkpoint");
assert!(body.contains(BEFORE)); assert!(body.contains(BEFORE));
assert!(!body.contains(AFTER)); 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 requests = model.requests.lock().unwrap();
let input = requests[0].items.iter().find_map(|item| match item { 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")), Item::Message { role: agen::llm_client::types::Role::User, .. } => item.as_text().filter(|text| text.contains("## Conversation overview/index")),
@@ -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 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 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 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!(journal_items, 1);
assert_eq!(catalog.store.subject(&subject.id).unwrap().unwrap().behavior_md, "BEHAVIOR_ORIGINAL_UNCHANGED"); assert_eq!(catalog.store.subject(&subject.id).unwrap().unwrap().behavior_md, "BEHAVIOR_ORIGINAL_UNCHANGED");
second.shutdown().await.unwrap(); second.shutdown().await.unwrap();
+1 -1
View File
@@ -1621,7 +1621,7 @@ allow = []
let history = |id: &str, item: LoggedItem| LoggedHistoryEntry { let history = |id: &str, item: LoggedItem| LoggedHistoryEntry {
item, item,
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId(id.into()), entry_id: LoggedSessionHistoryEntryId(id.into()),
origin: LoggedSessionHistoryOrigin::BackendInstruction { operation_id: None }, origin: LoggedSessionHistoryOrigin::BackendInstruction { operation_id: None },
derivation: None, derivation: None,
+1 -1
View File
@@ -3567,7 +3567,7 @@ mod completion_flow_tests {
session_store::LoggedHistoryEntry { session_store::LoggedHistoryEntry {
item: session_store::LoggedItem::from(item), item: session_store::LoggedItem::from(item),
metadata: session_store::LoggedSessionHistoryMetadata { metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: session_store::LoggedSessionHistoryEntryId::new(), entry_id: session_store::LoggedSessionHistoryEntryId::new(),
origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
@@ -35,7 +35,7 @@ fn message(id: &str, role: LoggedRole, text: &str) -> LoggedHistoryEntry {
content: vec![LoggedContentPart::Text { text: text.into() }], content: vec![LoggedContentPart::Text { text: text.into() }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId(id.into()), entry_id: LoggedSessionHistoryEntryId(id.into()),
origin: LoggedSessionHistoryOrigin::LocalHumanInput, origin: LoggedSessionHistoryOrigin::LocalHumanInput,
derivation: None, derivation: None,
+2 -2
View File
@@ -1867,7 +1867,7 @@ mod tests {
}], }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId("original".into()), entry_id: LoggedSessionHistoryEntryId("original".into()),
origin: LoggedSessionHistoryOrigin::LocalHumanInput, origin: LoggedSessionHistoryOrigin::LocalHumanInput,
derivation: None, derivation: None,
@@ -1881,7 +1881,7 @@ mod tests {
}], }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId("summary".into()), entry_id: LoggedSessionHistoryEntryId("summary".into()),
origin: LoggedSessionHistoryOrigin::DerivedSummary, origin: LoggedSessionHistoryOrigin::DerivedSummary,
derivation: Some(session_store::LoggedHistoryDerivation { derivation: Some(session_store::LoggedHistoryDerivation {
@@ -155,7 +155,7 @@ fn assert_generic(root: &Path, session: SessionId, segment: SegmentId) {
entry.item.history_text(), entry.item.history_text(),
old["entry"]["item"]["body"].as_str().unwrap() 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(); let usage = input.usage();
assert_eq!(usage.recording_material, InputEvidenceUse::Exclude); assert_eq!(usage.recording_material, InputEvidenceUse::Exclude);
assert_eq!(usage.memory_evidence, 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_eq!(tree_bytes(&root), before);
assert!(!temp.path().join(".runtime.schema-v12-staging").exists()); 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);
}
+1 -1
View File
@@ -398,7 +398,7 @@ mod tests {
content: vec![LoggedContentPart::Text { text: text.into() }], content: vec![LoggedContentPart::Text { text: text.into() }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId(id.to_string()), entry_id: LoggedSessionHistoryEntryId(id.to_string()),
origin: LoggedSessionHistoryOrigin::HumanInput { origin: LoggedSessionHistoryOrigin::HumanInput {
account_id: "account-1".to_string(), account_id: "account-1".to_string(),
+1 -1
View File
@@ -5033,7 +5033,7 @@ mod tests {
}], }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id, entry_id,
origin: LoggedSessionHistoryOrigin::LegacyUnknown, origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
+2 -2
View File
@@ -79,8 +79,8 @@ fn journal_input(
body: body.clone(), body: body.clone(),
presentation: PublicSystemPresentation::Context, presentation: PublicSystemPresentation::Context,
}; };
let metadata = SystemInputMetadata::new( let metadata = HistoryInputMetadata::new(
SystemInputUsePolicy { HistoryInputUsePolicy {
recording_material: InputEvidenceUse::Exclude, recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden, observation: InputObservationUse::Hidden,
+6 -6
View File
@@ -61,12 +61,12 @@ fn resident_input(
schema: &str, schema: &str,
payload: &impl serde::Serialize, payload: &impl serde::Serialize,
kind: &str, kind: &str,
) -> Result<session_store::LoggedSystemHistoryEntry, session_store::SystemInputError> { ) -> Result<session_store::LoggedSystemHistoryEntry, session_store::InputMetadataError> {
use session_store::*; use session_store::*;
let input = SystemInputEnvelope::new( let input = SystemInputEnvelope::new(
body.clone(), body.clone(),
SystemInputMetadata::new( HistoryInputMetadata::new(
SystemInputUsePolicy { HistoryInputUsePolicy {
recording_material: InputEvidenceUse::Exclude, recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden, observation: InputObservationUse::Hidden,
@@ -93,7 +93,7 @@ pub(crate) fn subject_behavior_input(
body: String, body: String,
behavior_md: String, behavior_md: String,
prompt_provenance: Option<session_store::PromptRenderProvenance>, prompt_provenance: Option<session_store::PromptRenderProvenance>,
) -> Result<session_store::LoggedSystemHistoryEntry, session_store::SystemInputError> { ) -> Result<session_store::LoggedSystemHistoryEntry, session_store::InputMetadataError> {
resident_input( resident_input(
body, body,
"subject:behavior", "subject:behavior",
@@ -108,7 +108,7 @@ pub(crate) fn subject_behavior_input(
pub(crate) fn resident_summary_input( pub(crate) fn resident_summary_input(
body: String, body: String,
prompt_provenance: Option<session_store::PromptRenderProvenance>, prompt_provenance: Option<session_store::PromptRenderProvenance>,
) -> Result<session_store::LoggedSystemHistoryEntry, session_store::SystemInputError> { ) -> Result<session_store::LoggedSystemHistoryEntry, session_store::InputMetadataError> {
resident_input( resident_input(
body, body,
"memory:resident-summary", "memory:resident-summary",
@@ -891,7 +891,7 @@ impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource
metadata: &crate::session_history::SessionHistoryMetadata, metadata: &crate::session_history::SessionHistoryMetadata,
) { ) {
if let Some(input) = metadata if let Some(input) = metadata
.system_input .input_metadata
.as_ref() .as_ref()
.and_then(|input| input.feature_payload()) .and_then(|input| input.feature_payload())
.and_then(|payload| { .and_then(|payload| {
+3 -3
View File
@@ -48,15 +48,15 @@ pub(crate) struct InvocationResultInput {
pub(crate) fn result_input( pub(crate) fn result_input(
result: FeatureInvocationResult, result: FeatureInvocationResult,
body: String, body: String,
) -> Result<session_store::LoggedSystemHistoryEntry, session_store::SystemInputError> { ) -> Result<session_store::LoggedSystemHistoryEntry, session_store::InputMetadataError> {
use session_store::*; use session_store::*;
let payload = FeatureInputPayload::encode( let payload = FeatureInputPayload::encode(
"feature:invocation-result", "feature:invocation-result",
1, 1,
&InvocationResultInput { result }, &InvocationResultInput { result },
)?; )?;
let metadata = SystemInputMetadata::new( let metadata = HistoryInputMetadata::new(
SystemInputUsePolicy { HistoryInputUsePolicy {
recording_material: InputEvidenceUse::Exclude, recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden, observation: InputObservationUse::Hidden,
+1 -1
View File
@@ -1499,7 +1499,7 @@ permission = "write"
is_error: false, is_error: false,
}, },
metadata: session_store::LoggedSessionHistoryMetadata { metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: session_store::LoggedSessionHistoryEntryId(entry_id.into()), entry_id: session_store::LoggedSessionHistoryEntryId(entry_id.into()),
origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
+1 -1
View File
@@ -1033,7 +1033,7 @@ mod tests {
metadata: &crate::session_history::SessionHistoryMetadata, metadata: &crate::session_history::SessionHistoryMetadata,
) { ) {
if let Some(input) = metadata if let Some(input) = metadata
.system_input .input_metadata
.as_ref() .as_ref()
.and_then(|input| input.feature_payload()) .and_then(|input| input.feature_payload())
.and_then(|payload| { .and_then(|payload| {
+165 -11
View File
@@ -352,11 +352,19 @@ impl SessionCapture {
let entries = entries let entries = entries
.into_iter() .into_iter()
.map(|entry| { .map(|entry| {
let (item, use_policy) = if let Some(input) = &entry.annotation.system_input { let (item, use_policy) = if let Some(input) = &entry.annotation.input_metadata {
( let item = if matches!(
Item::system_message(input.public_projection().body), entry.item,
Some(input.public_use_policy()), Item::Message {
) role: Role::System,
..
}
) {
Item::system_message(input.public_projection().body)
} else {
entry.item
};
(item, Some(input.public_use_policy()))
} else { } else {
(entry.item, None) (entry.item, None)
}; };
@@ -982,10 +990,10 @@ mod tests {
#[test] #[test]
fn generic_public_input_observation_and_memory_policies_survive_store_restore_independently() { fn generic_public_input_observation_and_memory_policies_survive_store_restore_independently() {
use session_store::{ use session_store::{
FeatureInputPayload, FsStore, InputCompactionUse, InputEvidenceUse, FeatureInputPayload, FsStore, HistoryInputMetadata, HistoryInputUsePolicy,
InputObservationUse, LogEntry, LoggedSystemHistoryEntry, PublicSystemItem, InputCompactionUse, InputEvidenceUse, InputObservationUse, LogEntry,
PublicSystemPresentation, Store, SystemInputEnvelope, SystemInputMetadata, LoggedSystemHistoryEntry, PublicSystemItem, PublicSystemPresentation, Store,
SystemInputUsePolicy, SystemInputEnvelope,
}; };
let root = tempfile::tempdir().unwrap(); let root = tempfile::tempdir().unwrap();
let store = FsStore::new(root.path()).unwrap(); let store = FsStore::new(root.path()).unwrap();
@@ -1046,8 +1054,8 @@ mod tests {
None, None,
SystemInputEnvelope::new( SystemInputEnvelope::new(
"private model text", "private model text",
SystemInputMetadata::new( HistoryInputMetadata::new(
SystemInputUsePolicy { HistoryInputUsePolicy {
recording_material: InputEvidenceUse::Exclude, recording_material: InputEvidenceUse::Exclude,
memory_evidence, memory_evidence,
observation, 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::<Vec<_>>();
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::<Vec<_>>();
assert_eq!(
ids,
sources
.iter()
.map(|id| format!("E{}", id.0))
.collect::<Vec<_>>()
);
}
}
#[test] #[test]
fn overview_contains_user_and_assistant_only() { fn overview_contains_user_and_assistant_only() {
let view = SessionCapture::new( let view = SessionCapture::new(
+5 -5
View File
@@ -24,7 +24,7 @@ pub(crate) fn test_logged_history_entry(item: impl Into<Item>) -> LoggedHistoryE
LoggedHistoryEntry { LoggedHistoryEntry {
item: session_store::LoggedItem::from(item.into()), item: session_store::LoggedItem::from(item.into()),
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown, origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
@@ -39,7 +39,7 @@ pub(crate) fn test_logged_system_entry(
session_store::LoggedSystemHistoryEntry { session_store::LoggedSystemHistoryEntry {
item, item,
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId::new(), entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown, origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
@@ -60,7 +60,7 @@ pub(crate) fn metadata(
derivation: Option<SessionHistoryDerivation>, derivation: Option<SessionHistoryDerivation>,
) -> SessionHistoryMetadata { ) -> SessionHistoryMetadata {
SessionHistoryMetadata { SessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: SessionHistoryEntryId::new(), entry_id: SessionHistoryEntryId::new(),
origin, origin,
derivation, derivation,
@@ -71,7 +71,7 @@ pub(crate) fn compaction_use(
metadata: &SessionHistoryMetadata, metadata: &SessionHistoryMetadata,
) -> session_store::InputCompactionUse { ) -> session_store::InputCompactionUse {
metadata metadata
.system_input .input_metadata
.as_ref() .as_ref()
.map(|input| input.usage().compaction) .map(|input| input.usage().compaction)
.unwrap_or(session_store::InputCompactionUse::Retain) .unwrap_or(session_store::InputCompactionUse::Retain)
@@ -93,7 +93,7 @@ pub(crate) fn history_entry_with_id(
HistoryEntry::new( HistoryEntry::new(
item, item,
SessionHistoryMetadata { SessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id, entry_id,
origin, origin,
derivation: None, derivation: None,
+3 -3
View File
@@ -16,7 +16,7 @@ pub(crate) struct SkillActivationInput {
pub(crate) fn activation_input( pub(crate) fn activation_input(
activation: &SkillActivationResponse, activation: &SkillActivationResponse,
) -> Result<session_store::LoggedSystemHistoryEntry, session_store::SystemInputError> { ) -> Result<session_store::LoggedSystemHistoryEntry, session_store::InputMetadataError> {
use session_store::*; use session_store::*;
let body = format!( let body = format!(
"Agent Skill `{}` activated from {}.\n\n{}", "Agent Skill `{}` activated from {}.\n\n{}",
@@ -29,8 +29,8 @@ pub(crate) fn activation_input(
name: activation.name.clone(), name: activation.name.clone(),
}, },
)?; )?;
let metadata = SystemInputMetadata::new( let metadata = HistoryInputMetadata::new(
SystemInputUsePolicy { HistoryInputUsePolicy {
recording_material: InputEvidenceUse::Exclude, recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden, observation: InputObservationUse::Hidden,
+1 -1
View File
@@ -1338,7 +1338,7 @@ mod tests {
is_error: false, is_error: false,
}, },
metadata: session_store::LoggedSessionHistoryMetadata { metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: session_store::LoggedSessionHistoryEntryId(entry_id.into()), entry_id: session_store::LoggedSessionHistoryEntryId(entry_id.into()),
origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
+3 -3
View File
@@ -1156,7 +1156,7 @@ enabled = true
LogEntry::AnnotatedSystemItem { entry, .. } LogEntry::AnnotatedSystemItem { entry, .. }
if entry if entry
.metadata .metadata
.system_input .input_metadata
.as_ref() .as_ref()
.and_then(|input| input.feature_payload()) .and_then(|input| input.feature_payload())
.is_some_and(|payload| payload.schema() == "journal:startup") => .is_some_and(|payload| payload.schema() == "journal:startup") =>
@@ -1178,7 +1178,7 @@ enabled = true
}; };
let payload = journal let payload = journal
.metadata .metadata
.system_input .input_metadata
.as_ref() .as_ref()
.unwrap() .unwrap()
.feature_payload() .feature_payload()
@@ -1277,7 +1277,7 @@ enabled = true
} }
assert_eq!(host.startup_calls.load(Ordering::SeqCst), 1); assert_eq!(host.startup_calls.load(Ordering::SeqCst), 1);
let logs = store.read_all(session, segment).unwrap(); 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 controller
.handle .handle
.send(Method::Shutdown { .send(Method::Shutdown {
+1 -1
View File
@@ -5502,7 +5502,7 @@ mod tests {
entry: session_store::LoggedHistoryEntry { entry: session_store::LoggedHistoryEntry {
item: session_store::LoggedItem::from(saved), item: session_store::LoggedItem::from(saved),
metadata: session_store::LoggedSessionHistoryMetadata { metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: session_store::LoggedSessionHistoryEntryId::new(), entry_id: session_store::LoggedSessionHistoryEntryId::new(),
origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
+26 -15
View File
@@ -2739,7 +2739,7 @@ pub trait SystemItemCommitter: Send + Sync {
extensions: Vec<SessionExtension>, extensions: Vec<SessionExtension>,
after_commit: &mut dyn FnMut(&HistoryEntry<SessionHistoryMetadata>), after_commit: &mut dyn FnMut(&HistoryEntry<SessionHistoryMetadata>),
) -> Result<HistoryEntry<SessionHistoryMetadata>, StoreError> { ) -> Result<HistoryEntry<SessionHistoryMetadata>, StoreError> {
entry.validate_system_input()?; entry.validate_input_metadata()?;
let history = HistoryEntry::new(entry.item.to_history_item(), entry.metadata.clone()); let history = HistoryEntry::new(entry.item.to_history_item(), entry.metadata.clone());
self.commit_usage_boundary( self.commit_usage_boundary(
LogEntry::AnnotatedSystemItem { LogEntry::AnnotatedSystemItem {
@@ -6950,7 +6950,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
}; };
for entry in &frozen_context { for entry in &frozen_context {
entry entry
.validate_system_input() .validate_input_metadata()
.map_err(|e| WorkerError::InvalidState(e.to_string()))?; .map_err(|e| WorkerError::InvalidState(e.to_string()))?;
} }
let history_entries = self.session.history().entries(); let history_entries = self.session.history().entries();
@@ -7469,8 +7469,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
new_history[0] new_history[0]
.as_text() .as_text()
.expect("compaction summary is text"), .expect("compaction summary is text"),
session_store::SystemInputMetadata::new( session_store::HistoryInputMetadata::new(
session_store::SystemInputUsePolicy { session_store::HistoryInputUsePolicy {
recording_material: session_store::InputEvidenceUse::Exclude, recording_material: session_store::InputEvidenceUse::Exclude,
memory_evidence: session_store::InputEvidenceUse::Exclude, memory_evidence: session_store::InputEvidenceUse::Exclude,
observation: session_store::InputObservationUse::Hidden, observation: session_store::InputObservationUse::Hidden,
@@ -7509,15 +7509,26 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
original_cursor += offset + 1; original_cursor += offset + 1;
HistoryEntry::new(item, original.annotation.clone()) HistoryEntry::new(item, original.annotation.clone())
} else { } else {
HistoryEntry::new( let mut metadata = new_history_metadata(
item,
new_history_metadata(
WorkerHistoryProvenance::DerivedSummary, WorkerHistoryProvenance::DerivedSummary,
Some(SessionHistoryDerivation { Some(SessionHistoryDerivation {
sources: derived_sources.clone(), 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::<Vec<_>>(); .collect::<Vec<_>>();
@@ -10790,9 +10801,9 @@ mod build_summary_prompt_tests {
producer: "arbitrary-summary-source".into(), producer: "arbitrary-summary-source".into(),
}, },
); );
entry.annotation.system_input = Some( entry.annotation.input_metadata = Some(
session_store::SystemInputMetadata::new( session_store::HistoryInputMetadata::new(
session_store::SystemInputUsePolicy { session_store::HistoryInputUsePolicy {
recording_material: session_store::InputEvidenceUse::Exclude, recording_material: session_store::InputEvidenceUse::Exclude,
memory_evidence: session_store::InputEvidenceUse::Exclude, memory_evidence: session_store::InputEvidenceUse::Exclude,
observation: session_store::InputObservationUse::Hidden, observation: session_store::InputObservationUse::Hidden,
@@ -14631,7 +14642,7 @@ permission = "write"
matches!( matches!(
entry, entry,
LogEntry::AnnotatedSystemItem { entry, .. } LogEntry::AnnotatedSystemItem { entry, .. }
if entry.metadata.system_input.as_ref().and_then(|input| input.feature_payload()).and_then(|payload| payload.decode::<crate::skill::SkillActivationInput>("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::<crate::skill::SkillActivationInput>("skill:activation", 1).ok()).is_some_and(|payload| payload.name == "triage-errors")
&& entry.item.history_text().contains("# Triage Errors") && entry.item.history_text().contains("# Triage Errors")
&& entry.item.history_text() == history_text && entry.item.history_text() == history_text
) )
@@ -16919,7 +16930,7 @@ permission = "write"
fn input_schema(entry: &session_store::LoggedSystemHistoryEntry) -> Option<&str> { fn input_schema(entry: &session_store::LoggedSystemHistoryEntry) -> Option<&str> {
entry entry
.metadata .metadata
.system_input .input_metadata
.as_ref()? .as_ref()?
.feature_payload() .feature_payload()
.map(|payload| payload.schema()) .map(|payload| payload.schema())
@@ -16953,8 +16964,8 @@ permission = "write"
None, None,
SystemInputEnvelope::new( SystemInputEnvelope::new(
"PRIOR_JOURNAL_REFERENCE", "PRIOR_JOURNAL_REFERENCE",
SystemInputMetadata::new( HistoryInputMetadata::new(
SystemInputUsePolicy { HistoryInputUsePolicy {
recording_material: InputEvidenceUse::Exclude, recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden, observation: InputObservationUse::Hidden,
+7 -7
View File
@@ -119,7 +119,7 @@ fn annotated(item: Item) -> session_store::LoggedHistoryEntry {
session_store::LoggedHistoryEntry { session_store::LoggedHistoryEntry {
item: session_store::LoggedItem::from(item), item: session_store::LoggedItem::from(item),
metadata: session_store::LoggedSessionHistoryMetadata { metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: session_store::LoggedSessionHistoryEntryId::new(), entry_id: session_store::LoggedSessionHistoryEntryId::new(),
origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None, derivation: None,
@@ -2220,7 +2220,7 @@ fn summary_entry(
origin: session_store::LoggedSessionHistoryOrigin, origin: session_store::LoggedSessionHistoryOrigin,
) -> session_store::LoggedHistoryEntry { ) -> session_store::LoggedHistoryEntry {
let mut entry = annotated(item); let mut entry = annotated(item);
entry.metadata.system_input = matches!( entry.metadata.input_metadata = matches!(
&entry.item, &entry.item,
session_store::LoggedItem::Message { session_store::LoggedItem::Message {
role: session_store::LoggedRole::System, role: session_store::LoggedRole::System,
@@ -2228,8 +2228,8 @@ fn summary_entry(
} }
) )
.then(|| { .then(|| {
session_store::SystemInputMetadata::new( session_store::HistoryInputMetadata::new(
session_store::SystemInputUsePolicy { session_store::HistoryInputUsePolicy {
recording_material: session_store::InputEvidenceUse::Exclude, recording_material: session_store::InputEvidenceUse::Exclude,
memory_evidence: session_store::InputEvidenceUse::Exclude, memory_evidence: session_store::InputEvidenceUse::Exclude,
observation: session_store::InputObservationUse::Hidden, observation: session_store::InputObservationUse::Hidden,
@@ -2256,7 +2256,7 @@ fn compaction_use(
metadata: &session_store::LoggedSessionHistoryMetadata, metadata: &session_store::LoggedSessionHistoryMetadata,
) -> session_store::InputCompactionUse { ) -> session_store::InputCompactionUse {
metadata metadata
.system_input .input_metadata
.as_ref() .as_ref()
.map(|input| input.usage().compaction) .map(|input| input.usage().compaction)
.unwrap_or(session_store::InputCompactionUse::Retain) .unwrap_or(session_store::InputCompactionUse::Retain)
@@ -3167,8 +3167,8 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
None, None,
session_store::SystemInputEnvelope::new( session_store::SystemInputEnvelope::new(
"OLD_TYPED_STARTUP_JOURNAL", "OLD_TYPED_STARTUP_JOURNAL",
session_store::SystemInputMetadata::new( session_store::HistoryInputMetadata::new(
session_store::SystemInputUsePolicy { session_store::HistoryInputUsePolicy {
recording_material: session_store::InputEvidenceUse::Exclude, recording_material: session_store::InputEvidenceUse::Exclude,
memory_evidence: session_store::InputEvidenceUse::Exclude, memory_evidence: session_store::InputEvidenceUse::Exclude,
observation: session_store::InputObservationUse::Hidden, observation: session_store::InputObservationUse::Hidden,
@@ -231,7 +231,7 @@ fn results(worker: &Worker<Trace, TestStore>) -> Vec<FeatureInvocationResult> {
} }
entry entry
.metadata .metadata
.system_input .input_metadata
.as_ref()? .as_ref()?
.feature_payload()? .feature_payload()?
.decode::<SavedResult>("feature:invocation-result", 1) .decode::<SavedResult>("feature:invocation-result", 1)
+19 -15
View File
@@ -44108,7 +44108,7 @@ mod tests {
content: vec![LoggedContentPart::Text { text: text.into() }], content: vec![LoggedContentPart::Text { text: text.into() }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId(id.to_string()), entry_id: LoggedSessionHistoryEntryId(id.to_string()),
origin: LoggedSessionHistoryOrigin::HumanInput { origin: LoggedSessionHistoryOrigin::HumanInput {
account_id: "account-1".to_string(), account_id: "account-1".to_string(),
@@ -44298,7 +44298,7 @@ mod tests {
content: vec![LoggedContentPart::Text { text: text.into() }], content: vec![LoggedContentPart::Text { text: text.into() }],
}, },
metadata: LoggedSessionHistoryMetadata { metadata: LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: LoggedSessionHistoryEntryId(id.to_string()), entry_id: LoggedSessionHistoryEntryId(id.to_string()),
origin: LoggedSessionHistoryOrigin::HumanInput { origin: LoggedSessionHistoryOrigin::HumanInput {
account_id: "account-1".to_string(), account_id: "account-1".to_string(),
@@ -48550,19 +48550,23 @@ mod tests {
let worker_root = root let worker_root = root
.join(".test-embedded-runtime-store/workers") .join(".test-embedded-runtime-store/workers")
.join(&worker.worker_id); .join(&worker.worker_id);
fs::create_dir_all(worker_root.join("session/segments")).unwrap(); use session_store::Store;
fs::write( let store = session_store::WorkerSessionStore::new(worker_root.join("session")).unwrap();
worker_root.join("session/session.json"), let session = session_store::new_session_id();
serde_json::to_vec_pretty(&json!({ let segment = session_store::new_segment_id();
"schema_version": 1, store
"session_id": session_store::new_session_id() .create_segment(
})) session,
.unwrap(), segment,
) &[session_store::LogEntry::AnnotatedSegmentStart {
.unwrap(); ts: 1,
fs::write( session_id: session,
worker_root.join("session/segments/segment-a.jsonl"), system_prompt: Some("backend job retained session evidence".into()),
b"backend job retained session evidence\n", config: Default::default(),
history: vec![],
forked_from: None,
compacted_from: None,
}],
) )
.unwrap(); .unwrap();
} }
@@ -562,7 +562,7 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup()
}; };
let payload = entry let payload = entry
.metadata .metadata
.system_input .input_metadata
.as_ref()? .as_ref()?
.feature_payload()? .feature_payload()?
.decode::<SavedJournal>("journal:startup", 1) .decode::<SavedJournal>("journal:startup", 1)
@@ -479,7 +479,7 @@ fn seed_session(
vec![session_store::LoggedHistoryEntry { vec![session_store::LoggedHistoryEntry {
item, item,
metadata: session_store::LoggedSessionHistoryMetadata { metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: session_store::LoggedSessionHistoryEntryId("read-source-original".into()), entry_id: session_store::LoggedSessionHistoryEntryId("read-source-original".into()),
origin, origin,
derivation: None, derivation: None,
@@ -1707,7 +1707,7 @@ impl SubjectRemovalFixture {
.map(|text| session_store::LoggedHistoryEntry { .map(|text| session_store::LoggedHistoryEntry {
item: message(text), item: message(text),
metadata: session_store::LoggedSessionHistoryMetadata { metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: session_store::LoggedSessionHistoryEntryId(text.into()), entry_id: session_store::LoggedSessionHistoryEntryId(text.into()),
origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput, origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput,
derivation: None, derivation: None,
@@ -67,7 +67,7 @@ impl Fixture {
content: vec![session_store::LoggedContentPart::Text { text: text.into() }], content: vec![session_store::LoggedContentPart::Text { text: text.into() }],
}, },
metadata: session_store::LoggedSessionHistoryMetadata { metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None, input_metadata: None,
entry_id: session_store::LoggedSessionHistoryEntryId(format!("original-{index}")), entry_id: session_store::LoggedSessionHistoryEntryId(format!("original-{index}")),
origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput, origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput,
derivation: None, derivation: None,