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

feat: persist System input policies through public consumers

Browse Source
This commit is contained in:
Hare
2026-10-20 02:44:28 +09:00
parent 19514a5d6a
commit bb5c50f9ab
40 changed files with 988 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/session-store/src/fs_store.rs
+4
View File
@@ -361,6 +361,7 @@ impl Store for FsStore {
segment_id: SegmentId,
entry: &LogEntry,
) -> Result<(), StoreError> {
entry.validate_system_inputs()?;
let _retention = self.retention().lock_live(session_id)?;
let line = serde_json::to_string(entry)?;
self.append_line(&self.log_path(session_id, segment_id), &line)
@@ -448,6 +449,9 @@ impl Store for FsStore {
segment_id: SegmentId,
entries: &[LogEntry],
) -> Result<(), StoreError> {
for entry in entries {
entry.validate_system_inputs()?;
}
let _retention = self.retention().lock_live(session_id)?;
let path = self.log_path(session_id, segment_id);
if let Some(parent) = path.parent() {
crates/session-store/src/history.rs
+192
View File
@@ -67,6 +67,10 @@ pub enum LoggedSessionHistoryOrigin {
ToolOutput {
worker: LoggedWorkerSubject,
},
/// Descriptive producer identity; use policies live in separate metadata.
Provided {
producer: String,
},
/// Host-produced opening context summary; not auto-read or TaskStore material.
CompactionSummary,
/// Frozen Host Journal reference, never independent observation evidence.
@@ -86,14 +90,26 @@ pub struct LoggedSessionHistoryMetadata {
pub origin: LoggedSessionHistoryOrigin,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub derivation: Option<LoggedHistoryDerivation>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub system_input: Option<crate::SystemInputMetadata>,
}
impl LoggedSessionHistoryMetadata {
fn validate_system_input(&self) -> Result<(), crate::SystemInputError> {
if matches!(self.origin, LoggedSessionHistoryOrigin::Provided { .. })
&& self.system_input.is_none()
{
return Err(crate::SystemInputError::MissingMetadata);
}
Ok(())
}
pub fn legacy_unknown() -> Self {
Self {
entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
system_input: None,
}
}
}
@@ -101,20 +117,140 @@ impl LoggedSessionHistoryMetadata {
/// Persisted item and metadata are one value so transforms cannot reorder or
/// truncate one without the other.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(try_from = "HistoryEntryWire")]
pub struct LoggedHistoryEntry {
pub item: LoggedItem,
pub metadata: LoggedSessionHistoryMetadata,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct HistoryEntryWire {
item: LoggedItem,
metadata: LoggedSessionHistoryMetadata,
}
impl TryFrom<HistoryEntryWire> for LoggedHistoryEntry {
type Error = crate::SystemInputError;
fn try_from(wire: HistoryEntryWire) -> Result<Self, Self::Error> {
let entry = Self {
item: wire.item,
metadata: wire.metadata,
};
entry.validate_system_input()?;
Ok(entry)
}
}
impl LoggedHistoryEntry {
pub fn validate_system_input(&self) -> Result<(), crate::SystemInputError> {
self.metadata.validate_system_input()?;
if self.metadata.system_input.is_some() {
let LoggedItem::Message {
role: crate::LoggedRole::System,
content,
} = &self.item
else {
return Err(crate::SystemInputError::InconsistentRecord);
};
let [crate::LoggedContentPart::Text { text }] = content.as_slice() else {
return Err(crate::SystemInputError::InconsistentRecord);
};
if text.len() > crate::system_input::MAX_SYSTEM_INPUT_BYTES {
return Err(crate::SystemInputError::ResourceLimit);
}
}
Ok(())
}
}
/// Typed system-item history record. The typed system event remains available
/// to client replay while its model-visible projection carries the same stable
/// metadata used by live history.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(try_from = "SystemEntryWire")]
pub struct LoggedSystemHistoryEntry {
pub item: crate::SystemItem,
pub metadata: LoggedSessionHistoryMetadata,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct SystemEntryWire {
item: crate::SystemItem,
metadata: LoggedSessionHistoryMetadata,
}
impl TryFrom<SystemEntryWire> for LoggedSystemHistoryEntry {
type Error = crate::SystemInputError;
fn try_from(wire: SystemEntryWire) -> Result<Self, Self::Error> {
let entry = Self {
item: wire.item,
metadata: wire.metadata,
};
entry.validate_system_input()?;
Ok(entry)
}
}
impl LoggedSystemHistoryEntry {
pub fn from_input(
origin: LoggedSessionHistoryOrigin,
derivation: Option<LoggedHistoryDerivation>,
input: crate::SystemInputEnvelope,
) -> Self {
let (body, system_input) = input.into_parts();
Self {
item: crate::SystemItem::Input { body },
metadata: LoggedSessionHistoryMetadata {
entry_id: LoggedSessionHistoryEntryId::new(),
origin,
derivation,
system_input: Some(system_input),
},
}
}
pub fn validate_system_input(&self) -> Result<(), crate::SystemInputError> {
self.metadata.validate_system_input()?;
match (&self.item, &self.metadata.system_input) {
(crate::SystemItem::Input { body }, Some(_)) => {
if body.len() > crate::system_input::MAX_SYSTEM_INPUT_BYTES {
return Err(crate::SystemInputError::ResourceLimit);
}
}
(crate::SystemItem::Input { .. }, None) => {
return Err(crate::SystemInputError::MissingMetadata);
}
(_, Some(_)) => return Err(crate::SystemInputError::InconsistentRecord),
(_, None) => {}
}
Ok(())
}
pub fn to_history_entry(&self) -> LoggedHistoryEntry {
LoggedHistoryEntry {
item: self.item.to_history_item().into(),
metadata: self.metadata.clone(),
}
}
pub fn public_item(&self) -> crate::PublicSystemItem {
if matches!(self.item, crate::SystemItem::Input { .. }) {
return self
.metadata
.system_input
.as_ref()
.and_then(crate::SystemInputMetadata::public_item)
.cloned()
.unwrap_or_else(|| self.item.public_item());
}
self.item.public_item()
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -132,6 +268,7 @@ mod tests {
}],
},
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::HumanInput {
account_id: "account-1".into(),
@@ -150,6 +287,61 @@ mod tests {
);
}
#[test]
fn external_system_records_reject_missing_policy_role_mismatch_and_overlimit_body() {
let valid = serde_json::json!({
"item": { "kind": "input", "body": "context" },
"metadata": {
"entry_id": "stable-entry", "origin": { "kind": "provided", "producer": "owner" },
"system_input": { "usage": {
"recording_material": "exclude", "memory_evidence": "exclude",
"observation": "hidden", "compaction": "reference_input"
} }
}
});
let restored: LoggedSystemHistoryEntry = serde_json::from_value(valid.clone()).unwrap();
let flat = serde_json::to_value(restored.to_history_entry()).unwrap();
assert!(serde_json::from_value::<LoggedHistoryEntry>(flat.clone()).is_ok());
for generic in [true, false] {
let wire = if generic { &valid } else { &flat };
let decode = |wire| {
if generic {
serde_json::from_value::<LoggedSystemHistoryEntry>(wire).is_err()
} else {
serde_json::from_value::<LoggedHistoryEntry>(wire).is_err()
}
};
let mut missing = wire.clone();
missing["metadata"]
.as_object_mut()
.unwrap()
.remove("system_input");
assert!(decode(missing));
let mut absent_policy = wire.clone();
absent_policy["metadata"]["system_input"]
.as_object_mut()
.unwrap()
.remove("usage");
assert!(decode(absent_policy));
let mut invalid_role = wire.clone();
invalid_role["item"] =
serde_json::to_value(LoggedItem::from(agen::Item::user_message("not system")))
.unwrap();
assert!(decode(invalid_role));
let mut oversized = wire.clone();
if generic {
oversized["item"]["body"] = "x"
.repeat(crate::system_input::MAX_SYSTEM_INPUT_BYTES + 1)
.into();
} else {
oversized["item"]["content"][0]["text"] = "x"
.repeat(crate::system_input::MAX_SYSTEM_INPUT_BYTES + 1)
.into();
}
assert!(decode(oversized));
}
}
#[test]
fn annotated_segment_start_is_restore_visible_without_projecting_metadata() {
let session_id = uuid::Uuid::now_v7();
crates/session-store/src/legacy_session_log.rs
+1
View File
@@ -167,6 +167,7 @@ fn legacy_metadata(
identity.extend_from_slice(&(line_index as u64).to_be_bytes());
identity.extend_from_slice(&(item_index as u64).to_be_bytes());
LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId(format!(
"l-{}",
base64::Engine::encode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, identity)
crates/session-store/src/legacy_session_log/tests.rs
+1
View File
@@ -22,6 +22,7 @@ fn seed(
content: vec![crate::LoggedContentPart::Text { text: text.into() }],
},
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId("summary".into()),
origin,
derivation: Some(crate::LoggedHistoryDerivation {
crates/session-store/src/public_index.rs
+2
View File
@@ -1296,6 +1296,7 @@ mod tests {
LoggedHistoryEntry {
item,
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId(id.to_owned()),
origin,
derivation: None,
@@ -1537,6 +1538,7 @@ mod tests {
prompt_provenance: None,
},
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId("system-item".into()),
origin: LoggedSessionHistoryOrigin::BackendInstruction {
operation_id: None,
crates/session-store/src/public_material.rs
+104 -15
View File
@@ -148,15 +148,24 @@ struct SegmentHeader {
file_len: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PublicMaterialUse {
Recording,
MemoryEvidence,
}
/// A fixed material catalog. Opening a new catalog observes newly appended
/// records; reading a body through this catalog verifies its saved digest and
/// byte length again. Missing/corrected/deleted sources are explicit failures.
/// Explicit evidence permission applies to model text, not the private payload;
/// Web display and observation instead use the producer's public projection.
pub struct SessionPublicMaterialReader {
session_id: SessionId,
active_segment_id: SegmentId,
layout: Layout,
limits: PublicMaterialLimits,
materials: Vec<Material>,
purpose: PublicMaterialUse,
retention: Option<crate::retention::SessionRetentionStore>,
}
@@ -180,7 +189,13 @@ impl SessionPublicMaterialReader {
active_segment_id: SegmentId,
limits: PublicMaterialLimits,
) -> Result<Self, PublicMaterialError> {
Self::open_worker_guarded(session_root, active_segment_id, limits, None)
Self::open_worker_guarded(
session_root,
active_segment_id,
limits,
None,
PublicMaterialUse::Recording,
)
}
/// Uses the Host's aggregate-external control root for the whole scan and every body read.
@@ -191,7 +206,30 @@ impl SessionPublicMaterialReader {
limits: PublicMaterialLimits,
retention: crate::retention::SessionRetentionStore,
) -> Result<Self, PublicMaterialError> {
Self::open_worker_guarded(session_root, active_segment_id, limits, Some(retention))
Self::open_worker_for_use(
session_root,
active_segment_id,
limits,
retention,
PublicMaterialUse::Recording,
)
}
/// Selects the evidence purpose independently of public display and observation.
pub fn open_worker_for_use(
session_root: &Path,
active_segment_id: SegmentId,
limits: PublicMaterialLimits,
retention: crate::retention::SessionRetentionStore,
purpose: PublicMaterialUse,
) -> Result<Self, PublicMaterialError> {
Self::open_worker_guarded(
session_root,
active_segment_id,
limits,
Some(retention),
purpose,
)
}
fn open_worker_guarded(
@@ -199,6 +237,7 @@ impl SessionPublicMaterialReader {
active_segment_id: SegmentId,
limits: PublicMaterialLimits,
retention: Option<crate::retention::SessionRetentionStore>,
purpose: PublicMaterialUse,
) -> Result<Self, PublicMaterialError> {
let store = WorkerSessionStore::open_read_only(session_root)
.map_err(|_| PublicMaterialError::Unavailable)?;
@@ -212,6 +251,7 @@ impl SessionPublicMaterialReader {
Layout::Worker(session_root.into()),
limits,
retention,
purpose,
)
}
@@ -237,6 +277,25 @@ impl SessionPublicMaterialReader {
active_segment_id: SegmentId,
limits: PublicMaterialLimits,
retention: crate::retention::SessionRetentionStore,
) -> Result<Self, PublicMaterialError> {
Self::open_fs_for_use(
sessions_root,
session_id,
active_segment_id,
limits,
retention,
PublicMaterialUse::Recording,
)
}
/// Uses retained originals with the independently selected evidence policy.
pub fn open_fs_for_use(
sessions_root: &Path,
session_id: SessionId,
active_segment_id: SegmentId,
limits: PublicMaterialLimits,
retention: crate::retention::SessionRetentionStore,
purpose: PublicMaterialUse,
) -> Result<Self, PublicMaterialError> {
Self::open(
session_id,
@@ -244,6 +303,7 @@ impl SessionPublicMaterialReader {
Layout::Fs(sessions_root.join(session_id.to_string())),
limits,
Some(retention),
purpose,
)
}
@@ -263,6 +323,7 @@ impl SessionPublicMaterialReader {
layout: Layout,
limits: PublicMaterialLimits,
retention: Option<crate::retention::SessionRetentionStore>,
purpose: PublicMaterialUse,
) -> Result<Self, PublicMaterialError> {
let _guard = retention
.as_ref()
@@ -435,23 +496,19 @@ impl SessionPublicMaterialReader {
} => turn = *total_turn_count,
_ => {}
}
let system_history;
let entries: &[LoggedHistoryEntry] = match &record {
LogEntry::AnnotatedSegmentStart { history, .. }
| LogEntry::AnnotatedUserInput { history, .. } => history,
LogEntry::AnnotatedAssistantItem { entry, .. }
| LogEntry::AnnotatedToolResult { entry, .. }
| LogEntry::ToolResultCorrected { entry, .. } => std::slice::from_ref(entry),
LogEntry::AnnotatedSystemItem { entry, .. } => {
system_history = entry.to_history_entry();
std::slice::from_ref(&system_history)
}
_ => &[],
};
if let LogEntry::AnnotatedSystemItem { entry, .. } = &record {
charge(&mut examined, limits.max_entries)?;
if entry.metadata.entry_id.0.is_empty()
|| !history_seen.insert(entry.metadata.entry_id.0.clone())
{
return Err(PublicMaterialError::Inconsistent);
}
history_ids.push(entry.metadata.entry_id.0.clone());
}
for (history_index, entry) in entries.iter().enumerate() {
charge(&mut examined, limits.max_entries)?;
let id = &entry.metadata.entry_id.0;
@@ -468,7 +525,7 @@ impl SessionPublicMaterialReader {
if in_run && ai_materialized(&entry.item) {
safe = true;
}
let Some((sha256, total_bytes)) = canonical_metadata(entry)? else {
let Some((sha256, total_bytes)) = canonical_metadata(entry, purpose)? else {
continue;
};
let location = Location {
@@ -621,6 +678,7 @@ impl SessionPublicMaterialReader {
layout,
limits,
materials,
purpose,
retention: retention.clone(),
})
}
@@ -660,6 +718,7 @@ impl SessionPublicMaterialReader {
let Some((_, record)) = reader.next()? else {
return Err(PublicMaterialError::Missing);
};
let system_history;
let (body, origin) = if let Some(boundary) = run_boundary(&record) {
if boundary_reference(
&record,
@@ -683,12 +742,16 @@ impl SessionPublicMaterialReader {
LogEntry::AnnotatedAssistantItem { entry, .. }
| LogEntry::AnnotatedToolResult { entry, .. }
| LogEntry::ToolResultCorrected { entry, .. } => Some(entry),
LogEntry::AnnotatedSystemItem { entry, .. } => {
system_history = entry.to_history_entry();
Some(&system_history)
}
_ => None,
}
.filter(|e| format!("E{}", e.metadata.entry_id.0) == descriptor.identity.entry_ref)
.ok_or(PublicMaterialError::Changed)?;
(
canonical_body(entry).ok_or(PublicMaterialError::Changed)?,
canonical_body(entry, self.purpose).ok_or(PublicMaterialError::Changed)?,
PublicMaterialOrigin::History {
origin: entry.metadata.origin.clone(),
},
@@ -828,7 +891,32 @@ fn boundary_reference(
.unwrap_or_else(|| format!("El-run-{session}-{segment}-{offset}"))
}
fn canonical_body(entry: &LoggedHistoryEntry) -> Option<CanonicalBody<'_>> {
fn canonical_body(
entry: &LoggedHistoryEntry,
purpose: PublicMaterialUse,
) -> Option<CanonicalBody<'_>> {
if let Some(input) = &entry.metadata.system_input {
let usage = input.usage();
let allowed = match purpose {
PublicMaterialUse::Recording => usage.recording_material,
PublicMaterialUse::MemoryEvidence => usage.memory_evidence,
};
if allowed == crate::InputEvidenceUse::Exclude {
return None;
}
if let LoggedItem::Message {
role: LoggedRole::System,
content,
} = &entry.item
{
return Some(CanonicalBody::Message {
origin: &entry.metadata.origin,
role: LoggedRole::System,
content,
});
}
return None;
}
if matches!(
entry.metadata.origin,
LoggedSessionHistoryOrigin::DerivedSummary
@@ -880,8 +968,9 @@ fn canonical_body(entry: &LoggedHistoryEntry) -> Option<CanonicalBody<'_>> {
fn canonical_metadata(
entry: &LoggedHistoryEntry,
purpose: PublicMaterialUse,
) -> Result<Option<(String, u64)>, PublicMaterialError> {
let Some(body) = canonical_body(entry) else {
let Some(body) = canonical_body(entry, purpose) else {
return Ok(None);
};
let mut sink = BodySink::new(0, 0);
crates/session-store/src/public_material/tests.rs
+164
View File
@@ -9,6 +9,7 @@ fn entry(id: &str, item: LoggedItem) -> LoggedHistoryEntry {
LoggedHistoryEntry {
item,
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId(id.into()),
origin: LoggedSessionHistoryOrigin::LocalHumanInput,
derivation: None,
@@ -591,6 +592,7 @@ fn reasoning_proves_rollback_safety_but_private_and_derived_items_are_not_materi
},
},
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId("startup-journal".into()),
origin: LoggedSessionHistoryOrigin::DerivedSummary,
derivation: None,
@@ -1243,3 +1245,165 @@ fn compaction_without_exact_derivation_or_with_invalid_turn_is_rejected() {
));
}
}
#[test]
fn generic_system_material_uses_explicit_evidence_purpose_not_producer_or_body() {
use crate::{
FeatureInputPayload, InputCompactionUse, InputEvidenceUse, InputObservationUse,
LoggedSystemHistoryEntry, SystemInputEnvelope, SystemInputMetadata, SystemInputUsePolicy,
};
for worker_layout in [false, true] {
let temp = tempfile::tempdir().unwrap();
let session = crate::new_session_id();
let segment = crate::new_segment_id();
let fork = crate::new_segment_id();
let make = |producer: &str, recording_material, memory_evidence, body: &str| {
LoggedSystemHistoryEntry::from_input(
LoggedSessionHistoryOrigin::Provided {
producer: producer.into(),
},
None,
SystemInputEnvelope::new(
body,
SystemInputMetadata::new(
SystemInputUsePolicy {
recording_material,
memory_evidence,
observation: InputObservationUse::Hidden,
compaction: InputCompactionUse::Retain,
},
Some(
FeatureInputPayload::encode(
"unknown:future",
99,
&serde_json::json!({"private": "not-public"}),
)
.unwrap(),
),
None,
)
.unwrap(),
)
.unwrap(),
)
};
let inputs = [
make(
"same-owner",
InputEvidenceUse::Include,
InputEvidenceUse::Exclude,
"do not extract this; original material",
),
make(
"same-owner",
InputEvidenceUse::Exclude,
InputEvidenceUse::Include,
"memory-only original",
),
make(
"renamed-owner",
InputEvidenceUse::Exclude,
InputEvidenceUse::Exclude,
"reference without warning text",
),
];
let root = if worker_layout {
temp.path().join("worker")
} else {
temp.path().join(session.to_string())
};
let store: Box<dyn Store> = if worker_layout {
Box::new(WorkerSessionStore::new(&root).unwrap())
} else {
Box::new(FsStore::new(temp.path()).unwrap())
};
store
.create_segment(session, segment, &[seed(session, vec![])])
.unwrap();
for entry in &inputs {
store
.append(
session,
segment,
&LogEntry::AnnotatedSystemItem {
ts: 2,
entry: entry.clone(),
extensions: vec![],
},
)
.unwrap();
}
store
.append(
session,
segment,
&LogEntry::TurnEnd {
ts: 3,
turn_count: 1,
},
)
.unwrap();
let mut fork_seed = seed(
session,
inputs
.iter()
.map(LoggedSystemHistoryEntry::to_history_entry)
.collect(),
);
if let LogEntry::AnnotatedSegmentStart { forked_from, .. } = &mut fork_seed {
*forked_from = Some(crate::SegmentOrigin {
segment_id: segment,
at_turn_index: 1,
});
}
store.create_segment(session, fork, &[fork_seed]).unwrap();
let open = |active, purpose| {
if worker_layout {
SessionPublicMaterialReader::open_worker_for_use(
&root,
active,
Default::default(),
crate::retention::fs_retention(temp.path()),
purpose,
)
.unwrap()
} else {
SessionPublicMaterialReader::open_fs_for_use(
temp.path(),
session,
active,
Default::default(),
crate::retention::fs_retention(temp.path()),
purpose,
)
.unwrap()
}
};
for (purpose, selected, expected_text) in [
(
PublicMaterialUse::Recording,
0,
"do not extract this; original material",
),
(PublicMaterialUse::MemoryEvidence, 1, "memory-only original"),
] {
let original = open(segment, purpose);
let inherited = open(fork, purpose);
assert_eq!(
ids(&original),
vec![format!("E{}", inputs[selected].metadata.entry_id.0)]
);
assert_eq!(ids(&inherited), ids(&original));
let descriptor = inherited.descriptors().next().unwrap();
let body = inherited
.read_content(descriptor, 0, 16 * 1024)
.unwrap()
.content;
let value: serde_json::Value = serde_json::from_str(&body).unwrap();
assert_eq!(value["role"], "system");
assert_eq!(value["content"][0]["text"], expected_text);
assert!(!body.contains("not-public"));
assert!(!body.contains("unknown:future"));
}
}
}
crates/session-store/src/public_snapshot.rs
+44 -11
View File
@@ -27,7 +27,7 @@ use thiserror::Error;
use crate::{
LogEntry, LoggedContentPart, LoggedHistoryEntry, LoggedItem, LoggedRole,
LoggedSessionHistoryOrigin, SegmentId, SegmentOrigin, SessionId, StoreError, SystemItem,
LoggedSessionHistoryOrigin, SegmentId, SegmentOrigin, SessionId, StoreError,
WorkerAggregateStore, WorkerSessionStore, WorkerStoreError,
};
@@ -1165,7 +1165,7 @@ fn capture_tool_correction(
let LogEntry::ToolResultCorrected { entry, .. } = record else {
return false;
};
if let Some(data) = project_item(&entry.metadata.entry_id.0, &entry.item) {
if let Some(data) = project_history_item(entry) {
// We read backwards: the first correction seen is the latest one.
corrections
.entry(entry.metadata.entry_id.0.clone())
@@ -1600,12 +1600,12 @@ pub(crate) fn project_session_snapshot_for_segment(
} => extend_history(&mut entries, history, Some(segments), *ts),
LogEntry::AnnotatedAssistantItem { ts, entry }
| LogEntry::AnnotatedToolResult { ts, entry } => {
if let Some(data) = project_item(&entry.metadata.entry_id.0, &entry.item) {
if let Some(data) = project_history_item(entry) {
entries.push(history_entry(entry, *ts, data));
}
}
LogEntry::ToolResultCorrected { ts, entry } => {
if let Some(data) = project_item(&entry.metadata.entry_id.0, &entry.item) {
if let Some(data) = project_history_item(entry) {
if let Some(existing) = entries
.iter_mut()
.find(|existing| existing.entry_id == entry.metadata.entry_id.0)
@@ -1618,11 +1618,16 @@ pub(crate) fn project_session_snapshot_for_segment(
}
}
LogEntry::AnnotatedSystemItem { ts, entry, .. } => entries.push(system_entry(
&entry.item,
&entry.public_item(),
entry.metadata.entry_id.0.clone(),
*ts,
provenance(&entry.metadata.origin),
derivation_ids(entry),
entry
.metadata
.system_input
.as_ref()
.map(crate::SystemInputMetadata::public_use_policy),
)),
LogEntry::RunYielded {
ts,
@@ -1775,7 +1780,7 @@ fn extend_history(
segments: input_segments.cloned().unwrap_or_default(),
}
} else {
let Some(data) = project_item(&entry.metadata.entry_id.0, &entry.item) else {
let Some(data) = project_history_item(entry) else {
continue;
};
data
@@ -1790,6 +1795,11 @@ fn history_entry(
data: SessionSnapshotEntryData,
) -> SessionSnapshotEntry {
SessionSnapshotEntry {
use_policy: entry
.metadata
.system_input
.as_ref()
.map(crate::SystemInputMetadata::public_use_policy),
entry_id: entry.metadata.entry_id.0.clone(),
timestamp,
provenance: provenance(&entry.metadata.origin),
@@ -1833,6 +1843,7 @@ fn run_transition_entry(
data: SessionSnapshotEntryData,
) -> SessionSnapshotEntry {
SessionSnapshotEntry {
use_policy: None,
entry_id: persisted_entry_id
.map(|entry_id| entry_id.0.clone())
.unwrap_or_else(|| legacy_entry_id(session_key, segment_id, log_index, 0)),
@@ -1852,6 +1863,7 @@ fn legacy_entry(
data: SessionSnapshotEntryData,
) -> SessionSnapshotEntry {
SessionSnapshotEntry {
use_policy: None,
entry_id: legacy_entry_id(session_key, segment_id, log_index, item_index),
timestamp,
provenance: SessionEntryProvenance::LegacyUnknown,
@@ -1884,9 +1896,8 @@ fn provenance(origin: &LoggedSessionHistoryOrigin) -> SessionEntryProvenance {
LoggedSessionHistoryOrigin::FlowInstruction { .. } => {
SessionEntryProvenance::FlowInstruction
}
LoggedSessionHistoryOrigin::BackendInstruction { .. } => {
SessionEntryProvenance::BackendInstruction
}
LoggedSessionHistoryOrigin::BackendInstruction { .. }
| LoggedSessionHistoryOrigin::Provided { .. } => SessionEntryProvenance::BackendInstruction,
LoggedSessionHistoryOrigin::ModelOutput { .. } => SessionEntryProvenance::ModelOutput,
LoggedSessionHistoryOrigin::ToolOutput { .. } => SessionEntryProvenance::ToolOutput,
LoggedSessionHistoryOrigin::CompactionSummary
@@ -1914,6 +1925,18 @@ pub fn session_tool_attachment_id(
URL_SAFE_NO_PAD.encode(hash.finalize())
}
fn project_history_item(entry: &LoggedHistoryEntry) -> Option<SessionSnapshotEntryData> {
if let Some(input) = &entry.metadata.system_input {
let public = input.public_projection();
return Some(SessionSnapshotEntryData::SystemItem {
item_kind: public.kind.clone(),
content: public.body.clone(),
data: Some(serde_json::to_value(public).expect("public system input is serializable")),
});
}
project_item(&entry.metadata.entry_id.0, &entry.item)
}
fn project_item(entry_id: &str, item: &LoggedItem) -> Option<SessionSnapshotEntryData> {
match item {
LoggedItem::Message { role, content } => {
@@ -1986,17 +2009,19 @@ fn project_item(entry_id: &str, item: &LoggedItem) -> Option<SessionSnapshotEntr
}
fn system_entry(
item: &SystemItem,
item: &crate::PublicSystemItem,
entry_id: String,
timestamp: u64,
provenance: SessionEntryProvenance,
derived_from: Vec<String>,
use_policy: Option<protocol::SessionEntryUsePolicy>,
) -> SessionSnapshotEntry {
let public = item.public_item();
let public = item.clone();
let data = Some(serde_json::to_value(&public).expect("public system input is serializable"));
let item_kind = public.kind;
let content = public.body;
SessionSnapshotEntry {
use_policy,
entry_id,
timestamp,
provenance,
@@ -2285,6 +2310,7 @@ mod tests {
content: vec![LoggedContentPart::Text { text: text.into() }],
},
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId(entry_id.into()),
origin,
derivation: None,
@@ -3583,6 +3609,7 @@ mod tests {
.map(|item| LoggedHistoryEntry {
item,
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
@@ -3637,6 +3664,7 @@ mod tests {
}],
},
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: user_entry_id,
origin: LoggedSessionHistoryOrigin::HumanInput {
account_id: "account-1".into(),
@@ -3706,6 +3734,7 @@ mod tests {
}],
},
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::FlowInstruction {
selector: "builtin:coder-review".into(),
@@ -3725,6 +3754,7 @@ mod tests {
}],
},
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: user_entry_id.clone(),
origin: origin.clone(),
derivation: Some(LoggedHistoryDerivation {
@@ -3755,6 +3785,7 @@ mod tests {
fn annotated_projection_preserves_identity_and_provenance() {
let session_id = crate::new_session_id();
let metadata = LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::ModelOutput {
worker: LoggedWorkerSubject {
@@ -4317,6 +4348,7 @@ mod tests {
is_error: false,
},
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId("ancestor-image".into()),
origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
@@ -4411,6 +4443,7 @@ mod tests {
is_error: false,
},
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId("tool-images".into()),
origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
crates/session-store/src/public_snapshot/run_stats.rs
+1
View File
@@ -103,6 +103,7 @@ fn finish(
stats.insert(
anchor.clone(),
SessionSnapshotEntry {
use_policy: None,
entry_id: format!("run-stats:{anchor}"),
timestamp: ended_at,
provenance: SessionEntryProvenance::LegacyUnknown,
crates/session-store/src/segment_log.rs
+18
View File
@@ -254,6 +254,23 @@ pub struct SegmentOrigin {
}
impl LogEntry {
/// Validate generic System records before a Store publishes any bytes.
pub fn validate_system_inputs(&self) -> Result<(), crate::SystemInputError> {
match self {
Self::AnnotatedSegmentStart { history, .. }
| Self::AnnotatedUserInput { history, .. } => {
for entry in history {
entry.validate_system_input()?;
}
}
Self::AnnotatedAssistantItem { entry, .. }
| Self::AnnotatedToolResult { entry, .. }
| Self::ToolResultCorrected { entry, .. } => entry.validate_system_input()?,
Self::AnnotatedSystemItem { entry, .. } => entry.validate_system_input()?,
_ => {}
}
Ok(())
}
/// Allocate the durable public identity for a newly committed Run
/// transition. Calling this repeatedly is idempotent so the exact value
/// written to storage is also the value published to live subscribers.
@@ -537,6 +554,7 @@ mod tests {
LoggedHistoryEntry {
item: LoggedItem::from(item),
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
crates/session-store/src/store.rs
+3
View File
@@ -27,6 +27,9 @@ pub enum StoreError {
#[error("serialization error: {0}")]
Serde(#[from] serde_json::Error),
#[error("invalid system input: {0}")]
InvalidSystemInput(#[from] crate::SystemInputError),
#[error("segment not found: {0}")]
NotFound(SegmentId),
crates/session-store/src/system_input.rs
+21
View File
@@ -36,6 +36,10 @@ pub enum SystemInputError {
UnsupportedPayload,
#[error("system input payload does not match the owner's typed contract")]
InvalidTypedPayload,
#[error("generic system input is missing explicit metadata")]
MissingMetadata,
#[error("system input metadata does not match its logical record")]
InconsistentRecord,
}
/// One private payload owned by a Feature's versioned typed decoder.
@@ -229,6 +233,23 @@ impl SystemInputMetadata {
pub fn public_item(&self) -> Option<&PublicSystemItem> {
self.public.as_ref()
}
pub fn public_use_policy(&self) -> protocol::SessionEntryUsePolicy {
let has_public = self.public.is_some();
protocol::SessionEntryUsePolicy {
observation_readable: has_public
&& self.usage.observation == InputObservationUse::Readable,
memory_evidence: has_public && self.usage.memory_evidence == InputEvidenceUse::Include,
}
}
pub fn public_projection(&self) -> PublicSystemItem {
self.public.clone().unwrap_or_else(|| PublicSystemItem {
kind: "input".into(),
body: String::new(),
presentation: crate::PublicSystemPresentation::Ignored,
})
}
}
/// System role is invariant; provenance and use policies are attached data.
crates/session-store/src/system_item.rs
+8 -3
View File
@@ -92,6 +92,9 @@ pub struct PromptRenderProvenance {
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum SystemItem {
/// Generic System body; private metadata and public presentation are supplied
/// independently by the producer in the same annotated history record.
Input { body: String },
/// One-time derived Journal reference context. The bounded Host snapshot is
/// retained as JSON independently of its rendered body and prompt projection.
JournalStartup {
@@ -227,9 +230,9 @@ impl SystemItem {
event: event.clone(),
},
Self::TaskReminder { .. } => PublicSystemPresentation::TaskReminder,
Self::LegacyKnowledgeIgnored { .. } | Self::LegacyIgnored { .. } => {
PublicSystemPresentation::Ignored
}
Self::Input { .. }
| Self::LegacyKnowledgeIgnored { .. }
| Self::LegacyIgnored { .. } => PublicSystemPresentation::Ignored,
Self::JournalStartup { .. }
| Self::ResidentSummaryRefresh { .. }
| Self::SubjectBehaviorRefresh { .. }
@@ -253,6 +256,7 @@ impl SystemItem {
/// for this item. Returns the variant's stored `body` verbatim.
pub fn history_text(&self) -> String {
match self {
SystemItem::Input { body } => body.clone(),
SystemItem::JournalStartup { body, .. } => body.clone(),
SystemItem::Notification { body, .. } => body.clone(),
SystemItem::WorkerEvent { body, .. } => body.clone(),
@@ -280,6 +284,7 @@ impl SystemItem {
/// wire — keep flexible.
pub fn kind_label(&self) -> &'static str {
match self {
SystemItem::Input { .. } => "input",
SystemItem::JournalStartup { .. } => "journal_startup",
SystemItem::Notification { .. } => "notification",
SystemItem::WorkerEvent { .. } => "worker_event",
crates/session-store/src/worker_session_store.rs
+5
View File
@@ -742,6 +742,7 @@ impl Store for WorkerSessionStore {
segment_id: SegmentId,
entry: &LogEntry,
) -> Result<(), StoreError> {
entry.validate_system_inputs()?;
let _retention = self.lock_publication(session_id)?;
self.ensure_session(session_id, true)?;
self.append_log_entry(&self.log_path(segment_id), entry)
@@ -800,6 +801,9 @@ impl Store for WorkerSessionStore {
segment_id: SegmentId,
entries: &[LogEntry],
) -> Result<(), StoreError> {
for entry in entries {
entry.validate_system_inputs()?;
}
let _retention = self.lock_publication(session_id)?;
self.ensure_session(session_id, true)?;
let mut content = Vec::new();
@@ -1430,6 +1434,7 @@ mod tests {
LoggedHistoryEntry {
item: LoggedItem::from(item),
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
crates/session-store/tests/fs_store_test.rs
+176
View File
@@ -11,6 +11,7 @@ fn annotated(item: Item) -> LoggedHistoryEntry {
LoggedHistoryEntry {
item: LoggedItem::from(item),
metadata: LoggedSessionHistoryMetadata {
system_input: None,
entry_id: LoggedSessionHistoryEntryId::new(),
origin: LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,
@@ -421,3 +422,178 @@ fn lookup_session_of_finds_owning_session() {
assert_eq!(store.lookup_session_of(segid).unwrap(), Some(sid));
}
// Storage/projection component test: both production layouts, temporary originals,
// no Feature, model, runtime or Host authority is substituted into this boundary.
#[test]
fn generic_system_append_and_fork_preserve_metadata_without_public_payload_leak() {
use session_store::{
FeatureInputPayload, InputCompactionUse, InputEvidenceUse, InputObservationUse,
LoggedHistoryDerivation, LoggedSystemHistoryEntry, PublicSystemItem,
PublicSystemPresentation, SystemInputEnvelope, SystemInputMetadata, SystemInputUsePolicy,
WorkerSessionStore,
};
for worker_layout in [false, true] {
let dir = tempfile::tempdir().unwrap();
let sid = new_session_id();
let segment = new_segment_id();
let fork = new_segment_id();
let store: Box<dyn Store> = if worker_layout {
Box::new(WorkerSessionStore::new(dir.path()).unwrap())
} else {
Box::new(FsStore::new(dir.path()).unwrap())
};
store
.create_segment(sid, segment, &[nil_session_start(1, sid)])
.unwrap();
let entry = LoggedSystemHistoryEntry::from_input(
LoggedSessionHistoryOrigin::Provided {
producer: "new-feature".into(),
},
Some(LoggedHistoryDerivation {
sources: vec![LoggedSessionHistoryEntryId("source-entry".into())],
}),
SystemInputEnvelope::new(
"private model body",
SystemInputMetadata::new(
SystemInputUsePolicy {
recording_material: InputEvidenceUse::Exclude,
memory_evidence: InputEvidenceUse::Exclude,
observation: InputObservationUse::Hidden,
compaction: InputCompactionUse::ReferenceInput,
},
Some(
FeatureInputPayload::encode(
"new-feature:context",
101,
&serde_json::json!({"private": "secret-detail"}),
)
.unwrap(),
),
Some(PublicSystemItem {
kind: "context".into(),
body: "explicit public body".into(),
presentation: PublicSystemPresentation::Context,
}),
)
.unwrap(),
)
.unwrap(),
);
store
.append(
sid,
segment,
&LogEntry::AnnotatedSystemItem {
ts: 2,
entry: entry.clone(),
extensions: vec![],
},
)
.unwrap();
let saved = store.read_all(sid, segment).unwrap();
let LogEntry::AnnotatedSystemItem {
entry: restored, ..
} = &saved[1]
else {
panic!("missing system input")
};
assert_eq!(restored.to_history_entry(), entry.to_history_entry());
assert_eq!(
collect_state(&saved).history[0].as_text(),
Some("private model body")
);
let mut fork_seed = nil_session_start(3, sid);
if let LogEntry::AnnotatedSegmentStart {
history,
forked_from,
..
} = &mut fork_seed
{
history.push(restored.to_history_entry());
*forked_from = Some(session_store::SegmentOrigin {
segment_id: segment,
at_turn_index: 0,
});
}
store.create_segment(sid, fork, &[fork_seed]).unwrap();
let fork_saved = store.read_all(sid, fork).unwrap();
let LogEntry::AnnotatedSegmentStart { history, .. } = &fork_saved[0] else {
unreachable!()
};
assert_eq!(history, &[entry.to_history_entry()]);
let original = session_store::public_snapshot::project_session_snapshot(sid, &saved);
let inherited = session_store::public_snapshot::project_session_snapshot(sid, &fork_saved);
let public = serde_json::to_string(&original.entries).unwrap();
assert!(public.contains("explicit public body"));
for private in ["private model body", "secret-detail", "new-feature:context"] {
assert!(!public.contains(private), "private field leaked: {private}");
}
assert_eq!(original.entries.len(), 1);
assert_eq!(inherited.entries.len(), 1);
assert_eq!(original.entries[0].entry_id, inherited.entries[0].entry_id);
assert_eq!(
original.entries[0].derived_from,
inherited.entries[0].derived_from
);
assert_eq!(
original.entries[0].provenance,
inherited.entries[0].provenance
);
assert_eq!(
serde_json::to_value(&original.entries[0].data).unwrap(),
serde_json::to_value(&inherited.entries[0].data).unwrap()
);
}
}
#[test]
fn invalid_generic_system_input_rejects_append_create_and_external_json_without_changing_originals()
{
use session_store::{StoreError, SystemItem, WorkerSessionStore};
for worker_layout in [false, true] {
let dir = tempfile::tempdir().unwrap();
let sid = new_session_id();
let segment = new_segment_id();
let store: Box<dyn Store> = if worker_layout {
Box::new(WorkerSessionStore::new(dir.path()).unwrap())
} else {
Box::new(FsStore::new(dir.path()).unwrap())
};
let seed = nil_session_start(1, sid);
store
.create_segment(sid, segment, std::slice::from_ref(&seed))
.unwrap();
let path = if worker_layout {
dir.path().join("segments").join(format!("{segment}.jsonl"))
} else {
dir.path()
.join(sid.to_string())
.join(format!("{segment}.jsonl"))
};
let before = std::fs::read(&path).unwrap();
let invalid = LogEntry::AnnotatedSystemItem {
ts: 2,
extensions: vec![],
entry: session_store::LoggedSystemHistoryEntry {
item: SystemItem::Input {
body: "no policy".into(),
},
metadata: LoggedSessionHistoryMetadata::legacy_unknown(),
},
};
assert!(matches!(
store.append(sid, segment, &invalid),
Err(StoreError::InvalidSystemInput(_))
));
assert!(matches!(
store.create_segment(sid, segment, &[seed, invalid.clone()]),
Err(StoreError::InvalidSystemInput(_))
));
assert!(
serde_json::from_value::<LogEntry>(serde_json::to_value(invalid).unwrap()).is_err()
);
assert_eq!(std::fs::read(path).unwrap(), before);
assert_eq!(store.read_all(sid, segment).unwrap().len(), 1);
}
}
crates/session-store/tests/session_test.rs
+1
View File
@@ -24,6 +24,7 @@ fn annotated(items: &[Item]) -> Vec<session_store::LoggedHistoryEntry> {
.map(|item| session_store::LoggedHistoryEntry {
item: session_store::LoggedItem::from(item),
metadata: session_store::LoggedSessionHistoryMetadata {
system_input: None,
entry_id: session_store::LoggedSessionHistoryEntryId::new(),
origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
derivation: None,