diff --git a/config/grid-agent.example.json b/config/grid-agent.example.json index 3f113e4..873c0c4 100644 --- a/config/grid-agent.example.json +++ b/config/grid-agent.example.json @@ -32,5 +32,16 @@ "stable_reset_seconds": 120, "jitter_basis_points": 2000, "offline_work_capacity": 128 + }, + "conversation": { + "persistence_enabled": false, + "max_active_sessions": 512, + "max_turns_per_session": 64, + "max_session_bytes": 262144, + "max_total_bytes": 8388608, + "max_tool_results_per_session": 16, + "max_tool_result_bytes": 16384, + "max_persisted_bytes": 16777216, + "max_summary_bytes": 8192 } } diff --git a/crates/metacrate-grid-agent/README.md b/crates/metacrate-grid-agent/README.md index ce5c616..a426bdc 100644 --- a/crates/metacrate-grid-agent/README.md +++ b/crates/metacrate-grid-agent/README.md @@ -48,3 +48,6 @@ backend authorization are specified in The reconnect state machine, generation fencing, offline-work contract, and shutdown deadline are specified in [`../../docs/grid-agent-session.md`](../../docs/grid-agent-session.md). +Per-avatar/channel expiry, compaction, redaction, optional atomic persistence, +and metadata-only operator controls are specified in +[`../../docs/grid-agent-conversation.md`](../../docs/grid-agent-conversation.md). diff --git a/crates/metacrate-grid-agent/src/config.rs b/crates/metacrate-grid-agent/src/config.rs index 7c560fa..378ba85 100644 --- a/crates/metacrate-grid-agent/src/config.rs +++ b/crates/metacrate-grid-agent/src/config.rs @@ -202,6 +202,14 @@ pub struct BehaviorSettings { pub heartbeat: Duration, } +/// Conversation-memory settings. Persistence is opt-in and uses +/// `storage_path/conversations`. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct ConversationSettings { + pub persistence_enabled: bool, + pub limits: crate::conversation::ConversationLimits, +} + /// Fully resolved configuration. It cannot be constructed without validation. #[derive(Clone, Debug, Eq, PartialEq)] pub struct AgentConfig { @@ -214,6 +222,7 @@ pub struct AgentConfig { pub storage_path: PathBuf, pub behavior: BehaviorSettings, pub reconnect: crate::session::ReconnectPolicy, + pub conversation: ConversationSettings, } impl AgentConfig { @@ -243,6 +252,10 @@ impl AgentConfig { self.reconnect .validate() .map_err(|_| ConfigError::InvalidReconnect)?; + self.conversation + .limits + .validate() + .map_err(|_| ConfigError::InvalidConversationMemory)?; if self.mode != OperatingMode::OfflineFake && self.grid.is_none() { return Err(ConfigError::Missing { field: "grid", @@ -416,6 +429,7 @@ pub enum ConfigError { maximum: usize, }, InvalidReconnect, + InvalidConversationMemory, } impl fmt::Display for ConfigError { @@ -466,6 +480,9 @@ impl fmt::Display for ConfigError { "unsafe {field}={value}; expected {minimum}..={maximum}" ), Self::InvalidReconnect => formatter.write_str("invalid reconnect policy bounds"), + Self::InvalidConversationMemory => { + formatter.write_str("invalid conversation-memory bounds") + } } } } @@ -486,6 +503,7 @@ struct FileConfig { storage_path: Option, behavior: RawBehavior, reconnect: RawReconnect, + conversation: RawConversation, } #[derive(Clone, Default, Deserialize)] @@ -551,6 +569,20 @@ struct RawReconnect { offline_work_capacity: Option, } +#[derive(Clone, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct RawConversation { + persistence_enabled: Option, + max_active_sessions: Option, + max_turns_per_session: Option, + max_session_bytes: Option, + max_total_bytes: Option, + max_tool_results_per_session: Option, + max_tool_result_bytes: Option, + max_persisted_bytes: Option, + max_summary_bytes: Option, +} + fn read_config(path: &Path) -> Result { let bytes = read_bounded_regular_file("configuration file", path, MAX_CONFIG_BYTES)?; serde_json::from_slice(&bytes).map_err(|error| ConfigError::InvalidSchema { @@ -721,6 +753,44 @@ fn resolve( .offline_work_capacity .unwrap_or(reconnect_defaults.offline_work_capacity), }; + let conversation_defaults = ConversationSettings::default(); + let conversation = ConversationSettings { + persistence_enabled: raw.conversation.persistence_enabled.unwrap_or(false), + limits: crate::conversation::ConversationLimits { + max_active_sessions: raw + .conversation + .max_active_sessions + .unwrap_or(conversation_defaults.limits.max_active_sessions), + max_turns_per_session: raw + .conversation + .max_turns_per_session + .unwrap_or(conversation_defaults.limits.max_turns_per_session), + max_session_bytes: raw + .conversation + .max_session_bytes + .unwrap_or(conversation_defaults.limits.max_session_bytes), + max_total_bytes: raw + .conversation + .max_total_bytes + .unwrap_or(conversation_defaults.limits.max_total_bytes), + max_tool_results_per_session: raw + .conversation + .max_tool_results_per_session + .unwrap_or(conversation_defaults.limits.max_tool_results_per_session), + max_tool_result_bytes: raw + .conversation + .max_tool_result_bytes + .unwrap_or(conversation_defaults.limits.max_tool_result_bytes), + max_persisted_bytes: raw + .conversation + .max_persisted_bytes + .unwrap_or(conversation_defaults.limits.max_persisted_bytes), + max_summary_bytes: raw + .conversation + .max_summary_bytes + .unwrap_or(conversation_defaults.limits.max_summary_bytes), + }, + }; let config = AgentConfig { mode, @@ -741,6 +811,7 @@ fn resolve( )?, }, reconnect, + conversation, }; config.validate()?; Ok(config) @@ -1127,6 +1198,19 @@ mod tests { Err(ConfigError::InvalidReconnect) )); let _ = fs::remove_file(reconnect); + + let conversation = temporary_file( + "unsafe-conversation.json", + r#"{"conversation":{"max_active_sessions":0}}"#, + ); + assert!(matches!( + ConfigLoader::new() + .with_file(&conversation) + .with_environment(offline_environment()) + .load(), + Err(ConfigError::InvalidConversationMemory) + )); + let _ = fs::remove_file(conversation); } #[test] diff --git a/crates/metacrate-grid-agent/src/conversation.rs b/crates/metacrate-grid-agent/src/conversation.rs new file mode 100644 index 0000000..c0735de --- /dev/null +++ b/crates/metacrate-grid-agent/src/conversation.rs @@ -0,0 +1,1295 @@ +//! Isolated, bounded per-avatar conversation memory. + +#![allow(clippy::missing_errors_doc)] + +use crate::llm::CompletionMessage; +use crate::types::{ + BoundedText, MAX_CONVERSATION_MESSAGES, MAX_IDENTIFIER_BYTES, MAX_MESSAGE_BYTES, MessageRole, +}; +use libremetaverse_types::UUID; +use serde::{Deserialize, Serialize}; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; +use std::error::Error; +use std::fmt; +use std::fs::{self, File, OpenOptions}; +use std::io::{Read as _, Write as _}; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::time::Instant; +use url::Url; + +const PUBLIC_TTL: Duration = Duration::from_mins(30); +const DIRECT_IM_TTL: Duration = Duration::from_hours(24); +const SNAPSHOT_SCHEMA: u32 = 1; +const MAX_ACTIVE_SESSIONS: usize = 4_096; +const MAX_SESSION_BYTES: usize = 1024 * 1024; +const MAX_TOTAL_BYTES: usize = 64 * 1024 * 1024; +const MAX_TOOL_RESULTS: usize = 64; +const MAX_PERSISTED_BYTES: usize = 64 * 1024 * 1024; +const MAX_SUMMARY_BYTES: usize = 64 * 1024; +const RETAINED_SNAPSHOTS: usize = 2; +const MAX_MEMORY_EVENTS: usize = 8_192; + +/// A trust-separated conversation channel. Group and conference channels are +/// deliberately unrepresentable until their authority semantics are defined. +#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ConversationChannel { + PublicChat, + DirectIm, +} + +impl ConversationChannel { + #[must_use] + pub const fn inactivity_ttl(self) -> Duration { + match self { + Self::PublicChat => PUBLIC_TTL, + Self::DirectIm => DIRECT_IM_TTL, + } + } +} + +/// Immutable identity for one avatar/channel history. +#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)] +pub struct ConversationKey { + pub avatar_id: UUID, + pub channel: ConversationChannel, +} + +impl ConversationKey { + pub fn new(avatar_id: UUID, channel: ConversationChannel) -> Result { + if avatar_id == UUID::zero() { + return Err(ConversationError::InvalidAvatar); + } + Ok(Self { avatar_id, channel }) + } +} + +/// Dynamic limits below the crate hard ceilings. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ConversationLimits { + pub max_active_sessions: usize, + pub max_turns_per_session: usize, + pub max_session_bytes: usize, + pub max_total_bytes: usize, + pub max_tool_results_per_session: usize, + pub max_tool_result_bytes: usize, + pub max_persisted_bytes: usize, + pub max_summary_bytes: usize, +} + +impl Default for ConversationLimits { + fn default() -> Self { + Self { + max_active_sessions: 512, + max_turns_per_session: 64, + max_session_bytes: 256 * 1024, + max_total_bytes: 8 * 1024 * 1024, + max_tool_results_per_session: 16, + max_tool_result_bytes: 16 * 1024, + max_persisted_bytes: 16 * 1024 * 1024, + max_summary_bytes: 8 * 1024, + } + } +} + +impl ConversationLimits { + pub fn validate(&self) -> Result<(), ConversationError> { + if self.max_active_sessions == 0 + || self.max_active_sessions > MAX_ACTIVE_SESSIONS + || self.max_turns_per_session < 4 + || self.max_turns_per_session > MAX_CONVERSATION_MESSAGES + || self.max_session_bytes < 1024 + || self.max_session_bytes > MAX_SESSION_BYTES + || self.max_total_bytes < self.max_session_bytes + || self.max_total_bytes > MAX_TOTAL_BYTES + || self.max_tool_results_per_session == 0 + || self.max_tool_results_per_session > MAX_TOOL_RESULTS + || self.max_tool_result_bytes == 0 + || self.max_tool_result_bytes > MAX_MESSAGE_BYTES + || self.max_persisted_bytes < self.max_total_bytes + || self.max_persisted_bytes > MAX_PERSISTED_BYTES + || self.max_summary_bytes == 0 + || self.max_summary_bytes > MAX_SUMMARY_BYTES + || self.max_summary_bytes >= self.max_session_bytes + { + return Err(ConversationError::UnsafeLimits); + } + Ok(()) + } +} + +/// Optional local snapshot directory. Each flush publishes a new immutable +/// generation, avoiding platform-specific replace semantics. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ConversationPersistence { + pub directory: PathBuf, +} + +/// Source of monotonic expiry and wall-clock persistence timestamps. +pub trait ConversationClock: Send + Sync { + fn monotonic_now(&self) -> Instant; + fn wall_now(&self) -> SystemTime; +} + +#[derive(Clone, Copy, Debug, Default)] +pub struct SystemConversationClock; + +impl ConversationClock for SystemConversationClock { + fn monotonic_now(&self) -> Instant { + Instant::now() + } + + fn wall_now(&self) -> SystemTime { + SystemTime::now() + } +} + +/// Whether text remains untrusted prompt data after storage and compaction. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum MemoryTrust { + Untrusted, + TrustedOutput, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum RecordKind { + Message, + ToolSummary, + ActionResult, + FactualSummary, +} + +/// Admitted memory record. Constructors encode the only supported data classes; +/// hidden reasoning and raw binary assets have no representation. +#[derive(Clone)] +pub struct MemoryRecord { + role: MessageRole, + kind: RecordKind, + trust: MemoryTrust, + text: String, +} + +impl fmt::Debug for MemoryRecord { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("MemoryRecord") + .field("role", &self.role) + .field("kind", &self.kind) + .field("trust", &self.trust) + .field("bytes", &self.text.len()) + .finish() + } +} + +impl MemoryRecord { + #[must_use] + pub fn avatar_message(text: impl Into) -> Self { + Self { + role: MessageRole::Avatar, + kind: RecordKind::Message, + trust: MemoryTrust::Untrusted, + text: text.into(), + } + } + + /// Stores only the assistant response that was actually shown to the + /// resident. Model scratchpads and hidden reasoning must not cross this API. + #[must_use] + pub fn agent_visible_response(text: impl Into) -> Self { + Self { + role: MessageRole::Agent, + kind: RecordKind::Message, + trust: MemoryTrust::TrustedOutput, + text: text.into(), + } + } + + #[must_use] + pub fn tool_summary(text: impl Into, trust: MemoryTrust) -> Self { + Self { + role: MessageRole::Tool, + kind: RecordKind::ToolSummary, + trust, + text: text.into(), + } + } + + #[must_use] + pub fn action_result(text: impl Into, trust: MemoryTrust) -> Self { + Self { + role: MessageRole::Tool, + kind: RecordKind::ActionResult, + trust, + text: text.into(), + } + } +} + +/// Stable, content-free reasons suitable for audit and observability. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum MemoryReason { + PublicExpired, + DirectImExpired, + TurnCompacted, + ByteCompacted, + ToolResultCompacted, + SessionLimitEviction, + TotalByteEviction, + OperatorDeleted, + OperatorExpired, + CorruptSnapshotQuarantined, + UnsupportedSnapshotQuarantined, + SnapshotTooLargeQuarantined, + PermissionsNotVerified, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct MemoryEvent { + pub reason: MemoryReason, + pub session_id: Option>, + pub avatar_id: Option, + pub channel: Option, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ConversationError { + UnsafeLimits, + InvalidAvatar, + EmptyContent, + ContentTooLarge, + ToolResultTooLarge, + IdentifierGeneration, + PersistenceUnavailable, + PersistenceTooLarge, + Boundary(crate::types::BoundaryError), +} + +impl fmt::Display for ConversationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::UnsafeLimits => formatter.write_str("unsafe conversation-memory limits"), + Self::InvalidAvatar => formatter.write_str("conversation avatar UUID is invalid"), + Self::EmptyContent => formatter.write_str("conversation content must not be empty"), + Self::ContentTooLarge => formatter.write_str("conversation content exceeds its bound"), + Self::ToolResultTooLarge => formatter.write_str("tool result exceeds its memory bound"), + Self::IdentifierGeneration => formatter.write_str("cannot generate a conversation ID"), + Self::PersistenceUnavailable => { + formatter.write_str("conversation persistence is unavailable") + } + Self::PersistenceTooLarge => { + formatter.write_str("conversation snapshot exceeds its storage bound") + } + Self::Boundary(error) => { + write!(formatter, "conversation boundary rejected data: {error}") + } + } + } +} + +impl Error for ConversationError {} + +impl From for ConversationError { + fn from(value: crate::types::BoundaryError) -> Self { + Self::Boundary(value) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct MemoryUpdate { + pub session_id: BoundedText, + pub sequence: u64, + pub events: Vec, +} + +/// Safe operator projection. Message content is intentionally absent. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ConversationMetadata { + pub session_id: BoundedText, + pub avatar_id: UUID, + pub channel: ConversationChannel, + pub created_unix_millis: u64, + pub last_active_unix_millis: u64, + pub turns: usize, + pub bytes: usize, +} + +/// LLM-only projection for one exact UUID/channel key. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ConversationContext { + pub session_id: BoundedText, + entries: Vec, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct ContextEntry { + role: MessageRole, + kind: RecordKind, + trust: MemoryTrust, + text: String, +} + +impl ConversationContext { + pub fn llm_messages(&self) -> Result, ConversationError> { + self.entries + .iter() + .map(|entry| { + let (role, prefix) = match (entry.kind, entry.trust) { + (RecordKind::FactualSummary, MemoryTrust::Untrusted) => { + (MessageRole::Avatar, "[untrusted historical summary] ") + } + (RecordKind::FactualSummary, MemoryTrust::TrustedOutput) => { + (MessageRole::Avatar, "[historical summary] ") + } + (RecordKind::ToolSummary, MemoryTrust::Untrusted) => { + (MessageRole::Avatar, "[untrusted tool data] ") + } + (RecordKind::ToolSummary, MemoryTrust::TrustedOutput) => { + (MessageRole::Avatar, "[tool summary] ") + } + (RecordKind::ActionResult, MemoryTrust::Untrusted) => { + (MessageRole::Avatar, "[untrusted action data] ") + } + (RecordKind::ActionResult, MemoryTrust::TrustedOutput) => { + (MessageRole::Avatar, "[action result] ") + } + (RecordKind::Message, _) => (entry.role, ""), + }; + CompletionMessage::text(role, format!("{prefix}{}", entry.text)).map_err(|error| { + match error { + crate::llm::LlmError::Boundary(boundary) => { + ConversationError::Boundary(boundary) + } + _ => ConversationError::ContentTooLarge, + } + }) + }) + .collect() + } + + #[must_use] + pub fn turn_count(&self) -> usize { + self.entries.len() + } +} + +struct StoredEntry { + sequence: u64, + unix_millis: u64, + role: MessageRole, + kind: RecordKind, + trust: MemoryTrust, + text: BoundedText, +} + +impl StoredEntry { + fn bytes(&self) -> usize { + self.text.len() + } + + fn is_tool_result(&self) -> bool { + matches!( + self.kind, + RecordKind::ToolSummary | RecordKind::ActionResult + ) + } +} + +struct StoredSession { + id: BoundedText, + key: ConversationKey, + created_unix_millis: u64, + last_active_unix_millis: u64, + last_active: Instant, + entries: VecDeque, + bytes: usize, + tool_results: usize, +} + +impl StoredSession { + fn metadata(&self) -> ConversationMetadata { + ConversationMetadata { + session_id: self.id.clone(), + avatar_id: self.key.avatar_id, + channel: self.key.channel, + created_unix_millis: self.created_unix_millis, + last_active_unix_millis: self.last_active_unix_millis, + turns: self.entries.len(), + bytes: self.bytes, + } + } +} + +struct StoreState { + sessions: BTreeMap, + total_bytes: usize, + next_sequence: u64, + snapshot_generation: u64, + events: VecDeque, +} + +/// Thread-safe conversation store. A single mutex assigns total ordering to +/// concurrent turns and keeps all aggregate bounds atomic. +pub struct ConversationStore { + limits: ConversationLimits, + persistence: Option, + clock: Arc, + state: Mutex, +} + +impl fmt::Debug for ConversationStore { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + let state = self.lock_state(); + formatter + .debug_struct("ConversationStore") + .field("limits", &self.limits) + .field("persistence_enabled", &self.persistence.is_some()) + .field("active_sessions", &state.sessions.len()) + .field("total_bytes", &state.total_bytes) + .finish_non_exhaustive() + } +} + +impl ConversationStore { + pub fn from_config(config: &crate::config::AgentConfig) -> Result { + let persistence = + config + .conversation + .persistence_enabled + .then(|| ConversationPersistence { + directory: config.storage_path.join("conversations"), + }); + Self::open(config.conversation.limits.clone(), persistence) + } + + pub fn open( + limits: ConversationLimits, + persistence: Option, + ) -> Result { + Self::open_with_clock(limits, persistence, Arc::new(SystemConversationClock)) + } + + pub fn open_with_clock( + limits: ConversationLimits, + persistence: Option, + clock: Arc, + ) -> Result { + limits.validate()?; + let mut state = StoreState { + sessions: BTreeMap::new(), + total_bytes: 0, + next_sequence: 1, + snapshot_generation: 0, + events: VecDeque::new(), + }; + if let Some(settings) = &persistence { + prepare_directory(&settings.directory, &mut state.events)?; + load_latest_snapshot(&limits, settings, clock.as_ref(), &mut state)?; + } + Ok(Self { + limits, + persistence, + clock, + state: Mutex::new(state), + }) + } + + pub fn append( + &self, + key: ConversationKey, + record: MemoryRecord, + ) -> Result { + if key.avatar_id == UUID::zero() { + return Err(ConversationError::InvalidAvatar); + } + let MemoryRecord { + role, + kind, + trust, + text: raw_text, + } = record; + let sanitized = redact_sensitive(&raw_text); + if sanitized.is_empty() { + return Err(ConversationError::EmptyContent); + } + if sanitized.len() > MAX_MESSAGE_BYTES { + return Err(ConversationError::ContentTooLarge); + } + if kind != RecordKind::Message && sanitized.len() > self.limits.max_tool_result_bytes { + return Err(ConversationError::ToolResultTooLarge); + } + let text = BoundedText::new("conversation.content", sanitized)?; + let monotonic_now = self.clock.monotonic_now(); + let wall_now = unix_millis(self.clock.wall_now()); + let mut state = self.lock_state(); + let event_start = state.events.len(); + expire_idle_locked(&mut state, monotonic_now, None); + if !state.sessions.contains_key(&key) { + while state.sessions.len() >= self.limits.max_active_sessions { + evict_oldest(&mut state, MemoryReason::SessionLimitEviction, None); + } + let session = StoredSession { + id: new_session_id()?, + key, + created_unix_millis: wall_now, + last_active_unix_millis: wall_now, + last_active: monotonic_now, + entries: VecDeque::new(), + bytes: 0, + tool_results: 0, + }; + state.sessions.insert(key, session); + } + let effective_wall = state.sessions.get(&key).map_or(wall_now, |session| { + wall_now.max(session.last_active_unix_millis) + }); + let sequence = state.next_sequence; + state.next_sequence = state + .next_sequence + .checked_add(1) + .ok_or(ConversationError::IdentifierGeneration)?; + let entry = StoredEntry { + sequence, + unix_millis: effective_wall, + role, + kind, + trust, + text, + }; + let entry_bytes = entry.bytes(); + { + let session = state + .sessions + .get_mut(&key) + .ok_or(ConversationError::InvalidAvatar)?; + session.entries.push_back(entry); + session.bytes = session.bytes.saturating_add(entry_bytes); + session.tool_results = session + .tool_results + .saturating_add(usize::from(kind != RecordKind::Message)); + session.last_active = monotonic_now; + session.last_active_unix_millis = effective_wall; + } + state.total_bytes = state.total_bytes.saturating_add(entry_bytes); + compact_session(&mut state, key, &self.limits)?; + while state.total_bytes > self.limits.max_total_bytes { + evict_oldest(&mut state, MemoryReason::TotalByteEviction, Some(key)); + } + let session_id = state + .sessions + .get(&key) + .map(|session| session.id.clone()) + .ok_or(ConversationError::ContentTooLarge)?; + let events = state.events.iter().skip(event_start).cloned().collect(); + Ok(MemoryUpdate { + session_id, + sequence, + events, + }) + } + + pub fn context(&self, key: ConversationKey) -> Option { + let now = self.clock.monotonic_now(); + let mut state = self.lock_state(); + expire_idle_locked(&mut state, now, Some(key)); + state.sessions.get(&key).map(|session| ConversationContext { + session_id: session.id.clone(), + entries: session + .entries + .iter() + .map(|entry| ContextEntry { + role: entry.role, + kind: entry.kind, + trust: entry.trust, + text: entry.text.as_str().to_owned(), + }) + .collect(), + }) + } + + #[must_use] + pub fn list_metadata(&self) -> Vec { + let now = self.clock.monotonic_now(); + let mut state = self.lock_state(); + expire_idle_locked(&mut state, now, None); + state + .sessions + .values() + .map(StoredSession::metadata) + .collect() + } + + pub fn delete(&self, key: ConversationKey) -> bool { + self.remove_operator(key, MemoryReason::OperatorDeleted) + } + + pub fn expire(&self, key: ConversationKey) -> bool { + self.remove_operator(key, MemoryReason::OperatorExpired) + } + + pub fn expire_idle(&self) -> Vec { + let now = self.clock.monotonic_now(); + let mut state = self.lock_state(); + let start = state.events.len(); + expire_idle_locked(&mut state, now, None); + state.events.iter().skip(start).cloned().collect() + } + + pub fn drain_events(&self) -> Vec { + self.lock_state().events.drain(..).collect() + } + + pub fn flush(&self) -> Result<(), ConversationError> { + let Some(persistence) = &self.persistence else { + return Ok(()); + }; + let mut state = self.lock_state(); + write_snapshot(&self.limits, persistence, &mut state) + } + + fn remove_operator(&self, key: ConversationKey, reason: MemoryReason) -> bool { + let mut state = self.lock_state(); + let Some(session) = state.sessions.remove(&key) else { + return false; + }; + state.total_bytes = state.total_bytes.saturating_sub(session.bytes); + push_event(&mut state.events, event_for(reason, &session)); + true + } + + fn lock_state(&self) -> MutexGuard<'_, StoreState> { + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +fn compact_session( + state: &mut StoreState, + key: ConversationKey, + limits: &ConversationLimits, +) -> Result<(), ConversationError> { + loop { + let reason = { + let session = state + .sessions + .get(&key) + .ok_or(ConversationError::InvalidAvatar)?; + if session.entries.len() > limits.max_turns_per_session { + Some(MemoryReason::TurnCompacted) + } else if session.bytes > limits.max_session_bytes { + Some(MemoryReason::ByteCompacted) + } else if session.tool_results > limits.max_tool_results_per_session { + Some(MemoryReason::ToolResultCompacted) + } else { + None + } + }; + let Some(reason) = reason else { break }; + let removed = { + let session = state + .sessions + .get_mut(&key) + .ok_or(ConversationError::InvalidAvatar)?; + let remove_count = (session.entries.len() / 3).max(1); + let mut removed = Vec::with_capacity(remove_count); + for _ in 0..remove_count { + if let Some(entry) = session.entries.pop_front() { + session.bytes = session.bytes.saturating_sub(entry.bytes()); + session.tool_results = session + .tool_results + .saturating_sub(usize::from(entry.is_tool_result())); + removed.push(entry); + } + } + removed + }; + let removed_bytes = removed.iter().map(StoredEntry::bytes).sum::(); + state.total_bytes = state.total_bytes.saturating_sub(removed_bytes); + let summary = summarize_removed(&removed, limits.max_summary_bytes)?; + if let Some(summary) = summary.filter(|entry| entry.bytes() < removed_bytes) { + let summary_bytes = summary.bytes(); + let session = state + .sessions + .get_mut(&key) + .ok_or(ConversationError::InvalidAvatar)?; + session.entries.push_front(summary); + session.bytes = session.bytes.saturating_add(summary_bytes); + state.total_bytes = state.total_bytes.saturating_add(summary_bytes); + } + if let Some(session) = state.sessions.get(&key) { + push_event(&mut state.events, event_for(reason, session)); + } + } + Ok(()) +} + +fn summarize_removed( + removed: &[StoredEntry], + maximum: usize, +) -> Result, ConversationError> { + if removed.is_empty() { + return Ok(None); + } + let trust = if removed + .iter() + .any(|entry| entry.trust == MemoryTrust::Untrusted) + { + MemoryTrust::Untrusted + } else { + MemoryTrust::TrustedOutput + }; + let removed_bytes = removed.iter().map(StoredEntry::bytes).sum::(); + let target = maximum.min(removed_bytes.saturating_sub(1)); + if target < 16 { + return Ok(None); + } + let mut text = String::from("Earlier: "); + for entry in removed { + let label = match entry.role { + MessageRole::Avatar => "avatar", + MessageRole::Agent => "agent", + MessageRole::Tool => "tool", + MessageRole::System => "data", + }; + let fragment = format!("{label}: {}; ", entry.text.as_str()); + if text.len().saturating_add(fragment.len()) > target { + let remaining = target.saturating_sub(text.len()); + if remaining > 3 { + let mut fragment = fragment; + truncate_utf8(&mut fragment, remaining); + text.push_str(&fragment); + } + break; + } + text.push_str(&fragment); + } + truncate_utf8(&mut text, target.min(MAX_MESSAGE_BYTES)); + Ok(Some(StoredEntry { + sequence: removed.last().map_or(0, |entry| entry.sequence), + unix_millis: removed.last().map_or(0, |entry| entry.unix_millis), + role: MessageRole::Avatar, + kind: RecordKind::FactualSummary, + trust, + text: BoundedText::new("conversation.summary", text)?, + })) +} + +fn expire_idle_locked(state: &mut StoreState, now: Instant, only: Option) { + let expired = state + .sessions + .iter() + .filter(|(key, session)| { + only.is_none_or(|wanted| wanted == **key) + && now.saturating_duration_since(session.last_active) + >= key.channel.inactivity_ttl() + }) + .map(|(key, _)| *key) + .collect::>(); + for key in expired { + if let Some(session) = state.sessions.remove(&key) { + state.total_bytes = state.total_bytes.saturating_sub(session.bytes); + let reason = match key.channel { + ConversationChannel::PublicChat => MemoryReason::PublicExpired, + ConversationChannel::DirectIm => MemoryReason::DirectImExpired, + }; + push_event(&mut state.events, event_for(reason, &session)); + } + } +} + +fn evict_oldest(state: &mut StoreState, reason: MemoryReason, preserve: Option) { + let candidate = state + .sessions + .iter() + .filter(|(key, _)| preserve.is_none_or(|preserved| preserved != **key)) + .min_by_key(|(key, session)| (session.last_active, **key)) + .map(|(key, _)| *key) + .or_else(|| state.sessions.keys().next().copied()); + if let Some(key) = candidate + && let Some(session) = state.sessions.remove(&key) + { + state.total_bytes = state.total_bytes.saturating_sub(session.bytes); + push_event(&mut state.events, event_for(reason, &session)); + } +} + +fn event_for(reason: MemoryReason, session: &StoredSession) -> MemoryEvent { + MemoryEvent { + reason, + session_id: Some(session.id.clone()), + avatar_id: Some(session.key.avatar_id), + channel: Some(session.key.channel), + } +} + +fn push_event(events: &mut VecDeque, event: MemoryEvent) { + if events.len() == MAX_MEMORY_EVENTS { + events.pop_front(); + } + events.push_back(event); +} + +fn new_session_id() -> Result, ConversationError> { + let id = UUID::secure_random().map_err(|_| ConversationError::IdentifierGeneration)?; + BoundedText::new("conversation.session_id", id.to_string()).map_err(ConversationError::Boundary) +} + +fn unix_millis(time: SystemTime) -> u64 { + time.duration_since(UNIX_EPOCH) + .unwrap_or(Duration::ZERO) + .as_millis() + .try_into() + .unwrap_or(u64::MAX) +} + +fn truncate_utf8(text: &mut String, maximum: usize) { + if text.len() <= maximum { + return; + } + let mut boundary = maximum; + while !text.is_char_boundary(boundary) { + boundary = boundary.saturating_sub(1); + } + text.truncate(boundary); +} + +fn redact_sensitive(text: &str) -> String { + let mut redact_next = false; + text.split_whitespace() + .map(|token| { + if redact_next { + redact_next = false; + return "[REDACTED]".to_owned(); + } + let trimmed = token.trim_matches(|character: char| { + matches!(character, ',' | '.' | ';' | ')' | '(' | '[' | ']') + }); + let lower = trimmed.to_ascii_lowercase(); + if Url::parse(trimmed).is_ok_and(|url| matches!(url.scheme(), "http" | "https")) { + "[REDACTED URL]".to_owned() + } else if matches!( + lower.trim_end_matches(':'), + "bearer" | "authorization" | "api_key" | "apikey" | "password" | "token" + ) { + redact_next = true; + "[REDACTED]".to_owned() + } else if lower.contains("api_key=") + || lower.contains("apikey=") + || lower.contains("password=") + || lower.contains("token=") + || lower.contains("authorization:") + { + "[REDACTED]".to_owned() + } else { + token.to_owned() + } + }) + .collect::>() + .join(" ") +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Snapshot { + schema: u32, + generation: u64, + next_sequence: u64, + sessions: Vec, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct SnapshotSession { + session_id: String, + avatar_id: String, + channel: ConversationChannel, + created_unix_millis: u64, + last_active_unix_millis: u64, + entries: Vec, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct SnapshotEntry { + sequence: u64, + unix_millis: u64, + role: String, + kind: RecordKind, + trust: MemoryTrust, + text: String, +} + +fn prepare_directory( + directory: &Path, + _events: &mut VecDeque, +) -> Result<(), ConversationError> { + if directory.exists() { + let metadata = fs::symlink_metadata(directory) + .map_err(|_| ConversationError::PersistenceUnavailable)?; + if !metadata.is_dir() || metadata.file_type().is_symlink() { + return Err(ConversationError::PersistenceUnavailable); + } + } else { + fs::create_dir_all(directory).map_err(|_| ConversationError::PersistenceUnavailable)?; + } + restrict_directory(directory)?; + #[cfg(windows)] + push_event( + _events, + MemoryEvent { + reason: MemoryReason::PermissionsNotVerified, + session_id: None, + avatar_id: None, + channel: None, + }, + ); + Ok(()) +} + +#[cfg(unix)] +fn restrict_directory(directory: &Path) -> Result<(), ConversationError> { + use std::os::unix::fs::PermissionsExt as _; + fs::set_permissions(directory, fs::Permissions::from_mode(0o700)) + .map_err(|_| ConversationError::PersistenceUnavailable) +} + +#[cfg(not(unix))] +fn restrict_directory(_directory: &Path) -> Result<(), ConversationError> { + Ok(()) +} + +fn load_latest_snapshot( + limits: &ConversationLimits, + persistence: &ConversationPersistence, + clock: &dyn ConversationClock, + state: &mut StoreState, +) -> Result<(), ConversationError> { + let mut snapshots = snapshot_paths(&persistence.directory)?; + snapshots.sort_by_key(|item| std::cmp::Reverse(item.0)); + for (generation, path) in snapshots { + let metadata = + fs::symlink_metadata(&path).map_err(|_| ConversationError::PersistenceUnavailable)?; + if !metadata.is_file() + || metadata.file_type().is_symlink() + || metadata.len() > limits.max_persisted_bytes as u64 + { + quarantine(&path, MemoryReason::SnapshotTooLargeQuarantined, state); + continue; + } + let capacity = + usize::try_from(metadata.len()).map_err(|_| ConversationError::PersistenceTooLarge)?; + let mut bytes = Vec::with_capacity(capacity); + File::open(&path) + .and_then(|file| { + file.take(limits.max_persisted_bytes as u64 + 1) + .read_to_end(&mut bytes) + }) + .map_err(|_| ConversationError::PersistenceUnavailable)?; + let Ok(snapshot) = serde_json::from_slice::(&bytes) else { + quarantine(&path, MemoryReason::CorruptSnapshotQuarantined, state); + continue; + }; + if snapshot.schema != SNAPSHOT_SCHEMA || snapshot.generation != generation { + quarantine(&path, MemoryReason::UnsupportedSnapshotQuarantined, state); + continue; + } + if restore_snapshot(limits, snapshot, clock, state).is_ok() { + return Ok(()); + } + quarantine(&path, MemoryReason::CorruptSnapshotQuarantined, state); + } + Ok(()) +} + +#[allow(clippy::too_many_lines)] // One linear validator keeps every persisted field fail-closed. +fn restore_snapshot( + limits: &ConversationLimits, + snapshot: Snapshot, + clock: &dyn ConversationClock, + state: &mut StoreState, +) -> Result<(), ConversationError> { + let monotonic_now = clock.monotonic_now(); + let wall_now = unix_millis(clock.wall_now()); + let mut restored = StoreState { + sessions: BTreeMap::new(), + total_bytes: 0, + next_sequence: snapshot.next_sequence.max(1), + snapshot_generation: snapshot.generation, + events: VecDeque::new(), + }; + let mut restored_session_ids = BTreeSet::new(); + let mut restored_sequences = BTreeSet::new(); + let mut maximum_sequence = 0_u64; + if snapshot.sessions.len() > MAX_ACTIVE_SESSIONS { + return Err(ConversationError::PersistenceTooLarge); + } + for persisted in snapshot.sessions { + let avatar_id = UUID::new_with_string(persisted.avatar_id) + .map_err(|_| ConversationError::InvalidAvatar)?; + let key = ConversationKey::new(avatar_id, persisted.channel)?; + let session_id = BoundedText::new("conversation.session_id", persisted.session_id)?; + if restored.sessions.contains_key(&key) + || !restored_session_ids.insert(session_id.as_str().to_owned()) + || persisted.created_unix_millis > persisted.last_active_unix_millis + { + return Err(ConversationError::PersistenceUnavailable); + } + let elapsed_millis = wall_now.saturating_sub(persisted.last_active_unix_millis); + let elapsed = Duration::from_millis(elapsed_millis); + if elapsed >= key.channel.inactivity_ttl() { + continue; + } + let last_active = monotonic_now.checked_sub(elapsed).unwrap_or(monotonic_now); + let mut session = StoredSession { + id: session_id, + key, + created_unix_millis: persisted.created_unix_millis.min(wall_now), + last_active_unix_millis: persisted.last_active_unix_millis.min(wall_now), + last_active, + entries: VecDeque::new(), + bytes: 0, + tool_results: 0, + }; + if persisted.entries.len() > MAX_CONVERSATION_MESSAGES { + return Err(ConversationError::PersistenceTooLarge); + } + let mut previous_sequence = 0_u64; + for entry in persisted.entries { + if entry.sequence == 0 || !restored_sequences.insert(entry.sequence) { + return Err(ConversationError::PersistenceUnavailable); + } + if entry.sequence <= previous_sequence + || entry.unix_millis < persisted.created_unix_millis + || entry.unix_millis > persisted.last_active_unix_millis + { + return Err(ConversationError::PersistenceUnavailable); + } + previous_sequence = entry.sequence; + maximum_sequence = maximum_sequence.max(entry.sequence); + let role = match entry.role.as_str() { + "avatar" => MessageRole::Avatar, + "agent" => MessageRole::Agent, + "tool" => MessageRole::Tool, + _ => return Err(ConversationError::PersistenceUnavailable), + }; + let sanitized = redact_sensitive(&entry.text); + if sanitized != entry.text + || sanitized.is_empty() + || sanitized.len() > MAX_MESSAGE_BYTES + { + return Err(ConversationError::PersistenceUnavailable); + } + let valid_semantics = matches!( + (role, entry.kind, entry.trust), + ( + MessageRole::Avatar, + RecordKind::Message, + MemoryTrust::Untrusted + ) | ( + MessageRole::Agent, + RecordKind::Message, + MemoryTrust::TrustedOutput + ) | ( + MessageRole::Tool, + RecordKind::ToolSummary | RecordKind::ActionResult, + _ + ) | (MessageRole::Avatar, RecordKind::FactualSummary, _) + ); + if !valid_semantics { + return Err(ConversationError::PersistenceUnavailable); + } + let text = BoundedText::new("conversation.content", sanitized)?; + session.bytes = session.bytes.saturating_add(text.len()); + session.tool_results = session + .tool_results + .saturating_add(usize::from(entry.kind != RecordKind::Message)); + session.entries.push_back(StoredEntry { + sequence: entry.sequence, + unix_millis: entry.unix_millis.min(wall_now), + role, + kind: entry.kind, + trust: entry.trust, + text, + }); + } + restored.total_bytes = restored.total_bytes.saturating_add(session.bytes); + restored.sessions.insert(key, session); + compact_session(&mut restored, key, limits)?; + while restored.sessions.len() > limits.max_active_sessions { + evict_oldest(&mut restored, MemoryReason::SessionLimitEviction, None); + } + while restored.total_bytes > limits.max_total_bytes { + evict_oldest(&mut restored, MemoryReason::TotalByteEviction, None); + } + } + if restored.next_sequence <= maximum_sequence { + return Err(ConversationError::PersistenceUnavailable); + } + for event in state.events.drain(..) { + push_event(&mut restored.events, event); + } + *state = restored; + Ok(()) +} + +fn write_snapshot( + limits: &ConversationLimits, + persistence: &ConversationPersistence, + state: &mut StoreState, +) -> Result<(), ConversationError> { + let generation = state + .snapshot_generation + .checked_add(1) + .ok_or(ConversationError::IdentifierGeneration)?; + let snapshot = Snapshot { + schema: SNAPSHOT_SCHEMA, + generation, + next_sequence: state.next_sequence, + sessions: state + .sessions + .values() + .map(|session| SnapshotSession { + session_id: session.id.as_str().to_owned(), + avatar_id: session.key.avatar_id.to_string(), + channel: session.key.channel, + created_unix_millis: session.created_unix_millis, + last_active_unix_millis: session.last_active_unix_millis, + entries: session + .entries + .iter() + .map(|entry| SnapshotEntry { + sequence: entry.sequence, + unix_millis: entry.unix_millis, + role: match entry.role { + MessageRole::Avatar => "avatar", + MessageRole::Agent => "agent", + MessageRole::Tool => "tool", + MessageRole::System => "data", + } + .to_owned(), + kind: entry.kind, + trust: entry.trust, + text: entry.text.as_str().to_owned(), + }) + .collect(), + }) + .collect(), + }; + let bytes = + serde_json::to_vec(&snapshot).map_err(|_| ConversationError::PersistenceUnavailable)?; + if bytes.len() > limits.max_persisted_bytes { + return Err(ConversationError::PersistenceTooLarge); + } + let final_path = persistence + .directory + .join(format!("memory-{generation:020}.json")); + let temporary_path = persistence.directory.join(format!( + ".memory-{generation:020}-{}.tmp", + UUID::secure_random() + .map_err(|_| ConversationError::IdentifierGeneration)? + .to_string() + )); + let mut file = secure_new_file(&temporary_path)?; + let result = (|| { + file.write_all(&bytes)?; + file.sync_all()?; + fs::rename(&temporary_path, &final_path)?; + sync_directory(&persistence.directory)?; + Ok::<(), std::io::Error>(()) + })(); + if result.is_err() { + let _ = fs::remove_file(&temporary_path); + return Err(ConversationError::PersistenceUnavailable); + } + state.snapshot_generation = generation; + prune_snapshots(&persistence.directory, limits.max_persisted_bytes)?; + Ok(()) +} + +#[cfg(unix)] +fn secure_new_file(path: &Path) -> Result { + use std::os::unix::fs::OpenOptionsExt as _; + OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(path) + .map_err(|_| ConversationError::PersistenceUnavailable) +} + +#[cfg(not(unix))] +fn secure_new_file(path: &Path) -> Result { + OpenOptions::new() + .write(true) + .create_new(true) + .open(path) + .map_err(|_| ConversationError::PersistenceUnavailable) +} + +#[cfg(unix)] +fn sync_directory(directory: &Path) -> Result<(), std::io::Error> { + File::open(directory)?.sync_all() +} + +#[cfg(not(unix))] +fn sync_directory(_directory: &Path) -> Result<(), std::io::Error> { + Ok(()) +} + +fn snapshot_paths(directory: &Path) -> Result, ConversationError> { + let entries = fs::read_dir(directory).map_err(|_| ConversationError::PersistenceUnavailable)?; + Ok(entries + .filter_map(Result::ok) + .filter_map(|entry| { + let name = entry.file_name().to_string_lossy().into_owned(); + name.strip_prefix("memory-") + .and_then(|value| value.strip_suffix(".json")) + .and_then(|value| value.parse::().ok()) + .map(|generation| (generation, entry.path())) + }) + .collect()) +} + +fn prune_snapshots(directory: &Path, maximum_total: usize) -> Result<(), ConversationError> { + let mut snapshots = snapshot_paths(directory)?; + snapshots.sort_by_key(|item| std::cmp::Reverse(item.0)); + let mut retained_bytes = 0_usize; + for (index, (_, path)) in snapshots.into_iter().enumerate() { + let length = fs::metadata(&path) + .map_err(|_| ConversationError::PersistenceUnavailable)? + .len() + .try_into() + .unwrap_or(usize::MAX); + let retain = + index < RETAINED_SNAPSHOTS && retained_bytes.saturating_add(length) <= maximum_total; + if retain { + retained_bytes = retained_bytes.saturating_add(length); + } else { + fs::remove_file(path).map_err(|_| ConversationError::PersistenceUnavailable)?; + } + } + Ok(()) +} + +fn quarantine(path: &Path, reason: MemoryReason, state: &mut StoreState) { + let suffix = UUID::secure_random().map_or_else(|_| "unknown".to_owned(), |id| id.to_string()); + let name = path.file_name().map_or_else( + || "snapshot".into(), + |name| name.to_string_lossy().into_owned(), + ); + let target = path.with_file_name(format!("quarantine-{name}-{suffix}.corrupt")); + let _ = fs::rename(path, target); + push_event( + &mut state.events, + MemoryEvent { + reason, + session_id: None, + avatar_id: None, + channel: None, + }, + ); +} diff --git a/crates/metacrate-grid-agent/src/conversation_tests.rs b/crates/metacrate-grid-agent/src/conversation_tests.rs new file mode 100644 index 0000000..698638e --- /dev/null +++ b/crates/metacrate-grid-agent/src/conversation_tests.rs @@ -0,0 +1,555 @@ +use crate::conversation::*; +use crate::{ContentPart, MessageRole}; +use libremetaverse_types::UUID; +use std::fs; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +fn avatar(number: u64) -> UUID { + UUID::new_with_u_int64(number).expect("fixture UUID") +} + +fn key(number: u64, channel: ConversationChannel) -> ConversationKey { + ConversationKey::new(avatar(number), channel).expect("nonzero fixture") +} + +#[derive(Debug)] +struct FakeClock { + wall_millis: AtomicU64, +} + +impl FakeClock { + fn new(wall_millis: u64) -> Self { + Self { + wall_millis: AtomicU64::new(wall_millis), + } + } + + fn advance(&self, duration: Duration) { + self.wall_millis.fetch_add( + u64::try_from(duration.as_millis()).expect("fixture duration"), + Ordering::AcqRel, + ); + } + + fn rewind(&self, duration: Duration) { + self.wall_millis.fetch_sub( + u64::try_from(duration.as_millis()).expect("fixture duration"), + Ordering::AcqRel, + ); + } +} + +impl ConversationClock for FakeClock { + fn monotonic_now(&self) -> tokio::time::Instant { + tokio::time::Instant::now() + } + + fn wall_now(&self) -> SystemTime { + UNIX_EPOCH + Duration::from_millis(self.wall_millis.load(Ordering::Acquire)) + } +} + +fn limits() -> ConversationLimits { + ConversationLimits { + max_active_sessions: 256, + max_turns_per_session: 256, + max_session_bytes: 256 * 1024, + max_total_bytes: 1024 * 1024, + max_tool_results_per_session: 64, + max_tool_result_bytes: 16 * 1024, + max_persisted_bytes: 2 * 1024 * 1024, + max_summary_bytes: 4 * 1024, + } +} + +fn text_messages(context: &ConversationContext) -> Vec<(MessageRole, String)> { + context + .llm_messages() + .expect("valid LLM projection") + .into_iter() + .map(|message| { + let ContentPart::Text(text) = &message.content.as_slice()[0] else { + panic!("memory emits only text") + }; + (message.role, text.as_str().to_owned()) + }) + .collect() +} + +#[tokio::test(start_paused = true)] +async fn exact_public_and_im_expiry_boundaries_create_fresh_ids() { + let clock = Arc::new(FakeClock::new(1_800_000_000_000)); + let store = + ConversationStore::open_with_clock(limits(), None, clock.clone()).expect("bounded store"); + let public = key(1, ConversationChannel::PublicChat); + let direct = key(1, ConversationChannel::DirectIm); + let public_id = store + .append(public, MemoryRecord::avatar_message("public first")) + .expect("public append") + .session_id; + let direct_id = store + .append(direct, MemoryRecord::avatar_message("direct first")) + .expect("direct append") + .session_id; + + tokio::time::advance(Duration::from_mins(30) - Duration::from_millis(1)).await; + clock.advance(Duration::from_mins(30) - Duration::from_millis(1)); + assert_eq!( + store.context(public).expect("not expired").session_id, + public_id + ); + tokio::time::advance(Duration::from_millis(1)).await; + clock.advance(Duration::from_millis(1)); + let replacement = store + .append(public, MemoryRecord::avatar_message("public replacement")) + .expect("fresh public session"); + assert_ne!(replacement.session_id, public_id); + assert_eq!( + text_messages(&store.context(public).expect("new context")).len(), + 1 + ); + assert_eq!( + store.context(direct).expect("direct remains").session_id, + direct_id + ); + + tokio::time::advance(Duration::from_hours(23) + Duration::from_mins(30)).await; + clock.advance(Duration::from_hours(23) + Duration::from_mins(30)); + let replacement = store + .append(direct, MemoryRecord::avatar_message("direct replacement")) + .expect("fresh direct session"); + assert_ne!(replacement.session_id, direct_id); + assert_eq!( + text_messages(&store.context(direct).expect("new context")).len(), + 1 + ); +} + +#[test] +fn avatar_and_channel_histories_are_strictly_separate_and_redacted() { + let store = ConversationStore::open(limits(), None).expect("bounded store"); + let alice_public = key(10, ConversationChannel::PublicChat); + let alice_im = key(10, ConversationChannel::DirectIm); + let bob_public = key(11, ConversationChannel::PublicChat); + store + .append( + alice_public, + MemoryRecord::avatar_message("alice-public https://grid.test/CAPS/secret?token=x"), + ) + .expect("alice public"); + store + .append( + alice_im, + MemoryRecord::avatar_message("alice-im password=hunter2 Bearer swordfish"), + ) + .expect("alice im"); + store + .append(bob_public, MemoryRecord::avatar_message("bob-public")) + .expect("bob public"); + + let public = text_messages(&store.context(alice_public).expect("alice public context")); + assert_eq!( + public, + vec![(MessageRole::Avatar, "alice-public [REDACTED URL]".into())] + ); + let direct = text_messages(&store.context(alice_im).expect("alice direct context")); + assert_eq!( + direct, + vec![( + MessageRole::Avatar, + "alice-im [REDACTED] [REDACTED] [REDACTED]".into() + )] + ); + let bob = text_messages(&store.context(bob_public).expect("bob context")); + assert_eq!(bob, vec![(MessageRole::Avatar, "bob-public".into())]); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn concurrent_appends_receive_one_total_order() { + let store = Arc::new(ConversationStore::open(limits(), None).expect("bounded store")); + let session_key = key(20, ConversationChannel::DirectIm); + let mut tasks = Vec::new(); + for number in 0..128 { + let store = Arc::clone(&store); + tasks.push(tokio::spawn(async move { + let text = format!("message-{number}"); + let sequence = store + .append(session_key, MemoryRecord::avatar_message(text.clone())) + .expect("concurrent append") + .sequence; + (sequence, text) + })); + } + let mut ordered_messages = std::collections::BTreeMap::new(); + for task in tasks { + let (sequence, text) = task.await.expect("append task"); + ordered_messages.insert(sequence, text); + } + assert_eq!( + ordered_messages.keys().copied().collect::>(), + (1..=128).collect::>() + ); + assert_eq!( + store.context(session_key).expect("context").turn_count(), + 128 + ); + let context_messages = text_messages(&store.context(session_key).expect("ordered context")); + assert_eq!(context_messages.len(), 128); + assert_eq!( + context_messages + .iter() + .map(|(_, text)| text) + .collect::>(), + ordered_messages.values().collect::>() + ); +} + +#[test] +fn compaction_eviction_and_thousands_of_senders_stay_bounded() { + let bounded = ConversationLimits { + max_active_sessions: 32, + max_turns_per_session: 8, + max_session_bytes: 2048, + max_total_bytes: 32 * 2048, + max_tool_results_per_session: 2, + max_tool_result_bytes: 512, + max_persisted_bytes: 128 * 1024, + max_summary_bytes: 256, + }; + let store = ConversationStore::open(bounded.clone(), None).expect("bounded store"); + let first = key(1, ConversationChannel::PublicChat); + for index in 0..40 { + let record = if index % 2 == 0 { + MemoryRecord::avatar_message(format!("untrusted instruction {index}")) + } else { + MemoryRecord::tool_summary(format!("tool result {index}"), MemoryTrust::Untrusted) + }; + store.append(first, record).expect("compacted append"); + } + let messages = text_messages(&store.context(first).expect("compacted context")); + assert!(messages.len() <= bounded.max_turns_per_session); + assert!(messages.iter().any(|(role, text)| { + *role == MessageRole::Avatar && text.starts_with("[untrusted historical summary]") + })); + + for sender in 2..=5_000 { + store + .append( + key(sender, ConversationChannel::PublicChat), + MemoryRecord::avatar_message("bounded sender"), + ) + .expect("bounded sender append"); + } + let metadata = store.list_metadata(); + assert!(metadata.len() <= bounded.max_active_sessions); + assert!(metadata.iter().map(|item| item.bytes).sum::() <= bounded.max_total_bytes); + assert!(store.drain_events().len() <= 8_192); +} + +fn temporary_directory() -> PathBuf { + std::env::temp_dir().join(format!( + "metacrate-conversation-{}", + UUID::secure_random().expect("temporary UUID").to_string() + )) +} + +#[tokio::test(start_paused = true)] +async fn persistence_recovers_ids_handles_clock_jumps_and_expires_by_wall_time() { + let directory = temporary_directory(); + let persistence = ConversationPersistence { + directory: directory.clone(), + }; + let clock = Arc::new(FakeClock::new(1_900_000_000_000)); + let public = key(30, ConversationChannel::PublicChat); + let direct = key(30, ConversationChannel::DirectIm); + let store = + ConversationStore::open_with_clock(limits(), Some(persistence.clone()), clock.clone()) + .expect("persistent store"); + let public_id = store + .append(public, MemoryRecord::avatar_message("persist public")) + .expect("public") + .session_id; + let direct_id = store + .append(direct, MemoryRecord::avatar_message("persist direct")) + .expect("direct") + .session_id; + store.flush().expect("atomic snapshot"); + drop(store); + + clock.rewind(Duration::from_hours(1)); + let recovered = + ConversationStore::open_with_clock(limits(), Some(persistence.clone()), clock.clone()) + .expect("backward wall jump is safe"); + assert_eq!( + recovered + .context(public) + .expect("public recovered") + .session_id, + public_id + ); + assert_eq!( + recovered + .context(direct) + .expect("direct recovered") + .session_id, + direct_id + ); + recovered + .append( + direct, + MemoryRecord::agent_visible_response("response during backward wall jump"), + ) + .expect("monotonic append across wall jump"); + recovered.flush().expect("consistent jumped-clock snapshot"); + drop(recovered); + + clock.advance(Duration::from_hours(2)); + let recovered = ConversationStore::open_with_clock(limits(), Some(persistence), clock) + .expect("restart after elapsed wall time"); + assert!(recovered.context(public).is_none()); + assert_eq!( + recovered + .context(direct) + .expect("direct remains") + .session_id, + direct_id + ); + drop(recovered); + fs::remove_dir_all(directory).expect("remove scoped temporary directory"); +} + +#[test] +fn corrupt_newest_snapshot_is_quarantined_and_older_snapshot_recovers() { + let directory = temporary_directory(); + let persistence = ConversationPersistence { + directory: directory.clone(), + }; + let session_key = key(40, ConversationChannel::DirectIm); + let store = ConversationStore::open(limits(), Some(persistence.clone())).expect("store"); + let session_id = store + .append( + session_key, + MemoryRecord::avatar_message("first durable turn"), + ) + .expect("append") + .session_id; + store.flush().expect("first snapshot"); + store + .append( + session_key, + MemoryRecord::agent_visible_response("second durable turn"), + ) + .expect("second append"); + store.flush().expect("second snapshot"); + drop(store); + let newest = fs::read_dir(&directory) + .expect("snapshot directory") + .filter_map(Result::ok) + .map(|entry| entry.path()) + .filter(|path| { + path.extension() + .is_some_and(|extension| extension == "json") + }) + .max() + .expect("newest snapshot"); + fs::write(&newest, b"{truncated").expect("corrupt fixture"); + + let recovered = ConversationStore::open(limits(), Some(persistence)).expect("safe recovery"); + assert_eq!( + recovered + .context(session_key) + .expect("older snapshot") + .session_id, + session_id + ); + assert!( + recovered + .drain_events() + .iter() + .any(|event| event.reason == MemoryReason::CorruptSnapshotQuarantined) + ); + assert!( + fs::read_dir(&directory) + .expect("quarantine directory") + .filter_map(Result::ok) + .any(|entry| entry + .path() + .extension() + .is_some_and(|extension| extension == "corrupt")) + ); + drop(recovered); + fs::remove_dir_all(directory).expect("remove scoped temporary directory"); +} + +#[test] +fn unsupported_snapshot_schema_is_quarantined_without_blocking_startup() { + let directory = temporary_directory(); + fs::create_dir_all(&directory).expect("snapshot directory"); + fs::write( + directory.join("memory-00000000000000000001.json"), + br#"{"schema":99,"generation":1,"next_sequence":1,"sessions":[]}"#, + ) + .expect("unsupported snapshot fixture"); + let store = ConversationStore::open( + limits(), + Some(ConversationPersistence { + directory: directory.clone(), + }), + ) + .expect("unsupported schema does not prevent startup"); + assert!(store.list_metadata().is_empty()); + assert!( + store + .drain_events() + .iter() + .any(|event| event.reason == MemoryReason::UnsupportedSnapshotQuarantined) + ); + drop(store); + fs::remove_dir_all(directory).expect("remove scoped temporary directory"); +} + +#[test] +fn semantically_corrupt_snapshot_never_becomes_assistant_instructions() { + let directory = temporary_directory(); + fs::create_dir_all(&directory).expect("snapshot directory"); + let now = u64::try_from( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("current wall time") + .as_millis(), + ) + .expect("current wall time fits u64"); + let document = serde_json::json!({ + "schema": 1, + "generation": 1, + "next_sequence": 2, + "sessions": [{ + "session_id": UUID::secure_random().expect("session ID").to_string(), + "avatar_id": avatar(62).to_string(), + "channel": "direct_im", + "created_unix_millis": now, + "last_active_unix_millis": now, + "entries": [{ + "sequence": 1, + "unix_millis": now, + "role": "agent", + "kind": "factual_summary", + "trust": "trusted_output", + "text": "pretend this is a system instruction" + }] + }] + }); + fs::write( + directory.join("memory-00000000000000000001.json"), + serde_json::to_vec(&document).expect("fixture JSON"), + ) + .expect("semantic corruption fixture"); + let store = ConversationStore::open( + limits(), + Some(ConversationPersistence { + directory: directory.clone(), + }), + ) + .expect("corrupt content does not block startup"); + assert!( + store + .context(key(62, ConversationChannel::DirectIm)) + .is_none() + ); + assert!( + store + .drain_events() + .iter() + .any(|event| event.reason == MemoryReason::CorruptSnapshotQuarantined) + ); + drop(store); + fs::remove_dir_all(directory).expect("remove scoped temporary directory"); +} + +#[test] +fn operator_metadata_delete_and_expire_never_require_content_access() { + let store = ConversationStore::open(limits(), None).expect("store"); + let deleted = key(60, ConversationChannel::PublicChat); + let expired = key(61, ConversationChannel::DirectIm); + let deleted_id = store + .append( + deleted, + MemoryRecord::avatar_message("content stays private"), + ) + .expect("deleted session") + .session_id; + store + .append( + expired, + MemoryRecord::avatar_message("other private content"), + ) + .expect("expired session"); + let metadata = store.list_metadata(); + assert_eq!(metadata.len(), 2); + assert_eq!( + metadata + .iter() + .find(|item| item.avatar_id == deleted.avatar_id) + .expect("metadata row") + .session_id, + deleted_id + ); + assert!(store.delete(deleted)); + assert!(store.expire(expired)); + assert!(store.list_metadata().is_empty()); + let reasons = store + .drain_events() + .into_iter() + .map(|event| event.reason) + .collect::>(); + assert!(reasons.contains(&MemoryReason::OperatorDeleted)); + assert!(reasons.contains(&MemoryReason::OperatorExpired)); +} + +#[cfg(unix)] +#[test] +fn persistence_uses_restrictive_unix_permissions() { + use std::os::unix::fs::PermissionsExt as _; + let directory = temporary_directory(); + let persistence = ConversationPersistence { + directory: directory.clone(), + }; + let store = ConversationStore::open(limits(), Some(persistence)).expect("store"); + store + .append( + key(50, ConversationChannel::DirectIm), + MemoryRecord::avatar_message("private"), + ) + .expect("append"); + store.flush().expect("snapshot"); + assert_eq!( + fs::metadata(&directory) + .expect("directory metadata") + .permissions() + .mode() + & 0o777, + 0o700 + ); + let snapshot = fs::read_dir(&directory) + .expect("snapshot directory") + .filter_map(Result::ok) + .map(|entry| entry.path()) + .find(|path| { + path.extension() + .is_some_and(|extension| extension == "json") + }) + .expect("snapshot"); + assert_eq!( + fs::metadata(snapshot) + .expect("file metadata") + .permissions() + .mode() + & 0o777, + 0o600 + ); + drop(store); + fs::remove_dir_all(directory).expect("remove scoped temporary directory"); +} diff --git a/crates/metacrate-grid-agent/src/lib.rs b/crates/metacrate-grid-agent/src/lib.rs index 2fbc68d..901a56d 100644 --- a/crates/metacrate-grid-agent/src/lib.rs +++ b/crates/metacrate-grid-agent/src/lib.rs @@ -6,6 +6,7 @@ pub mod backend; pub mod config; +pub mod conversation; pub mod llm; pub mod policy; pub mod service; @@ -13,6 +14,8 @@ pub mod session; pub mod tool_loop; pub mod types; +#[cfg(test)] +mod conversation_tests; #[cfg(test)] mod policy_tests; #[cfg(test)] @@ -25,9 +28,15 @@ pub use backend::{ #[cfg(feature = "live-grid")] pub use backend::{LibremetaverseClientOwner, LibremetaverseSessionBackend}; pub use config::{ - AgentConfig, BehaviorSettings, ConfigError, ConfigLoader, EndpointUrl, Environment, - GridConnection, Limits, LlmConnection, MapEnvironment, OperatingMode, SecretString, - StdEnvironment, Timeouts, + AgentConfig, BehaviorSettings, ConfigError, ConfigLoader, ConversationSettings, EndpointUrl, + Environment, GridConnection, Limits, LlmConnection, MapEnvironment, OperatingMode, + SecretString, StdEnvironment, Timeouts, +}; +pub use conversation::{ + ConversationChannel, ConversationClock, ConversationContext, ConversationError, + ConversationKey, ConversationLimits, ConversationMetadata, ConversationPersistence, + ConversationStore, MemoryEvent, MemoryReason, MemoryRecord, MemoryTrust, MemoryUpdate, + SystemConversationClock, }; pub use llm::{ Completion, CompletionMessage, ContentPart, ImageDetail, LlmClient, LlmError, diff --git a/crates/metacrate-grid-agent/tests/conversation_memory.rs b/crates/metacrate-grid-agent/tests/conversation_memory.rs new file mode 100644 index 0000000..996e084 --- /dev/null +++ b/crates/metacrate-grid-agent/tests/conversation_memory.rs @@ -0,0 +1,149 @@ +use libremetaverse_types::UUID; +use libremetaverse_types::compat::CancellationTokenSource; +use metacrate_grid_agent::{ + AgentConfig, ConversationChannel, ConversationKey, ConversationLimits, ConversationStore, + LlmClient, LlmTransportLimits, MemoryRecord, +}; +use serde_json::{Value, json}; +use std::sync::Arc; +use std::time::Duration; +use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; +use tokio::net::TcpListener; +use tokio::sync::mpsc; + +const MAX_REQUEST_BYTES: usize = 1024 * 1024; + +fn avatar(number: u64) -> UUID { + UUID::new_with_u_int64(number).expect("fixture UUID") +} + +fn key(number: u64, channel: ConversationChannel) -> ConversationKey { + ConversationKey::new(avatar(number), channel).expect("nonzero fixture") +} + +async fn capture_server( + request_count: usize, +) -> (String, mpsc::Receiver, tokio::task::JoinHandle<()>) { + let listener = TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)) + .await + .expect("bind fake LLM"); + let address = listener.local_addr().expect("listener address"); + let (sender, receiver) = mpsc::channel(request_count); + let task = tokio::spawn(async move { + for _ in 0..request_count { + let (mut stream, _) = listener.accept().await.expect("LLM connection"); + let request = read_request(&mut stream).await; + sender.send(request).await.expect("capture receiver"); + let response = serde_json::to_vec(&json!({ + "choices": [{"message": {"content": "ok", "tool_calls": []}}], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2} + })) + .expect("response JSON"); + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + response.len() + ); + stream + .write_all(head.as_bytes()) + .await + .expect("response head"); + stream.write_all(&response).await.expect("response body"); + stream.shutdown().await.expect("response shutdown"); + } + }); + (format!("http://{address}/exact/chat"), receiver, task) +} + +async fn read_request(stream: &mut tokio::net::TcpStream) -> Value { + let mut bytes = Vec::new(); + let header_end = loop { + assert!(bytes.len() < MAX_REQUEST_BYTES, "bounded request headers"); + let mut chunk = [0_u8; 2048]; + let count = stream.read(&mut chunk).await.expect("request read"); + assert!(count > 0, "complete request headers"); + bytes.extend_from_slice(&chunk[..count]); + if let Some(position) = bytes.windows(4).position(|window| window == b"\r\n\r\n") { + break position + 4; + } + }; + let headers = std::str::from_utf8(&bytes[..header_end]).expect("UTF-8 headers"); + let content_length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().expect("content length")) + }) + .expect("content-length header"); + assert!(content_length <= MAX_REQUEST_BYTES, "bounded request body"); + while bytes.len() - header_end < content_length { + let mut chunk = [0_u8; 4096]; + let count = stream.read(&mut chunk).await.expect("request body read"); + assert!(count > 0, "complete request body"); + bytes.extend_from_slice(&chunk[..count]); + } + serde_json::from_slice(&bytes[header_end..header_end + content_length]).expect("request JSON") +} + +#[tokio::test] +async fn actual_llm_requests_never_cross_avatar_or_channel_boundaries() { + let (endpoint, mut requests, server) = capture_server(3).await; + let config = AgentConfig::offline(&endpoint, "test-key").expect("offline config"); + let limits = LlmTransportLimits { + connect_timeout: Duration::from_secs(1), + request_timeout: Duration::from_secs(2), + read_idle_timeout: Duration::from_secs(1), + pool_idle_timeout: Duration::from_secs(1), + total_timeout: Duration::from_secs(3), + max_prompt_bytes: 64 * 1024, + max_response_bytes: 64 * 1024, + max_concurrent_requests: 2, + max_retries: 0, + max_retry_delay: Duration::from_millis(10), + }; + let client = Arc::new(LlmClient::new(config.llm, limits).expect("LLM client")); + let store = ConversationStore::open(ConversationLimits::default(), None).expect("store"); + let alice_public = key(1, ConversationChannel::PublicChat); + let alice_direct = key(1, ConversationChannel::DirectIm); + let bob_public = key(2, ConversationChannel::PublicChat); + store + .append( + alice_public, + MemoryRecord::avatar_message("alice-public-only"), + ) + .expect("alice public"); + store + .append( + alice_direct, + MemoryRecord::avatar_message("alice-direct-only"), + ) + .expect("alice direct"); + store + .append(bob_public, MemoryRecord::avatar_message("bob-public-only")) + .expect("bob public"); + + for session_key in [alice_public, alice_direct, bob_public] { + let messages = store + .context(session_key) + .expect("isolated context") + .llm_messages() + .expect("LLM messages"); + client + .complete(&messages, &[], &CancellationTokenSource::new().token()) + .await + .expect("fake completion"); + } + + let expected = ["alice-public-only", "alice-direct-only", "bob-public-only"]; + for own_text in expected { + let request = requests.recv().await.expect("captured request"); + let serialized = serde_json::to_string(&request).expect("request string"); + assert!(serialized.contains(own_text)); + for other_text in expected { + if other_text != own_text { + assert!(!serialized.contains(other_text)); + } + } + } + server.await.expect("capture server"); +} diff --git a/crates/metacrate-grid-agent/tests/dependency_policy.rs b/crates/metacrate-grid-agent/tests/dependency_policy.rs index 0e13be1..91a6905 100644 --- a/crates/metacrate-grid-agent/tests/dependency_policy.rs +++ b/crates/metacrate-grid-agent/tests/dependency_policy.rs @@ -43,10 +43,10 @@ fn package_has_only_reviewed_rust_dependencies_and_no_build_script() { #[test] fn runtime_source_has_no_subprocess_or_native_abi_escape_hatch() { let source = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("src"); - let mut files = Vec::with_capacity(8); + let mut files = Vec::with_capacity(16); collect_rust_files(&source, &mut files); assert!( - files.len() <= 12, + files.len() <= 14, "source-file count needs a reviewed bound update" ); for path in files { diff --git a/docs/grid-agent-architecture.md b/docs/grid-agent-architecture.md index f381f82..413571d 100644 --- a/docs/grid-agent-architecture.md +++ b/docs/grid-agent-architecture.md @@ -27,6 +27,7 @@ observable receivers. | Offline session work | live supervisor | 1,024 hard / `reconnect.offline_work_capacity` | read-only queue; mutations rejected while not ready | | Body / message | typed boundary owners | 8 MiB / 64 KiB hard ceilings, with lower configured limits | rejected before enqueue | | Conversation / tool calls | request owner | 256 messages / 64 calls, with lower configured limits | rejected before request | +| Per-avatar conversation memory | `ConversationStore` mutex | 4,096 sessions / 64 MiB hard, lower `conversation` limits | monotonic expiry, deterministic compaction/LRU eviction | | LLM request slots | shared `LlmClient` semaphore | 256 hard / configured concurrent requests | async acquire or cancellation | | Reasoning/tool session | `ToolLoop` caller | 32 turns / 256 calls hard, with lower configured limits | total timeout, cancellation, or supersession | | Policy tools / approvals / schedules | `PolicyGateway` mutex | 64 tools / 4,096 approval records / 1,024 scheduler grants hard | deny before opaque authorization | @@ -84,6 +85,10 @@ cleanup, fencing old events and late LLM/tool results. See are refused, response bodies are bounded while streaming, bearer secrets are redacted, and provider/model discovery does not exist. Proposed calls cross `ToolExecutor` only after registered-name and schema validation. +- Conversation context is keyed by immutable avatar UUID and either public chat + or direct IM. Group channels are not representable. The LLM projection can + retrieve only one exact key, and recovered/untrusted summaries remain user-role + prompt data rather than system authority. - Signals and console output belong to the binary. The reusable core relies on no terminal, Unix socket, Unix signal, separator, or fixed platform path. @@ -116,3 +121,5 @@ The origin/capability matrix and opaque mutation boundary are documented in [`grid-agent-policy.md`](grid-agent-policy.md). The live lifecycle and generation contract is documented in [`grid-agent-session.md`](grid-agent-session.md). +The conversation isolation and persistence contract is documented in +[`grid-agent-conversation.md`](grid-agent-conversation.md). diff --git a/docs/grid-agent-conversation.md b/docs/grid-agent-conversation.md new file mode 100644 index 0000000..f6757a1 --- /dev/null +++ b/docs/grid-agent-conversation.md @@ -0,0 +1,55 @@ +# Grid-agent conversation memory + +`ConversationStore` owns bounded conversational context independently of the +grid connection generation. Its key is the immutable avatar UUID plus exactly +one channel kind: public chat or direct IM. Group and conference conversations +are deliberately not representable. Public sessions expire at exactly 30 +minutes of monotonic inactivity and direct IM sessions at exactly 24 hours. A +subsequent turn creates a cryptographically random new session ID and receives +none of the expired transcript. + +The mutex-protected store assigns one sequence order to concurrent turns. It +stores wall-clock timestamps, normalized avatar/agent/tool roles, visible agent +responses, bounded tool summaries, and externally meaningful action results. +Avatar input and untrusted tool data remain untrusted when compacted. There is +no system-message, model-scratchpad, hidden-reasoning, credential, capability +URL, or binary-asset input variant. Text is bounded and sensitive URL/token +forms are redacted before allocation in the store. + +Configured limits lower hard ceilings for active sessions, turns, per-session +bytes, aggregate bytes, tool-result count and size, summary bytes, and total +snapshot storage. Old records compact deterministically into a smaller factual +summary followed by recent context. Session and aggregate pressure evict the +least-recently-active key with UUID/channel tie-breaking. Expiry, compaction, +eviction, quarantine, and operator deletion publish stable reason codes through +a bounded event queue. + +`list_metadata` returns session ID, UUID, channel, timestamps, turn count, and +byte count but never content. Operators can delete or expire an exact key. +`context` is the separate LLM-facing projection and can read only that key; +untrusted summaries are emitted as explicitly marked user-role data. + +Persistence is local and opt-in through `conversation.persistence_enabled`. +`ConversationStore::from_config` uses `storage_path/conversations`; callers +flush at their durability boundary. A flush writes and syncs a new immutable, +versioned generation before an atomic rename, then retains only generations +whose aggregate bytes fit the configured storage ceiling. Linux and other Unix +targets force directory mode 0700 and file mode 0600. Rust standard library +does not expose a portable Windows ACL editor, so Windows emits the explicit +`PermissionsNotVerified` event and operators must restrict the directory ACL to +the service identity. + +Restart recovery validates schema, UUIDs, IDs, unique sequences, timestamps, +roles, bounds, and redaction before any record can become LLM context. A +truncated, corrupt, oversized, or unsupported generation is quarantined and an +older valid generation is tried. If the wall clock moved backwards, recovered +age is zero; forward elapsed time is applied to the channel TTL. Neither case +can grant policy authority or prevent startup. + +Focused gates: + +```sh +cargo test --locked -p metacrate-grid-agent --lib conversation_tests +cargo test --locked -p metacrate-grid-agent --test conversation_memory +cargo clippy --locked -p metacrate-grid-agent --all-targets -- -D warnings +```