1296 lines
44 KiB
Rust
1296 lines
44 KiB
Rust
//! 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<Self, ConversationError> {
|
|
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<String>) -> 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<String>) -> Self {
|
|
Self {
|
|
role: MessageRole::Agent,
|
|
kind: RecordKind::Message,
|
|
trust: MemoryTrust::TrustedOutput,
|
|
text: text.into(),
|
|
}
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn tool_summary(text: impl Into<String>, trust: MemoryTrust) -> Self {
|
|
Self {
|
|
role: MessageRole::Tool,
|
|
kind: RecordKind::ToolSummary,
|
|
trust,
|
|
text: text.into(),
|
|
}
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn action_result(text: impl Into<String>, 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<BoundedText<MAX_IDENTIFIER_BYTES>>,
|
|
pub avatar_id: Option<UUID>,
|
|
pub channel: Option<ConversationChannel>,
|
|
}
|
|
|
|
#[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<crate::types::BoundaryError> for ConversationError {
|
|
fn from(value: crate::types::BoundaryError) -> Self {
|
|
Self::Boundary(value)
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct MemoryUpdate {
|
|
pub session_id: BoundedText<MAX_IDENTIFIER_BYTES>,
|
|
pub sequence: u64,
|
|
pub events: Vec<MemoryEvent>,
|
|
}
|
|
|
|
/// Safe operator projection. Message content is intentionally absent.
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct ConversationMetadata {
|
|
pub session_id: BoundedText<MAX_IDENTIFIER_BYTES>,
|
|
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<MAX_IDENTIFIER_BYTES>,
|
|
entries: Vec<ContextEntry>,
|
|
}
|
|
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
struct ContextEntry {
|
|
role: MessageRole,
|
|
kind: RecordKind,
|
|
trust: MemoryTrust,
|
|
text: String,
|
|
}
|
|
|
|
impl ConversationContext {
|
|
pub fn llm_messages(&self) -> Result<Vec<CompletionMessage>, 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<MAX_MESSAGE_BYTES>,
|
|
}
|
|
|
|
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<MAX_IDENTIFIER_BYTES>,
|
|
key: ConversationKey,
|
|
created_unix_millis: u64,
|
|
last_active_unix_millis: u64,
|
|
last_active: Instant,
|
|
entries: VecDeque<StoredEntry>,
|
|
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<ConversationKey, StoredSession>,
|
|
total_bytes: usize,
|
|
next_sequence: u64,
|
|
snapshot_generation: u64,
|
|
events: VecDeque<MemoryEvent>,
|
|
}
|
|
|
|
/// 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<ConversationPersistence>,
|
|
clock: Arc<dyn ConversationClock>,
|
|
state: Mutex<StoreState>,
|
|
}
|
|
|
|
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<Self, ConversationError> {
|
|
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<ConversationPersistence>,
|
|
) -> Result<Self, ConversationError> {
|
|
Self::open_with_clock(limits, persistence, Arc::new(SystemConversationClock))
|
|
}
|
|
|
|
pub fn open_with_clock(
|
|
limits: ConversationLimits,
|
|
persistence: Option<ConversationPersistence>,
|
|
clock: Arc<dyn ConversationClock>,
|
|
) -> Result<Self, ConversationError> {
|
|
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<MemoryUpdate, ConversationError> {
|
|
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<ConversationContext> {
|
|
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<ConversationMetadata> {
|
|
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<MemoryEvent> {
|
|
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<MemoryEvent> {
|
|
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::<usize>();
|
|
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<Option<StoredEntry>, 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::<usize>();
|
|
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<ConversationKey>) {
|
|
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::<Vec<_>>();
|
|
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<ConversationKey>) {
|
|
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<MemoryEvent>, event: MemoryEvent) {
|
|
if events.len() == MAX_MEMORY_EVENTS {
|
|
events.pop_front();
|
|
}
|
|
events.push_back(event);
|
|
}
|
|
|
|
fn new_session_id() -> Result<BoundedText<MAX_IDENTIFIER_BYTES>, 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::<Vec<_>>()
|
|
.join(" ")
|
|
}
|
|
|
|
#[derive(Serialize, Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct Snapshot {
|
|
schema: u32,
|
|
generation: u64,
|
|
next_sequence: u64,
|
|
sessions: Vec<SnapshotSession>,
|
|
}
|
|
|
|
#[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<SnapshotEntry>,
|
|
}
|
|
|
|
#[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<MemoryEvent>,
|
|
) -> 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::<Snapshot>(&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<File, ConversationError> {
|
|
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<File, ConversationError> {
|
|
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<Vec<(u64, PathBuf)>, 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::<u64>().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,
|
|
},
|
|
);
|
|
}
|