From 096911a15680c0275a065edc7077d094b57dbad0 Mon Sep 17 00:00:00 2001 From: Chili Palmer Date: Sun, 23 Aug 2026 19:49:42 +0200 Subject: [PATCH] Implement conversation browser and safe deletion (#138) --- .../metacrate-grid-agent/src/control_plane.rs | 100 +++++- .../src/control_runtime.rs | 220 ++++++++++++- .../src/control_runtime_tests.rs | 147 ++++++++- .../metacrate-grid-agent/src/conversation.rs | 292 ++++++++++++++++- .../src/conversation_tests.rs | 73 +++++ .../metacrate-grid-agent/src/interaction.rs | 165 +++++++++- crates/metacrate-grid-agent/src/lib.rs | 32 +- crates/metacrate-grid-agent/src/main.rs | 4 + .../metacrate-grid-agent/src/memory_admin.rs | 182 ++++++++++- crates/metacrate-grid-agent/src/tui.rs | 304 ++++++++++++++++-- vendor/mentra/src/agent.rs | 8 + vendor/mentra/src/memory/journal/ops.rs | 9 + vendor/mentra/src/memory/journal/tests.rs | 37 +++ 13 files changed, 1513 insertions(+), 60 deletions(-) diff --git a/crates/metacrate-grid-agent/src/control_plane.rs b/crates/metacrate-grid-agent/src/control_plane.rs index 80d4e79..53817c5 100644 --- a/crates/metacrate-grid-agent/src/control_plane.rs +++ b/crates/metacrate-grid-agent/src/control_plane.rs @@ -172,7 +172,7 @@ impl PageRequest { } } -#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] pub struct Page { pub items: Vec, pub next_cursor: Option, @@ -220,6 +220,66 @@ pub struct SessionMetadataView { pub bytes: usize, } +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ConversationStatusView { + Active, + Expired, +} + +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ConversationFilterView { + pub avatar: Option, + pub channel: Option, + pub status: Option, + pub active_from_unix_millis: Option, + pub active_to_unix_millis: Option, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct ConversationMetadataView { + pub session_id: String, + pub avatar_id: String, + pub channel: ConversationChannelView, + pub status: ConversationStatusView, + pub created_unix_millis: u64, + pub last_active_unix_millis: u64, + pub turns: usize, + pub bytes: usize, + pub compacted: bool, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct ConversationEntryView { + pub sequence: u64, + pub unix_millis: u64, + pub role: String, + pub record_kind: String, + pub trust: String, + pub text: String, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct TranscriptEntryView { + pub entry_id: String, + pub parent_id: Option, + pub kind: String, + pub active: bool, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct ConversationDetailView { + pub conversation: ConversationMetadataView, + pub entries: Page, + pub mentra_agent_id: Option, + pub profile_generation: Option, + pub active_transcript_entries: usize, + pub archived_transcript_entries: usize, + pub transcript_entries: Vec, + pub transcript_busy: bool, +} + #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] pub struct ScheduledJobView { pub job_id: String, @@ -331,6 +391,8 @@ pub enum ControlPayload { Metrics(MetricsSnapshot), Runtime(RuntimeView), Sessions(Page), + Conversations(Page), + ConversationDetail(ConversationDetailView), ScheduledJobs(Page), PendingApprovals(Page), AuditEvents(Page), @@ -361,6 +423,17 @@ pub enum ControlRequest { ListSessions { page: PageRequest, }, + ListConversations { + page: PageRequest, + filter: ConversationFilterView, + }, + GetConversation { + session_id: String, + page: PageRequest, + }, + DeleteConversation { + session_id: String, + }, ListScheduledJobs { page: PageRequest, }, @@ -449,6 +522,8 @@ impl ControlRequest { | Self::Metrics | Self::Runtime | Self::ListSessions { .. } + | Self::ListConversations { .. } + | Self::GetConversation { .. } | Self::ListScheduledJobs { .. } | Self::ListPendingApprovals { .. } | Self::ListAuditEvents { .. } @@ -470,6 +545,21 @@ impl ControlRequest { | Self::ListAuditEvents { page } | Self::ListObservabilityEvents { page } | Self::ListMemoryAgents { page } => page.valid(), + Self::ListConversations { page, filter } => { + page.valid() + && filter + .avatar + .as_ref() + .is_none_or(|value| valid_identifier(value)) + && filter + .active_from_unix_millis + .zip(filter.active_to_unix_millis) + .is_none_or(|(from, to)| from <= to) + } + Self::GetConversation { session_id, page } => { + valid_identifier(session_id) && page.valid() + } + Self::DeleteConversation { session_id } => valid_identifier(session_id), Self::BrowseMemories { mentra_agent_id, cursor, @@ -1888,6 +1978,11 @@ fn memory_filter_valid(filter: &MemoryFilterView) -> bool { fn payload_valid(payload: &ControlPayload, maximum_frame_bytes: usize) -> bool { let page_is_bounded = match payload { ControlPayload::Sessions(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE), + ControlPayload::Conversations(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE), + ControlPayload::ConversationDetail(detail) => { + detail.entries.items.len() <= usize::from(MAX_PAGE_SIZE) + && detail.transcript_entries.len() <= usize::from(MAX_PAGE_SIZE) + } ControlPayload::ScheduledJobs(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE), ControlPayload::PendingApprovals(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE), ControlPayload::AuditEvents(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE), @@ -1928,6 +2023,9 @@ fn request_name(request: &ControlRequest) -> &'static str { ControlRequest::Metrics => "metrics", ControlRequest::Runtime => "runtime", ControlRequest::ListSessions { .. } => "list_sessions", + ControlRequest::ListConversations { .. } => "list_conversations", + ControlRequest::GetConversation { .. } => "get_conversation", + ControlRequest::DeleteConversation { .. } => "delete_conversation", ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs", ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals", ControlRequest::ListAuditEvents { .. } => "list_audit_events", diff --git a/crates/metacrate-grid-agent/src/control_runtime.rs b/crates/metacrate-grid-agent/src/control_runtime.rs index bc9a25a..3f63907 100644 --- a/crates/metacrate-grid-agent/src/control_runtime.rs +++ b/crates/metacrate-grid-agent/src/control_runtime.rs @@ -6,9 +6,11 @@ use crate::behavior::{BehaviorIngress, BehaviorMode}; use crate::control_plane::{ AuditEventView, CONTROL_PROTOCOL_VERSION, ControlContext, ControlError, ControlErrorCode, ControlFuture, ControlPayload, ControlRequest, ControlTarget, ConversationChannelView, - HealthView, MemoryAgentView, MemoryCursorView, MemoryDetailView, MemoryFilterView, - MemoryHealthView, MemoryKindView, MemoryPageView, MemoryRecordView, MemorySortView, Page, - PageRequest, PendingApprovalView, RuntimeView, ScheduledJobView, SessionMetadataView, + ConversationDetailView, ConversationEntryView, ConversationFilterView, + ConversationMetadataView, ConversationStatusView, HealthView, MemoryAgentView, + MemoryCursorView, MemoryDetailView, MemoryFilterView, MemoryHealthView, MemoryKindView, + MemoryPageView, MemoryRecordView, MemorySortView, Page, PageRequest, PendingApprovalView, + RuntimeView, ScheduledJobView, SessionMetadataView, TranscriptEntryView, }; use crate::conversation::{ConversationChannel, ConversationKey, ConversationStore}; use crate::observability::{ @@ -90,6 +92,7 @@ pub struct AgentControlTarget { build: Mutex>>, vision: Mutex>>, memory: Mutex>>, + conversation_admin: Mutex>>, behavior: BehaviorIngress, commands: mpsc::Sender, command_capacity: usize, @@ -136,6 +139,7 @@ impl AgentControlTarget { build: Mutex::new(None), vision: Mutex::new(None), memory: Mutex::new(None), + conversation_admin: Mutex::new(None), behavior, commands, command_capacity, @@ -165,6 +169,13 @@ impl AgentControlTarget { *lock(&self.memory) = Some(memory); } + pub fn attach_conversation_control( + &self, + responder: Arc, + ) { + *lock(&self.conversation_admin) = Some(responder); + } + pub fn update_session(&self, session: SessionStatus) { let mut state = lock(&self.state); state.session = session; @@ -198,6 +209,113 @@ impl AgentControlTarget { lock(&self.state).service_state = "stopping"; } + fn get_conversation( + &self, + context: &ControlContext, + session_id: &str, + page: &PageRequest, + ) -> Result { + require_operator(context)?; + let detail = self + .conversations + .operator_detail(session_id) + .ok_or_else(|| { + control_error(ControlErrorCode::NotFound, "conversation not found", false) + })?; + let offset = usize::try_from(page.cursor.unwrap_or(0)).unwrap_or(usize::MAX); + let entries = page_values( + page, + detail + .entries + .into_iter() + .map(|entry| ConversationEntryView { + sequence: entry.sequence, + unix_millis: entry.unix_millis, + role: entry.role.to_owned(), + record_kind: entry.kind.to_owned(), + trust: match entry.trust { + crate::conversation::MemoryTrust::Untrusted => "untrusted", + crate::conversation::MemoryTrust::TrustedOutput => "trusted_output", + } + .to_owned(), + text: entry.text, + }) + .collect(), + ); + let conversation_admin = { lock(&self.conversation_admin).clone() }; + let transcript = if let Some(admin) = conversation_admin { + match admin.conversation_transcript(session_id, offset, usize::from(page.limit)) { + Ok(value) => Some(value), + Err(error) if error == "conversation transcript not found" => None, + Err(error) => return Err(memory_error(error)), + } + } else { + None + }; + let transcript_busy = transcript.as_ref().is_some_and(|value| value.busy); + Ok(ControlPayload::ConversationDetail(ConversationDetailView { + conversation: conversation_metadata_view(&detail.metadata, detail.status, true), + entries, + mentra_agent_id: transcript + .as_ref() + .map(|value| value.mentra_agent_id.clone()), + profile_generation: transcript.as_ref().map(|value| value.profile_generation), + active_transcript_entries: transcript.as_ref().map_or(0, |value| value.active_entries), + archived_transcript_entries: transcript + .as_ref() + .map_or(0, |value| value.archived_entries), + transcript_entries: transcript.map_or_else(Vec::new, |value| { + value + .entries + .into_iter() + .map(|entry| TranscriptEntryView { + entry_id: entry.entry_id, + parent_id: entry.parent_id, + kind: entry.kind.to_owned(), + active: entry.active, + }) + .collect() + }), + transcript_busy, + })) + } + + async fn delete_conversation( + &self, + context: &ControlContext, + session_id: &str, + ) -> Result { + require_operator(context)?; + if self.conversations.operator_detail(session_id).is_none() { + return Err(control_error( + ControlErrorCode::NotFound, + "conversation not found", + false, + )); + } + let conversation_admin = { lock(&self.conversation_admin).clone() }; + if let Some(admin) = conversation_admin { + admin + .clear_conversation_transcript(session_id) + .await + .map_err(memory_error)?; + } + if !self.conversations.delete_session(session_id).map_err(|_| { + control_error( + ControlErrorCode::Internal, + "conversation deletion could not be persisted", + true, + ) + })? { + return Err(control_error( + ControlErrorCode::NotFound, + "conversation not found", + false, + )); + } + Ok(ControlPayload::Completed) + } + fn enqueue( &self, operation: &'static str, @@ -315,6 +433,26 @@ impl AgentControlTarget { .collect(); Ok(ControlPayload::Sessions(page_values(&page, values))) } + ControlRequest::ListConversations { page, filter } => { + let operator = context.role == crate::control_plane::ControlRole::Operator; + let values = self + .conversations + .list_operator_metadata() + .into_iter() + .filter(|value| conversation_matches(value, &filter)) + .map(|value| { + conversation_metadata_view(&value.metadata, value.status, operator) + }) + .collect(); + Ok(ControlPayload::Conversations(page_values(&page, values))) + } + ControlRequest::GetConversation { .. } | ControlRequest::DeleteConversation { .. } => { + Err(control_error( + ControlErrorCode::Internal, + "async conversation request reached synchronous runtime", + false, + )) + } ControlRequest::ListScheduledJobs { page } => { let values = lock(&self.jobs) .iter() @@ -694,7 +832,15 @@ impl ControlTarget for AgentControlTarget { let operation = runtime_request_name(&request); let correlation = request_correlation(&request); let domain_event = runtime_domain_event(&request).ok().flatten(); - let result = self.execute_now(&context, request); + let result = match request { + ControlRequest::GetConversation { session_id, page } => { + self.get_conversation(&context, &session_id, &page) + } + ControlRequest::DeleteConversation { session_id } => { + self.delete_conversation(&context, &session_id).await + } + request => self.execute_now(&context, request), + }; if let Some(observability) = lock(&self.observability).as_ref() { let (severity, result_code) = match &result { Ok(_) => (EventSeverity::Info, "accepted"), @@ -836,6 +982,8 @@ fn request_correlation(request: &ControlRequest) -> CorrelationIds { Some(format!("approval-{approval_id}")) } ControlRequest::ForgetMemory { record_id, .. } => Some(record_id.clone()), + ControlRequest::GetConversation { session_id, .. } + | ControlRequest::DeleteConversation { session_id } => Some(session_id.clone()), _ => None, }; CorrelationIds { @@ -850,6 +998,9 @@ const fn runtime_request_name(request: &ControlRequest) -> &'static str { ControlRequest::Metrics => "metrics", ControlRequest::Runtime => "runtime", ControlRequest::ListSessions { .. } => "list_sessions", + ControlRequest::ListConversations { .. } => "list_conversations", + ControlRequest::GetConversation { .. } => "get_conversation", + ControlRequest::DeleteConversation { .. } => "delete_conversation", ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs", ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals", ControlRequest::ListAuditEvents { .. } => "list_audit_events", @@ -910,6 +1061,67 @@ fn page_values(request: &PageRequest, values: Vec) -> Page { } } +fn conversation_matches( + value: &crate::conversation::OperatorConversationMetadata, + filter: &ConversationFilterView, +) -> bool { + let metadata = &value.metadata; + filter.status.is_none_or(|status| { + status + == match value.status { + crate::conversation::ConversationStatus::Active => ConversationStatusView::Active, + crate::conversation::ConversationStatus::Expired => ConversationStatusView::Expired, + } + }) && filter + .avatar + .as_ref() + .is_none_or(|avatar| metadata.avatar_id.to_string().eq_ignore_ascii_case(avatar)) + && filter.channel.as_ref().is_none_or(|channel| match channel { + ConversationChannelView::PublicChat => { + metadata.channel == ConversationChannel::PublicChat + } + ConversationChannelView::DirectIm => metadata.channel == ConversationChannel::DirectIm, + }) + && filter + .active_from_unix_millis + .is_none_or(|from| metadata.last_active_unix_millis >= from) + && filter + .active_to_unix_millis + .is_none_or(|to| metadata.last_active_unix_millis <= to) +} + +fn conversation_metadata_view( + metadata: &crate::conversation::ConversationMetadata, + status: crate::conversation::ConversationStatus, + operator: bool, +) -> ConversationMetadataView { + ConversationMetadataView { + session_id: if operator { + metadata.session_id.as_str().to_owned() + } else { + pseudonymous_identifier(metadata.session_id.as_str()) + }, + avatar_id: if operator { + metadata.avatar_id.to_string() + } else { + pseudonymous_identifier(&metadata.avatar_id.to_string()) + }, + channel: match metadata.channel { + ConversationChannel::PublicChat => ConversationChannelView::PublicChat, + ConversationChannel::DirectIm => ConversationChannelView::DirectIm, + }, + status: match status { + crate::conversation::ConversationStatus::Active => ConversationStatusView::Active, + crate::conversation::ConversationStatus::Expired => ConversationStatusView::Expired, + }, + created_unix_millis: metadata.created_unix_millis, + last_active_unix_millis: metadata.last_active_unix_millis, + turns: metadata.turns, + bytes: metadata.bytes, + compacted: metadata.compacted, + } +} + const fn behavior_name(mode: BehaviorMode) -> &'static str { match mode { BehaviorMode::Offline => "offline", diff --git a/crates/metacrate-grid-agent/src/control_runtime_tests.rs b/crates/metacrate-grid-agent/src/control_runtime_tests.rs index 77af690..f4aa3a0 100644 --- a/crates/metacrate-grid-agent/src/control_runtime_tests.rs +++ b/crates/metacrate-grid-agent/src/control_runtime_tests.rs @@ -1,10 +1,14 @@ use crate::behavior::{BehaviorController, BehaviorError, EmbodimentFuture, EmbodimentSink}; use crate::control_plane::{ CONTROL_PROTOCOL_VERSION, ControlLimits, ControlPayload, ControlPlane, ControlRequest, - ControlRequestEnvelope, MemoryFilterView, MemorySortView, PageRequest, + ControlRequestEnvelope, ConversationFilterView, ConversationStatusView, MemoryFilterView, + MemorySortView, PageRequest, }; use crate::control_runtime::{AgentControlTarget, RuntimeControlCommand}; -use crate::conversation::{ConversationLimits, ConversationStore}; +use crate::conversation::{ + ConversationChannel, ConversationKey, ConversationLimits, ConversationStore, + MemoryRecord as ConversationRecord, +}; use crate::perception::WorldPosition; use crate::policy::{MemoryPolicyAudit, PolicyAuditSink, PolicyGateway, PolicyLimits}; use crate::{ @@ -146,6 +150,38 @@ async fn production_target_projects_state_and_routes_real_mutations() { let conversations = Arc::new( ConversationStore::open(ConversationLimits::default(), None).expect("conversations"), ); + let conversation_id = conversations + .append( + ConversationKey::new( + UUID::new_with_u_int64(55).expect("avatar"), + ConversationChannel::DirectIm, + ) + .expect("key"), + ConversationRecord::avatar_message("bounded operator transcript"), + ) + .expect("conversation") + .session_id; + conversations + .append( + ConversationKey::new( + UUID::new_with_u_int64(55).expect("avatar"), + ConversationChannel::DirectIm, + ) + .expect("key"), + ConversationRecord::agent_visible_response("bounded response"), + ) + .expect("second turn"); + let other_conversation_id = conversations + .append( + ConversationKey::new( + UUID::new_with_u_int64(56).expect("avatar"), + ConversationChannel::PublicChat, + ) + .expect("key"), + ConversationRecord::avatar_message("other avatar transcript"), + ) + .expect("other conversation") + .session_id; let (target, mut commands) = AgentControlTarget::new(conversations, policy, audit, behavior.ingress(), 8) .expect("target"); @@ -226,6 +262,108 @@ async fn production_target_projects_state_and_routes_real_mutations() { assert!(vision.active_capture().is_none(), "pause cancels capture"); let observer = plane.connect("observer-token").expect("observer"); + let observer_conversations = observer + .request(envelope( + "observer-conversations", + ControlRequest::ListConversations { + page: PageRequest::default(), + filter: ConversationFilterView::default(), + }, + )) + .await; + let Ok(ControlPayload::Conversations(observer_conversations)) = observer_conversations.result + else { + panic!("observer conversation metadata") + }; + assert_eq!(observer_conversations.items.len(), 2); + assert_ne!( + observer_conversations.items[0].session_id, + conversation_id.as_str() + ); + assert!( + observer + .request(envelope( + "observer-conversation-detail", + ControlRequest::GetConversation { + session_id: conversation_id.as_str().into(), + page: PageRequest::default(), + }, + )) + .await + .result + .is_err() + ); + assert!( + observer + .request(envelope( + "observer-conversation-delete", + ControlRequest::DeleteConversation { + session_id: conversation_id.as_str().into(), + }, + )) + .await + .result + .is_err() + ); + let operator_conversations = operator + .request(envelope( + "operator-conversations", + ControlRequest::ListConversations { + page: PageRequest::default(), + filter: ConversationFilterView { + status: Some(ConversationStatusView::Active), + ..ConversationFilterView::default() + }, + }, + )) + .await; + assert!(matches!( + operator_conversations.result, + Ok(ControlPayload::Conversations(_)) + )); + let detail = operator + .request(envelope( + "operator-conversation-detail", + ControlRequest::GetConversation { + session_id: conversation_id.as_str().into(), + page: PageRequest { + cursor: None, + limit: 1, + }, + }, + )) + .await; + let Ok(ControlPayload::ConversationDetail(detail)) = detail.result else { + panic!("operator conversation detail") + }; + assert_eq!(detail.entries.items.len(), 1); + assert_eq!(detail.entries.next_cursor, Some(1)); + assert_eq!(detail.entries.items[0].text, "bounded operator transcript"); + assert!( + operator + .request(envelope( + "operator-conversation-delete", + ControlRequest::DeleteConversation { + session_id: conversation_id.as_str().into() + }, + )) + .await + .result + .is_ok() + ); + assert!( + operator + .request(envelope( + "other-conversation-detail", + ControlRequest::GetConversation { + session_id: other_conversation_id.as_str().into(), + page: PageRequest::default(), + }, + )) + .await + .result + .is_ok() + ); let health = observer .request(envelope("memory-health", ControlRequest::MemoryHealth)) .await; @@ -327,9 +465,8 @@ async fn production_target_projects_state_and_routes_real_mutations() { .any(|event| event.family == "control_command") ); assert!(events.items.iter().all(|event| { - !serde_json::to_string(event) - .expect("event JSON") - .contains("operator-token") + let json = serde_json::to_string(event).expect("event JSON"); + !json.contains("operator-token") && !json.contains("bounded operator transcript") })); behavior.shutdown().await.expect("shutdown behavior"); } diff --git a/crates/metacrate-grid-agent/src/conversation.rs b/crates/metacrate-grid-agent/src/conversation.rs index d105aa5..f3bff6c 100644 --- a/crates/metacrate-grid-agent/src/conversation.rs +++ b/crates/metacrate-grid-agent/src/conversation.rs @@ -30,6 +30,7 @@ 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; +const MAX_RETIRED_SESSIONS: usize = 4_096; /// A trust-separated conversation channel. Group and conference channels are /// deliberately unrepresentable until their authority semantics are defined. @@ -266,6 +267,7 @@ pub enum ConversationError { IdentifierGeneration, PersistenceUnavailable, PersistenceTooLarge, + StaleSession, Boundary(crate::types::BoundaryError), } @@ -284,6 +286,7 @@ impl fmt::Display for ConversationError { Self::PersistenceTooLarge => { formatter.write_str("conversation snapshot exceeds its storage bound") } + Self::StaleSession => formatter.write_str("conversation session is no longer active"), Self::Boundary(error) => { write!(formatter, "conversation boundary rejected data: {error}") } @@ -316,6 +319,19 @@ pub struct ConversationMetadata { pub last_active_unix_millis: u64, pub turns: usize, pub bytes: usize, + pub compacted: bool, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ConversationStatus { + Active, + Expired, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct OperatorConversationMetadata { + pub metadata: ConversationMetadata, + pub status: ConversationStatus, } /// LLM-only projection for one exact UUID/channel key. @@ -325,6 +341,23 @@ pub struct ConversationContext { entries: Vec, } +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ConversationEntryProjection { + pub sequence: u64, + pub unix_millis: u64, + pub role: &'static str, + pub kind: &'static str, + pub trust: MemoryTrust, + pub text: String, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ConversationDetailProjection { + pub metadata: ConversationMetadata, + pub status: ConversationStatus, + pub entries: Vec, +} + #[derive(Clone, Debug, Eq, PartialEq)] struct ContextEntry { role: MessageRole, @@ -420,12 +453,17 @@ impl StoredSession { last_active_unix_millis: self.last_active_unix_millis, turns: self.entries.len(), bytes: self.bytes, + compacted: self + .entries + .iter() + .any(|entry| entry.kind == RecordKind::FactualSummary), } } } struct StoreState { sessions: BTreeMap, + retired: VecDeque, total_bytes: usize, next_sequence: u64, snapshot_generation: u64, @@ -481,6 +519,7 @@ impl ConversationStore { limits.validate()?; let mut state = StoreState { sessions: BTreeMap::new(), + retired: VecDeque::new(), total_bytes: 0, next_sequence: 1, snapshot_generation: 0, @@ -502,6 +541,24 @@ impl ConversationStore { &self, key: ConversationKey, record: MemoryRecord, + ) -> Result { + self.append_checked(key, None, record) + } + + pub fn append_if_session( + &self, + key: ConversationKey, + session_id: &str, + record: MemoryRecord, + ) -> Result { + self.append_checked(key, Some(session_id), record) + } + + fn append_checked( + &self, + key: ConversationKey, + expected_session_id: Option<&str>, + record: MemoryRecord, ) -> Result { if key.avatar_id == UUID::zero() { return Err(ConversationError::InvalidAvatar); @@ -528,6 +585,14 @@ impl ConversationStore { let mut state = self.lock_state(); let event_start = state.events.len(); expire_idle_locked(&mut state, monotonic_now, None); + if expected_session_id.is_some_and(|expected| { + state + .sessions + .get(&key) + .is_none_or(|session| session.id.as_str() != expected) + }) { + return Err(ConversationError::StaleSession); + } if !state.sessions.contains_key(&key) { while state.sessions.len() >= self.limits.max_active_sessions { evict_oldest(&mut state, MemoryReason::SessionLimitEviction, None); @@ -627,6 +692,140 @@ impl ConversationStore { .collect() } + #[must_use] + pub fn list_operator_metadata(&self) -> Vec { + let now = self.clock.monotonic_now(); + let mut state = self.lock_state(); + expire_idle_locked(&mut state, now, None); + let mut values = state + .sessions + .values() + .map(|session| OperatorConversationMetadata { + metadata: session.metadata(), + status: ConversationStatus::Active, + }) + .chain( + state + .retired + .iter() + .cloned() + .map(|metadata| OperatorConversationMetadata { + metadata, + status: ConversationStatus::Expired, + }), + ) + .collect::>(); + values.sort_by(|left, right| { + right + .metadata + .last_active_unix_millis + .cmp(&left.metadata.last_active_unix_millis) + .then_with(|| { + left.metadata + .session_id + .as_str() + .cmp(right.metadata.session_id.as_str()) + }) + }); + values + } + + pub fn operator_detail(&self, session_id: &str) -> Option { + let now = self.clock.monotonic_now(); + let mut state = self.lock_state(); + expire_idle_locked(&mut state, now, None); + state + .sessions + .values() + .find(|session| session.id.as_str() == session_id) + .map(|session| ConversationDetailProjection { + metadata: session.metadata(), + status: ConversationStatus::Active, + entries: session + .entries + .iter() + .map(|entry| ConversationEntryProjection { + sequence: entry.sequence, + unix_millis: entry.unix_millis, + role: match entry.role { + MessageRole::Avatar => "avatar", + MessageRole::Agent => "agent", + MessageRole::Tool => "tool", + MessageRole::System => "data", + }, + kind: match entry.kind { + RecordKind::Message => "message", + RecordKind::ToolSummary => "tool_summary", + RecordKind::ActionResult => "action_result", + RecordKind::FactualSummary => "factual_summary", + }, + trust: entry.trust, + text: entry.text.as_str().to_owned(), + }) + .collect(), + }) + .or_else(|| { + state + .retired + .iter() + .find(|metadata| metadata.session_id.as_str() == session_id) + .cloned() + .map(|metadata| ConversationDetailProjection { + metadata, + status: ConversationStatus::Expired, + entries: Vec::new(), + }) + }) + } + + pub fn delete_session(&self, session_id: &str) -> Result { + let mut state = self.lock_state(); + let Some(key) = state + .sessions + .iter() + .find_map(|(key, session)| (session.id.as_str() == session_id).then_some(*key)) + else { + if let Some(index) = state + .retired + .iter() + .position(|metadata| metadata.session_id.as_str() == session_id) + { + let previous = state + .retired + .remove(index) + .ok_or(ConversationError::PersistenceUnavailable)?; + if let Some(persistence) = &self.persistence + && let Err(error) = write_snapshot(&self.limits, persistence, &mut state) + { + state.retired.insert(index, previous); + let _ = write_snapshot(&self.limits, persistence, &mut state); + return Err(error); + } + return Ok(true); + } + return Ok(false); + }; + let session = state + .sessions + .remove(&key) + .ok_or(ConversationError::PersistenceUnavailable)?; + state.total_bytes = state.total_bytes.saturating_sub(session.bytes); + push_event( + &mut state.events, + event_for(MemoryReason::OperatorDeleted, &session), + ); + if let Some(persistence) = &self.persistence + && let Err(error) = write_snapshot(&self.limits, persistence, &mut state) + { + state.total_bytes = state.total_bytes.saturating_add(session.bytes); + state.sessions.insert(key, session); + state.events.pop_back(); + let _ = write_snapshot(&self.limits, persistence, &mut state); + return Err(error); + } + Ok(true) + } + pub fn delete(&self, key: ConversationKey) -> bool { self.remove_operator(key, MemoryReason::OperatorDeleted) } @@ -661,6 +860,9 @@ impl ConversationStore { return false; }; state.total_bytes = state.total_bytes.saturating_sub(session.bytes); + if reason == MemoryReason::OperatorExpired { + retain_expired(&mut state, session.metadata()); + } push_event(&mut state.events, event_for(reason, &session)); true } @@ -801,6 +1003,7 @@ fn expire_idle_locked(state: &mut StoreState, now: Instant, only: Option MemoryReason::PublicExpired, ConversationChannel::DirectIm => MemoryReason::DirectImExpired, }; + retain_expired(state, session.metadata()); push_event(&mut state.events, event_for(reason, &session)); } } @@ -818,10 +1021,18 @@ fn evict_oldest(state: &mut StoreState, reason: MemoryReason, preserve: Option MemoryEvent { MemoryEvent { reason, @@ -904,6 +1115,21 @@ struct Snapshot { generation: u64, next_sequence: u64, sessions: Vec, + #[serde(default)] + retired_sessions: Vec, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct SnapshotRetiredSession { + session_id: String, + avatar_id: String, + channel: ConversationChannel, + created_unix_millis: u64, + last_active_unix_millis: u64, + turns: usize, + bytes: usize, + compacted: bool, } #[derive(Serialize, Deserialize)] @@ -1021,6 +1247,7 @@ fn restore_snapshot( let wall_now = unix_millis(clock.wall_now()); let mut restored = StoreState { sessions: BTreeMap::new(), + retired: VecDeque::new(), total_bytes: 0, next_sequence: snapshot.next_sequence.max(1), snapshot_generation: snapshot.generation, @@ -1029,6 +1256,31 @@ fn restore_snapshot( let mut restored_session_ids = BTreeSet::new(); let mut restored_sequences = BTreeSet::new(); let mut maximum_sequence = 0_u64; + if snapshot.retired_sessions.len() > MAX_RETIRED_SESSIONS { + return Err(ConversationError::PersistenceTooLarge); + } + for persisted in snapshot.retired_sessions { + let avatar_id = UUID::new_with_string(persisted.avatar_id) + .map_err(|_| ConversationError::InvalidAvatar)?; + let session_id = BoundedText::new("conversation.session_id", persisted.session_id)?; + if !restored_session_ids.insert(session_id.as_str().to_owned()) + || persisted.created_unix_millis > persisted.last_active_unix_millis + || persisted.turns > MAX_CONVERSATION_MESSAGES + || persisted.bytes > MAX_SESSION_BYTES + { + return Err(ConversationError::PersistenceUnavailable); + } + restored.retired.push_back(ConversationMetadata { + session_id, + avatar_id, + channel: persisted.channel, + created_unix_millis: persisted.created_unix_millis.min(wall_now), + last_active_unix_millis: persisted.last_active_unix_millis.min(wall_now), + turns: persisted.turns, + bytes: persisted.bytes, + compacted: persisted.compacted, + }); + } if snapshot.sessions.len() > MAX_ACTIVE_SESSIONS { return Err(ConversationError::PersistenceTooLarge); } @@ -1037,15 +1289,38 @@ fn restore_snapshot( .map_err(|_| ConversationError::InvalidAvatar)?; let key = ConversationKey::new(avatar_id, persisted.channel)?; let session_id = BoundedText::new("conversation.session_id", persisted.session_id)?; + let persisted_bytes = persisted + .entries + .iter() + .try_fold(0_usize, |total, entry| total.checked_add(entry.text.len())) + .ok_or(ConversationError::PersistenceTooLarge)?; 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 + || persisted.entries.len() > MAX_CONVERSATION_MESSAGES + || persisted_bytes > limits.max_session_bytes { 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() { + retain_expired( + &mut restored, + ConversationMetadata { + session_id, + avatar_id, + channel: key.channel, + created_unix_millis: persisted.created_unix_millis.min(wall_now), + last_active_unix_millis: persisted.last_active_unix_millis.min(wall_now), + turns: persisted.entries.len(), + bytes: persisted_bytes, + compacted: persisted + .entries + .iter() + .any(|entry| entry.kind == RecordKind::FactualSummary), + }, + ); continue; } let last_active = monotonic_now.checked_sub(elapsed).unwrap_or(monotonic_now); @@ -1059,9 +1334,6 @@ fn restore_snapshot( 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) { @@ -1183,6 +1455,20 @@ fn write_snapshot( .collect(), }) .collect(), + retired_sessions: state + .retired + .iter() + .map(|metadata| SnapshotRetiredSession { + session_id: metadata.session_id.as_str().to_owned(), + avatar_id: metadata.avatar_id.to_string(), + channel: metadata.channel, + created_unix_millis: metadata.created_unix_millis, + last_active_unix_millis: metadata.last_active_unix_millis, + turns: metadata.turns, + bytes: metadata.bytes, + compacted: metadata.compacted, + }) + .collect(), }; let bytes = serde_json::to_vec(&snapshot).map_err(|_| ConversationError::PersistenceUnavailable)?; diff --git a/crates/metacrate-grid-agent/src/conversation_tests.rs b/crates/metacrate-grid-agent/src/conversation_tests.rs index 698638e..b82ecc8 100644 --- a/crates/metacrate-grid-agent/src/conversation_tests.rs +++ b/crates/metacrate-grid-agent/src/conversation_tests.rs @@ -509,6 +509,79 @@ fn operator_metadata_delete_and_expire_never_require_content_access() { assert!(reasons.contains(&MemoryReason::OperatorExpired)); } +#[test] +fn expired_metadata_and_exact_delete_survive_restart_without_touching_other_sessions() { + let directory = temporary_directory(); + let persistence = ConversationPersistence { + directory: directory.clone(), + }; + let store = ConversationStore::open(limits(), Some(persistence.clone())).unwrap(); + let expired = key(70, ConversationChannel::PublicChat); + let retained = key(71, ConversationChannel::DirectIm); + let expired_id = store + .append(expired, MemoryRecord::avatar_message("old")) + .unwrap() + .session_id; + let retained_id = store + .append(retained, MemoryRecord::avatar_message("keep")) + .unwrap() + .session_id; + assert!(store.expire(expired)); + store.flush().unwrap(); + drop(store); + + let reopened = ConversationStore::open(limits(), Some(persistence)).unwrap(); + let rows = reopened.list_operator_metadata(); + assert!( + rows.iter().any(|row| row.metadata.session_id == expired_id + && row.status == ConversationStatus::Expired) + ); + assert!( + rows.iter().any(|row| row.metadata.session_id == retained_id + && row.status == ConversationStatus::Active) + ); + assert!(reopened.delete_session(expired_id.as_str()).unwrap()); + assert!(reopened.context(retained).is_some()); + drop(reopened); + + let reopened = ConversationStore::open( + limits(), + Some(ConversationPersistence { + directory: directory.clone(), + }), + ) + .unwrap(); + assert!( + !reopened + .list_operator_metadata() + .iter() + .any(|row| row.metadata.session_id == expired_id) + ); + assert!(reopened.context(retained).is_some()); + drop(reopened); + fs::remove_dir_all(directory).unwrap(); +} + +#[test] +fn stale_completion_cannot_recreate_a_deleted_session() { + let store = ConversationStore::open(limits(), None).unwrap(); + let conversation = key(72, ConversationChannel::DirectIm); + let session_id = store + .append(conversation, MemoryRecord::avatar_message("request")) + .unwrap() + .session_id; + assert!(store.delete_session(session_id.as_str()).unwrap()); + assert_eq!( + store.append_if_session( + conversation, + session_id.as_str(), + MemoryRecord::agent_visible_response("late") + ), + Err(ConversationError::StaleSession) + ); + assert!(store.list_metadata().is_empty()); +} + #[cfg(unix)] #[test] fn persistence_uses_restrictive_unix_permissions() { diff --git a/crates/metacrate-grid-agent/src/interaction.rs b/crates/metacrate-grid-agent/src/interaction.rs index f1f78d3..38eee54 100644 --- a/crates/metacrate-grid-agent/src/interaction.rs +++ b/crates/metacrate-grid-agent/src/interaction.rs @@ -1228,8 +1228,9 @@ async fn process_batch( ) .await; let outcome = if delivered == parts.len() { - let _ = conversation.append( + let _ = conversation.append_if_session( conversation_key, + update.session_id.as_str(), MemoryRecord::agent_visible_response(safe.clone()), ); DeliveryOutcome::Succeeded @@ -1781,6 +1782,7 @@ pub struct PolicyLlmResponder { runtime: Arc, agents: Mutex>, memory_admin: Arc, + conversation_mappings: Arc, active_tools: Arc>>, gateway: Arc, backend: Arc, @@ -1789,6 +1791,24 @@ pub struct PolicyLlmResponder { storage_path: std::path::PathBuf, } +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct MentraTranscriptEntry { + pub entry_id: String, + pub parent_id: Option, + pub kind: &'static str, + pub active: bool, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct MentraTranscriptMetadata { + pub mentra_agent_id: String, + pub profile_generation: u64, + pub active_entries: usize, + pub archived_entries: usize, + pub entries: Vec, + pub busy: bool, +} + #[derive(Clone)] struct MentraAgentHandle { id: String, @@ -1800,6 +1820,7 @@ struct ActiveMentraRequest { executor: Arc, cancellation: CancellationToken, mentra_cancellation: mentra::runtime::CancellationToken, + session_id: String, } struct AutonomousApprovalReviewer { @@ -2009,6 +2030,12 @@ impl PolicyLlmResponder { ) .map_err(InteractionError::Persistence)?, ); + let conversation_mappings = Arc::new( + crate::memory_admin::MentraConversationMappings::new( + storage_path.join("conversation-agents.json"), + ) + .map_err(InteractionError::Persistence)?, + ); let mut builder = mentra::Runtime::builder() .with_runtime_identifier(MENTRA_RUNTIME_IDENTIFIER) .with_store(store) @@ -2047,6 +2074,7 @@ impl PolicyLlmResponder { runtime: Arc::new(runtime), agents: Mutex::new(agents), memory_admin, + conversation_mappings, active_tools, gateway, backend, @@ -2072,6 +2100,7 @@ impl PolicyLlmResponder { .register(&request.sender_id.to_string(), &key, &handle.id) .map_err(|_| InteractionModelError::Failed)?; } + self.register_conversation(request, &key, &handle.id)?; return Ok(Arc::clone(&handle.agent)); } let tools = self.gateway.tools_for(context, (self.now)()); @@ -2118,13 +2147,146 @@ impl PolicyLlmResponder { .register(&request.sender_id.to_string(), &key, &id) .map_err(|_| InteractionModelError::Failed)?; } + self.register_conversation(request, &key, &id)?; Ok(agent) } + fn register_conversation( + &self, + request: &ResponseRequest, + logical_agent_id: &str, + mentra_agent_id: &str, + ) -> Result<(), InteractionModelError> { + let profile_generation = self + .conversation_mappings + .profile_generation(request.session_id.as_str(), logical_agent_id); + self.conversation_mappings + .register(crate::memory_admin::ConversationAgentMapping { + session_id: request.session_id.as_str().to_owned(), + avatar_id: request.sender_id.to_string(), + channel: match request.channel { + InteractionChannel::PublicChat => "public_chat", + InteractionChannel::DirectIm => "direct_im", + } + .to_owned(), + logical_agent_id: logical_agent_id.to_owned(), + mentra_agent_id: mentra_agent_id.to_owned(), + profile_generation, + }) + .map_err(|_| InteractionModelError::Failed) + } + #[must_use] pub fn memory_admin(&self) -> Arc { Arc::clone(&self.memory_admin) } + + pub fn conversation_transcript( + &self, + session_id: &str, + offset: usize, + limit: usize, + ) -> Result { + let mapping = self + .conversation_mappings + .get(session_id) + .ok_or_else(|| "conversation transcript not found".to_owned())?; + let agent = { + let agents = self + .agents + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + agents + .values() + .find(|handle| handle.id == mapping.mentra_agent_id) + .map(|handle| Arc::clone(&handle.agent)) + .ok_or_else(|| "conversation agent not active".to_owned())? + }; + let Ok(agent) = agent.try_lock() else { + return Ok(MentraTranscriptMetadata { + mentra_agent_id: mapping.mentra_agent_id, + profile_generation: mapping.profile_generation, + active_entries: 0, + archived_entries: 0, + entries: Vec::new(), + busy: true, + }); + }; + let transcript = agent.transcript(); + let active_entries = transcript.items().len(); + let archived_entries = transcript.archived().len(); + let entries = transcript + .items() + .iter() + .map(|item| (item, true)) + .chain(transcript.archived().iter().map(|item| (item, false))) + .skip(offset) + .take(limit) + .map(|(item, active)| MentraTranscriptEntry { + entry_id: item.id.to_string(), + parent_id: item.parent_id.as_ref().map(ToString::to_string), + kind: transcript_kind_name(&item.kind), + active, + }) + .collect(); + Ok(MentraTranscriptMetadata { + mentra_agent_id: mapping.mentra_agent_id, + profile_generation: mapping.profile_generation, + active_entries, + archived_entries, + entries, + busy: false, + }) + } + + pub async fn clear_conversation_transcript(&self, session_id: &str) -> Result { + let Some(mapping) = self.conversation_mappings.get(session_id) else { + return Ok(false); + }; + if let Some(active) = self + .active_tools + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&mapping.mentra_agent_id) + .cloned() + { + if active.session_id != session_id { + return Err("conversation agent is serving a different session".to_owned()); + } + active.mentra_cancellation.cancel(); + } + let agent = { + let agents = self + .agents + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + agents + .values() + .find(|handle| handle.id == mapping.mentra_agent_id) + .map(|handle| Arc::clone(&handle.agent)) + .ok_or_else(|| "conversation agent not active".to_owned())? + }; + agent + .lock() + .await + .clear_transcript() + .map_err(|error| error.to_string())?; + self.conversation_mappings.remove(session_id)?; + Ok(true) + } +} + +fn transcript_kind_name(kind: &mentra::TranscriptKind) -> &'static str { + match kind { + mentra::TranscriptKind::UserTurn => "user_turn", + mentra::TranscriptKind::AssistantTurn => "assistant_turn", + mentra::TranscriptKind::ToolExchange { .. } => "tool_exchange", + mentra::TranscriptKind::CanonicalContext => "canonical_context", + mentra::TranscriptKind::MemoryRecall => "memory_recall", + mentra::TranscriptKind::DelegationRequest { .. } => "delegation_request", + mentra::TranscriptKind::DelegationResult { .. } => "delegation_result", + mentra::TranscriptKind::CompactionSummary { .. } => "compaction_summary", + } } pub(crate) fn mentra_agent_name(request: &ResponseRequest) -> String { @@ -2196,6 +2358,7 @@ impl InteractionResponder for PolicyLlmResponder { executor, cancellation: cancellation.clone(), mentra_cancellation: mentra_cancellation.clone(), + session_id: request.session_id.as_str().to_owned(), }, ); let cancellation_bridge = { diff --git a/crates/metacrate-grid-agent/src/lib.rs b/crates/metacrate-grid-agent/src/lib.rs index 13a558b..8a08b2b 100644 --- a/crates/metacrate-grid-agent/src/lib.rs +++ b/crates/metacrate-grid-agent/src/lib.rs @@ -96,27 +96,31 @@ pub use control_plane::{ AuditEventView, CONTROL_PROTOCOL_VERSION, ControlContext, ControlError, ControlErrorBody, ControlErrorCode, ControlEvent, ControlEventKind, ControlFuture, ControlLimits, ControlPayload, ControlPlane, ControlRequest, ControlRequestEnvelope, ControlResponseEnvelope, ControlRole, - ControlSubscription, ControlTarget, ConversationChannelView, HealthView, - InProcessControlClient, MemoryAgentView, MemoryCursorView, MemoryDetailView, MemoryFilterView, - MemoryHealthView, MemoryKindView, MemoryPageView, MemoryRecordView, MemorySortView, Page, - PageRequest, PendingApprovalView, RuntimeView, ScheduledJobView, SessionMetadataView, - TcpControlClient, TcpControlConfig, TcpControlServer, + ControlSubscription, ControlTarget, ConversationChannelView, ConversationDetailView, + ConversationEntryView, ConversationFilterView, ConversationMetadataView, + ConversationStatusView, HealthView, InProcessControlClient, MemoryAgentView, MemoryCursorView, + MemoryDetailView, MemoryFilterView, MemoryHealthView, MemoryKindView, MemoryPageView, + MemoryRecordView, MemorySortView, Page, PageRequest, PendingApprovalView, RuntimeView, + ScheduledJobView, SessionMetadataView, TcpControlClient, TcpControlConfig, TcpControlServer, + TranscriptEntryView, }; pub use control_runtime::{AgentControlTarget, RuntimeControlCommand}; pub use conversation::{ - ConversationChannel, ConversationClock, ConversationContext, ConversationError, - ConversationKey, ConversationLimits, ConversationMetadata, ConversationPersistence, - ConversationStore, MemoryEvent, MemoryReason, MemoryRecord, MemoryTrust, MemoryUpdate, - SystemConversationClock, + ConversationChannel, ConversationClock, ConversationContext, ConversationDetailProjection, + ConversationEntryProjection, ConversationError, ConversationKey, ConversationLimits, + ConversationMetadata, ConversationPersistence, ConversationStatus, ConversationStore, + MemoryEvent, MemoryReason, MemoryRecord, MemoryTrust, MemoryUpdate, + OperatorConversationMetadata, SystemConversationClock, }; pub use interaction::{ DeliveryFuture, DeliveryOutcome, ImDialogKind, InboundInteraction, InboundSource, InferenceOutcome, InteractionChannel, InteractionCoordinator, InteractionDeliveryError, InteractionError, InteractionHandle, InteractionIngress, InteractionIntent, InteractionModelError, InteractionObservation, InteractionOrigin, InteractionPacingError, - InteractionResponder, InteractionSettings, InteractionSink, OutboundInteraction, PacerFuture, - PolicyLlmResponder, ResponderFuture, ResponsePacer, ResponseRequest, SuppressionReason, - VisibleResponse, split_utf8, + InteractionResponder, InteractionSettings, InteractionSink, MentraTranscriptEntry, + MentraTranscriptMetadata, OutboundInteraction, PacerFuture, PolicyLlmResponder, + ResponderFuture, ResponsePacer, ResponseRequest, SuppressionReason, VisibleResponse, + split_utf8, }; pub use landmarks::{ LANDMARK_CREATE_TOOL, LANDMARK_LIST_TOOL, LANDMARK_SCHEDULE_TOOL, LANDMARK_STATUS_TOOL, @@ -133,7 +137,9 @@ pub use landmarks::{ pub use llm::{ CompletionMessage, ContentPart, ImageDetail, LlmClient, LlmError, ToolDefinition, ToolSchema, }; -pub use memory_admin::{MemoryAgentMapping, MentraMemoryAdmin}; +pub use memory_admin::{ + ConversationAgentMapping, MemoryAgentMapping, MentraConversationMappings, MentraMemoryAdmin, +}; pub use observability::{ AgentMetrics, CorrelationIds, DiagnosticDirection, EventDraft, EventFamily, EventOrigin, EventSeverity, EventSubscription, JournalConfig, LatencyMetrics, MetricsSnapshot, diff --git a/crates/metacrate-grid-agent/src/main.rs b/crates/metacrate-grid-agent/src/main.rs index 734cda7..ae6d2ba 100644 --- a/crates/metacrate-grid-agent/src/main.rs +++ b/crates/metacrate-grid-agent/src/main.rs @@ -416,6 +416,7 @@ async fn run_live( control_target.attach_build_control(live.build_control.clone()); control_target.attach_vision_control(live.vision.clone()); control_target.attach_memory_control(live.memory_admin.clone()); + control_target.attach_conversation_control(live.conversation_admin.clone()); control_target.update_session(handle.status()); let erased_target: Arc = control_target.clone(); let (control_plane, integrated_client, control_server) = match config.mode { @@ -734,6 +735,7 @@ struct LiveInteractions { audit: Arc, observability: Arc, memory_admin: Arc, + conversation_admin: Arc, _appearance_recovery: libremetaverse_types::compat::Subscription, _landmark_intake: metacrate_grid_agent::LibremetaverseLandmarkIntake, landmark_roaming: metacrate_grid_agent::LandmarkRoamingHandle, @@ -906,6 +908,7 @@ async fn start_live_interactions( config.storage_path.join("mentra"), )?); let memory_admin = responder.memory_admin(); + let conversation_admin = responder.clone(); let vision_limits = config.vision; let scene_source = Arc::new(LibremetaverseSceneSource::new(owner, vision_limits)); scene_source.start_prefetch(); @@ -940,6 +943,7 @@ async fn start_live_interactions( audit, observability, memory_admin, + conversation_admin, _appearance_recovery: appearance_recovery, _landmark_intake: landmark_intake, landmark_roaming, diff --git a/crates/metacrate-grid-agent/src/memory_admin.rs b/crates/metacrate-grid-agent/src/memory_admin.rs index 29a6378..e817187 100644 --- a/crates/metacrate-grid-agent/src/memory_admin.rs +++ b/crates/metacrate-grid-agent/src/memory_admin.rs @@ -13,6 +13,7 @@ use std::path::{Path, PathBuf}; use std::sync::{Mutex, MutexGuard}; const MAX_MAPPINGS: usize = 4_096; +const MAX_MAPPING_BYTES: u64 = 8 * 1024 * 1024; #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] pub struct MemoryAgentMapping { @@ -21,6 +22,100 @@ pub struct MemoryAgentMapping { pub mentra_agent_id: String, } +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct ConversationAgentMapping { + pub session_id: String, + pub avatar_id: String, + pub channel: String, + pub logical_agent_id: String, + pub mentra_agent_id: String, + pub profile_generation: u64, +} + +pub struct MentraConversationMappings { + path: PathBuf, + mappings: Mutex>, +} + +impl MentraConversationMappings { + pub fn new(path: PathBuf) -> Result { + let values: Vec = load_values(&path)?; + if values.len() > MAX_MAPPINGS { + return Err("excessive conversation-agent mappings".to_owned()); + } + let mut mappings = BTreeMap::new(); + for mapping in values { + if !valid_conversation_mapping(&mapping) + || mappings + .insert(mapping.session_id.clone(), mapping) + .is_some() + { + return Err("invalid or duplicate conversation-agent mapping".to_owned()); + } + } + Ok(Self { + path, + mappings: Mutex::new(mappings), + }) + } + + pub fn register(&self, mapping: ConversationAgentMapping) -> Result<(), String> { + if !valid_conversation_mapping(&mapping) { + return Err("invalid conversation-agent mapping".to_owned()); + } + let mut mappings = lock(&self.mappings); + if !mappings.contains_key(&mapping.session_id) && mappings.len() >= MAX_MAPPINGS { + return Err("conversation-agent mapping limit reached".to_owned()); + } + if mappings.get(&mapping.session_id) == Some(&mapping) { + return Ok(()); + } + let key = mapping.session_id.clone(); + let previous = mappings.insert(key.clone(), mapping); + if let Err(error) = persist_values(&self.path, mappings.values()) { + if let Some(previous) = previous { + mappings.insert(key, previous); + } else { + mappings.remove(&key); + } + return Err(error); + } + Ok(()) + } + + pub fn get(&self, session_id: &str) -> Option { + lock(&self.mappings).get(session_id).cloned() + } + + pub fn profile_generation(&self, session_id: &str, logical_agent_id: &str) -> u64 { + let mappings = lock(&self.mappings); + mappings.get(session_id).map_or_else( + || { + mappings + .values() + .filter(|mapping| mapping.logical_agent_id == logical_agent_id) + .map(|mapping| mapping.profile_generation) + .max() + .unwrap_or(0) + .saturating_add(1) + }, + |mapping| mapping.profile_generation, + ) + } + + pub fn remove(&self, session_id: &str) -> Result { + let mut mappings = lock(&self.mappings); + let Some(previous) = mappings.remove(session_id) else { + return Ok(false); + }; + if let Err(error) = persist_values(&self.path, mappings.values()) { + mappings.insert(session_id.to_owned(), previous); + return Err(error); + } + Ok(true) + } +} + pub struct MentraMemoryAdmin { store: HybridRuntimeStore, mappings_path: PathBuf, @@ -160,21 +255,7 @@ impl MentraMemoryAdmin { } fn load_mappings(path: &Path) -> Result, String> { - let bytes = match std::fs::read(path) { - Ok(bytes) => bytes, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => { - match std::fs::read(path.with_extension("json.bak")) { - Ok(bytes) => bytes, - Err(backup) if backup.kind() == std::io::ErrorKind::NotFound => { - return Ok(BTreeMap::new()); - } - Err(backup) => return Err(backup.to_string()), - } - } - Err(error) => return Err(error.to_string()), - }; - let values: Vec = - serde_json::from_slice(&bytes).map_err(|e| e.to_string())?; + let values: Vec = load_values(path)?; if values.len() > MAX_MAPPINGS || values.iter().any(|mapping| !valid_mapping(mapping)) { return Err("invalid or excessive memory-agent mappings".to_owned()); } @@ -190,6 +271,33 @@ fn load_mappings(path: &Path) -> Result, St Ok(mappings) } +fn load_values Deserialize<'de>>(path: &Path) -> Result, String> { + let bytes = match read_mapping_file(path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + match read_mapping_file(&path.with_extension("json.bak")) { + Ok(bytes) => bytes, + Err(backup) if backup.kind() == std::io::ErrorKind::NotFound => { + return Ok(Vec::new()); + } + Err(backup) => return Err(backup.to_string()), + } + } + Err(error) => return Err(error.to_string()), + }; + serde_json::from_slice(&bytes).map_err(|error| error.to_string()) +} + +fn read_mapping_file(path: &Path) -> std::io::Result> { + if std::fs::metadata(path)?.len() > MAX_MAPPING_BYTES { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "mapping file exceeds its bound", + )); + } + std::fs::read(path) +} + fn valid_mapping(mapping: &MemoryAgentMapping) -> bool { [ mapping.avatar_id.as_str(), @@ -203,6 +311,13 @@ fn valid_mapping(mapping: &MemoryAgentMapping) -> bool { fn persist_mappings<'a>( path: &Path, mappings: impl Iterator, +) -> Result<(), String> { + persist_values(path, mappings) +} + +fn persist_values<'a, T: Serialize + 'a>( + path: &Path, + mappings: impl Iterator, ) -> Result<(), String> { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent).map_err(|error| error.to_string())?; @@ -229,6 +344,20 @@ fn persist_mappings<'a>( Ok(()) } +fn valid_conversation_mapping(mapping: &ConversationAgentMapping) -> bool { + mapping.profile_generation > 0 + && [ + mapping.session_id.as_str(), + mapping.avatar_id.as_str(), + mapping.channel.as_str(), + mapping.logical_agent_id.as_str(), + mapping.mentra_agent_id.as_str(), + ] + .iter() + .all(|value| !value.is_empty() && value.len() <= 256 && !value.contains('\0')) + && matches!(mapping.channel.as_str(), "public_chat" | "direct_im") +} + fn lock(value: &Mutex) -> MutexGuard<'_, T> { value .lock() @@ -271,6 +400,29 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[test] + fn conversation_mapping_is_explicit_persisted_and_exactly_removed() { + let root = + std::env::temp_dir().join(format!("metacrate-conversation-map-{}", std::process::id())); + let path = root.join("conversation-agents.json"); + let mappings = MentraConversationMappings::new(path.clone()).unwrap(); + let mapping = ConversationAgentMapping { + session_id: "session-a".into(), + avatar_id: "avatar-a".into(), + channel: "direct_im".into(), + logical_agent_id: "logical-a".into(), + mentra_agent_id: "mentra-a".into(), + profile_generation: 1, + }; + mappings.register(mapping.clone()).unwrap(); + drop(mappings); + let reopened = MentraConversationMappings::new(path).unwrap(); + assert_eq!(reopened.get("session-a"), Some(mapping)); + assert!(reopened.remove("session-a").unwrap()); + assert!(reopened.get("session-a").is_none()); + let _ = std::fs::remove_dir_all(root); + } + fn record(id: &str, agent_id: &str) -> MemoryRecord { MemoryRecord { record_id: id.to_owned(), diff --git a/crates/metacrate-grid-agent/src/tui.rs b/crates/metacrate-grid-agent/src/tui.rs index 6c876a3..a5d03ea 100644 --- a/crates/metacrate-grid-agent/src/tui.rs +++ b/crates/metacrate-grid-agent/src/tui.rs @@ -7,7 +7,8 @@ use crate::control_plane::{ AuditEventView, CONTROL_PROTOCOL_VERSION, ControlEvent, ControlEventKind, ControlPayload, - ControlRequest, ControlRequestEnvelope, ControlResponseEnvelope, HealthView, + ControlRequest, ControlRequestEnvelope, ControlResponseEnvelope, ConversationDetailView, + ConversationFilterView, ConversationMetadataView, ConversationStatusView, HealthView, InProcessControlClient, MemoryAgentView, MemoryCursorView, MemoryDetailView, MemoryFilterView, MemoryHealthView, MemoryKindView, MemoryRecordView, MemorySortView, Page, PageRequest, PendingApprovalView, RuntimeView, ScheduledJobView, SessionMetadataView, TcpControlClient, @@ -339,6 +340,8 @@ pub struct OperatorSnapshot { pub runtime: Option, pub metrics: Option, pub sessions: Vec, + pub conversations: Vec, + pub conversation_detail: Option, pub schedules: Vec, pub approvals: Vec, pub memory_health: Option, @@ -358,6 +361,8 @@ impl Default for OperatorSnapshot { runtime: None, metrics: None, sessions: Vec::new(), + conversations: Vec::new(), + conversation_detail: None, schedules: Vec::new(), approvals: Vec::new(), memory_health: None, @@ -397,6 +402,9 @@ pub enum OperatorCommand { avatar_id: String, direct_im: bool, }, + DeleteConversation { + session_id: String, + }, ToggleSchedule { job_id: String, enabled: bool, @@ -432,6 +440,9 @@ impl OperatorCommand { crate::control_plane::ConversationChannelView::PublicChat }, }, + Self::DeleteConversation { session_id } => ControlRequest::DeleteConversation { + session_id: session_id.clone(), + }, Self::ToggleSchedule { job_id, enabled } => ControlRequest::SetRoamingJob { job_id: job_id.clone(), enabled: *enabled, @@ -456,7 +467,9 @@ impl OperatorCommand { | Self::Reconnect | Self::ExpireSession { .. } | Self::ToggleSchedule { .. } => CommandConfirmation::Normal, - Self::ForgetMemory { .. } | Self::Shutdown => CommandConfirmation::HighRisk, + Self::ForgetMemory { .. } | Self::DeleteConversation { .. } | Self::Shutdown => { + CommandConfirmation::HighRisk + } } } } @@ -491,6 +504,10 @@ pub enum TuiInput { FilterBackspace, FilterCommit, FilterCancel, + ConversationLoad, + ConversationClose, + ConversationNextPage, + ConversationPreviousPage, MemoryLoad, MemoryNextAgent, MemoryNextPage, @@ -653,6 +670,10 @@ pub struct OperatorTui { memory_current_offset: u64, memory_next_offset: Option, memory_refresh_pending: bool, + conversation_cursors: Vec, + conversation_next_cursor: Option, + conversation_detail_cursor: u64, + conversation_detail_next: Option, } impl Default for OperatorTui { @@ -683,6 +704,10 @@ impl Default for OperatorTui { memory_current_offset: 0, memory_next_offset: None, memory_refresh_pending: false, + conversation_cursors: Vec::new(), + conversation_next_cursor: None, + conversation_detail_cursor: 0, + conversation_detail_next: None, } } } @@ -752,9 +777,45 @@ impl OperatorTui { self.filter.query = query.join(" "); } + fn conversation_request(&self) -> ControlRequest { + let mut filter = ConversationFilterView::default(); + for token in self.filter.query.split_whitespace() { + if let Some(value) = token.strip_prefix("avatar:") { + filter.avatar = (!value.is_empty()).then(|| value.to_owned()); + } else if let Some(value) = token.strip_prefix("channel:") { + filter.channel = match value { + "public" | "public_chat" => { + Some(crate::control_plane::ConversationChannelView::PublicChat) + } + "im" | "direct_im" => { + Some(crate::control_plane::ConversationChannelView::DirectIm) + } + _ => None, + }; + } else if let Some(value) = token.strip_prefix("status:") { + filter.status = match value { + "active" => Some(ConversationStatusView::Active), + "expired" => Some(ConversationStatusView::Expired), + _ => None, + }; + } else if let Some(value) = token.strip_prefix("from:") { + filter.active_from_unix_millis = value.parse().ok(); + } else if let Some(value) = token.strip_prefix("to:") { + filter.active_to_unix_millis = value.parse().ok(); + } + } + ControlRequest::ListConversations { + page: PageRequest { + cursor: self.conversation_cursors.last().copied(), + limit: 50, + }, + filter, + } + } + fn visible_rows(&self) -> usize { match self.screen { - OperatorScreen::Sessions => self.snapshot.sessions.len(), + OperatorScreen::Sessions => self.snapshot.conversations.len(), OperatorScreen::Memories => self.snapshot.memories.len(), OperatorScreen::Roaming => self.snapshot.schedules.len(), OperatorScreen::Approvals => self.snapshot.approvals.len(), @@ -801,6 +862,15 @@ impl OperatorTui { record_id: record.record_id.clone(), }); } + if self.screen == OperatorScreen::Sessions && key == 'd' { + return self + .snapshot + .conversations + .get(self.selected_row()) + .map(|session| OperatorCommand::DeleteConversation { + session_id: session.session_id.clone(), + }); + } match key { 'p' => Some(OperatorCommand::Pause), 'u' => Some(OperatorCommand::Resume), @@ -822,11 +892,12 @@ impl OperatorTui { }), 'e' => self .snapshot - .sessions + .conversations .get(self.selected[Self::screen_index(OperatorScreen::Sessions)]) .map(|session| OperatorCommand::ExpireSession { avatar_id: session.avatar_id.clone(), - direct_im: session.channel == "direct_im", + direct_im: session.channel + == crate::control_plane::ConversationChannelView::DirectIm, }), 't' => self .snapshot @@ -853,6 +924,9 @@ impl OperatorTui { self.scroll = 0; if self.screen == OperatorScreen::Memories { TuiAction::Request(ControlRequest::ListMemoryAgents { page: page() }) + } else if self.screen == OperatorScreen::Sessions { + self.conversation_cursors.clear(); + TuiAction::Request(self.conversation_request()) } else { TuiAction::None } @@ -867,17 +941,32 @@ impl OperatorTui { self.scroll = 0; if self.screen == OperatorScreen::Memories { TuiAction::Request(ControlRequest::ListMemoryAgents { page: page() }) + } else if self.screen == OperatorScreen::Sessions { + self.conversation_cursors.clear(); + TuiAction::Request(self.conversation_request()) } else { TuiAction::None } } TuiInput::ScrollUp => { + if self.screen == OperatorScreen::Sessions + && self.snapshot.conversation_detail.is_some() + { + self.scroll = self.scroll.saturating_sub(1); + return TuiAction::None; + } let index = Self::screen_index(self.screen); self.selected[index] = self.selected[index].saturating_sub(1); self.scroll = self.selected[index]; TuiAction::None } TuiInput::ScrollDown => { + if self.screen == OperatorScreen::Sessions + && self.snapshot.conversation_detail.is_some() + { + self.scroll = self.scroll.saturating_add(1); + return TuiAction::None; + } let index = Self::screen_index(self.screen); self.selected[index] = self.selected[index] .saturating_add(1) @@ -945,6 +1034,8 @@ impl OperatorTui { self.filter_editing = true; self.status = if self.screen == OperatorScreen::Memories { "memory search/filter: type query, Enter applies, Esc clears".into() + } else if self.screen == OperatorScreen::Sessions { + "conversation filter: avatar:/channel:/status:/from:/to:".into() } else { "event/audit filter: type query, Enter applies, Esc clears".into() }; @@ -993,6 +1084,69 @@ impl OperatorTui { TuiAction::None } } + TuiInput::ConversationLoad => self + .snapshot + .conversations + .get(self.selected_row()) + .map_or(TuiAction::None, |session| { + TuiAction::Request(ControlRequest::GetConversation { + session_id: session.session_id.clone(), + page: PageRequest { + cursor: None, + limit: 50, + }, + }) + }), + TuiInput::ConversationClose => { + self.snapshot.conversation_detail = None; + self.conversation_detail_cursor = 0; + self.conversation_detail_next = None; + self.scroll = 0; + TuiAction::None + } + TuiInput::ConversationNextPage => { + if let Some(detail) = &self.snapshot.conversation_detail { + if let Some(cursor) = self.conversation_detail_next { + self.conversation_detail_cursor = cursor; + return TuiAction::Request(ControlRequest::GetConversation { + session_id: detail.conversation.session_id.clone(), + page: PageRequest { + cursor: Some(cursor), + limit: 50, + }, + }); + } + return TuiAction::None; + } + if let Some(cursor) = self.conversation_next_cursor { + self.conversation_cursors.push(cursor); + TuiAction::Request(self.conversation_request()) + } else { + TuiAction::None + } + } + TuiInput::ConversationPreviousPage => { + if let Some(detail) = &self.snapshot.conversation_detail { + if self.conversation_detail_cursor > 0 { + self.conversation_detail_cursor = + self.conversation_detail_cursor.saturating_sub(50); + return TuiAction::Request(ControlRequest::GetConversation { + session_id: detail.conversation.session_id.clone(), + page: PageRequest { + cursor: (self.conversation_detail_cursor > 0) + .then_some(self.conversation_detail_cursor), + limit: 50, + }, + }); + } + return TuiAction::None; + } + if self.conversation_cursors.pop().is_some() { + TuiAction::Request(self.conversation_request()) + } else { + TuiAction::None + } + } TuiInput::MemoryLoad => { if let (Some(agent), Some(record)) = ( self.snapshot.memory_agents.get(self.memory_agent), @@ -1135,7 +1289,7 @@ impl OperatorTui { ControlRequest::Runtime, ControlRequest::Metrics, ControlRequest::MemoryHealth, - ControlRequest::ListSessions { page: page() }, + self.conversation_request(), ControlRequest::ListScheduledJobs { page: page() }, ControlRequest::ListPendingApprovals { page: page() }, ControlRequest::ListAuditEvents { page: page() }, @@ -1269,6 +1423,26 @@ impl OperatorTui { .get(self.selected[index]) .map(|item| item.session_id.clone()); self.snapshot.sessions = items; + self.snapshot.conversations = self + .snapshot + .sessions + .iter() + .map(|item| ConversationMetadataView { + session_id: item.session_id.clone(), + avatar_id: item.avatar_id.clone(), + channel: if item.channel == "direct_im" { + crate::control_plane::ConversationChannelView::DirectIm + } else { + crate::control_plane::ConversationChannelView::PublicChat + }, + status: ConversationStatusView::Active, + created_unix_millis: item.created_unix_millis, + last_active_unix_millis: item.last_active_unix_millis, + turns: item.turns, + bytes: item.bytes, + compacted: false, + }) + .collect(); self.selected[index] = identity .and_then(|id| { self.snapshot @@ -1279,6 +1453,39 @@ impl OperatorTui { .unwrap_or(0) .min(self.snapshot.sessions.len().saturating_sub(1)); } + ControlPayload::Conversations(Page { items, next_cursor }) => { + let index = Self::screen_index(OperatorScreen::Sessions); + let identity = self + .snapshot + .conversations + .get(self.selected[index]) + .map(|item| item.session_id.clone()); + self.snapshot.conversations = items; + self.conversation_next_cursor = next_cursor; + self.snapshot.conversation_detail = None; + self.selected[index] = identity + .and_then(|id| { + self.snapshot + .conversations + .iter() + .position(|item| item.session_id == id) + }) + .unwrap_or(0) + .min(self.snapshot.conversations.len().saturating_sub(1)); + } + ControlPayload::ConversationDetail(detail) => { + self.conversation_detail_next = detail.entries.next_cursor.or_else(|| { + let total = detail + .active_transcript_entries + .saturating_add(detail.archived_transcript_entries); + let end = usize::try_from(self.conversation_detail_cursor) + .unwrap_or(usize::MAX) + .saturating_add(detail.transcript_entries.len()); + (end < total).then(|| u64::try_from(end).unwrap_or(u64::MAX)) + }); + self.snapshot.conversation_detail = Some(detail); + self.scroll = 0; + } ControlPayload::ScheduledJobs(Page { items, .. }) => { let index = Self::screen_index(OperatorScreen::Roaming); let identity = self @@ -1701,7 +1908,7 @@ impl OperatorTui { fn draw_sessions(&self, frame: &mut Frame<'_>, area: Rect, color: bool) { let rows = self .snapshot - .sessions + .conversations .iter() .enumerate() .skip(self.selected_row().saturating_sub(5)) @@ -1712,24 +1919,32 @@ impl OperatorTui { vec![ item.session_id.clone(), item.avatar_id.clone(), - item.channel.clone(), + format!("{:?}", item.channel).to_lowercase(), item.turns.to_string(), - item.bytes.to_string(), + format!("{} {:?}", item.bytes, item.status).to_lowercase(), ], ) }); - let detail = self.snapshot.sessions.get(self.selected_row()).map_or_else( - || vec![Line::raw("No active sessions")], + let detail = self.snapshot.conversations.get(self.selected_row()).map_or_else( + || vec![Line::raw("No conversations")], |item| { - vec![ + let mut lines = vec![ Line::raw(format!("session: {}", item.session_id)), Line::raw(format!("avatar: {}", item.avatar_id)), Line::raw(format!( - "created: {} last active: {}", - item.created_unix_millis, item.last_active_unix_millis + "status: {:?} compacted={} created: {} last: {}", + item.status, item.compacted, item.created_unix_millis, item.last_active_unix_millis )), - Line::raw("[e] expire conversation (confirmation required)"), - ] + Line::raw("Enter detail Esc list n/b page Up/Down scroll / filter [e] expire [d] delete"), + ]; + if let Some(detail) = &self.snapshot.conversation_detail + && detail.conversation.session_id == item.session_id + { + lines.push(Line::raw(format!("Meta entries={} Mentra active={} archived={} running={}", detail.entries.items.len(), detail.active_transcript_entries, detail.archived_transcript_entries, detail.transcript_busy))); + lines.extend(detail.entries.items.iter().map(|entry| Line::raw(format!("{} {} {}: {}", entry.sequence, entry.role, entry.record_kind, entry.text)))); + lines.extend(detail.transcript_entries.iter().map(|entry| Line::raw(format!("Mentra {} {} active={}", entry.entry_id, entry.kind, entry.active)))); + } + lines.into_iter().skip(self.scroll).collect() }, ); Self::draw_table_detail( @@ -1737,7 +1952,7 @@ impl OperatorTui { area, color, "Sessions", - ["Session", "Avatar", "Channel", "Turns", "Bytes"], + ["Session", "Avatar", "Channel", "Turns", "Bytes/status"], rows, detail, ); @@ -2935,6 +3150,7 @@ async fn worker_send( .map_err(|error| TuiError(format!("{:?}: {}", error.code, error.message))) } +#[allow(clippy::too_many_lines)] // One exhaustive key map keeps screen precedence visible. fn map_event(app: &OperatorTui, value: &event::Event) -> Option { let event::Event::Key(key) = value else { return matches!(value, event::Event::Resize(_, _)).then_some(TuiInput::Refresh); @@ -2980,6 +3196,12 @@ fn map_event(app: &OperatorTui, value: &event::Event) -> Option { }; } match key.code { + event::KeyCode::Esc + if app.screen == OperatorScreen::Sessions + && app.snapshot.conversation_detail.is_some() => + { + Some(TuiInput::ConversationClose) + } event::KeyCode::Tab | event::KeyCode::Right => Some(TuiInput::NextScreen), event::KeyCode::BackTab | event::KeyCode::Left => Some(TuiInput::PreviousScreen), event::KeyCode::Up => Some(TuiInput::ScrollUp), @@ -2987,7 +3209,7 @@ fn map_event(app: &OperatorTui, value: &event::Event) -> Option { event::KeyCode::Char('/') if matches!( app.screen, - OperatorScreen::Timeline | OperatorScreen::Memories + OperatorScreen::Timeline | OperatorScreen::Memories | OperatorScreen::Sessions ) => { Some(TuiInput::BeginFilter) @@ -2995,15 +3217,24 @@ fn map_event(app: &OperatorTui, value: &event::Event) -> Option { event::KeyCode::Enter if app.screen == OperatorScreen::Memories => { Some(TuiInput::MemoryLoad) } + event::KeyCode::Enter if app.screen == OperatorScreen::Sessions => { + Some(TuiInput::ConversationLoad) + } event::KeyCode::Char('g') if app.screen == OperatorScreen::Memories => { Some(TuiInput::MemoryNextAgent) } event::KeyCode::Char('n') if app.screen == OperatorScreen::Memories => { Some(TuiInput::MemoryNextPage) } + event::KeyCode::Char('n') if app.screen == OperatorScreen::Sessions => { + Some(TuiInput::ConversationNextPage) + } event::KeyCode::Char('b') if app.screen == OperatorScreen::Memories => { Some(TuiInput::MemoryPreviousPage) } + event::KeyCode::Char('b') if app.screen == OperatorScreen::Sessions => { + Some(TuiInput::ConversationPreviousPage) + } event::KeyCode::Char('k') if app.screen == OperatorScreen::Memories => { Some(TuiInput::MemoryCycleKind) } @@ -3088,6 +3319,43 @@ mod ratatui_private_tests { app.command_shortcut('e'), Some(OperatorCommand::ExpireSession { avatar_id, .. }) if avatar_id == "avatar-b" )); + assert!(matches!( + app.command_shortcut('d'), + Some(OperatorCommand::DeleteConversation { session_id }) if session_id == "b" + )); + assert!(matches!( + app.reduce(TuiInput::ConversationLoad), + TuiAction::Request(ControlRequest::GetConversation { session_id, .. }) if session_id == "b" + )); + app.snapshot.conversation_detail = Some(ConversationDetailView { + conversation: app.snapshot.conversations[app.selected_row()].clone(), + entries: Page { + items: Vec::new(), + next_cursor: Some(50), + }, + mentra_agent_id: None, + profile_generation: None, + active_transcript_entries: 0, + archived_transcript_entries: 0, + transcript_entries: Vec::new(), + transcript_busy: false, + }); + app.conversation_detail_next = Some(50); + assert!(matches!( + app.reduce(TuiInput::ConversationNextPage), + TuiAction::Request(ControlRequest::GetConversation { + page: PageRequest { + cursor: Some(50), + .. + }, + .. + }) + )); + assert_eq!(app.reduce(TuiInput::ConversationClose), TuiAction::None); + assert_eq!( + app.snapshot.conversations[app.selected_row()].session_id, + "b" + ); app.apply(ControlPayload::Sessions(Page { items: (0..20).map(|index| session(&index.to_string())).collect(), diff --git a/vendor/mentra/src/agent.rs b/vendor/mentra/src/agent.rs index 9e49ea4..14d777b 100644 --- a/vendor/mentra/src/agent.rs +++ b/vendor/mentra/src/agent.rs @@ -415,6 +415,14 @@ impl Agent { self.memory.transcript() } + /// Clears active and archived transcript entries while preserving the + /// agent identity, configuration, and long-term memory namespace. + /// Clears active and archived working transcript state while preserving + /// this agent's identity and durable long-term-memory namespace. + pub fn clear_transcript(&mut self) -> Result<(), RuntimeError> { + self.memory.clear_transcript() + } + /// The transcript entry the next turn will continue from. pub fn leaf(&self) -> Option<&crate::transcript::EntryId> { self.transcript().leaf() diff --git a/vendor/mentra/src/memory/journal/ops.rs b/vendor/mentra/src/memory/journal/ops.rs index c707859..cecad1b 100644 --- a/vendor/mentra/src/memory/journal/ops.rs +++ b/vendor/mentra/src/memory/journal/ops.rs @@ -165,6 +165,15 @@ impl AgentMemory { &self.history_cache } + pub fn clear_transcript(&mut self) -> Result<(), RuntimeError> { + self.state.transcript = AgentTranscript::default(); + self.state.pending_turn = None; + self.state.resumable_user_message = None; + self.state.run = None; + self.sync_history_cache(); + self.persist() + } + pub fn revision(&self) -> u64 { self.state.revision } diff --git a/vendor/mentra/src/memory/journal/tests.rs b/vendor/mentra/src/memory/journal/tests.rs index 8169986..658cdc5 100644 --- a/vendor/mentra/src/memory/journal/tests.rs +++ b/vendor/mentra/src/memory/journal/tests.rs @@ -5,6 +5,7 @@ use serde_json::json; use crate::{ AgentTranscript, ContentBlock, Message, TranscriptKind, memory::journal::{AgentMemory, AgentMemoryState, CompactionOutcome, PendingTurnState}, + memory::{MemoryRecord, MemoryRecordKind, MemoryStore}, runtime::VolatileRuntimeStore, }; @@ -42,6 +43,42 @@ fn begin_run_commit_and_finish_persist_memory_state() { assert!(memory.resumable_user_message().is_none()); } +#[test] +fn clear_transcript_persists_an_empty_history() { + let store = Arc::new(VolatileRuntimeStore::new()); + let mut memory = AgentMemory::new("agent-clear", store.clone(), AgentMemoryState::default()); + store + .upsert_records(&[MemoryRecord { + record_id: "fact:kept".into(), + agent_id: "agent-clear".into(), + kind: MemoryRecordKind::Fact, + content: "durable preference".into(), + source_revision: 1, + created_at: 1, + metadata_json: "{}".into(), + source: None, + pinned: true, + score: None, + }]) + .unwrap(); + memory + .append_message(Message::user(ContentBlock::text("private turn"))) + .unwrap(); + memory.clear_transcript().unwrap(); + assert!(memory.transcript().is_empty()); + assert!(memory.history().is_empty()); + assert!( + memory.revision() >= 2, + "clear is persisted as a new revision" + ); + assert!( + store + .get_record("agent-clear", "fact:kept") + .unwrap() + .is_some() + ); +} + #[test] fn rollback_and_compaction_update_memory_state() { let store = Arc::new(VolatileRuntimeStore::new());