diff --git a/crates/session-store/README.md b/crates/session-store/README.md index 1aecdad4a..4a3b9bbd3 100644 --- a/crates/session-store/README.md +++ b/crates/session-store/README.md @@ -11,5 +11,7 @@ checks or Runtime catalog lifecycle management. - [`worker_session_store`]: storage boundary for a Session bound to a Worker - [`worker_metadata`]: current Worker state and active Segment switching - [`system_item`]: records of system input presented to the model +- [`capture`]: Worker-independent immutable public values and bounded body transport; + authorization, source I/O, and delivery lifecycle stay with the Host - [`public_snapshot`] / [`public_index`]: public projections of stored history - [`event_trace`]: diagnostic records independent of history replay diff --git a/crates/session-store/src/capture.rs b/crates/session-store/src/capture.rs new file mode 100644 index 000000000..4c9b4d3bc --- /dev/null +++ b/crates/session-store/src/capture.rs @@ -0,0 +1,948 @@ +//! Exploration domain for immutable Session captures, with no Workspace/Memory mutation capability. +//! +//! A capture is an ordered projection of committed entries; exploration excludes reasoning and +//! system prompts, including Host compaction summaries and other derived context. Full inherited +//! summary input belongs to compaction, not public observation or Memory evidence. Overview, +//! search, bounded reads, and evidence conversion share `SessionEntryRef`. +//! `SessionEntryRef` is a contextual UI reference. `ScopedSessionEntryRef` additionally binds +//! the Session and exact immutable capture, and is required for portable checked operations. +//! Appends preserve existing entry identities. Sparse real user/assistant anchors and intervening +//! entry counts index long histories. Search with an empty filter also provides bounded range lists. +//! +//! Host-fixed snapshot exploration, Memory candidate evidence validation, and current Worker +//! observation share only this domain; they do not enable or modify each other's Features. +//! The provider owns authorization to recapture a current Worker. A capture itself proves neither +//! completion nor approval and grants no business-mutation authority. + +use std::sync::Arc; + +use crate::{ + LoggedSessionHistoryMetadata as SessionHistoryMetadata, + LoggedSessionHistoryOrigin as HistoryOrigin, +}; +mod body; +use agen::{HistoryEntry, Item, Role}; +pub use body::*; +use protocol::{ + SessionContentPart, SessionEntryProvenance, SessionMessageRole, SessionSnapshot, + SessionSnapshotEntryData, +}; +use serde::{Deserialize, Serialize}; + +const DEFAULT_SEARCH_LIMIT: usize = 20; +const MAX_SEARCH_LIMIT: usize = 50; +const DEFAULT_READ_MAX_ITEMS: usize = 40; +const MAX_READ_MAX_ITEMS: usize = 80; +const DEFAULT_READ_MAX_BYTES: usize = 32 * 1024; +const OVERVIEW_ANCHOR_STRIDE: usize = 8; + +/// Append-stable contextual UI reference derived from a logical entry identity. +/// Portable consumers must bind it with `SessionCapture::scoped_ref` before use. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct SessionEntryRef(String); + +impl SessionEntryRef { + pub fn from_history_entry_id(entry_id: &crate::LoggedSessionHistoryEntryId) -> Self { + Self(format!("E{}", entry_id.0)) + } + + pub fn parse(value: &str) -> Option { + let suffix = value.strip_prefix('E')?; + if suffix.is_empty() + || suffix.len() > 64 + || !suffix + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-') + { + return None; + } + Some(Self(value.to_string())) + } + + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl std::fmt::Display for SessionEntryRef { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(&self.0) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReferenceKind { + User, + Assistant, + System, + Tool, +} + +impl ReferenceKind { + pub fn as_str(self) -> &'static str { + match self { + Self::User => "user", + Self::Assistant => "assistant", + Self::System => "system", + Self::Tool => "tool", + } + } + + pub fn parse(value: &str) -> Option { + match value { + "user" => Some(Self::User), + "assistant" | "agent" => Some(Self::Assistant), + "system" => Some(Self::System), + "tool" => Some(Self::Tool), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ToolPart { + Input, + Output, + Both, +} + +impl ToolPart { + pub fn parse(value: &str) -> Option { + match value { + "input" => Some(Self::Input), + "output" => Some(Self::Output), + "both" => Some(Self::Both), + _ => None, + } + } + + pub fn as_str(self) -> &'static str { + match self { + Self::Input => "input", + Self::Output => "output", + Self::Both => "both", + } + } + + pub fn matches(self, actual: ToolPart) -> bool { + matches!(self, ToolPart::Both) || self == actual + } +} + +#[derive(Debug, Clone)] +pub struct OverviewItem { + pub id: SessionEntryRef, + pub origin: SessionEntryProvenance, + pub entry_range: [u64; 2], + pub kind: ReferenceKind, + pub label: String, + pub text: String, + pub intervening_entries: usize, +} + +#[derive(Debug, Clone)] +pub struct ReferenceEntry { + pub attachment_count: usize, + use_policy: Option, + pub id: SessionEntryRef, + pub origin: SessionEntryProvenance, + pub entry_range: [u64; 2], + pub kind: ReferenceKind, + pub tool_part: Option, + pub tool_name: Option, + pub label: String, + pub summary: String, + search_text: String, +} + +#[derive(Debug, Clone, Default)] +pub struct SearchOptions { + pub query: String, + pub kind: Option, + pub tool_part: Option, + pub tool_name: Option, + pub limit: Option, + pub min_entry_index: Option, + pub from: Option, + pub through: Option, + pub offset: usize, +} + +#[derive(Debug, Clone)] +pub struct SearchHit { + pub id: SessionEntryRef, + pub origin: SessionEntryProvenance, + pub kind: ReferenceKind, + pub tool_part: Option, + pub tool_name: Option, + pub entry_range: [u64; 2], + pub label: String, + pub summary: String, +} + +#[derive(Debug, Clone, Copy)] +pub enum ReadSelector<'a> { + Id(&'a str), + EntryRange([u64; 2]), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReadDetail { + Compact, + Full, +} + +#[derive(Debug, Clone)] +pub struct ReadOptions { + pub include_tools: bool, + pub tool_part: ToolPart, + pub detail: ReadDetail, + pub max_items: usize, + pub max_bytes: usize, +} + +impl Default for ReadOptions { + fn default() -> Self { + Self { + include_tools: true, + tool_part: ToolPart::Both, + detail: ReadDetail::Compact, + max_items: DEFAULT_READ_MAX_ITEMS, + max_bytes: DEFAULT_READ_MAX_BYTES, + } + } +} + +#[derive(Debug, Clone)] +pub struct ReadEntry { + pub id: SessionEntryRef, + pub origin: SessionEntryProvenance, + pub kind: ReferenceKind, + pub tool_part: Option, + pub tool_name: Option, + pub entry_range: [u64; 2], + pub label: String, + pub text: String, +} + +#[derive(Debug, Clone)] +pub struct ReadResult { + pub entries: Vec, + pub truncated: bool, +} + +#[derive(Debug, Clone)] +pub struct SessionEntryEvidence { + pub segment_id: String, + pub entry_ref: SessionEntryRef, + pub origin: SessionEntryProvenance, + pub entry_range: [u64; 2], + pub kind: ReferenceKind, + pub tool_part: Option, + pub label: String, + pub summary: String, + pub excerpt: String, +} + +#[derive(Debug, Clone)] +struct CapturedHistoryEntry { + use_policy: Option, + item: Item, + entry_id: crate::LoggedSessionHistoryEntryId, + origin: SessionEntryProvenance, + dto: CaptureEntryDto, +} + +#[derive(Debug, Clone)] +pub struct SessionCapture { + session_id: String, + segment_id: String, + capture_id: String, + entries: Arc>, + overview: Vec, + index: Vec, +} + +impl SessionCapture { + pub fn from_session_snapshot( + session_id: impl Into, + segment_id: impl Into, + snapshot: SessionSnapshot, + ) -> Self { + let segment_id = segment_id.into(); + let entries = snapshot + .entries + .into_iter() + .enumerate() + .filter_map(|(index, entry)| { + let policy = CaptureUsePolicy { + recording_material: false, + memory_evidence: entry.use_policy.is_none_or(|p| p.memory_evidence), + observation_readable: entry.use_policy.is_none_or(|p| p.observation_readable), + }; + let explicit = entry.use_policy.is_some(); + let item = match entry.data { + SessionSnapshotEntryData::UserInput { segments } => { + Item::user_message(protocol::Segment::flatten_to_text(&segments)) + } + SessionSnapshotEntryData::Message { role, content } => Item::Message { + id: None, + status: None, + role: match role { + SessionMessageRole::User => Role::User, + SessionMessageRole::Assistant => Role::Assistant, + }, + content: content + .into_iter() + .map(|part| match part { + SessionContentPart::Text { text } => { + agen::ContentPart::Text { text } + } + SessionContentPart::Refusal { refusal } => { + agen::ContentPart::Refusal { refusal } + } + }) + .collect(), + }, + SessionSnapshotEntryData::ToolCall { + call_id, + name, + arguments, + } => Item::tool_call(call_id, name, arguments), + SessionSnapshotEntryData::ToolResult { + call_id, + summary, + content, + is_error, + .. + } => Item::tool_result_item(call_id, summary, content, is_error), + SessionSnapshotEntryData::SystemItem { content, .. } if explicit => { + Item::system_message(content) + } + _ => return None, + }; + let body = CaptureBody::project(&item, &entry.provenance, explicit)?; + if !policy.any() { + return None; + } + Some(CaptureEntryDto { + entry_id: entry.entry_id, + order: index as u64, + source: CaptureSource { + segment_id: segment_id.clone(), + entry_index: index as u64, + provenance: entry.provenance, + history_origin: None, + derived_from: entry.derived_from, + }, + policy, + body, + }) + }) + .collect(); + Self::from_projected(session_id.into(), segment_id, entries) + } + + pub fn new( + session_id: impl Into, + segment_id: impl Into, + items: Vec, + ) -> Self { + let segment_id = segment_id.into(); + let entries = items + .into_iter() + .enumerate() + .filter_map(|(index, item)| { + let origin = SessionEntryProvenance::LegacyUnknown; + let body = CaptureBody::project(&item, &origin, false)?; + Some(CaptureEntryDto { + entry_id: format!("{index:08}"), + order: index as u64, + source: CaptureSource { + segment_id: segment_id.clone(), + entry_index: index as u64, + provenance: origin, + history_origin: None, + derived_from: Vec::new(), + }, + policy: CaptureUsePolicy::PUBLIC, + body, + }) + }) + .collect(); + Self::from_projected(session_id.into(), segment_id, entries) + } + + pub fn from_history_entries( + session_id: impl Into, + segment_id: impl Into, + entries: Vec>, + ) -> Self { + let segment_id = segment_id.into(); + let entries = entries + .into_iter() + .enumerate() + .filter_map(|(index, entry)| { + let origin = public_provenance(&entry.annotation.origin); + let mut item = entry.item; + let mut explicit = false; + let policy = if let Some(input) = entry.annotation.input_metadata { + explicit = true; + if matches!( + item, + Item::Message { + role: Role::System, + .. + } + ) { + // Only the explicit public projection crosses this boundary. + let public = input.public_item()?; + item = Item::system_message(public.body.clone()); + } else if input.feature_payload().is_some() || input.public_item().is_some() { + return None; + } + CaptureUsePolicy::from(input.usage()) + } else { + if matches!(entry.annotation.origin, HistoryOrigin::Provided { .. }) { + return None; + } + CaptureUsePolicy::PUBLIC + }; + if !policy.any() { + return None; + } + let body = CaptureBody::project(&item, &origin, explicit)?; + Some(CaptureEntryDto { + entry_id: entry.annotation.entry_id.0, + order: index as u64, + source: CaptureSource { + segment_id: segment_id.clone(), + entry_index: index as u64, + provenance: origin, + history_origin: Some(entry.annotation.origin), + derived_from: entry + .annotation + .derivation + .map(|d| d.sources.into_iter().map(|id| id.0).collect()) + .unwrap_or_default(), + }, + policy, + body, + }) + }) + .collect(); + Self::from_projected(session_id.into(), segment_id, entries) + } + + fn from_projected(session_id: String, segment_id: String, dtos: Vec) -> Self { + let dto = CaptureDto { + version: CAPTURE_VERSION, + session_id: session_id.clone(), + segment_id: segment_id.clone(), + entries: dtos, + }; + let capture_id = dto.fingerprint(); + let entries: Vec<_> = dto + .entries + .into_iter() + .map(|dto| CapturedHistoryEntry { + use_policy: Some(protocol::SessionEntryUsePolicy { + observation_readable: dto.policy.observation_readable, + memory_evidence: dto.policy.memory_evidence, + }), + item: dto.body.to_item(), + entry_id: crate::LoggedSessionHistoryEntryId(dto.entry_id.clone()), + origin: dto.source.provenance.clone(), + dto, + }) + .collect(); + let mut overview = Vec::new(); + let mut index = Vec::new(); + + for entry in &entries { + let item = &entry.item; + let entry_range = [entry.dto.order; 2]; + let attachment_count = match entry.dto.body { + CaptureBody::ToolResult { + attachment_count, .. + } => attachment_count, + _ => 0, + }; + match item { + Item::Message { role, content, .. } => { + let Some(kind) = message_reference_kind(&entry.origin, role, entry.use_policy) + else { + continue; + }; + let text = content + .iter() + .map(|p| p.as_text()) + .collect::>() + .join(""); + let label = format!("{} message", kind.as_str()); + let summary = truncate_chars(&text, 240); + let id = SessionEntryRef::from_history_entry_id(&entry.entry_id); + index.push(ReferenceEntry { + use_policy: entry.use_policy, + attachment_count, + id: id.clone(), + origin: entry.origin.clone(), + entry_range, + kind, + tool_part: None, + tool_name: None, + label: label.clone(), + summary: summary.clone(), + search_text: text.clone(), + }); + if entry.use_policy.is_none_or(|p| p.observation_readable) + && matches!(kind, ReferenceKind::User | ReferenceKind::Assistant) + { + overview.push(OverviewItem { + id: id.clone(), + origin: entry.origin.clone(), + entry_range, + kind, + label, + text, + intervening_entries: 0, + }); + } + } + Item::ToolCall { + name, arguments, .. + } => { + let text = format!("{name}\n{arguments}"); + index.push(ReferenceEntry { + use_policy: entry.use_policy, + attachment_count, + id: SessionEntryRef::from_history_entry_id(&entry.entry_id), + origin: entry.origin.clone(), + entry_range, + kind: ReferenceKind::Tool, + tool_part: Some(ToolPart::Input), + tool_name: Some(name.clone()), + label: format!("tool input: {name}"), + summary: format!("Tool call {name}"), + search_text: text, + }); + } + Item::ToolResult { + summary, + content, + attachments: _, + .. + } => { + let attachment_marker = if attachment_count == 0 { + String::new() + } else { + format!("\n[{} image attachment(s)]", attachment_count) + }; + let text = format!( + "{summary}\n{}{attachment_marker}", + content.as_deref().unwrap_or_default(), + ); + index.push(ReferenceEntry { + use_policy: entry.use_policy, + attachment_count, + id: SessionEntryRef::from_history_entry_id(&entry.entry_id), + origin: entry.origin.clone(), + entry_range, + kind: ReferenceKind::Tool, + tool_part: Some(ToolPart::Output), + tool_name: None, + label: "tool output".to_string(), + summary: truncate_chars(summary, 240), + search_text: text, + }); + } + Item::AssistantResponseBoundary { .. } | Item::Reasoning { .. } => {} + } + } + + if overview.len() > 2 { + let last = overview.len() - 1; + overview = overview + .into_iter() + .enumerate() + .filter_map(|(index, entry)| { + (index == 0 || index == last || index % OVERVIEW_ANCHOR_STRIDE == 0) + .then_some(entry) + }) + .collect(); + } + + for overview_index in 0..overview.len().saturating_sub(1) { + let current_entry = overview[overview_index].entry_range[0]; + let next_entry = overview[overview_index + 1].entry_range[0]; + overview[overview_index].intervening_entries = index + .iter() + .filter(|entry| { + let entry_index = entry.entry_range[0]; + entry_index > current_entry && entry_index < next_entry + }) + .count(); + } + + Self { + session_id, + segment_id, + capture_id, + entries: Arc::new(entries), + overview, + index, + } + } + + /// Public source entries are immutable; callers retain their own business decisions. + pub fn entries(&self) -> impl ExactSizeIterator { + self.entries.iter().map(|entry| &entry.dto) + } + + pub fn overview(&self) -> &[OverviewItem] { + &self.overview + } + + pub fn source_index_for_ref(&self, reference: &SessionEntryRef) -> Option { + self.index + .iter() + .find(|entry| entry.id == *reference) + .map(|entry| entry.entry_range[0]) + } + + pub fn search(&self, options: &SearchOptions) -> Vec { + if self.validate_search(options).is_err() { + return Vec::new(); + } + let query = options.query.trim().to_lowercase(); + let limit = options + .limit + .unwrap_or(DEFAULT_SEARCH_LIMIT) + .clamp(1, MAX_SEARCH_LIMIT); + let tool_name = options.tool_name.as_deref(); + let min_entry_index = options + .from + .as_ref() + .and_then(|reference| self.source_index_for_ref(reference)) + .unwrap_or_else(|| options.min_entry_index.unwrap_or(0)); + let max_entry_index = options + .through + .as_ref() + .and_then(|reference| self.source_index_for_ref(reference)) + .unwrap_or(u64::MAX); + let mut skipped = 0usize; + let mut hits = Vec::new(); + + for entry in &self.index { + if entry.use_policy.is_some_and(|p| !p.observation_readable) { + continue; + } + if entry.entry_range[0] < min_entry_index || entry.entry_range[0] > max_entry_index { + continue; + } + if let Some(kind) = options.kind { + if entry.kind != kind { + continue; + } + } + if let Some(part) = options.tool_part { + if entry.kind != ReferenceKind::Tool { + continue; + } + let Some(actual) = entry.tool_part else { + continue; + }; + if !part.matches(actual) { + continue; + } + } + if let Some(name) = tool_name { + if entry.tool_name.as_deref() != Some(name) { + continue; + } + } + if !query.is_empty() && !entry.search_text.to_lowercase().contains(&query) { + continue; + } + if skipped < options.offset { + skipped += 1; + continue; + } + hits.push(SearchHit { + id: entry.id.clone(), + origin: entry.origin.clone(), + kind: entry.kind, + tool_part: entry.tool_part, + tool_name: entry.tool_name.clone(), + entry_range: entry.entry_range, + label: entry.label.clone(), + summary: entry.summary.clone(), + }); + if hits.len() >= limit { + break; + } + } + + hits + } + + pub fn read(&self, selector: ReadSelector<'_>, options: ReadOptions) -> ReadResult { + let max_items = options.max_items.clamp(1, MAX_READ_MAX_ITEMS); + let max_bytes = options.max_bytes.max(1); + let mut entries = Vec::new(); + let mut bytes = 0usize; + let mut truncated = false; + + let selected: Vec<&ReferenceEntry> = match selector { + ReadSelector::Id(id) => self + .index + .iter() + .filter(|entry| entry.id.as_str() == id) + .collect(), + ReadSelector::EntryRange([start, end]) => self + .index + .iter() + .filter(|entry| entry.entry_range[0] >= start && entry.entry_range[0] <= end) + .collect(), + }; + + for entry in selected { + if entry.use_policy.is_some_and(|p| !p.observation_readable) { + continue; + } + if entries.len() >= max_items { + truncated = true; + break; + } + if entry.kind == ReferenceKind::Tool { + if !options.include_tools { + continue; + } + if let Some(actual) = entry.tool_part { + if !options.tool_part.matches(actual) { + continue; + } + } + } + let Some(item) = self + .entries + .iter() + .find(|captured| captured.dto.order == entry.entry_range[0]) + .map(|entry| &entry.item) + else { + continue; + }; + let remaining = max_bytes.saturating_sub(bytes); + if remaining == 0 { + truncated = true; + break; + } + let rendered = render_item(item, entry, options.detail, usize::MAX); + truncated |= rendered.len() > remaining; + let text = truncate_bytes(&rendered, remaining); + bytes = bytes.saturating_add(text.len()); + entries.push(ReadEntry { + id: entry.id.clone(), + origin: entry.origin.clone(), + kind: entry.kind, + tool_part: entry.tool_part, + tool_name: entry.tool_name.clone(), + entry_range: entry.entry_range, + label: entry.label.clone(), + text, + }); + if truncated { + break; + } + } + + ReadResult { entries, truncated } + } + + /// Resolves the committed tool-call input for a provider call id. This is + /// used by explicit-memory receipts only after the containing run is durable, + /// so a model-authored claim can cite its own committed tool invocation + /// without inventing a SessionEntryRef. + pub fn evidence_for_tool_call(&self, call_id: &str) -> Option { + let captured = self.entries.iter().find(|entry| { + matches!( + &entry.item, + Item::ToolCall { + call_id: stored_call_id, + .. + } if stored_call_id == call_id + ) + })?; + let reference = self.index.iter().find(|entry| { + entry.entry_range[0] == captured.dto.order + && entry.kind == ReferenceKind::Tool + && entry.tool_part == Some(ToolPart::Input) + })?; + self.evidence_for(reference.id.as_str()) + } + + pub fn evidence_for(&self, id: &str) -> Option { + let entry = self.index.iter().find(|entry| entry.id.as_str() == id)?; + if entry.use_policy.is_some_and(|p| !p.memory_evidence) { + return None; + } + let item = &self + .entries + .iter() + .find(|captured| captured.dto.order == entry.entry_range[0])? + .item; + let excerpt = render_item(item, entry, ReadDetail::Compact, 2 * 1024); + Some(SessionEntryEvidence { + segment_id: self + .entries + .iter() + .find(|captured| captured.dto.order == entry.entry_range[0])? + .dto + .source + .segment_id + .clone(), + entry_ref: entry.id.clone(), + origin: entry.origin.clone(), + entry_range: entry.entry_range, + kind: entry.kind, + tool_part: entry.tool_part, + label: entry.label.clone(), + summary: entry.summary.clone(), + excerpt, + }) + } +} + +fn public_provenance(origin: &HistoryOrigin) -> SessionEntryProvenance { + match origin { + HistoryOrigin::HumanInput { .. } | HistoryOrigin::LocalHumanInput => { + SessionEntryProvenance::HumanInput + } + HistoryOrigin::WorkerInput { .. } => SessionEntryProvenance::WorkerInput, + HistoryOrigin::FlowInstruction { .. } => SessionEntryProvenance::FlowInstruction, + HistoryOrigin::BackendInstruction { .. } | HistoryOrigin::Provided { .. } => { + SessionEntryProvenance::BackendInstruction + } + HistoryOrigin::ModelOutput { .. } => SessionEntryProvenance::ModelOutput, + HistoryOrigin::ToolOutput { .. } => SessionEntryProvenance::ToolOutput, + HistoryOrigin::CompactionSummary + | HistoryOrigin::DerivedSummary + | HistoryOrigin::JournalContext => SessionEntryProvenance::DerivedSummary, + HistoryOrigin::LegacyUnknown => SessionEntryProvenance::LegacyUnknown, + } +} + +fn message_reference_kind( + origin: &SessionEntryProvenance, + provider_role: &Role, + use_policy: Option, +) -> Option { + if use_policy.is_some() { + return match provider_role { + Role::System => Some(ReferenceKind::System), + Role::User => Some(ReferenceKind::User), + Role::Assistant => Some(ReferenceKind::Assistant), + }; + } + match origin { + SessionEntryProvenance::HumanInput | SessionEntryProvenance::WorkerInput => { + Some(ReferenceKind::User) + } + SessionEntryProvenance::ModelOutput => Some(ReferenceKind::Assistant), + SessionEntryProvenance::ToolOutput => Some(ReferenceKind::Tool), + SessionEntryProvenance::LegacyUnknown => match provider_role { + Role::User => Some(ReferenceKind::User), + Role::Assistant => Some(ReferenceKind::Assistant), + Role::System => None, + }, + // Flow/backend/system content remains out of the observation surface + // even when represented with a provider user/system role. + SessionEntryProvenance::FlowInstruction + | SessionEntryProvenance::BackendInstruction + | SessionEntryProvenance::DerivedSummary => None, + } +} + +fn render_item( + item: &Item, + entry: &ReferenceEntry, + detail: ReadDetail, + max_bytes: usize, +) -> String { + let text = match item { + Item::Message { role, content, .. } => { + let text = content + .iter() + .map(|p| p.as_text()) + .collect::>() + .join(""); + format!("[{} {:?}] {text}", entry.id, role) + } + Item::ToolCall { + name, arguments, .. + } => match detail { + ReadDetail::Compact => format!("[{} ToolInput {name}] (arguments omitted)", entry.id), + ReadDetail::Full => format!("[{} ToolInput {name}]\narguments: {arguments}", entry.id), + }, + Item::ToolResult { + summary, + content, + attachments: _, + is_error, + .. + } => { + let attachment_line = if entry.attachment_count == 0 { + String::new() + } else { + format!("\nattachments: {} image(s)", entry.attachment_count) + }; + match detail { + ReadDetail::Compact => format!( + "[{} ToolOutput{}]\nsummary: {summary}\ncontent: (omitted){attachment_line}", + entry.id, + if *is_error { " error" } else { "" }, + ), + ReadDetail::Full => format!( + "[{} ToolOutput{}]\nsummary: {summary}\ncontent: {}{attachment_line}", + entry.id, + if *is_error { " error" } else { "" }, + content.as_deref().unwrap_or_default(), + ), + } + } + Item::AssistantResponseBoundary { .. } => { + format!("[{} Assistant response boundary omitted]", entry.id) + } + Item::Reasoning { .. } => format!("[{} Reasoning omitted]", entry.id), + }; + truncate_bytes(&text, max_bytes) +} + +fn truncate_chars(text: &str, max_chars: usize) -> String { + if max_chars == 0 { + return "… [truncated]".to_string(); + } + if text.chars().count() <= max_chars { + return text.to_string(); + } + let mut out = text.chars().take(max_chars).collect::(); + out.push_str("… [truncated]"); + out +} + +fn truncate_bytes(text: &str, max_bytes: usize) -> String { + if text.len() <= max_bytes { + return text.to_owned(); + } + let marker = "… [truncated]"; + let mut end = max_bytes.saturating_sub(marker.len()); + while !text.is_char_boundary(end) { + end -= 1; + } + let mut out = text[..end].to_owned(); + if max_bytes >= marker.len() { + out.push_str(marker); + } + out +} diff --git a/crates/session-store/src/capture/body.rs b/crates/session-store/src/capture/body.rs new file mode 100644 index 000000000..747586889 --- /dev/null +++ b/crates/session-store/src/capture/body.rs @@ -0,0 +1,934 @@ +//! Portable public capture values. DTOs are transport data, not authorization handles. +//! Checked construction validates an explicitly public projection; adapters must never put +//! a provider System body or private Feature payload in `PublicSystem`. + +use super::*; +use serde::de::Error as _; +use sha2::{Digest, Sha256}; +use std::collections::HashSet; +use std::io::Write; + +pub const CAPTURE_VERSION: u32 = 1; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CaptureUsePolicy { + pub recording_material: bool, + pub memory_evidence: bool, + pub observation_readable: bool, +} + +impl CaptureUsePolicy { + pub const PUBLIC: Self = Self { + recording_material: true, + memory_evidence: true, + observation_readable: true, + }; + pub fn any(self) -> bool { + self.recording_material || self.memory_evidence || self.observation_readable + } + pub fn allows(self, purpose: CapturePurpose) -> bool { + match purpose { + CapturePurpose::Recording => self.recording_material, + CapturePurpose::MemoryEvidence => self.memory_evidence, + CapturePurpose::Observation => self.observation_readable, + } + } +} + +impl From for CaptureUsePolicy { + fn from(policy: crate::HistoryInputUsePolicy) -> Self { + Self { + recording_material: policy.recording_material == crate::InputEvidenceUse::Include, + memory_evidence: policy.memory_evidence == crate::InputEvidenceUse::Include, + observation_readable: policy.observation == crate::InputObservationUse::Readable, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum CapturePurpose { + Recording, + MemoryEvidence, + Observation, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum CaptureRole { + User, + Assistant, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum CaptureContentPart { + Text { text: String }, + Refusal { refusal: String }, +} + +impl CaptureContentPart { + fn text(&self) -> &str { + match self { + Self::Text { text } => text, + Self::Refusal { refusal } => refusal, + } + } +} + +/// The type has no reasoning, raw System prompt, Feature payload, or image bytes variant. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum CaptureBody { + Message { + role: CaptureRole, + content: Vec, + }, + PublicSystem { + text: String, + }, + ToolCall { + call_id: String, + name: String, + arguments: String, + }, + ToolResult { + call_id: String, + summary: String, + content: Option, + is_error: bool, + attachment_count: usize, + }, +} + +impl CaptureBody { + pub(super) fn project( + item: &Item, + origin: &SessionEntryProvenance, + explicit: bool, + ) -> Option { + match item { + Item::Message { role, content, .. } => { + message_reference_kind( + origin, + role, + explicit.then_some(protocol::SessionEntryUsePolicy { + observation_readable: true, + memory_evidence: true, + }), + )?; + if *role == Role::System { + if !explicit { + return None; + } + return Some(Self::PublicSystem { + text: content.iter().map(|p| p.as_text()).collect::(), + }); + } + Some(Self::Message { + role: if *role == Role::User { + CaptureRole::User + } else { + CaptureRole::Assistant + }, + content: content + .iter() + .map(|p| match p { + agen::ContentPart::Text { text } => { + CaptureContentPart::Text { text: text.clone() } + } + agen::ContentPart::Refusal { refusal } => CaptureContentPart::Refusal { + refusal: refusal.clone(), + }, + }) + .collect(), + }) + } + Item::ToolCall { + call_id, + name, + arguments, + .. + } => Some(Self::ToolCall { + call_id: call_id.clone(), + name: name.clone(), + arguments: arguments.clone(), + }), + Item::ToolResult { + call_id, + summary, + content, + is_error, + attachments, + .. + } => Some(Self::ToolResult { + call_id: call_id.clone(), + summary: summary.clone(), + content: content.clone(), + is_error: *is_error, + attachment_count: attachments.len(), + }), + Item::Reasoning { .. } | Item::AssistantResponseBoundary { .. } => None, + } + } + + pub(super) fn to_item(&self) -> Item { + match self { + Self::Message { role, content } => Item::Message { + id: None, + status: None, + role: match role { + CaptureRole::User => Role::User, + CaptureRole::Assistant => Role::Assistant, + }, + content: content + .iter() + .map(|p| match p { + CaptureContentPart::Text { text } => { + agen::ContentPart::Text { text: text.clone() } + } + CaptureContentPart::Refusal { refusal } => agen::ContentPart::Refusal { + refusal: refusal.clone(), + }, + }) + .collect(), + }, + Self::PublicSystem { text } => Item::system_message(text.clone()), + Self::ToolCall { + call_id, + name, + arguments, + } => Item::tool_call(call_id.clone(), name.clone(), arguments.clone()), + Self::ToolResult { + call_id, + summary, + content, + is_error, + .. + } => { + Item::tool_result_item(call_id.clone(), summary.clone(), content.clone(), *is_error) + } + } + } + + fn fields(&self) -> Vec<(RawField, &str)> { + match self { + Self::Message { content, .. } => content + .iter() + .enumerate() + .map(|(i, p)| (RawField::MessagePart(i), p.text())) + .collect(), + Self::PublicSystem { text } => vec![(RawField::PublicSystem, text)], + Self::ToolCall { arguments, .. } => vec![(RawField::ToolArguments, arguments)], + Self::ToolResult { + summary, content, .. + } => { + let mut fields = vec![(RawField::ToolSummary, summary.as_str())]; + if let Some(content) = content { + fields.push((RawField::ToolContent, content)); + } + fields + } + } + } + + fn metadata(&self, field: RawField) -> CaptureBodyMetadata { + match self { + Self::Message { role, content } => CaptureBodyMetadata::Message { + role: *role, + refusal: matches!(field, RawField::MessagePart(i) if matches!(content.get(i), Some(CaptureContentPart::Refusal { .. }))), + }, + Self::PublicSystem { .. } => CaptureBodyMetadata::PublicSystem, + Self::ToolCall { call_id, name, .. } => CaptureBodyMetadata::ToolCall { + call_id: call_id.clone(), + name: name.clone(), + }, + Self::ToolResult { + call_id, + content, + is_error, + attachment_count, + .. + } => CaptureBodyMetadata::ToolResult { + call_id: call_id.clone(), + has_content: content.is_some(), + is_error: *is_error, + attachment_count: *attachment_count, + }, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CaptureSource { + pub segment_id: String, + pub entry_index: u64, + pub provenance: SessionEntryProvenance, + /// Full original origin when supplied by the committed-history adapter. + #[serde(deserialize_with = "strict_origin")] + pub history_origin: Option, + pub derived_from: Vec, +} + +// Existing history origin types predate strict DTOs. Comparing their canonical value +// rejects unknown nested keys instead of silently accepting/losing them at this boundary. +fn strict_origin<'de, D: serde::Deserializer<'de>>( + deserializer: D, +) -> Result, D::Error> { + let value = Option::::deserialize(deserializer)?; + value + .map(|value| { + let origin: crate::LoggedSessionHistoryOrigin = + serde_json::from_value(value.clone()).map_err(D::Error::custom)?; + if serde_json::to_value(&origin).map_err(D::Error::custom)? != value { + return Err(D::Error::custom("noncanonical or unknown origin fields")); + } + Ok(origin) + }) + .transpose() +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CaptureEntryDto { + pub entry_id: String, + /// Position in this captured view, independent of the original segment position. + pub order: u64, + pub source: CaptureSource, + pub policy: CaptureUsePolicy, + pub body: CaptureBody, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CaptureDto { + pub version: u32, + pub session_id: String, + pub segment_id: String, + pub entries: Vec, +} + +#[derive(Debug, Clone, Copy)] +pub struct CaptureLimits { + pub max_bytes: usize, + pub max_entries: usize, + pub max_text_bytes: usize, + pub max_identity_bytes: usize, + pub max_derivation_sources: usize, +} + +impl Default for CaptureLimits { + fn default() -> Self { + Self { + max_bytes: 16 * 1024 * 1024, + max_entries: 100_000, + max_text_bytes: 8 * 1024 * 1024, + max_identity_bytes: 4096, + max_derivation_sources: 1024, + } + } +} + +#[derive(Debug, thiserror::Error)] +pub enum CaptureError { + #[error("unsupported capture representation version")] + UnsupportedVersion, + #[error("capture exceeds caller resource limits")] + ResourceLimit, + #[error("invalid capture: {0}")] + Invalid(&'static str), + #[error("reference or cursor belongs to another capture/session/purpose")] + ForeignReference, + #[error("unknown entry reference")] + UnknownReference, + #[error("entry is not eligible for this purpose")] + PolicyDenied, + #[error("invalid entry range or UTF-8 byte offset")] + InvalidRange, + #[error("budget cannot fit the next item and continuation metadata")] + BudgetTooSmall, + #[error("invalid capture JSON: {0}")] + Json(#[from] serde_json::Error), +} + +struct ByteCounter { + bytes: usize, + limit: usize, +} +impl Write for ByteCounter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.bytes = self + .bytes + .checked_add(buf.len()) + .ok_or_else(|| std::io::Error::other("capture byte limit"))?; + if self.bytes > self.limit { + return Err(std::io::Error::other("capture byte limit")); + } + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +fn json_size(value: &T, limit: usize) -> Result { + let mut counter = ByteCounter { bytes: 0, limit }; + serde_json::to_writer(&mut counter, value).map_err(|_| CaptureError::ResourceLimit)?; + Ok(counter.bytes) +} + +fn identity(value: &str, limits: CaptureLimits) -> Result<(), CaptureError> { + if value.is_empty() + || value.len() > limits.max_identity_bytes + || value.chars().any(char::is_control) + { + return Err(CaptureError::Invalid("identity")); + } + Ok(()) +} + +impl CaptureDto { + pub fn validate(&self, limits: CaptureLimits) -> Result<(), CaptureError> { + if self.version != CAPTURE_VERSION { + return Err(CaptureError::UnsupportedVersion); + } + if self.entries.len() > limits.max_entries { + return Err(CaptureError::ResourceLimit); + } + identity(&self.session_id, limits)?; + identity(&self.segment_id, limits)?; + let mut ids = HashSet::new(); + let mut sources = HashSet::new(); + let mut previous = None; + for entry in &self.entries { + identity(&entry.entry_id, limits)?; + if SessionEntryRef::parse(&format!("E{}", entry.entry_id)).is_none() + || !ids.insert(&entry.entry_id) + { + return Err(CaptureError::Invalid("entry identity")); + } + if previous.is_some_and(|order| order >= entry.order) { + return Err(CaptureError::Invalid("entry order")); + } + previous = Some(entry.order); + identity(&entry.source.segment_id, limits)?; + if !sources.insert((&entry.source.segment_id, entry.source.entry_index)) { + return Err(CaptureError::Invalid("duplicate source position")); + } + if !entry.policy.any() { + return Err(CaptureError::Invalid("unusable retained entry")); + } + if entry.source.derived_from.len() > limits.max_derivation_sources { + return Err(CaptureError::ResourceLimit); + } + for source in &entry.source.derived_from { + identity(source, limits)?; + } + if let Some(origin) = &entry.source.history_origin { + if public_provenance(origin) != entry.source.provenance { + return Err(CaptureError::Invalid("origin mismatch")); + } + json_size(origin, limits.max_text_bytes)?; + } + for (_, text) in entry.body.fields() { + if text.len() > limits.max_text_bytes { + return Err(CaptureError::ResourceLimit); + } + } + match &entry.body { + CaptureBody::Message { content, .. } if content.is_empty() => { + return Err(CaptureError::Invalid("empty message parts")); + } + CaptureBody::ToolCall { call_id, name, .. } => { + identity(call_id, limits)?; + identity(name, limits)?; + } + CaptureBody::ToolResult { call_id, .. } => identity(call_id, limits)?, + _ => {} + } + } + json_size(self, limits.max_bytes)?; + Ok(()) + } + + pub fn encode(&self, limits: CaptureLimits) -> Result, CaptureError> { + self.validate(limits)?; + Ok(serde_json::to_vec(self)?) + } + + pub fn decode(bytes: &[u8], limits: CaptureLimits) -> Result { + if bytes.len() > limits.max_bytes { + return Err(CaptureError::ResourceLimit); + } + let dto: Self = serde_json::from_slice(bytes)?; + dto.validate(limits)?; + Ok(dto) + } + + pub(super) fn fingerprint(&self) -> String { + // These DTOs have only finite, infallibly serializable JSON primitives. + let bytes = serde_json::to_vec(self).expect("capture DTO serialization"); + Sha256::digest(bytes) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ScopedSessionEntryRef { + pub session_id: String, + pub capture_id: String, + pub entry_ref: SessionEntryRef, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RawField { + MessagePart(usize), + PublicSystem, + ToolArguments, + ToolSummary, + ToolContent, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RawRead { + pub reference: ScopedSessionEntryRef, + pub purpose: CapturePurpose, + pub field: RawField, + pub offset: usize, + pub text: String, + pub next_offset: Option, + pub total_bytes: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum CaptureBodyMetadata { + Message { + role: CaptureRole, + refusal: bool, + }, + PublicSystem, + ToolCall { + call_id: String, + name: String, + }, + ToolResult { + call_id: String, + has_content: bool, + is_error: bool, + attachment_count: usize, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CaptureCursor { + pub session_id: String, + pub capture_id: String, + pub purpose: CapturePurpose, + pub entry_position: usize, + pub field_index: usize, + pub byte_offset: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CaptureSliceItem { + pub reference: ScopedSessionEntryRef, + pub order: u64, + pub source: CaptureSource, + pub policy: CaptureUsePolicy, + pub metadata: CaptureBodyMetadata, + pub field: RawField, + pub offset: usize, + pub total_bytes: usize, + /// Raw source bytes, not a rendered preview or ellipsis-bearing truncation. + pub text: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CaptureSlice { + pub version: u32, + pub session_id: String, + pub capture_id: String, + pub purpose: CapturePurpose, + pub items: Vec, + pub next_cursor: Option, +} + +/// Bytes and value remain paired and immutable so the size proof cannot go stale. +#[derive(Debug, Clone)] +pub struct BudgetedCaptureSlice { + slice: CaptureSlice, + json: Vec, +} +impl BudgetedCaptureSlice { + pub fn slice(&self) -> &CaptureSlice { + &self.slice + } + pub fn as_json(&self) -> &[u8] { + &self.json + } + pub fn encoded_len(&self) -> usize { + self.json.len() + } +} + +impl SessionCapture { + pub fn session_id(&self) -> &str { + &self.session_id + } + pub fn segment_id(&self) -> &str { + &self.segment_id + } + pub fn capture_id(&self) -> &str { + &self.capture_id + } + + pub fn to_dto(&self) -> CaptureDto { + CaptureDto { + version: CAPTURE_VERSION, + session_id: self.session_id.clone(), + segment_id: self.segment_id.clone(), + entries: self.entries.iter().map(|e| e.dto.clone()).collect(), + } + } + + pub fn try_from_dto(dto: CaptureDto, limits: CaptureLimits) -> Result { + dto.validate(limits)?; + Ok(Self::from_projected( + dto.session_id, + dto.segment_id, + dto.entries, + )) + } + + pub fn encode(&self, limits: CaptureLimits) -> Result, CaptureError> { + self.to_dto().encode(limits) + } + pub fn decode(bytes: &[u8], limits: CaptureLimits) -> Result { + Self::try_from_dto(CaptureDto::decode(bytes, limits)?, limits) + } + + pub fn scoped_ref( + &self, + reference: &SessionEntryRef, + ) -> Result { + if self.source_index_for_ref(reference).is_none() { + return Err(CaptureError::UnknownReference); + } + Ok(ScopedSessionEntryRef { + session_id: self.session_id.clone(), + capture_id: self.capture_id.clone(), + entry_ref: reference.clone(), + }) + } + + pub fn resolve_ref( + &self, + reference: &ScopedSessionEntryRef, + purpose: CapturePurpose, + ) -> Result<&CaptureEntryDto, CaptureError> { + if reference.session_id != self.session_id || reference.capture_id != self.capture_id { + return Err(CaptureError::ForeignReference); + } + let entry = self + .entries + .iter() + .find(|entry| { + SessionEntryRef::from_history_entry_id(&entry.entry_id) == reference.entry_ref + }) + .ok_or(CaptureError::UnknownReference)?; + if !entry.dto.policy.allows(purpose) { + return Err(CaptureError::PolicyDenied); + } + Ok(&entry.dto) + } + + pub fn checked_range( + &self, + from: &ScopedSessionEntryRef, + through: &ScopedSessionEntryRef, + purpose: CapturePurpose, + ) -> Result, CaptureError> { + let start = self.resolve_ref(from, purpose)?.order; + let end = self.resolve_ref(through, purpose)?.order; + if start > end { + return Err(CaptureError::InvalidRange); + } + Ok(self + .entries + .iter() + .map(|entry| &entry.dto) + .filter(|entry| { + entry.order >= start && entry.order <= end && entry.policy.allows(purpose) + }) + .collect()) + } + + pub(super) fn validate_search(&self, options: &SearchOptions) -> Result<(), CaptureError> { + let from = options + .from + .as_ref() + .map(|r| { + self.source_index_for_ref(r) + .ok_or(CaptureError::UnknownReference) + }) + .transpose()?; + let through = options + .through + .as_ref() + .map(|r| { + self.source_index_for_ref(r) + .ok_or(CaptureError::UnknownReference) + }) + .transpose()?; + if from.unwrap_or(options.min_entry_index.unwrap_or(0)) > through.unwrap_or(u64::MAX) { + return Err(CaptureError::InvalidRange); + } + Ok(()) + } + + pub fn search_checked(&self, options: &SearchOptions) -> Result, CaptureError> { + self.validate_search(options)?; + Ok(self.search(options)) + } + + pub fn read_checked( + &self, + selector: ReadSelector<'_>, + options: ReadOptions, + ) -> Result { + match selector { + ReadSelector::Id(id) => { + let reference = SessionEntryRef::parse(id).ok_or(CaptureError::UnknownReference)?; + self.source_index_for_ref(&reference) + .ok_or(CaptureError::UnknownReference)?; + } + ReadSelector::EntryRange([start, end]) => { + if start > end + || !self.index.iter().any(|entry| entry.entry_range[0] == start) + || !self.index.iter().any(|entry| entry.entry_range[0] == end) + { + return Err(CaptureError::InvalidRange); + } + } + } + Ok(self.read(selector, options)) + } + + pub fn read_raw( + &self, + reference: &ScopedSessionEntryRef, + purpose: CapturePurpose, + field: RawField, + offset: usize, + max_bytes: usize, + ) -> Result { + let entry = self.resolve_ref(reference, purpose)?; + let text = entry + .body + .fields() + .into_iter() + .find(|(f, _)| *f == field) + .map(|(_, text)| text) + .ok_or(CaptureError::InvalidRange)?; + if offset > text.len() || !text.is_char_boundary(offset) { + return Err(CaptureError::InvalidRange); + } + let mut end = offset.saturating_add(max_bytes).min(text.len()); + while !text.is_char_boundary(end) { + end -= 1; + } + if end == offset && offset < text.len() { + return Err(CaptureError::BudgetTooSmall); + } + Ok(RawRead { + reference: reference.clone(), + purpose, + field, + offset, + text: text[offset..end].to_owned(), + next_offset: (end < text.len()).then_some(end), + total_bytes: text.len(), + }) + } + + fn cursor( + &self, + purpose: CapturePurpose, + entry_position: usize, + field_index: usize, + byte_offset: usize, + ) -> CaptureCursor { + CaptureCursor { + session_id: self.session_id.clone(), + capture_id: self.capture_id.clone(), + purpose, + entry_position, + field_index, + byte_offset, + } + } + + fn normalize_cursor(&self, mut cursor: CaptureCursor) -> Option { + while let Some(entry) = self.entries.get(cursor.entry_position) { + if entry.dto.policy.allows(cursor.purpose) + && cursor.field_index < entry.dto.body.fields().len() + { + return Some(cursor); + } + cursor.entry_position += 1; + cursor.field_index = 0; + cursor.byte_offset = 0; + } + None + } + + /// The returned JSON includes every item's source/policy, escaped body, and continuation. + /// Both the item count and the exact encoded length are bounded, with no Job-specific limit. + pub fn slice( + &self, + purpose: CapturePurpose, + cursor: Option<&CaptureCursor>, + max_items: usize, + max_bytes: usize, + ) -> Result { + if max_items == 0 { + return Err(CaptureError::BudgetTooSmall); + } + let start = if let Some(cursor) = cursor { + if cursor.session_id != self.session_id + || cursor.capture_id != self.capture_id + || cursor.purpose != purpose + { + return Err(CaptureError::ForeignReference); + } + let entry = self + .entries + .get(cursor.entry_position) + .ok_or(CaptureError::InvalidRange)?; + if !entry.dto.policy.allows(purpose) { + return Err(CaptureError::PolicyDenied); + } + let fields = entry.dto.body.fields(); + let text = fields + .get(cursor.field_index) + .ok_or(CaptureError::InvalidRange)? + .1; + if cursor.byte_offset > text.len() || !text.is_char_boundary(cursor.byte_offset) { + return Err(CaptureError::InvalidRange); + } + Some(cursor.clone()) + } else { + self.normalize_cursor(self.cursor(purpose, 0, 0, 0)) + }; + let mut slice = CaptureSlice { + version: CAPTURE_VERSION, + session_id: self.session_id.clone(), + capture_id: self.capture_id.clone(), + purpose, + items: Vec::new(), + next_cursor: start, + }; + if json_size(&slice, max_bytes).is_err() { + return Err(CaptureError::BudgetTooSmall); + } + while slice.items.len() < max_items { + let Some(current) = slice.next_cursor.clone() else { + break; + }; + let dto = &self.entries[current.entry_position].dto; + let fields = dto.body.fields(); + let (field, text) = fields[current.field_index]; + let reference = ScopedSessionEntryRef { + session_id: self.session_id.clone(), + capture_id: self.capture_id.clone(), + entry_ref: SessionEntryRef(format!("E{}", dto.entry_id)), + }; + let item = CaptureSliceItem { + reference, + order: dto.order, + source: dto.source.clone(), + policy: dto.policy, + metadata: dto.body.metadata(field), + field, + offset: current.byte_offset, + total_bytes: text.len(), + text: String::new(), + }; + slice.items.push(item); + let item_index = slice.items.len() - 1; + // Search only the partial interval: its cursor stays in the same field, + // making encoded size monotonic even when completing a field drops metadata. + let remaining = &text[current.byte_offset..]; + let mut candidate_bytes = remaining.len().min(max_bytes); + while !remaining.is_char_boundary(candidate_bytes) { + candidate_bytes -= 1; + } + let mut boundaries: Vec = remaining[..candidate_bytes] + .char_indices() + .map(|(i, _)| i) + .collect(); + boundaries.push(candidate_bytes); + let mut accepted = None; + let mut low = 0; + let mut high = boundaries.len(); + while low < high { + let mid = low + (high - low) / 2; + let end = current.byte_offset + boundaries[mid]; + slice.items[item_index].text = text[current.byte_offset..end].to_owned(); + slice.next_cursor = + Some(self.cursor(purpose, current.entry_position, current.field_index, end)); + if json_size(&slice, max_bytes).is_ok() { + accepted = Some(end); + low = mid + 1; + } else { + high = mid; + } + } + // A completed field can fit even when its partial continuation does not. + let completed = self.normalize_cursor(self.cursor( + purpose, + current.entry_position, + current.field_index + 1, + 0, + )); + if remaining.len() <= max_bytes { + slice.items[item_index].text = remaining.to_owned(); + slice.next_cursor = completed.clone(); + if json_size(&slice, max_bytes).is_ok() { + accepted = Some(text.len()); + } + } + let Some(end) = accepted.filter(|end| *end > current.byte_offset || text.is_empty()) + else { + slice.items.pop(); + slice.next_cursor = Some(current); + if slice.items.is_empty() { + return Err(CaptureError::BudgetTooSmall); + } + break; + }; + slice.items[item_index].text = text[current.byte_offset..end].to_owned(); + slice.next_cursor = if end == text.len() { + completed + } else { + Some(self.cursor(purpose, current.entry_position, current.field_index, end)) + }; + } + let json = serde_json::to_vec(&slice)?; + if json.len() > max_bytes { + return Err(CaptureError::BudgetTooSmall); + } + Ok(BudgetedCaptureSlice { slice, json }) + } +} diff --git a/crates/session-store/src/lib.rs b/crates/session-store/src/lib.rs index d767defca..15abc32dd 100644 --- a/crates/session-store/src/lib.rs +++ b/crates/session-store/src/lib.rs @@ -38,6 +38,7 @@ pub(crate) fn read_without_atime_bounded(path: &Path, max_bytes: u64) -> io::Res } } +pub mod capture; pub mod event_trace; pub mod fs_store; pub mod history; diff --git a/crates/session-store/tests/capture_contract.rs b/crates/session-store/tests/capture_contract.rs new file mode 100644 index 000000000..a605d6d6e --- /dev/null +++ b/crates/session-store/tests/capture_contract.rs @@ -0,0 +1,594 @@ +//! Public consumer boundary: real capture adapters, checked DTO codec and pure readers. +//! No Worker, persistence, live Session provider, or substituted capture implementation. + +use agen::{HistoryEntry, Item}; +use protocol::{ + SessionEntryProvenance, SessionSnapshot, SessionSnapshotEntry, SessionSnapshotEntryData, +}; +use session_store::capture::*; +use session_store::{ + FeatureInputPayload, HistoryInputMetadata, HistoryInputUsePolicy, InputCompactionUse, + InputEvidenceUse, InputObservationUse, LoggedHistoryDerivation, LoggedSessionHistoryEntryId, + LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, PublicSystemItem, + PublicSystemPresentation, +}; + +fn history( + id: &str, + item: Item, + input_metadata: Option, +) -> HistoryEntry { + HistoryEntry::new( + item, + LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId(id.into()), + origin: LoggedSessionHistoryOrigin::LocalHumanInput, + derivation: Some(LoggedHistoryDerivation { + sources: vec![LoggedSessionHistoryEntryId("earlier-entry".into())], + }), + input_metadata, + }, + ) +} + +fn usage(recording: bool, memory: bool, observation: bool) -> HistoryInputUsePolicy { + HistoryInputUsePolicy { + recording_material: if recording { + InputEvidenceUse::Include + } else { + InputEvidenceUse::Exclude + }, + memory_evidence: if memory { + InputEvidenceUse::Include + } else { + InputEvidenceUse::Exclude + }, + observation: if observation { + InputObservationUse::Readable + } else { + InputObservationUse::Hidden + }, + compaction: InputCompactionUse::Retain, + } +} + +fn reference(capture: &SessionCapture, id: &str) -> ScopedSessionEntryRef { + capture + .scoped_ref(&SessionEntryRef::parse(id).unwrap()) + .unwrap() +} + +#[test] +fn public_consumer_roundtrip_retains_body_original_source_order_origin_and_independent_policy() { + let entries = vec![ + history( + "recording", + Item::user_message("record-only"), + Some(HistoryInputMetadata::new(usage(true, false, false), None, None).unwrap()), + ), + history( + "memory", + Item::user_message("memory-only"), + Some(HistoryInputMetadata::new(usage(false, true, false), None, None).unwrap()), + ), + history( + "observation", + Item::user_message("observation-only"), + Some(HistoryInputMetadata::new(usage(false, false, true), None, None).unwrap()), + ), + ]; + let capture = SessionCapture::from_history_entries("synthetic-session", "selected", entries); + let mut dto = capture.to_dto(); + // The checked boundary also accepts entries retained from different original segments. + dto.entries[0].source.segment_id = "original-a".into(); + dto.entries[0].source.entry_index = 80; + dto.entries[1].source.segment_id = "original-b".into(); + dto.entries[1].source.entry_index = 3; + let capture = SessionCapture::try_from_dto(dto.clone(), CaptureLimits::default()).unwrap(); + let encoded = capture.encode(CaptureLimits::default()).unwrap(); + let restored = SessionCapture::decode(&encoded, CaptureLimits::default()).unwrap(); + assert_eq!(restored.to_dto(), dto); + assert_eq!(restored.capture_id(), capture.capture_id()); + for (id, allowed) in [ + ("Erecording", CapturePurpose::Recording), + ("Ememory", CapturePurpose::MemoryEvidence), + ("Eobservation", CapturePurpose::Observation), + ] { + let reference = reference(&capture, id); + assert_eq!( + restored.resolve_ref(&reference, allowed).unwrap(), + capture.resolve_ref(&reference, allowed).unwrap() + ); + for purpose in [ + CapturePurpose::Recording, + CapturePurpose::MemoryEvidence, + CapturePurpose::Observation, + ] { + assert_eq!( + restored.resolve_ref(&reference, purpose).is_ok(), + purpose == allowed + ); + } + } + assert_eq!(restored.overview().len(), 1); + assert_eq!(restored.search(&SearchOptions::default()).len(), 1); + assert!(restored.evidence_for("Erecording").is_none()); + let evidence = restored.evidence_for("Ememory").unwrap(); + assert_eq!(evidence.segment_id, "original-b"); + assert!(evidence.excerpt.contains("memory-only")); + let recorded = restored + .slice(CapturePurpose::Recording, None, 10, 16_000) + .unwrap(); + assert_eq!(recorded.slice().items.len(), 1); + assert_eq!(recorded.slice().items[0].text, "record-only"); + for (purpose, expected) in [ + (CapturePurpose::MemoryEvidence, "memory-only"), + (CapturePurpose::Observation, "observation-only"), + ] { + let slice = restored.slice(purpose, None, 10, 16_000).unwrap(); + assert_eq!(slice.slice().items.len(), 1); + assert_eq!(slice.slice().items[0].text, expected); + assert!( + restored + .read_raw( + &reference(&restored, "Erecording"), + purpose, + RawField::MessagePart(0), + 0, + 100 + ) + .is_err() + ); + } +} + +#[test] +fn common_data_drops_reasoning_raw_system_and_private_payload_even_when_unindexed() { + let payload = FeatureInputPayload::encode( + "capture-test", + 1, + &serde_json::json!({"secret": "PRIVATE-PAYLOAD"}), + ) + .unwrap(); + let public = PublicSystemItem { + kind: "notice".into(), + body: "public notice".into(), + presentation: PublicSystemPresentation::Ignored, + }; + let system = + HistoryInputMetadata::new(usage(true, true, true), Some(payload.clone()), Some(public)) + .unwrap(); + let private = HistoryInputMetadata::new(usage(true, true, true), Some(payload), None).unwrap(); + let capture = SessionCapture::from_history_entries( + "s", + "segment", + vec![ + history("raw", Item::system_message("RAW-SYSTEM-PROMPT"), None), + history("reasoning", Item::reasoning("HIDDEN-REASONING"), None), + history( + "private", + Item::system_message("PRIVATE-SYSTEM-BODY"), + Some(private), + ), + history( + "public", + Item::system_message("MODEL-SYSTEM-BODY"), + Some(system), + ), + history("ordinary", Item::user_message("ordinary"), None), + ], + ); + let encoded = String::from_utf8(capture.encode(CaptureLimits::default()).unwrap()).unwrap(); + let retained_debug = format!("{capture:?}"); + for secret in [ + "RAW-SYSTEM-PROMPT", + "HIDDEN-REASONING", + "PRIVATE-SYSTEM-BODY", + "MODEL-SYSTEM-BODY", + "PRIVATE-PAYLOAD", + ] { + assert!(!encoded.contains(secret)); + assert!(!retained_debug.contains(secret)); + } + assert_eq!(capture.to_dto().entries.len(), 2); + assert!(encoded.contains("public notice")); + // Filtered entries do not renumber the public source positions. + assert_eq!(capture.to_dto().entries[0].order, 3); + assert_eq!( + capture.source_index_for_ref(&SessionEntryRef::parse("Eordinary").unwrap()), + Some(4) + ); +} + +#[test] +fn protocol_projection_does_not_invent_recording_permission() { + let capture = SessionCapture::from_session_snapshot( + "s", + "seg", + SessionSnapshot { + pending_submissions: Default::default(), + entries: vec![SessionSnapshotEntry { + entry_id: "projected".into(), + timestamp: 0, + provenance: SessionEntryProvenance::HumanInput, + derived_from: vec!["source".into()], + use_policy: Some(protocol::SessionEntryUsePolicy { + observation_readable: true, + memory_evidence: false, + }), + data: SessionSnapshotEntryData::Message { + role: protocol::SessionMessageRole::User, + content: vec![protocol::SessionContentPart::Text { + text: "projected".into(), + }], + }, + }], + }, + ); + let reference = reference(&capture, "Eprojected"); + assert!(matches!( + capture.resolve_ref(&reference, CapturePurpose::Recording), + Err(CaptureError::PolicyDenied) + )); + assert!( + capture + .resolve_ref(&reference, CapturePurpose::Observation) + .is_ok() + ); + assert!(capture.evidence_for("Eprojected").is_none()); +} + +#[test] +fn malformed_dtos_and_oversized_representations_are_rejected() { + let capture = SessionCapture::new("s", "seg", vec![Item::user_message("body")]); + let original = serde_json::to_value(capture.to_dto()).unwrap(); + let mut invalids = Vec::new(); + let mut value = original.clone(); + value["version"] = 999.into(); + invalids.push(value); + let mut value = original.clone(); + value["unexpected"] = true.into(); + invalids.push(value); + let mut value = original.clone(); + value["entries"][0]["policy"]["recording_material"] = "include".into(); + invalids.push(value); + let mut value = original.clone(); + value["entries"][0]["policy"] + .as_object_mut() + .unwrap() + .remove("memory_evidence"); + invalids.push(value); + let mut value = original.clone(); + value["entries"][0]["body"] = serde_json::json!({"kind":"reasoning","text":"secret"}); + invalids.push(value); + let mut value = original.clone(); + value["entries"][0]["body"]["feature_payload"] = serde_json::json!({"private":true}); + invalids.push(value); + let mut value = original.clone(); + value["entries"][0]["body"]["role"] = "system".into(); + invalids.push(value); + let mut value = original.clone(); + value["entries"][0]["source"]["unknown"] = true.into(); + invalids.push(value); + let mut value = original.clone(); + value["entries"][0]["source"]["history_origin"] = + serde_json::json!({"kind":"legacy_unknown","unknown":true}); + invalids.push(value); + let mut value = original.clone(); + value["entries"][0]["entry_id"] = "bad/id".into(); + invalids.push(value); + let mut value = original.clone(); + value["entries"] + .as_array_mut() + .unwrap() + .push(original["entries"][0].clone()); + invalids.push(value); + for invalid in invalids { + let bytes = serde_json::to_vec(&invalid).unwrap(); + assert!( + SessionCapture::decode(&bytes, CaptureLimits::default()).is_err(), + "accepted {invalid}" + ); + } + let bytes = capture.encode(CaptureLimits::default()).unwrap(); + assert!( + SessionCapture::decode( + &bytes, + CaptureLimits { + max_bytes: bytes.len() - 1, + ..CaptureLimits::default() + } + ) + .is_err() + ); + assert!( + capture + .encode(CaptureLimits { + max_entries: 0, + ..CaptureLimits::default() + }) + .is_err() + ); + assert!( + capture + .encode(CaptureLimits { + max_text_bytes: 3, + ..CaptureLimits::default() + }) + .is_err() + ); + let mut reversed = capture.to_dto(); + let mut second = reversed.entries[0].clone(); + second.entry_id = "second".into(); + second.source.entry_index = 4; + second.order = 0; + reversed.entries[0].order = 1; + reversed.entries.push(second); + assert!(SessionCapture::try_from_dto(reversed, CaptureLimits::default()).is_err()); +} + +#[test] +fn raw_utf8_tool_input_and_result_are_lossless_without_preview_markers() { + let arguments = "\"quoted\"\\path\n日本語🙂".repeat(300); + let result = "result\n\t\\\"λ🙂".repeat(400); + let capture = SessionCapture::new( + "s", + "seg", + vec![ + Item::tool_call("call", "tool", arguments.clone()), + Item::tool_result_item("call", "summary", Some(result.clone()), false), + ], + ); + for (id, field, expected) in [ + ("E00000000", RawField::ToolArguments, arguments), + ("E00000001", RawField::ToolContent, result), + ] { + let reference = reference(&capture, id); + let mut offset = 0; + let mut actual = String::new(); + loop { + let part = capture + .read_raw(&reference, CapturePurpose::Observation, field, offset, 17) + .unwrap(); + assert!(part.text.len() <= 17); + assert_eq!(part.offset, offset); + actual.push_str(&part.text); + match part.next_offset { + Some(next) => { + assert!(next > offset); + offset = next; + } + None => break, + } + } + assert_eq!(actual, expected); + } + let unicode = SessionCapture::new("s", "seg", vec![Item::tool_call("call", "tool", "🙂")]); + let reference = reference(&unicode, "E00000000"); + assert!(matches!( + unicode.read_raw( + &reference, + CapturePurpose::Observation, + RawField::ToolArguments, + 1, + 8 + ), + Err(CaptureError::InvalidRange) + )); + assert!(matches!( + unicode.read_raw( + &reference, + CapturePurpose::Observation, + RawField::ToolArguments, + 0, + 3 + ), + Err(CaptureError::BudgetTooSmall) + )); + assert!(matches!( + unicode.read_raw( + &reference, + CapturePurpose::Observation, + RawField::ToolArguments, + 5, + 8 + ), + Err(CaptureError::InvalidRange) + )); +} + +#[test] +fn escaped_json_slice_budget_includes_metadata_and_bound_continuation() { + let body = "\"\\\n\t🙂日本語".repeat(500); + let capture = SessionCapture::new( + "synthetic-session", + "segment", + vec![Item::tool_call("call", "tool", body.clone())], + ); + let other_session = SessionCapture::new( + "other-session", + "segment", + vec![Item::tool_call("call", "tool", body.clone())], + ); + let changed = SessionCapture::new( + "synthetic-session", + "segment", + vec![Item::tool_call("call", "tool", "changed")], + ); + let mut cursor = None; + let mut actual = String::new(); + let budget = 1400; + loop { + let part = capture + .slice(CapturePurpose::Recording, cursor.as_ref(), 1, budget) + .unwrap(); + assert!(part.encoded_len() <= budget); + assert_eq!(part.encoded_len(), part.as_json().len()); + assert_eq!(serde_json::to_vec(part.slice()).unwrap(), part.as_json()); + assert_eq!(part.slice().items.len(), 1); + let item = &part.slice().items[0]; + assert_eq!(item.offset, actual.len()); + actual.push_str(&item.text); + cursor = part.slice().next_cursor.clone(); + if let Some(cursor) = &cursor { + assert!(matches!( + other_session.slice(CapturePurpose::Recording, Some(cursor), 1, budget), + Err(CaptureError::ForeignReference) + )); + assert!(matches!( + changed.slice(CapturePurpose::Recording, Some(cursor), 1, budget), + Err(CaptureError::ForeignReference) + )); + assert!(matches!( + capture.slice(CapturePurpose::Observation, Some(cursor), 1, budget), + Err(CaptureError::ForeignReference) + )); + } else { + break; + } + } + assert_eq!(actual, body); + assert!(matches!( + capture.slice(CapturePurpose::Recording, None, 1, 8), + Err(CaptureError::BudgetTooSmall) + )); + assert!(matches!( + capture.slice(CapturePurpose::Recording, None, 0, budget), + Err(CaptureError::BudgetTooSmall) + )); +} + +#[test] +fn snapshot_immutability_and_scoped_operations_reject_foreign_or_unknown_refs_and_bad_ranges() { + let mut source = vec![ + Item::user_message("before"), + Item::assistant_message("answer"), + ]; + let capture = SessionCapture::new("s", "seg", source.clone()); + let fixed_bytes = capture.encode(CaptureLimits::default()).unwrap(); + let first = reference(&capture, "E00000000"); + let second = reference(&capture, "E00000001"); + source.push(Item::user_message("appended")); + source[0] = Item::user_message("compacted"); + source.clear(); + assert_eq!( + capture.encode(CaptureLimits::default()).unwrap(), + fixed_bytes + ); + assert_eq!( + capture.clone().encode(CaptureLimits::default()).unwrap(), + fixed_bytes + ); + let foreign = SessionCapture::new( + "different", + "seg", + vec![ + Item::user_message("before"), + Item::assistant_message("answer"), + ], + ); + assert!(matches!( + foreign.resolve_ref(&first, CapturePurpose::Observation), + Err(CaptureError::ForeignReference) + )); + assert!(matches!( + capture.checked_range(&second, &first, CapturePurpose::Observation), + Err(CaptureError::InvalidRange) + )); + assert_eq!( + capture + .checked_range(&first, &second, CapturePurpose::Observation) + .unwrap() + .len(), + 2 + ); + let unknown = SessionEntryRef::parse("E99999999").unwrap(); + assert_eq!(capture.source_index_for_ref(&unknown), None); + assert!(matches!( + capture.scoped_ref(&unknown), + Err(CaptureError::UnknownReference) + )); + assert!( + capture + .search_checked(&SearchOptions { + from: Some(unknown.clone()), + ..Default::default() + }) + .is_err() + ); + assert!( + capture + .search(&SearchOptions { + through: Some(unknown), + ..Default::default() + }) + .is_empty() + ); + assert!( + capture + .read_checked(ReadSelector::EntryRange([1, 0]), ReadOptions::default()) + .is_err() + ); + assert!( + capture + .read_checked(ReadSelector::EntryRange([0, 99]), ReadOptions::default()) + .is_err() + ); +} + +#[test] +fn slices_preserve_empty_and_long_result_fields_under_item_and_metadata_budgets() { + let summary = "summary\n\"🙂".repeat(100); + let content = "content\t\\日本語".repeat(200); + let capture = SessionCapture::new( + "s", + "seg", + vec![ + Item::tool_result_item("first", summary.clone(), Some(content.clone()), true), + Item::tool_result_item("empty", "", Some(String::new()), false), + ], + ); + let mut cursor = None; + let mut fields: Vec<(String, RawField, String)> = Vec::new(); + loop { + let part = capture + .slice(CapturePurpose::Recording, cursor.as_ref(), 2, 1500) + .unwrap(); + assert!(part.slice().items.len() <= 2); + assert!(serde_json::to_vec(part.slice()).unwrap().len() <= 1500); + for item in &part.slice().items { + let id = item.reference.entry_ref.to_string(); + if let Some((last_id, last_field, text)) = fields.last_mut() { + if *last_id == id && *last_field == item.field { + assert_eq!(text.len(), item.offset); + text.push_str(&item.text); + continue; + } + } + assert_eq!(item.offset, 0); + fields.push((id, item.field, item.text.clone())); + } + cursor = part.slice().next_cursor.clone(); + if cursor.is_none() { + break; + } + } + assert_eq!( + fields, + vec![ + ("E00000000".into(), RawField::ToolSummary, summary), + ("E00000000".into(), RawField::ToolContent, content), + ("E00000001".into(), RawField::ToolSummary, String::new()), + ("E00000001".into(), RawField::ToolContent, String::new()), + ] + ); + let metadata_heavy = SessionCapture::new( + "s", + "seg", + vec![Item::tool_call("call", "name".repeat(200), "x")], + ); + assert!(matches!( + metadata_heavy.slice(CapturePurpose::Recording, None, 1, 700), + Err(CaptureError::BudgetTooSmall) + )); +} diff --git a/crates/worker/src/feature/builtin/flow_transition.rs b/crates/worker/src/feature/builtin/flow_transition.rs index e3a9fc196..4c9cd98e3 100644 --- a/crates/worker/src/feature/builtin/flow_transition.rs +++ b/crates/worker/src/feature/builtin/flow_transition.rs @@ -26,7 +26,7 @@ use flow::{ use manifest::{Scope, WorkerManifest}; use schemars::JsonSchema; use serde::Deserialize; -use session_store::{SegmentId, SessionId, Store, collect_state}; +use session_store::{SegmentId, SessionId, Store}; use uuid::Uuid; use crate::feature::builtin::session_explore::{SessionExploreFeature, SessionExploreState}; @@ -469,10 +469,15 @@ where .store .read_all(self.session_id, self.segment_id) .map_err(|error| format!("read committed parent session: {error}"))?; - let restored = collect_state(&entries); - Ok(SessionCapture::new( + let history = crate::session_history::restore_history_entries( + self.session_id, + self.segment_id, + &entries, + )?; + Ok(SessionCapture::from_history_entries( + self.session_id.to_string(), self.segment_id.to_string(), - restored.history, + history, )) } } diff --git a/crates/worker/src/feature/builtin/memory_lifecycle.rs b/crates/worker/src/feature/builtin/memory_lifecycle.rs index 94a2a2481..7db770eb0 100644 --- a/crates/worker/src/feature/builtin/memory_lifecycle.rs +++ b/crates/worker/src/feature/builtin/memory_lifecycle.rs @@ -457,6 +457,7 @@ impl MemoryLifecycleTask { return Ok(()); } let view = SessionCapture::from_history_entries( + capture.session_id.clone(), capture.segment_id.clone(), capture.history[history_start..history_end].to_vec(), ); @@ -1556,11 +1557,14 @@ permission = "write" #[tokio::test] async fn run_committed_background_task_stages_non_empty_extraction_and_commits_pointer() { let source = capture(2, 250); - let entry_ref = - SessionCapture::from_history_entries(source.segment_id.clone(), source.history.clone()) - .overview()[0] - .id - .to_string(); + let entry_ref = SessionCapture::from_history_entries( + source.session_id.clone(), + source.segment_id.clone(), + source.history.clone(), + ) + .overview()[0] + .id + .to_string(); let client = ScriptClient::new(vec![ stage_candidate_events("stage-1", &entry_ref), finish_events("finish-1", 1), @@ -1609,11 +1613,14 @@ permission = "write" #[tokio::test] async fn subjektiv_extraction_stages_once_and_commits_its_independent_pointer() { let source = capture(2, 250); - let entry_ref = - SessionCapture::from_history_entries(source.segment_id.clone(), source.history.clone()) - .overview()[0] - .id - .to_string(); + let entry_ref = SessionCapture::from_history_entries( + source.session_id.clone(), + source.segment_id.clone(), + source.history.clone(), + ) + .overview()[0] + .id + .to_string(); let client = ScriptClient::new(vec![ stage_candidate_events("stage-1", &entry_ref), finish_events("finish-1", 1), @@ -1715,11 +1722,14 @@ permission = "write" #[tokio::test] async fn subjektiv_candidate_remains_staged_when_later_extraction_fails_without_pointer() { let source = capture(2, 250); - let entry_ref = - SessionCapture::from_history_entries(source.segment_id.clone(), source.history.clone()) - .overview()[0] - .id - .to_string(); + let entry_ref = SessionCapture::from_history_entries( + source.session_id.clone(), + source.segment_id.clone(), + source.history.clone(), + ) + .overview()[0] + .id + .to_string(); let client = ScriptClient::new(vec![stage_candidate_events("stage-1", &entry_ref)]); let extension_writes = Arc::new(Mutex::new(Vec::new())); let (event_tx, _) = broadcast::channel(32); diff --git a/crates/worker/src/feature/builtin/memory_staging_output.rs b/crates/worker/src/feature/builtin/memory_staging_output.rs index ad5de09df..dd830349d 100644 --- a/crates/worker/src/feature/builtin/memory_staging_output.rs +++ b/crates/worker/src/feature/builtin/memory_staging_output.rs @@ -510,7 +510,11 @@ mod tests { fn state() -> MemoryStagingOutputState { MemoryStagingOutputState::new( - SessionCapture::new("segment-1", vec![Item::user_message("durable decision")]), + SessionCapture::new( + "session", + "segment-1", + vec![Item::user_message("durable decision")], + ), crate::worker::marker_workspace_client(None, "test-backend"), SourceRef { segment_id: "segment-1".to_string(), @@ -537,6 +541,7 @@ mod tests { #[test] fn render_input_uses_session_entry_refs_and_new_tool_names() { let view = SessionCapture::new( + "session", "segment-1", vec![ Item::user_message("preference"), @@ -586,7 +591,11 @@ mod tests { async fn subjektiv_destination_sends_committed_session_without_subject_scope() { let client = Arc::new(RecordingSubjektivClient::default()); let state = MemoryStagingOutputState::new_subjektiv( - SessionCapture::new("segment-1", vec![Item::user_message("durable decision")]), + SessionCapture::new( + "session", + "segment-1", + vec![Item::user_message("durable decision")], + ), { let mut host = crate::subjektiv::test_connection(client.clone()); host.context.session_id = "session-committed".into(); diff --git a/crates/worker/src/feature/builtin/session_explore.rs b/crates/worker/src/feature/builtin/session_explore.rs index aacc16081..1400f51b4 100644 --- a/crates/worker/src/feature/builtin/session_explore.rs +++ b/crates/worker/src/feature/builtin/session_explore.rs @@ -17,14 +17,14 @@ use crate::feature::{ ToolDeclaration, }; use crate::session_capture::{ - ReadDetail, ReadOptions, ReadSelector, ReferenceKind, SearchOptions, SessionCapture, - SessionEntryRef, ToolPart, + CapturePurpose, RawField, ReadDetail, ReadOptions, ReadSelector, ReferenceKind, SearchOptions, + SessionCapture, SessionEntryRef, ToolPart, }; const SHOW_OVERVIEW_DESCRIPTION: &str = "Show a sparse, bounded index of real user and assistant session entries."; const SEARCH_ENTRIES_DESCRIPTION: &str = "Search or compactly list a bounded range of committed session entries. Reasoning entries are never exposed."; -const READ_ENTRY_DESCRIPTION: &str = "Read one committed session entry by stable SessionEntryRef. Compact mode is the default; full mode includes bounded tool input or output."; +const READ_ENTRY_DESCRIPTION: &str = "Read one committed session entry by stable SessionEntryRef. Compact mode is the default; full mode is a bounded rendering, not raw source. Raw mode requires raw_field and supports UTF-8 byte_offset/next_offset without omission; use capture_id to bind continuation to this exact capture."; const DEFAULT_PAGE_LIMIT: usize = 20; const MAX_PAGE_LIMIT: usize = 100; const DEFAULT_READ_ITEMS: usize = 10; @@ -173,6 +173,38 @@ struct ReadEntryParams { mode: String, #[serde(default)] max_items: Option, + #[serde(default)] + raw_field: Option, + #[serde(default)] + part_index: Option, + #[serde(default)] + byte_offset: Option, + #[serde(default)] + max_bytes: Option, + #[serde(default)] + capture_id: Option, +} + +#[derive(Debug, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +enum RawEntryField { + MessagePart, + PublicSystem, + ToolArguments, + ToolSummary, + ToolContent, +} + +impl RawEntryField { + fn into_common(self, part_index: usize) -> RawField { + match self { + Self::MessagePart => RawField::MessagePart(part_index), + Self::PublicSystem => RawField::PublicSystem, + Self::ToolArguments => RawField::ToolArguments, + Self::ToolSummary => RawField::ToolSummary, + Self::ToolContent => RawField::ToolContent, + } + } } fn default_read_mode() -> String { @@ -243,25 +275,20 @@ impl Tool for SearchEntriesTool { let from = params.from.as_deref().map(parse_entry_ref).transpose()?; let through = params.through.as_deref().map(parse_entry_ref).transpose()?; let view = self.state.view(); - if let (Some(from), Some(through)) = (&from, &through) { - if view.source_index_for_ref(from) > view.source_index_for_ref(through) { - return Err(ToolError::InvalidArgument( - "SearchEntries from must not be after through".to_string(), - )); - } - } let limit = bounded_limit(params.limit, DEFAULT_PAGE_LIMIT, MAX_PAGE_LIMIT); - let hits = view.search(&SearchOptions { - query: params.query, - kind, - tool_part, - tool_name: params.tool_name, - limit: Some(limit), - min_entry_index: None, - from, - through, - offset: params.offset, - }); + let hits = view + .search_checked(&SearchOptions { + query: params.query, + kind, + tool_part, + tool_name: params.tool_name, + limit: Some(limit), + min_entry_index: None, + from, + through, + offset: params.offset, + }) + .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; let entries = hits .iter() .map(|hit| { @@ -272,6 +299,7 @@ impl Tool for SearchEntriesTool { "tool_name": hit.tool_name, "label": hit.label, "text": hit.summary, + "text_kind": "preview", }) }) .collect::>(); @@ -299,12 +327,62 @@ impl Tool for ReadEntryTool { ) -> Result { let params: ReadEntryParams = parse_input("ReadEntry", input_json)?; let entry_ref = parse_entry_ref(¶ms.entry_ref)?; + let view = self.state.view(); + if params + .capture_id + .as_deref() + .is_some_and(|id| id != view.capture_id()) + { + return Err(ToolError::InvalidArgument( + "capture identity does not match".into(), + )); + } + if params.mode == "raw" { + let field = params + .raw_field + .ok_or_else(|| ToolError::InvalidArgument("raw mode requires raw_field".into()))?; + if params.part_index.is_some() && !matches!(field, RawEntryField::MessagePart) { + return Err(ToolError::InvalidArgument( + "part_index is only valid for message_part".into(), + )); + } + let reference = view + .scoped_ref(&entry_ref) + .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; + let raw = view + .read_raw( + &reference, + CapturePurpose::Observation, + field.into_common(params.part_index.unwrap_or(0)), + params.byte_offset.unwrap_or(0), + params + .max_bytes + .unwrap_or(MAX_READ_BYTES) + .clamp(1, MAX_READ_BYTES), + ) + .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; + return json_output( + "Read raw session source bytes.".into(), + serde_json::json!({ + "text_kind": "raw", "raw": raw, + }), + ); + } + if params.raw_field.is_some() + || params.part_index.is_some() + || params.byte_offset.is_some() + || params.max_bytes.is_some() + { + return Err(ToolError::InvalidArgument( + "raw fields require raw mode".into(), + )); + } let detail = match params.mode.as_str() { "compact" => ReadDetail::Compact, "full" => ReadDetail::Full, other => { return Err(ToolError::InvalidArgument(format!( - "invalid ReadEntry mode {other:?}; expected compact or full" + "invalid ReadEntry mode {other:?}; expected compact, full, or raw" ))); } }; @@ -347,6 +425,7 @@ impl Tool for ReadEntryTool { serde_json::json!({ "entries": entries, "truncated": read.truncated, + "text_kind": "rendered", }), ) } @@ -408,6 +487,7 @@ mod tests { fn state() -> SessionExploreState { SessionExploreState::new(SessionCapture::new( + "session", "segment-1", vec![ Item::user_message("first"), @@ -479,4 +559,64 @@ mod tests { .unwrap(); assert!(output.content.unwrap().contains("second")); } + #[tokio::test] + async fn search_rejects_unknown_or_reversed_capture_endpoints() { + let search = search_entries_definition(state())().1; + for input in [ + r#"{"from":"E99999999"}"#, + r#"{"through":"E99999999"}"#, + r#"{"from":"E00000003","through":"E00000000"}"#, + ] { + assert!( + search + .execute(input, agen::tool::ToolExecutionContext::direct()) + .await + .is_err() + ); + } + } + + #[tokio::test] + async fn raw_tool_arguments_reassemble_without_preview_or_utf8_loss() { + let arguments = "日本語\n\"\\".repeat(100); + let state = SessionExploreState::new(SessionCapture::new( + "session", + "segment", + vec![Item::tool_call("call", "Read", arguments.clone())], + )); + let capture_id = state.view().capture_id().to_owned(); + let tool = read_entry_definition(state)().1; + let mut offset = 0; + let mut assembled = String::new(); + loop { + let input = serde_json::json!({ + "entry_ref": "E00000000", "mode": "raw", "raw_field": "tool_arguments", + "byte_offset": offset, "max_bytes": 17, "capture_id": capture_id, + }); + let output = tool + .execute( + &input.to_string(), + agen::tool::ToolExecutionContext::direct(), + ) + .await + .unwrap(); + let value: serde_json::Value = serde_json::from_str(&output.content.unwrap()).unwrap(); + assert_eq!(value["text_kind"], "raw"); + let raw = &value["raw"]; + assert_eq!(raw["offset"], offset); + let text = raw["text"].as_str().unwrap(); + assert!(text.len() <= 17); + assembled.push_str(text); + match raw["next_offset"].as_u64() { + Some(next) => { + assert!(next > offset as u64); + offset = next as usize; + } + None => break, + } + } + assert_eq!(assembled, arguments); + assert!(tool.execute(r#"{"entry_ref":"E00000000","mode":"raw","raw_field":"tool_arguments","capture_id":"other"}"#, + agen::tool::ToolExecutionContext::direct()).await.is_err()); + } } diff --git a/crates/worker/src/feature/builtin/subjektiv_memory.rs b/crates/worker/src/feature/builtin/subjektiv_memory.rs index cd82ca390..82ce970f9 100644 --- a/crates/worker/src/feature/builtin/subjektiv_memory.rs +++ b/crates/worker/src/feature/builtin/subjektiv_memory.rs @@ -39,13 +39,66 @@ use crate::hook::{ HookErrorCategory, HookExecutionPolicy, HookFailurePolicy, HookPreRequestAction, PreLlmRequest, PreRequestContext, RunCommitted, RunCommittedContext, }; -use crate::session_capture::{SessionCapture, SessionEntryEvidence}; +use crate::session_capture::{CaptureBody, SessionCapture, SessionEntryEvidence}; use crate::subjektiv::SubjektivHostConnection; #[cfg(test)] use crate::worker::WorkspaceClient; #[cfg(test)] use server_api::SubjektivMemoryBackendRequest; +// Pending-commit receipt interpretation belongs to Memory, not the shared capture. +struct CommittedMemoryCall { + call_id: String, + name: String, + arguments: String, + evidence: SessionEntryEvidence, +} + +fn committed_pending_tool_calls(view: &SessionCapture, names: &[&str]) -> Vec { + let pending: HashSet<_> = view + .entries() + .filter_map(|entry| { + let CaptureBody::ToolResult { + call_id, + content: Some(content), + is_error: false, + .. + } = &entry.body + else { + return None; + }; + (serde_json::from_str::(content) + .ok() + .and_then(|value| value.get("status").cloned()) + .as_ref() + .and_then(serde_json::Value::as_str) + == Some("pending_commit")) + .then_some(call_id.as_str()) + }) + .collect(); + view.entries() + .filter_map(|entry| { + let CaptureBody::ToolCall { + call_id, + name, + arguments, + } = &entry.body + else { + return None; + }; + if !names.contains(&name.as_str()) || !pending.contains(call_id.as_str()) { + return None; + } + Some(CommittedMemoryCall { + call_id: call_id.clone(), + name: name.clone(), + arguments: arguments.clone(), + evidence: view.evidence_for_tool_call(call_id)?, + }) + }) + .collect() +} + const QUERY_TOOL: &str = "SubjektivMemoryQuery"; const READ_TOOL: &str = "SubjektivMemoryRead"; const LIST_CHANGES_TOOL: &str = "SubjektivMemoryListChanges"; @@ -518,14 +571,18 @@ impl SubjektivMemoryState { .capture() .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; let session_id = capture.session_id.clone(); - let view = SessionCapture::from_history_entries(capture.segment_id, capture.history); - let committed = view - .committed_pending_tool_calls(&[REMEMBER_TOOL, PROPOSE_CHANGE_TOOL]) - .into_iter() - .any(|call| { - receipt_id(&session_id, &call.call_id) - .is_ok_and(|committed_receipt| committed_receipt == expected_receipt_id) - }); + let view = SessionCapture::from_history_entries( + capture.session_id.clone(), + capture.segment_id, + capture.history, + ); + let committed = + committed_pending_tool_calls(&view, &[REMEMBER_TOOL, PROPOSE_CHANGE_TOOL]) + .into_iter() + .any(|call| { + receipt_id(&session_id, &call.call_id) + .is_ok_and(|committed_receipt| committed_receipt == expected_receipt_id) + }); if committed { status.status = SubjektivMemoryReceiptStatus::PendingCommit; } @@ -623,10 +680,13 @@ fn replay_pending_committed_memory_with_policy( .capture .capture() .map_err(|error| HookError::new(HookErrorCategory::Dependency, error.to_string()))?; - let view = - SessionCapture::from_history_entries(capture.segment_id.clone(), capture.history.clone()); + let view = SessionCapture::from_history_entries( + capture.session_id.clone(), + capture.segment_id.clone(), + capture.history.clone(), + ); let mut failures = Vec::new(); - for call in view.committed_pending_tool_calls(&[REMEMBER_TOOL, PROPOSE_CHANGE_TOOL]) { + for call in committed_pending_tool_calls(&view, &[REMEMBER_TOOL, PROPOSE_CHANGE_TOOL]) { let receipt_id = match receipt_id(&capture.session_id, &call.call_id) { Ok(value) => value, Err(error) => { @@ -755,8 +815,11 @@ fn resolve_entry_refs( capture: &crate::feature::session::CommittedSessionCapture, entry_refs: &[String], ) -> Result, ToolError> { - let view = - SessionCapture::from_history_entries(capture.segment_id.clone(), capture.history.clone()); + let view = SessionCapture::from_history_entries( + capture.session_id.clone(), + capture.segment_id.clone(), + capture.history.clone(), + ); let mut seen = HashSet::new(); entry_refs .iter() diff --git a/crates/worker/src/feature/builtin/worker_observation.rs b/crates/worker/src/feature/builtin/worker_observation.rs index 9b6d5488c..d1c4d0cce 100644 --- a/crates/worker/src/feature/builtin/worker_observation.rs +++ b/crates/worker/src/feature/builtin/worker_observation.rs @@ -570,15 +570,8 @@ impl Tool for SearchSessionEntriesTool { let view = latest_view(&*self.provider, ¶ms.subject).await?; let from = params.from.as_deref().map(parse_entry_ref).transpose()?; let through = params.through.as_deref().map(parse_entry_ref).transpose()?; - if let (Some(from), Some(through)) = (&from, &through) { - if view.source_index_for_ref(from) > view.source_index_for_ref(through) { - return Err(ToolError::InvalidArgument( - "SearchSessionEntries from must not be after through".to_string(), - )); - } - } let entries = view - .search(&SearchOptions { + .search_checked(&SearchOptions { query: params.query, kind: params.kind.as_deref().map(parse_kind).transpose()?, tool_part: params @@ -593,6 +586,7 @@ impl Tool for SearchSessionEntriesTool { through, offset: params.offset, }) + .map_err(|error| ToolError::InvalidArgument(error.to_string()))? .into_iter() .map(|entry| { serde_json::json!({ @@ -604,6 +598,7 @@ impl Tool for SearchSessionEntriesTool { "tool_name": entry.tool_name, "label": entry.label, "text": entry.summary, + "text_kind": "preview", }) }) .collect::>(); @@ -687,7 +682,10 @@ async fn latest_view( .capture_worker_session(subject) .await .map_err(tool_error)?; + // Observation providers expose a stable subject scope rather than a raw + // Session handle. This scope is never used to select recording material. Ok(SessionCapture::from_session_snapshot( + format!("observation:{}", capture.segment_id), capture.segment_id, capture.session, )) diff --git a/crates/worker/src/session_capture.rs b/crates/worker/src/session_capture.rs index c255d7dad..979c8d44a 100644 --- a/crates/worker/src/session_capture.rs +++ b/crates/worker/src/session_capture.rs @@ -1,881 +1,14 @@ -//! Exploration domain for immutable Session captures, with no Workspace/Memory mutation capability. -//! -//! A capture is an ordered projection of committed entries; exploration excludes reasoning and -//! system prompts, including Host compaction summaries and other derived context. Full inherited -//! summary input belongs to compaction, not public observation or Memory evidence. Overview, -//! search, bounded reads, and evidence conversion share `SessionEntryRef`. -//! References identify entries in the selected Session and cannot be reused for another subject. -//! Appends preserve existing entry identities. Sparse real user/assistant anchors and intervening -//! entry counts index long histories. Search with an empty filter also provides bounded range lists. -//! -//! Host-fixed snapshot exploration, Memory candidate evidence validation, and current Worker -//! observation share only this domain; they do not enable or modify each other's Features. -//! The provider owns authorization to recapture a current Worker. A capture itself proves neither -//! completion nor approval and grants no business-mutation authority. +//! Host consumers share the pure immutable Session capture implementation. +//! Live authorization, append fences, and recapture remain with the calling Host. -use std::collections::HashSet; -use std::sync::Arc; - -use crate::session_history::{SessionHistoryMetadata, WorkerHistoryProvenance}; -use agen::{HistoryEntry, Item, Role}; -use protocol::{ - SessionContentPart, SessionEntryProvenance, SessionMessageRole, SessionSnapshot, - SessionSnapshotEntryData, -}; -use serde::{Deserialize, Serialize}; - -const DEFAULT_SEARCH_LIMIT: usize = 20; -const MAX_SEARCH_LIMIT: usize = 50; -const DEFAULT_READ_MAX_ITEMS: usize = 40; -const MAX_READ_MAX_ITEMS: usize = 80; -const DEFAULT_READ_MAX_BYTES: usize = 32 * 1024; -const OVERVIEW_ANCHOR_STRIDE: usize = 8; - -/// Append-stable exploration reference derived from an entry identity in the selected Session. -/// Shared by overview, search, read, and evidence conversion; it does not identify other Sessions' entries. -#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] -#[serde(transparent)] -pub struct SessionEntryRef(String); - -impl SessionEntryRef { - pub fn from_history_entry_id(entry_id: &crate::SessionHistoryEntryId) -> Self { - Self(format!("E{}", entry_id.0)) - } - - pub fn parse(value: &str) -> Option { - let suffix = value.strip_prefix('E')?; - if suffix.is_empty() - || suffix.len() > 64 - || !suffix - .bytes() - .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-') - { - return None; - } - Some(Self(value.to_string())) - } - - pub fn as_str(&self) -> &str { - &self.0 - } - - pub(crate) fn source_index(&self) -> Option { - self.0.strip_prefix('E')?.parse().ok() - } -} - -impl std::fmt::Display for SessionEntryRef { - fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter.write_str(&self.0) - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum ReferenceKind { - User, - Assistant, - System, - Tool, -} - -impl ReferenceKind { - pub(crate) fn as_str(self) -> &'static str { - match self { - Self::User => "user", - Self::Assistant => "assistant", - Self::System => "system", - Self::Tool => "tool", - } - } - - pub(crate) fn parse(value: &str) -> Option { - match value { - "user" => Some(Self::User), - "assistant" | "agent" => Some(Self::Assistant), - "system" => Some(Self::System), - "tool" => Some(Self::Tool), - _ => None, - } - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum ToolPart { - Input, - Output, - Both, -} - -impl ToolPart { - pub(crate) fn parse(value: &str) -> Option { - match value { - "input" => Some(Self::Input), - "output" => Some(Self::Output), - "both" => Some(Self::Both), - _ => None, - } - } - - fn matches(self, actual: ToolPart) -> bool { - matches!(self, ToolPart::Both) || self == actual - } -} - -#[derive(Debug, Clone)] -pub(crate) struct OverviewItem { - pub id: SessionEntryRef, - pub origin: SessionEntryProvenance, - pub entry_range: [u64; 2], - pub kind: ReferenceKind, - pub label: String, - pub text: String, - pub intervening_entries: usize, -} - -#[derive(Debug, Clone)] -pub(crate) struct ReferenceEntry { - use_policy: Option, - pub id: SessionEntryRef, - pub origin: SessionEntryProvenance, - pub entry_range: [u64; 2], - pub kind: ReferenceKind, - pub tool_part: Option, - pub tool_name: Option, - pub label: String, - pub summary: String, - search_text: String, -} - -#[derive(Debug, Clone, Default)] -pub(crate) struct SearchOptions { - pub query: String, - pub kind: Option, - pub tool_part: Option, - pub tool_name: Option, - pub limit: Option, - pub min_entry_index: Option, - pub from: Option, - pub through: Option, - pub offset: usize, -} - -#[derive(Debug, Clone)] -pub(crate) struct SearchHit { - pub id: SessionEntryRef, - pub origin: SessionEntryProvenance, - pub kind: ReferenceKind, - pub tool_part: Option, - pub tool_name: Option, - pub entry_range: [u64; 2], - pub label: String, - pub summary: String, -} - -#[derive(Debug, Clone, Copy)] -pub(crate) enum ReadSelector<'a> { - Id(&'a str), - #[cfg(test)] - EntryRange([u64; 2]), -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum ReadDetail { - Compact, - Full, -} - -#[derive(Debug, Clone)] -pub(crate) struct ReadOptions { - pub include_tools: bool, - pub tool_part: ToolPart, - pub detail: ReadDetail, - pub max_items: usize, - pub max_bytes: usize, -} - -impl Default for ReadOptions { - fn default() -> Self { - Self { - include_tools: true, - tool_part: ToolPart::Both, - detail: ReadDetail::Compact, - max_items: DEFAULT_READ_MAX_ITEMS, - max_bytes: DEFAULT_READ_MAX_BYTES, - } - } -} - -#[derive(Debug, Clone)] -pub(crate) struct ReadEntry { - pub id: SessionEntryRef, - pub origin: SessionEntryProvenance, - pub kind: ReferenceKind, - pub tool_part: Option, - pub tool_name: Option, - pub entry_range: [u64; 2], - pub label: String, - pub text: String, -} - -#[derive(Debug, Clone)] -pub(crate) struct ReadResult { - pub entries: Vec, - pub truncated: bool, -} - -#[derive(Debug, Clone)] -pub(crate) struct CommittedToolCall { - pub call_id: String, - pub name: String, - pub arguments: String, - pub evidence: SessionEntryEvidence, -} - -#[derive(Debug, Clone)] -pub(crate) struct SessionEntryEvidence { - pub segment_id: String, - pub entry_ref: SessionEntryRef, - pub origin: SessionEntryProvenance, - pub entry_range: [u64; 2], - pub kind: ReferenceKind, - pub tool_part: Option, - pub label: String, - pub summary: String, - pub excerpt: String, -} - -#[derive(Debug, Clone)] -struct CapturedHistoryEntry { - use_policy: Option, - item: Item, - entry_id: session_store::LoggedSessionHistoryEntryId, - origin: SessionEntryProvenance, -} - -#[derive(Debug, Clone)] -pub(crate) struct SessionCapture { - segment_id: String, - entries: Arc>, - overview: Vec, - index: Vec, -} - -impl SessionCapture { - pub(crate) fn from_session_snapshot( - segment_id: impl Into, - snapshot: SessionSnapshot, - ) -> Self { - let entries = snapshot - .entries - .into_iter() - .filter_map(|entry| { - let item = match entry.data { - SessionSnapshotEntryData::UserInput { segments } => { - Item::user_message(protocol::Segment::flatten_to_text(&segments)) - } - SessionSnapshotEntryData::Message { role, content } => { - let role = match role { - SessionMessageRole::User => Role::User, - SessionMessageRole::Assistant => Role::Assistant, - }; - Item::Message { - id: None, - role, - content: content - .into_iter() - .map(|part| match part { - SessionContentPart::Text { text } => { - agen::ContentPart::Text { text } - } - SessionContentPart::Refusal { refusal } => { - agen::ContentPart::Refusal { refusal } - } - }) - .collect(), - status: None, - } - } - SessionSnapshotEntryData::ToolCall { - call_id, - name, - arguments, - } => Item::tool_call(call_id, name, arguments), - SessionSnapshotEntryData::ToolResult { - call_id, - summary, - content, - is_error, - attachments: _, - } => Item::tool_result_item(call_id, summary, content, is_error), - SessionSnapshotEntryData::SystemItem { content, .. } => { - entry.use_policy?; - Item::system_message(content) - } - // Controller errors are not model-visible evidence. - SessionSnapshotEntryData::Invoke { .. } - | SessionSnapshotEntryData::Usage { .. } - | SessionSnapshotEntryData::RunStats { .. } - | SessionSnapshotEntryData::RunYielded { .. } - | SessionSnapshotEntryData::RunResumed { .. } - | SessionSnapshotEntryData::RunCancelled - | SessionSnapshotEntryData::RunError { .. } => return None, - }; - Some(CapturedHistoryEntry { - use_policy: entry.use_policy, - item, - entry_id: session_store::LoggedSessionHistoryEntryId(entry.entry_id), - origin: entry.provenance, - }) - }) - .collect(); - Self::from_captured_entries(segment_id, entries) - } - - pub(crate) fn new(segment_id: impl Into, items: Vec) -> Self { - let entries = items - .into_iter() - .enumerate() - .map(|(index, item)| CapturedHistoryEntry { - use_policy: None, - item, - entry_id: session_store::LoggedSessionHistoryEntryId(format!("{index:08}")), - origin: SessionEntryProvenance::LegacyUnknown, - }) - .collect(); - Self::from_captured_entries(segment_id, entries) - } - - pub(crate) fn from_history_entries( - segment_id: impl Into, - entries: Vec>, - ) -> Self { - let entries = entries - .into_iter() - .map(|entry| { - let (item, use_policy) = if let Some(input) = &entry.annotation.input_metadata { - let is_system = matches!( - entry.item, - Item::Message { - role: Role::System, - .. - } - ); - if is_system { - ( - Item::system_message(input.public_projection().body), - Some(input.public_system_use_policy()), - ) - } else { - (entry.item, Some(input.public_use_policy())) - } - } else { - (entry.item, None) - }; - CapturedHistoryEntry { - item, - use_policy, - entry_id: entry.annotation.entry_id, - origin: public_provenance(&entry.annotation.origin), - } - }) - .collect(); - Self::from_captured_entries(segment_id, entries) - } - - fn from_captured_entries( - segment_id: impl Into, - entries: Vec, - ) -> Self { - let segment_id = segment_id.into(); - let mut overview = Vec::new(); - let mut index = Vec::new(); - - for (idx, entry) in entries.iter().enumerate() { - let item = &entry.item; - let entry_range = [idx as u64, idx as u64]; - match item { - Item::Message { role, content, .. } => { - let Some(kind) = message_reference_kind(&entry.origin, role, entry.use_policy) - else { - continue; - }; - let text = content - .iter() - .map(|p| p.as_text()) - .collect::>() - .join(""); - let label = format!("{} message", kind.as_str()); - let summary = truncate_chars(&text, 240); - let id = SessionEntryRef::from_history_entry_id(&entry.entry_id); - index.push(ReferenceEntry { - use_policy: entry.use_policy, - id: id.clone(), - origin: entry.origin.clone(), - entry_range, - kind, - tool_part: None, - tool_name: None, - label: label.clone(), - summary: summary.clone(), - search_text: text.clone(), - }); - if entry.use_policy.is_none_or(|p| p.observation_readable) - && matches!(kind, ReferenceKind::User | ReferenceKind::Assistant) - { - overview.push(OverviewItem { - id: id.clone(), - origin: entry.origin.clone(), - entry_range, - kind, - label, - text, - intervening_entries: 0, - }); - } - } - Item::ToolCall { - name, arguments, .. - } => { - let text = format!("{name}\n{arguments}"); - index.push(ReferenceEntry { - use_policy: entry.use_policy, - id: SessionEntryRef::from_history_entry_id(&entry.entry_id), - origin: entry.origin.clone(), - entry_range, - kind: ReferenceKind::Tool, - tool_part: Some(ToolPart::Input), - tool_name: Some(name.clone()), - label: format!("tool input: {name}"), - summary: format!("Tool call {name}"), - search_text: text, - }); - } - Item::ToolResult { - summary, - content, - attachments, - .. - } => { - let attachment_marker = if attachments.is_empty() { - String::new() - } else { - format!("\n[{} image attachment(s)]", attachments.len()) - }; - let text = format!( - "{summary}\n{}{attachment_marker}", - content.as_deref().unwrap_or_default(), - ); - index.push(ReferenceEntry { - use_policy: entry.use_policy, - id: SessionEntryRef::from_history_entry_id(&entry.entry_id), - origin: entry.origin.clone(), - entry_range, - kind: ReferenceKind::Tool, - tool_part: Some(ToolPart::Output), - tool_name: None, - label: "tool output".to_string(), - summary: truncate_chars(summary, 240), - search_text: text, - }); - } - Item::AssistantResponseBoundary { .. } | Item::Reasoning { .. } => {} - } - } - - if overview.len() > 2 { - let last = overview.len() - 1; - overview = overview - .into_iter() - .enumerate() - .filter_map(|(index, entry)| { - (index == 0 || index == last || index % OVERVIEW_ANCHOR_STRIDE == 0) - .then_some(entry) - }) - .collect(); - } - - for overview_index in 0..overview.len().saturating_sub(1) { - let current_entry = overview[overview_index].entry_range[0]; - let next_entry = overview[overview_index + 1].entry_range[0]; - overview[overview_index].intervening_entries = index - .iter() - .filter(|entry| { - let entry_index = entry.entry_range[0]; - entry_index > current_entry && entry_index < next_entry - }) - .count(); - } - - Self { - segment_id, - entries: Arc::new(entries), - overview, - index, - } - } - - pub(crate) fn overview(&self) -> &[OverviewItem] { - &self.overview - } - - pub(crate) fn source_index_for_ref(&self, reference: &SessionEntryRef) -> Option { - self.index - .iter() - .find(|entry| entry.id == *reference) - .map(|entry| entry.entry_range[0]) - .or_else(|| reference.source_index()) - } - - pub(crate) fn search(&self, options: &SearchOptions) -> Vec { - let query = options.query.trim().to_lowercase(); - let limit = options - .limit - .unwrap_or(DEFAULT_SEARCH_LIMIT) - .clamp(1, MAX_SEARCH_LIMIT); - let tool_name = options.tool_name.as_deref(); - let min_entry_index = options - .from - .as_ref() - .and_then(|reference| self.source_index_for_ref(reference)) - .unwrap_or_else(|| options.min_entry_index.unwrap_or(0)); - let max_entry_index = options - .through - .as_ref() - .and_then(|reference| self.source_index_for_ref(reference)) - .unwrap_or(u64::MAX); - let mut skipped = 0usize; - let mut hits = Vec::new(); - - for entry in &self.index { - if entry.use_policy.is_some_and(|p| !p.observation_readable) { - continue; - } - if entry.entry_range[0] < min_entry_index || entry.entry_range[0] > max_entry_index { - continue; - } - if let Some(kind) = options.kind { - if entry.kind != kind { - continue; - } - } - if let Some(part) = options.tool_part { - if entry.kind != ReferenceKind::Tool { - continue; - } - let Some(actual) = entry.tool_part else { - continue; - }; - if !part.matches(actual) { - continue; - } - } - if let Some(name) = tool_name { - if entry.tool_name.as_deref() != Some(name) { - continue; - } - } - if !query.is_empty() && !entry.search_text.to_lowercase().contains(&query) { - continue; - } - if skipped < options.offset { - skipped += 1; - continue; - } - hits.push(SearchHit { - id: entry.id.clone(), - origin: entry.origin.clone(), - kind: entry.kind, - tool_part: entry.tool_part, - tool_name: entry.tool_name.clone(), - entry_range: entry.entry_range, - label: entry.label.clone(), - summary: entry.summary.clone(), - }); - if hits.len() >= limit { - break; - } - } - - hits - } - - pub(crate) fn read(&self, selector: ReadSelector<'_>, options: ReadOptions) -> ReadResult { - let max_items = options.max_items.clamp(1, MAX_READ_MAX_ITEMS); - let max_bytes = options.max_bytes.max(1); - let mut entries = Vec::new(); - let mut bytes = 0usize; - let mut truncated = false; - - let selected: Vec<&ReferenceEntry> = match selector { - ReadSelector::Id(id) => self - .index - .iter() - .filter(|entry| entry.id.as_str() == id) - .collect(), - #[cfg(test)] - ReadSelector::EntryRange([start, end]) => self - .index - .iter() - .filter(|entry| entry.entry_range[0] >= start && entry.entry_range[0] <= end) - .collect(), - }; - - for entry in selected { - if entry.use_policy.is_some_and(|p| !p.observation_readable) { - continue; - } - if entries.len() >= max_items { - truncated = true; - break; - } - if entry.kind == ReferenceKind::Tool { - if !options.include_tools { - continue; - } - if let Some(actual) = entry.tool_part { - if !options.tool_part.matches(actual) { - continue; - } - } - } - let Some(item) = self - .entries - .get(entry.entry_range[0] as usize) - .map(|entry| &entry.item) - else { - continue; - }; - let text = render_item(item, entry, options.detail, max_bytes.saturating_sub(bytes)); - bytes = bytes.saturating_add(text.len()); - entries.push(ReadEntry { - id: entry.id.clone(), - origin: entry.origin.clone(), - kind: entry.kind, - tool_part: entry.tool_part, - tool_name: entry.tool_name.clone(), - entry_range: entry.entry_range, - label: entry.label.clone(), - text, - }); - if bytes >= max_bytes { - truncated = true; - break; - } - } - - ReadResult { entries, truncated } - } - - /// Resolves the committed tool-call input for a provider call id. This is - /// used by explicit-memory receipts only after the containing run is durable, - /// so a model-authored claim can cite its own committed tool invocation - /// without inventing a SessionEntryRef. - pub(crate) fn evidence_for_tool_call(&self, call_id: &str) -> Option { - let entry_index = self.entries.iter().position(|entry| { - matches!( - &entry.item, - Item::ToolCall { - call_id: stored_call_id, - .. - } if stored_call_id == call_id - ) - })?; - let reference = self.index.iter().find(|entry| { - entry.entry_range[0] == entry_index as u64 - && entry.kind == ReferenceKind::Tool - && entry.tool_part == Some(ToolPart::Input) - })?; - self.evidence_for(reference.id.as_str()) - } - - pub(crate) fn committed_pending_tool_calls(&self, names: &[&str]) -> Vec { - let pending_call_ids = self - .entries - .iter() - .filter_map(|entry| { - let Item::ToolResult { - call_id, - content: Some(content), - is_error: false, - .. - } = &entry.item - else { - return None; - }; - (serde_json::from_str::(content) - .ok() - .and_then(|value| value.get("status").cloned()) - .as_ref() - .and_then(serde_json::Value::as_str) - == Some("pending_commit")) - .then_some(call_id.as_str()) - }) - .collect::>(); - self.entries - .iter() - .filter_map(|entry| { - let Item::ToolCall { - call_id, - name, - arguments, - .. - } = &entry.item - else { - return None; - }; - if !names.contains(&name.as_str()) || !pending_call_ids.contains(call_id.as_str()) { - return None; - } - Some(CommittedToolCall { - call_id: call_id.clone(), - name: name.clone(), - arguments: arguments.clone(), - evidence: self.evidence_for_tool_call(call_id)?, - }) - }) - .collect() - } - - pub(crate) fn evidence_for(&self, id: &str) -> Option { - let entry = self.index.iter().find(|entry| entry.id.as_str() == id)?; - if entry.use_policy.is_some_and(|p| !p.memory_evidence) { - return None; - } - let item = &self.entries.get(entry.entry_range[0] as usize)?.item; - let excerpt = render_item(item, entry, ReadDetail::Compact, 2 * 1024); - Some(SessionEntryEvidence { - segment_id: self.segment_id.clone(), - entry_ref: entry.id.clone(), - origin: entry.origin.clone(), - entry_range: entry.entry_range, - kind: entry.kind, - tool_part: entry.tool_part, - label: entry.label.clone(), - summary: entry.summary.clone(), - excerpt, - }) - } -} - -fn public_provenance(origin: &WorkerHistoryProvenance) -> SessionEntryProvenance { - match origin { - WorkerHistoryProvenance::HumanInput { .. } | WorkerHistoryProvenance::LocalHumanInput => { - SessionEntryProvenance::HumanInput - } - WorkerHistoryProvenance::WorkerInput { .. } => SessionEntryProvenance::WorkerInput, - WorkerHistoryProvenance::FlowInstruction { .. } => SessionEntryProvenance::FlowInstruction, - WorkerHistoryProvenance::BackendInstruction { .. } - | WorkerHistoryProvenance::Provided { .. } => SessionEntryProvenance::BackendInstruction, - WorkerHistoryProvenance::ModelOutput { .. } => SessionEntryProvenance::ModelOutput, - WorkerHistoryProvenance::ToolOutput { .. } => SessionEntryProvenance::ToolOutput, - WorkerHistoryProvenance::CompactionSummary - | WorkerHistoryProvenance::DerivedSummary - | WorkerHistoryProvenance::JournalContext => SessionEntryProvenance::DerivedSummary, - WorkerHistoryProvenance::LegacyUnknown => SessionEntryProvenance::LegacyUnknown, - } -} - -fn message_reference_kind( - origin: &SessionEntryProvenance, - provider_role: &Role, - use_policy: Option, -) -> Option { - if use_policy.is_some() { - return match provider_role { - Role::System => Some(ReferenceKind::System), - Role::User => Some(ReferenceKind::User), - Role::Assistant => Some(ReferenceKind::Assistant), - }; - } - match origin { - SessionEntryProvenance::HumanInput | SessionEntryProvenance::WorkerInput => { - Some(ReferenceKind::User) - } - SessionEntryProvenance::ModelOutput => Some(ReferenceKind::Assistant), - SessionEntryProvenance::ToolOutput => Some(ReferenceKind::Tool), - SessionEntryProvenance::LegacyUnknown => match provider_role { - Role::User => Some(ReferenceKind::User), - Role::Assistant => Some(ReferenceKind::Assistant), - Role::System => None, - }, - // Flow/backend/system content remains out of the observation surface - // even when represented with a provider user/system role. - SessionEntryProvenance::FlowInstruction - | SessionEntryProvenance::BackendInstruction - | SessionEntryProvenance::DerivedSummary => None, - } -} - -fn render_item( - item: &Item, - entry: &ReferenceEntry, - detail: ReadDetail, - max_bytes: usize, -) -> String { - let text = match item { - Item::Message { role, content, .. } => { - let text = content - .iter() - .map(|p| p.as_text()) - .collect::>() - .join(""); - format!("[{} {:?}] {text}", entry.id, role) - } - Item::ToolCall { - name, arguments, .. - } => match detail { - ReadDetail::Compact => format!("[{} ToolInput {name}] (arguments omitted)", entry.id), - ReadDetail::Full => format!("[{} ToolInput {name}]\narguments: {arguments}", entry.id), - }, - Item::ToolResult { - summary, - content, - attachments, - is_error, - .. - } => { - let attachment_line = if attachments.is_empty() { - String::new() - } else { - format!("\nattachments: {} image(s)", attachments.len()) - }; - match detail { - ReadDetail::Compact => format!( - "[{} ToolOutput{}]\nsummary: {summary}\ncontent: (omitted){attachment_line}", - entry.id, - if *is_error { " error" } else { "" }, - ), - ReadDetail::Full => format!( - "[{} ToolOutput{}]\nsummary: {summary}\ncontent: {}{attachment_line}", - entry.id, - if *is_error { " error" } else { "" }, - content.as_deref().unwrap_or_default(), - ), - } - } - Item::AssistantResponseBoundary { .. } => { - format!("[{} Assistant response boundary omitted]", entry.id) - } - Item::Reasoning { .. } => format!("[{} Reasoning omitted]", entry.id), - }; - truncate_chars(&text, max_bytes) -} - -fn truncate_chars(text: &str, max_chars: usize) -> String { - if max_chars == 0 { - return "… [truncated]".to_string(); - } - if text.chars().count() <= max_chars { - return text.to_string(); - } - let mut out = text.chars().take(max_chars).collect::(); - out.push_str("… [truncated]"); - out -} +pub use session_store::capture::*; #[cfg(test)] mod tests { use super::*; + use crate::session_history::WorkerHistoryProvenance; + use agen::Item; + use protocol::{SessionEntryProvenance, SessionSnapshot}; #[test] fn flow_user_role_is_excluded_while_explicit_human_origin_remains_evidence() { @@ -897,7 +30,7 @@ mod tests { }, ), ]; - let capture = SessionCapture::from_history_entries("segment", entries); + let capture = SessionCapture::from_history_entries("session", "segment", entries); let overview = capture.overview(); assert_eq!(overview.len(), 1); assert!(matches!( @@ -921,8 +54,8 @@ mod tests { }, ); let expected_ref = SessionEntryRef::from_history_entry_id(&retained.annotation.entry_id); - let before = SessionCapture::from_history_entries("old", vec![retained.clone()]); - let after = SessionCapture::from_history_entries("new", vec![retained]); + let before = SessionCapture::from_history_entries("session", "old", vec![retained.clone()]); + let after = SessionCapture::from_history_entries("session", "new", vec![retained]); assert_eq!(before.overview()[0].id, expected_ref); assert_eq!(after.overview()[0].id, expected_ref); assert_eq!( @@ -963,7 +96,7 @@ mod tests { .iter() .map(|entry| SessionEntryRef::from_history_entry_id(&entry.annotation.entry_id)) .collect::>(); - let capture = SessionCapture::from_history_entries("segment", entries); + let capture = SessionCapture::from_history_entries("session", "segment", entries); assert!(capture.overview().is_empty()); assert!( capture @@ -1102,8 +235,16 @@ mod tests { let snapshot: SessionSnapshot = serde_json::from_value(serde_json::to_value(snapshot).unwrap()).unwrap(); for capture in [ - SessionCapture::from_history_entries(segment.to_string(), restored), - SessionCapture::from_session_snapshot(segment.to_string(), snapshot), + SessionCapture::from_history_entries( + session.to_string(), + segment.to_string(), + restored, + ), + SessionCapture::from_session_snapshot( + session.to_string(), + segment.to_string(), + snapshot, + ), ] { let hits = capture.search(&SearchOptions { kind: Some(ReferenceKind::System), @@ -1228,13 +369,15 @@ mod tests { 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 capture = SessionCapture::from_history_entries( + session.to_string(), + fork.to_string(), + restored, ); + // Private derived content must be absent even from the saved + // body representation, not merely hidden by the UI index. + let encoded = serde_json::to_string(&capture.to_dto()).unwrap(); + assert!(!encoded.contains("PRIVATE_SYNTHETIC")); let snapshot = public_snapshot::project_session_snapshot(session, &saved); assert!( !serde_json::to_string(&snapshot) @@ -1243,7 +386,11 @@ mod tests { ); for capture in [ capture, - SessionCapture::from_session_snapshot(fork.to_string(), snapshot), + SessionCapture::from_session_snapshot( + session.to_string(), + fork.to_string(), + snapshot, + ), ] { assert_eq!( capture @@ -1315,6 +462,7 @@ mod tests { #[test] fn overview_contains_user_and_assistant_only() { let view = SessionCapture::new( + "session", "segment-1", vec![ Item::system_message("sys"), @@ -1335,6 +483,7 @@ mod tests { #[test] fn search_filters_tool_input_and_output() { let view = SessionCapture::new( + "session", "segment-1", vec![ Item::tool_call("c1", "Read", "{\"file\":\"Cargo.toml\"}"), @@ -1374,6 +523,7 @@ mod tests { #[test] fn read_by_entry_range_is_bounded_and_can_skip_tools() { let view = SessionCapture::new( + "session", "segment-1", vec![ Item::user_message("one"), @@ -1406,6 +556,7 @@ mod tests { #[test] fn tool_image_projection_exposes_only_bounded_metadata() { let view = SessionCapture::new( + "session", "segment-1", vec![Item::tool_result_item_with_attachments( "c1", @@ -1435,6 +586,7 @@ mod tests { #[test] fn system_prompt_and_reasoning_are_excluded_from_every_projection() { let view = SessionCapture::new( + "session", "segment-1", vec![ Item::system_message("raw secret system prompt"), @@ -1472,7 +624,7 @@ mod tests { let items = (0..20) .map(|index| Item::user_message(format!("message-{index}"))) .collect::>(); - let view = SessionCapture::new("segment-1", items); + let view = SessionCapture::new("session", "segment-1", items); let refs = view .overview() .iter() @@ -1490,6 +642,7 @@ mod tests { #[test] fn append_preserves_existing_session_entry_refs() { let first = SessionCapture::new( + "session", "segment-1", vec![ Item::user_message("first"), @@ -1497,6 +650,7 @@ mod tests { ], ); let appended = SessionCapture::new( + "session", "segment-1", vec![ Item::user_message("first"), @@ -1540,7 +694,7 @@ mod tests { #[test] fn evidence_projection_uses_entry_range_and_session_entry_ref() { - let view = SessionCapture::new("segment-1", vec![Item::user_message("hello")]); + let view = SessionCapture::new("session", "segment-1", vec![Item::user_message("hello")]); let source = view.evidence_for("E00000000").unwrap(); assert_eq!(source.segment_id, "segment-1"); assert_eq!(source.entry_range, [0, 0]); @@ -1550,6 +704,7 @@ mod tests { #[test] fn committed_tool_call_can_be_resolved_by_provider_call_id() { let view = SessionCapture::new( + "session", "segment-1", vec![Item::tool_call( "call-remember", @@ -1562,4 +717,111 @@ mod tests { assert_eq!(evidence.tool_part, Some(ToolPart::Input)); assert!(view.evidence_for_tool_call("missing").is_none()); } + // Real Store append and Segment replacement -> Worker replay -> the common + // fixed value used by Memory and compaction exploration. No live I/O remains + // available after the owned temporary source is removed. + #[test] + fn frozen_memory_and_exploration_survive_append_compaction_and_source_removal() { + use crate::feature::builtin::memory_staging_output::render_extract_input; + use crate::feature::builtin::session_explore::SessionExploreState; + use session_store::{FsStore, LogEntry, SegmentOrigin, Store}; + let root = tempfile::tempdir().unwrap(); + let store = FsStore::new(root.path()).unwrap(); + let session = session_store::new_session_id(); + let first = session_store::new_segment_id(); + let next = session_store::new_segment_id(); + let original = crate::session_history::history_entry( + Item::user_message("fixed 日本語 intent"), + WorkerHistoryProvenance::LocalHumanInput, + ); + let seed = |history, compacted_from| LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id: session, + system_prompt: Some("PRIVATE_SYSTEM".into()), + config: Default::default(), + history, + forked_from: None, + compacted_from, + }; + store + .create_segment( + session, + first, + &[seed( + vec![crate::session_history::to_logged_history_entry(&original)], + None, + )], + ) + .unwrap(); + let log = store.read_all(session, first).unwrap(); + let restored = + crate::session_history::restore_history_entries(session, first, &log).unwrap(); + let fixed = + SessionCapture::from_history_entries(session.to_string(), first.to_string(), restored); + let overview = fixed.overview()[0].clone(); + let evidence = fixed.evidence_for(overview.id.as_str()).unwrap(); + let extract = render_extract_input(&fixed); + let encoded = fixed.encode(CaptureLimits::default()).unwrap(); + let state = SessionExploreState::new(fixed.clone()); + store + .append( + session, + first, + &LogEntry::AnnotatedAssistantItem { + ts: 2, + entry: crate::session_history::to_logged_history_entry( + &crate::session_history::history_entry( + Item::assistant_message("later append"), + WorkerHistoryProvenance::LegacyUnknown, + ), + ), + }, + ) + .unwrap(); + let summary = crate::session_history::history_entry( + Item::system_message("PRIVATE_COMPACTED_SUMMARY"), + WorkerHistoryProvenance::DerivedSummary, + ); + store + .create_segment( + session, + next, + &[seed( + vec![crate::session_history::to_logged_history_entry(&summary)], + Some(SegmentOrigin { + segment_id: first, + at_turn_index: 0, + }), + )], + ) + .unwrap(); + drop(store); + drop(root); + assert_eq!(fixed.encode(CaptureLimits::default()).unwrap(), encoded); + assert_eq!(render_extract_input(state.view()), extract); + assert_eq!(state.view().overview()[0].id, overview.id); + assert_eq!(state.view().overview()[0].text, overview.text); + let retained = state.view().evidence_for(overview.id.as_str()).unwrap(); + assert_eq!(retained.entry_range, evidence.entry_range); + assert_eq!(retained.excerpt, evidence.excerpt); + assert!( + state + .view() + .search(&SearchOptions { + query: "later".into(), + ..Default::default() + }) + .is_empty() + ); + let restored = SessionCapture::decode(&encoded, CaptureLimits::default()).unwrap(); + let scoped = fixed.scoped_ref(&overview.id).unwrap(); + assert_eq!( + restored + .resolve_ref(&scoped, CapturePurpose::MemoryEvidence) + .unwrap() + .entry_id, + original.annotation.entry_id.0 + ); + assert!(!String::from_utf8(encoded).unwrap().contains("PRIVATE_")); + } } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index f46be3d4a..859784866 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -7501,6 +7501,7 @@ impl Worker { crate::feature::builtin::session_explore::SessionExploreFeature::new( crate::feature::builtin::session_explore::SessionExploreState::new( crate::session_capture::SessionCapture::from_history_entries( + self.session_id().to_string(), self.segment_id().to_string(), entries_to_summarise.clone(), ), @@ -17374,8 +17375,9 @@ permission = "write" producer: "arbitrary-context-provider".into() } ); - let view = - crate::session_capture::SessionCapture::from_history_entries("startup", replay); + let view = crate::session_capture::SessionCapture::from_history_entries( + "session", "startup", replay, + ); let extract = crate::feature::builtin::memory_staging_output::render_extract_input(&view); assert!(!extract.contains("PRIOR_JOURNAL_REFERENCE"));