Implement conversation browser and safe deletion (#138)
This commit is contained in:
@@ -172,7 +172,7 @@ impl PageRequest {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
|
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
|
||||||
pub struct Page<T> {
|
pub struct Page<T> {
|
||||||
pub items: Vec<T>,
|
pub items: Vec<T>,
|
||||||
pub next_cursor: Option<u64>,
|
pub next_cursor: Option<u64>,
|
||||||
@@ -220,6 +220,66 @@ pub struct SessionMetadataView {
|
|||||||
pub bytes: usize,
|
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<String>,
|
||||||
|
pub channel: Option<ConversationChannelView>,
|
||||||
|
pub status: Option<ConversationStatusView>,
|
||||||
|
pub active_from_unix_millis: Option<u64>,
|
||||||
|
pub active_to_unix_millis: Option<u64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[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<String>,
|
||||||
|
pub kind: String,
|
||||||
|
pub active: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
|
||||||
|
pub struct ConversationDetailView {
|
||||||
|
pub conversation: ConversationMetadataView,
|
||||||
|
pub entries: Page<ConversationEntryView>,
|
||||||
|
pub mentra_agent_id: Option<String>,
|
||||||
|
pub profile_generation: Option<u64>,
|
||||||
|
pub active_transcript_entries: usize,
|
||||||
|
pub archived_transcript_entries: usize,
|
||||||
|
pub transcript_entries: Vec<TranscriptEntryView>,
|
||||||
|
pub transcript_busy: bool,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
|
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
|
||||||
pub struct ScheduledJobView {
|
pub struct ScheduledJobView {
|
||||||
pub job_id: String,
|
pub job_id: String,
|
||||||
@@ -331,6 +391,8 @@ pub enum ControlPayload {
|
|||||||
Metrics(MetricsSnapshot),
|
Metrics(MetricsSnapshot),
|
||||||
Runtime(RuntimeView),
|
Runtime(RuntimeView),
|
||||||
Sessions(Page<SessionMetadataView>),
|
Sessions(Page<SessionMetadataView>),
|
||||||
|
Conversations(Page<ConversationMetadataView>),
|
||||||
|
ConversationDetail(ConversationDetailView),
|
||||||
ScheduledJobs(Page<ScheduledJobView>),
|
ScheduledJobs(Page<ScheduledJobView>),
|
||||||
PendingApprovals(Page<PendingApprovalView>),
|
PendingApprovals(Page<PendingApprovalView>),
|
||||||
AuditEvents(Page<AuditEventView>),
|
AuditEvents(Page<AuditEventView>),
|
||||||
@@ -361,6 +423,17 @@ pub enum ControlRequest {
|
|||||||
ListSessions {
|
ListSessions {
|
||||||
page: PageRequest,
|
page: PageRequest,
|
||||||
},
|
},
|
||||||
|
ListConversations {
|
||||||
|
page: PageRequest,
|
||||||
|
filter: ConversationFilterView,
|
||||||
|
},
|
||||||
|
GetConversation {
|
||||||
|
session_id: String,
|
||||||
|
page: PageRequest,
|
||||||
|
},
|
||||||
|
DeleteConversation {
|
||||||
|
session_id: String,
|
||||||
|
},
|
||||||
ListScheduledJobs {
|
ListScheduledJobs {
|
||||||
page: PageRequest,
|
page: PageRequest,
|
||||||
},
|
},
|
||||||
@@ -449,6 +522,8 @@ impl ControlRequest {
|
|||||||
| Self::Metrics
|
| Self::Metrics
|
||||||
| Self::Runtime
|
| Self::Runtime
|
||||||
| Self::ListSessions { .. }
|
| Self::ListSessions { .. }
|
||||||
|
| Self::ListConversations { .. }
|
||||||
|
| Self::GetConversation { .. }
|
||||||
| Self::ListScheduledJobs { .. }
|
| Self::ListScheduledJobs { .. }
|
||||||
| Self::ListPendingApprovals { .. }
|
| Self::ListPendingApprovals { .. }
|
||||||
| Self::ListAuditEvents { .. }
|
| Self::ListAuditEvents { .. }
|
||||||
@@ -470,6 +545,21 @@ impl ControlRequest {
|
|||||||
| Self::ListAuditEvents { page }
|
| Self::ListAuditEvents { page }
|
||||||
| Self::ListObservabilityEvents { page }
|
| Self::ListObservabilityEvents { page }
|
||||||
| Self::ListMemoryAgents { page } => page.valid(),
|
| 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 {
|
Self::BrowseMemories {
|
||||||
mentra_agent_id,
|
mentra_agent_id,
|
||||||
cursor,
|
cursor,
|
||||||
@@ -1888,6 +1978,11 @@ fn memory_filter_valid(filter: &MemoryFilterView) -> bool {
|
|||||||
fn payload_valid(payload: &ControlPayload, maximum_frame_bytes: usize) -> bool {
|
fn payload_valid(payload: &ControlPayload, maximum_frame_bytes: usize) -> bool {
|
||||||
let page_is_bounded = match payload {
|
let page_is_bounded = match payload {
|
||||||
ControlPayload::Sessions(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE),
|
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::ScheduledJobs(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE),
|
||||||
ControlPayload::PendingApprovals(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),
|
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::Metrics => "metrics",
|
||||||
ControlRequest::Runtime => "runtime",
|
ControlRequest::Runtime => "runtime",
|
||||||
ControlRequest::ListSessions { .. } => "list_sessions",
|
ControlRequest::ListSessions { .. } => "list_sessions",
|
||||||
|
ControlRequest::ListConversations { .. } => "list_conversations",
|
||||||
|
ControlRequest::GetConversation { .. } => "get_conversation",
|
||||||
|
ControlRequest::DeleteConversation { .. } => "delete_conversation",
|
||||||
ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs",
|
ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs",
|
||||||
ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals",
|
ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals",
|
||||||
ControlRequest::ListAuditEvents { .. } => "list_audit_events",
|
ControlRequest::ListAuditEvents { .. } => "list_audit_events",
|
||||||
|
|||||||
@@ -6,9 +6,11 @@ use crate::behavior::{BehaviorIngress, BehaviorMode};
|
|||||||
use crate::control_plane::{
|
use crate::control_plane::{
|
||||||
AuditEventView, CONTROL_PROTOCOL_VERSION, ControlContext, ControlError, ControlErrorCode,
|
AuditEventView, CONTROL_PROTOCOL_VERSION, ControlContext, ControlError, ControlErrorCode,
|
||||||
ControlFuture, ControlPayload, ControlRequest, ControlTarget, ConversationChannelView,
|
ControlFuture, ControlPayload, ControlRequest, ControlTarget, ConversationChannelView,
|
||||||
HealthView, MemoryAgentView, MemoryCursorView, MemoryDetailView, MemoryFilterView,
|
ConversationDetailView, ConversationEntryView, ConversationFilterView,
|
||||||
MemoryHealthView, MemoryKindView, MemoryPageView, MemoryRecordView, MemorySortView, Page,
|
ConversationMetadataView, ConversationStatusView, HealthView, MemoryAgentView,
|
||||||
PageRequest, PendingApprovalView, RuntimeView, ScheduledJobView, SessionMetadataView,
|
MemoryCursorView, MemoryDetailView, MemoryFilterView, MemoryHealthView, MemoryKindView,
|
||||||
|
MemoryPageView, MemoryRecordView, MemorySortView, Page, PageRequest, PendingApprovalView,
|
||||||
|
RuntimeView, ScheduledJobView, SessionMetadataView, TranscriptEntryView,
|
||||||
};
|
};
|
||||||
use crate::conversation::{ConversationChannel, ConversationKey, ConversationStore};
|
use crate::conversation::{ConversationChannel, ConversationKey, ConversationStore};
|
||||||
use crate::observability::{
|
use crate::observability::{
|
||||||
@@ -90,6 +92,7 @@ pub struct AgentControlTarget {
|
|||||||
build: Mutex<Option<Arc<dyn crate::build::BuildControl>>>,
|
build: Mutex<Option<Arc<dyn crate::build::BuildControl>>>,
|
||||||
vision: Mutex<Option<Arc<dyn crate::vision::VisionControl>>>,
|
vision: Mutex<Option<Arc<dyn crate::vision::VisionControl>>>,
|
||||||
memory: Mutex<Option<Arc<crate::memory_admin::MentraMemoryAdmin>>>,
|
memory: Mutex<Option<Arc<crate::memory_admin::MentraMemoryAdmin>>>,
|
||||||
|
conversation_admin: Mutex<Option<Arc<crate::interaction::PolicyLlmResponder>>>,
|
||||||
behavior: BehaviorIngress,
|
behavior: BehaviorIngress,
|
||||||
commands: mpsc::Sender<RuntimeControlCommand>,
|
commands: mpsc::Sender<RuntimeControlCommand>,
|
||||||
command_capacity: usize,
|
command_capacity: usize,
|
||||||
@@ -136,6 +139,7 @@ impl AgentControlTarget {
|
|||||||
build: Mutex::new(None),
|
build: Mutex::new(None),
|
||||||
vision: Mutex::new(None),
|
vision: Mutex::new(None),
|
||||||
memory: Mutex::new(None),
|
memory: Mutex::new(None),
|
||||||
|
conversation_admin: Mutex::new(None),
|
||||||
behavior,
|
behavior,
|
||||||
commands,
|
commands,
|
||||||
command_capacity,
|
command_capacity,
|
||||||
@@ -165,6 +169,13 @@ impl AgentControlTarget {
|
|||||||
*lock(&self.memory) = Some(memory);
|
*lock(&self.memory) = Some(memory);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn attach_conversation_control(
|
||||||
|
&self,
|
||||||
|
responder: Arc<crate::interaction::PolicyLlmResponder>,
|
||||||
|
) {
|
||||||
|
*lock(&self.conversation_admin) = Some(responder);
|
||||||
|
}
|
||||||
|
|
||||||
pub fn update_session(&self, session: SessionStatus) {
|
pub fn update_session(&self, session: SessionStatus) {
|
||||||
let mut state = lock(&self.state);
|
let mut state = lock(&self.state);
|
||||||
state.session = session;
|
state.session = session;
|
||||||
@@ -198,6 +209,113 @@ impl AgentControlTarget {
|
|||||||
lock(&self.state).service_state = "stopping";
|
lock(&self.state).service_state = "stopping";
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn get_conversation(
|
||||||
|
&self,
|
||||||
|
context: &ControlContext,
|
||||||
|
session_id: &str,
|
||||||
|
page: &PageRequest,
|
||||||
|
) -> Result<ControlPayload, ControlError> {
|
||||||
|
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<ControlPayload, ControlError> {
|
||||||
|
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(
|
fn enqueue(
|
||||||
&self,
|
&self,
|
||||||
operation: &'static str,
|
operation: &'static str,
|
||||||
@@ -315,6 +433,26 @@ impl AgentControlTarget {
|
|||||||
.collect();
|
.collect();
|
||||||
Ok(ControlPayload::Sessions(page_values(&page, values)))
|
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 } => {
|
ControlRequest::ListScheduledJobs { page } => {
|
||||||
let values = lock(&self.jobs)
|
let values = lock(&self.jobs)
|
||||||
.iter()
|
.iter()
|
||||||
@@ -694,7 +832,15 @@ impl ControlTarget for AgentControlTarget {
|
|||||||
let operation = runtime_request_name(&request);
|
let operation = runtime_request_name(&request);
|
||||||
let correlation = request_correlation(&request);
|
let correlation = request_correlation(&request);
|
||||||
let domain_event = runtime_domain_event(&request).ok().flatten();
|
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() {
|
if let Some(observability) = lock(&self.observability).as_ref() {
|
||||||
let (severity, result_code) = match &result {
|
let (severity, result_code) = match &result {
|
||||||
Ok(_) => (EventSeverity::Info, "accepted"),
|
Ok(_) => (EventSeverity::Info, "accepted"),
|
||||||
@@ -836,6 +982,8 @@ fn request_correlation(request: &ControlRequest) -> CorrelationIds {
|
|||||||
Some(format!("approval-{approval_id}"))
|
Some(format!("approval-{approval_id}"))
|
||||||
}
|
}
|
||||||
ControlRequest::ForgetMemory { record_id, .. } => Some(record_id.clone()),
|
ControlRequest::ForgetMemory { record_id, .. } => Some(record_id.clone()),
|
||||||
|
ControlRequest::GetConversation { session_id, .. }
|
||||||
|
| ControlRequest::DeleteConversation { session_id } => Some(session_id.clone()),
|
||||||
_ => None,
|
_ => None,
|
||||||
};
|
};
|
||||||
CorrelationIds {
|
CorrelationIds {
|
||||||
@@ -850,6 +998,9 @@ const fn runtime_request_name(request: &ControlRequest) -> &'static str {
|
|||||||
ControlRequest::Metrics => "metrics",
|
ControlRequest::Metrics => "metrics",
|
||||||
ControlRequest::Runtime => "runtime",
|
ControlRequest::Runtime => "runtime",
|
||||||
ControlRequest::ListSessions { .. } => "list_sessions",
|
ControlRequest::ListSessions { .. } => "list_sessions",
|
||||||
|
ControlRequest::ListConversations { .. } => "list_conversations",
|
||||||
|
ControlRequest::GetConversation { .. } => "get_conversation",
|
||||||
|
ControlRequest::DeleteConversation { .. } => "delete_conversation",
|
||||||
ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs",
|
ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs",
|
||||||
ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals",
|
ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals",
|
||||||
ControlRequest::ListAuditEvents { .. } => "list_audit_events",
|
ControlRequest::ListAuditEvents { .. } => "list_audit_events",
|
||||||
@@ -910,6 +1061,67 @@ fn page_values<T>(request: &PageRequest, values: Vec<T>) -> Page<T> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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 {
|
const fn behavior_name(mode: BehaviorMode) -> &'static str {
|
||||||
match mode {
|
match mode {
|
||||||
BehaviorMode::Offline => "offline",
|
BehaviorMode::Offline => "offline",
|
||||||
|
|||||||
@@ -1,10 +1,14 @@
|
|||||||
use crate::behavior::{BehaviorController, BehaviorError, EmbodimentFuture, EmbodimentSink};
|
use crate::behavior::{BehaviorController, BehaviorError, EmbodimentFuture, EmbodimentSink};
|
||||||
use crate::control_plane::{
|
use crate::control_plane::{
|
||||||
CONTROL_PROTOCOL_VERSION, ControlLimits, ControlPayload, ControlPlane, ControlRequest,
|
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::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::perception::WorldPosition;
|
||||||
use crate::policy::{MemoryPolicyAudit, PolicyAuditSink, PolicyGateway, PolicyLimits};
|
use crate::policy::{MemoryPolicyAudit, PolicyAuditSink, PolicyGateway, PolicyLimits};
|
||||||
use crate::{
|
use crate::{
|
||||||
@@ -146,6 +150,38 @@ async fn production_target_projects_state_and_routes_real_mutations() {
|
|||||||
let conversations = Arc::new(
|
let conversations = Arc::new(
|
||||||
ConversationStore::open(ConversationLimits::default(), None).expect("conversations"),
|
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) =
|
let (target, mut commands) =
|
||||||
AgentControlTarget::new(conversations, policy, audit, behavior.ingress(), 8)
|
AgentControlTarget::new(conversations, policy, audit, behavior.ingress(), 8)
|
||||||
.expect("target");
|
.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");
|
assert!(vision.active_capture().is_none(), "pause cancels capture");
|
||||||
|
|
||||||
let observer = plane.connect("observer-token").expect("observer");
|
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
|
let health = observer
|
||||||
.request(envelope("memory-health", ControlRequest::MemoryHealth))
|
.request(envelope("memory-health", ControlRequest::MemoryHealth))
|
||||||
.await;
|
.await;
|
||||||
@@ -327,9 +465,8 @@ async fn production_target_projects_state_and_routes_real_mutations() {
|
|||||||
.any(|event| event.family == "control_command")
|
.any(|event| event.family == "control_command")
|
||||||
);
|
);
|
||||||
assert!(events.items.iter().all(|event| {
|
assert!(events.items.iter().all(|event| {
|
||||||
!serde_json::to_string(event)
|
let json = serde_json::to_string(event).expect("event JSON");
|
||||||
.expect("event JSON")
|
!json.contains("operator-token") && !json.contains("bounded operator transcript")
|
||||||
.contains("operator-token")
|
|
||||||
}));
|
}));
|
||||||
behavior.shutdown().await.expect("shutdown behavior");
|
behavior.shutdown().await.expect("shutdown behavior");
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -30,6 +30,7 @@ const MAX_PERSISTED_BYTES: usize = 64 * 1024 * 1024;
|
|||||||
const MAX_SUMMARY_BYTES: usize = 64 * 1024;
|
const MAX_SUMMARY_BYTES: usize = 64 * 1024;
|
||||||
const RETAINED_SNAPSHOTS: usize = 2;
|
const RETAINED_SNAPSHOTS: usize = 2;
|
||||||
const MAX_MEMORY_EVENTS: usize = 8_192;
|
const MAX_MEMORY_EVENTS: usize = 8_192;
|
||||||
|
const MAX_RETIRED_SESSIONS: usize = 4_096;
|
||||||
|
|
||||||
/// A trust-separated conversation channel. Group and conference channels are
|
/// A trust-separated conversation channel. Group and conference channels are
|
||||||
/// deliberately unrepresentable until their authority semantics are defined.
|
/// deliberately unrepresentable until their authority semantics are defined.
|
||||||
@@ -266,6 +267,7 @@ pub enum ConversationError {
|
|||||||
IdentifierGeneration,
|
IdentifierGeneration,
|
||||||
PersistenceUnavailable,
|
PersistenceUnavailable,
|
||||||
PersistenceTooLarge,
|
PersistenceTooLarge,
|
||||||
|
StaleSession,
|
||||||
Boundary(crate::types::BoundaryError),
|
Boundary(crate::types::BoundaryError),
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -284,6 +286,7 @@ impl fmt::Display for ConversationError {
|
|||||||
Self::PersistenceTooLarge => {
|
Self::PersistenceTooLarge => {
|
||||||
formatter.write_str("conversation snapshot exceeds its storage bound")
|
formatter.write_str("conversation snapshot exceeds its storage bound")
|
||||||
}
|
}
|
||||||
|
Self::StaleSession => formatter.write_str("conversation session is no longer active"),
|
||||||
Self::Boundary(error) => {
|
Self::Boundary(error) => {
|
||||||
write!(formatter, "conversation boundary rejected data: {error}")
|
write!(formatter, "conversation boundary rejected data: {error}")
|
||||||
}
|
}
|
||||||
@@ -316,6 +319,19 @@ pub struct ConversationMetadata {
|
|||||||
pub last_active_unix_millis: u64,
|
pub last_active_unix_millis: u64,
|
||||||
pub turns: usize,
|
pub turns: usize,
|
||||||
pub bytes: 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.
|
/// LLM-only projection for one exact UUID/channel key.
|
||||||
@@ -325,6 +341,23 @@ pub struct ConversationContext {
|
|||||||
entries: Vec<ContextEntry>,
|
entries: Vec<ContextEntry>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[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<ConversationEntryProjection>,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||||
struct ContextEntry {
|
struct ContextEntry {
|
||||||
role: MessageRole,
|
role: MessageRole,
|
||||||
@@ -420,12 +453,17 @@ impl StoredSession {
|
|||||||
last_active_unix_millis: self.last_active_unix_millis,
|
last_active_unix_millis: self.last_active_unix_millis,
|
||||||
turns: self.entries.len(),
|
turns: self.entries.len(),
|
||||||
bytes: self.bytes,
|
bytes: self.bytes,
|
||||||
|
compacted: self
|
||||||
|
.entries
|
||||||
|
.iter()
|
||||||
|
.any(|entry| entry.kind == RecordKind::FactualSummary),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
struct StoreState {
|
struct StoreState {
|
||||||
sessions: BTreeMap<ConversationKey, StoredSession>,
|
sessions: BTreeMap<ConversationKey, StoredSession>,
|
||||||
|
retired: VecDeque<ConversationMetadata>,
|
||||||
total_bytes: usize,
|
total_bytes: usize,
|
||||||
next_sequence: u64,
|
next_sequence: u64,
|
||||||
snapshot_generation: u64,
|
snapshot_generation: u64,
|
||||||
@@ -481,6 +519,7 @@ impl ConversationStore {
|
|||||||
limits.validate()?;
|
limits.validate()?;
|
||||||
let mut state = StoreState {
|
let mut state = StoreState {
|
||||||
sessions: BTreeMap::new(),
|
sessions: BTreeMap::new(),
|
||||||
|
retired: VecDeque::new(),
|
||||||
total_bytes: 0,
|
total_bytes: 0,
|
||||||
next_sequence: 1,
|
next_sequence: 1,
|
||||||
snapshot_generation: 0,
|
snapshot_generation: 0,
|
||||||
@@ -502,6 +541,24 @@ impl ConversationStore {
|
|||||||
&self,
|
&self,
|
||||||
key: ConversationKey,
|
key: ConversationKey,
|
||||||
record: MemoryRecord,
|
record: MemoryRecord,
|
||||||
|
) -> Result<MemoryUpdate, ConversationError> {
|
||||||
|
self.append_checked(key, None, record)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn append_if_session(
|
||||||
|
&self,
|
||||||
|
key: ConversationKey,
|
||||||
|
session_id: &str,
|
||||||
|
record: MemoryRecord,
|
||||||
|
) -> Result<MemoryUpdate, ConversationError> {
|
||||||
|
self.append_checked(key, Some(session_id), record)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn append_checked(
|
||||||
|
&self,
|
||||||
|
key: ConversationKey,
|
||||||
|
expected_session_id: Option<&str>,
|
||||||
|
record: MemoryRecord,
|
||||||
) -> Result<MemoryUpdate, ConversationError> {
|
) -> Result<MemoryUpdate, ConversationError> {
|
||||||
if key.avatar_id == UUID::zero() {
|
if key.avatar_id == UUID::zero() {
|
||||||
return Err(ConversationError::InvalidAvatar);
|
return Err(ConversationError::InvalidAvatar);
|
||||||
@@ -528,6 +585,14 @@ impl ConversationStore {
|
|||||||
let mut state = self.lock_state();
|
let mut state = self.lock_state();
|
||||||
let event_start = state.events.len();
|
let event_start = state.events.len();
|
||||||
expire_idle_locked(&mut state, monotonic_now, None);
|
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) {
|
if !state.sessions.contains_key(&key) {
|
||||||
while state.sessions.len() >= self.limits.max_active_sessions {
|
while state.sessions.len() >= self.limits.max_active_sessions {
|
||||||
evict_oldest(&mut state, MemoryReason::SessionLimitEviction, None);
|
evict_oldest(&mut state, MemoryReason::SessionLimitEviction, None);
|
||||||
@@ -627,6 +692,140 @@ impl ConversationStore {
|
|||||||
.collect()
|
.collect()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[must_use]
|
||||||
|
pub fn list_operator_metadata(&self) -> Vec<OperatorConversationMetadata> {
|
||||||
|
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::<Vec<_>>();
|
||||||
|
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<ConversationDetailProjection> {
|
||||||
|
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<bool, ConversationError> {
|
||||||
|
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 {
|
pub fn delete(&self, key: ConversationKey) -> bool {
|
||||||
self.remove_operator(key, MemoryReason::OperatorDeleted)
|
self.remove_operator(key, MemoryReason::OperatorDeleted)
|
||||||
}
|
}
|
||||||
@@ -661,6 +860,9 @@ impl ConversationStore {
|
|||||||
return false;
|
return false;
|
||||||
};
|
};
|
||||||
state.total_bytes = state.total_bytes.saturating_sub(session.bytes);
|
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));
|
push_event(&mut state.events, event_for(reason, &session));
|
||||||
true
|
true
|
||||||
}
|
}
|
||||||
@@ -801,6 +1003,7 @@ fn expire_idle_locked(state: &mut StoreState, now: Instant, only: Option<Convers
|
|||||||
ConversationChannel::PublicChat => MemoryReason::PublicExpired,
|
ConversationChannel::PublicChat => MemoryReason::PublicExpired,
|
||||||
ConversationChannel::DirectIm => MemoryReason::DirectImExpired,
|
ConversationChannel::DirectIm => MemoryReason::DirectImExpired,
|
||||||
};
|
};
|
||||||
|
retain_expired(state, session.metadata());
|
||||||
push_event(&mut state.events, event_for(reason, &session));
|
push_event(&mut state.events, event_for(reason, &session));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -818,10 +1021,18 @@ fn evict_oldest(state: &mut StoreState, reason: MemoryReason, preserve: Option<C
|
|||||||
&& let Some(session) = state.sessions.remove(&key)
|
&& let Some(session) = state.sessions.remove(&key)
|
||||||
{
|
{
|
||||||
state.total_bytes = state.total_bytes.saturating_sub(session.bytes);
|
state.total_bytes = state.total_bytes.saturating_sub(session.bytes);
|
||||||
|
retain_expired(state, session.metadata());
|
||||||
push_event(&mut state.events, event_for(reason, &session));
|
push_event(&mut state.events, event_for(reason, &session));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn retain_expired(state: &mut StoreState, metadata: ConversationMetadata) {
|
||||||
|
if state.retired.len() == MAX_RETIRED_SESSIONS {
|
||||||
|
state.retired.pop_front();
|
||||||
|
}
|
||||||
|
state.retired.push_back(metadata);
|
||||||
|
}
|
||||||
|
|
||||||
fn event_for(reason: MemoryReason, session: &StoredSession) -> MemoryEvent {
|
fn event_for(reason: MemoryReason, session: &StoredSession) -> MemoryEvent {
|
||||||
MemoryEvent {
|
MemoryEvent {
|
||||||
reason,
|
reason,
|
||||||
@@ -904,6 +1115,21 @@ struct Snapshot {
|
|||||||
generation: u64,
|
generation: u64,
|
||||||
next_sequence: u64,
|
next_sequence: u64,
|
||||||
sessions: Vec<SnapshotSession>,
|
sessions: Vec<SnapshotSession>,
|
||||||
|
#[serde(default)]
|
||||||
|
retired_sessions: Vec<SnapshotRetiredSession>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[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)]
|
#[derive(Serialize, Deserialize)]
|
||||||
@@ -1021,6 +1247,7 @@ fn restore_snapshot(
|
|||||||
let wall_now = unix_millis(clock.wall_now());
|
let wall_now = unix_millis(clock.wall_now());
|
||||||
let mut restored = StoreState {
|
let mut restored = StoreState {
|
||||||
sessions: BTreeMap::new(),
|
sessions: BTreeMap::new(),
|
||||||
|
retired: VecDeque::new(),
|
||||||
total_bytes: 0,
|
total_bytes: 0,
|
||||||
next_sequence: snapshot.next_sequence.max(1),
|
next_sequence: snapshot.next_sequence.max(1),
|
||||||
snapshot_generation: snapshot.generation,
|
snapshot_generation: snapshot.generation,
|
||||||
@@ -1029,6 +1256,31 @@ fn restore_snapshot(
|
|||||||
let mut restored_session_ids = BTreeSet::new();
|
let mut restored_session_ids = BTreeSet::new();
|
||||||
let mut restored_sequences = BTreeSet::new();
|
let mut restored_sequences = BTreeSet::new();
|
||||||
let mut maximum_sequence = 0_u64;
|
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 {
|
if snapshot.sessions.len() > MAX_ACTIVE_SESSIONS {
|
||||||
return Err(ConversationError::PersistenceTooLarge);
|
return Err(ConversationError::PersistenceTooLarge);
|
||||||
}
|
}
|
||||||
@@ -1037,15 +1289,38 @@ fn restore_snapshot(
|
|||||||
.map_err(|_| ConversationError::InvalidAvatar)?;
|
.map_err(|_| ConversationError::InvalidAvatar)?;
|
||||||
let key = ConversationKey::new(avatar_id, persisted.channel)?;
|
let key = ConversationKey::new(avatar_id, persisted.channel)?;
|
||||||
let session_id = BoundedText::new("conversation.session_id", persisted.session_id)?;
|
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)
|
if restored.sessions.contains_key(&key)
|
||||||
|| !restored_session_ids.insert(session_id.as_str().to_owned())
|
|| !restored_session_ids.insert(session_id.as_str().to_owned())
|
||||||
|| persisted.created_unix_millis > persisted.last_active_unix_millis
|
|| 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);
|
return Err(ConversationError::PersistenceUnavailable);
|
||||||
}
|
}
|
||||||
let elapsed_millis = wall_now.saturating_sub(persisted.last_active_unix_millis);
|
let elapsed_millis = wall_now.saturating_sub(persisted.last_active_unix_millis);
|
||||||
let elapsed = Duration::from_millis(elapsed_millis);
|
let elapsed = Duration::from_millis(elapsed_millis);
|
||||||
if elapsed >= key.channel.inactivity_ttl() {
|
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;
|
continue;
|
||||||
}
|
}
|
||||||
let last_active = monotonic_now.checked_sub(elapsed).unwrap_or(monotonic_now);
|
let last_active = monotonic_now.checked_sub(elapsed).unwrap_or(monotonic_now);
|
||||||
@@ -1059,9 +1334,6 @@ fn restore_snapshot(
|
|||||||
bytes: 0,
|
bytes: 0,
|
||||||
tool_results: 0,
|
tool_results: 0,
|
||||||
};
|
};
|
||||||
if persisted.entries.len() > MAX_CONVERSATION_MESSAGES {
|
|
||||||
return Err(ConversationError::PersistenceTooLarge);
|
|
||||||
}
|
|
||||||
let mut previous_sequence = 0_u64;
|
let mut previous_sequence = 0_u64;
|
||||||
for entry in persisted.entries {
|
for entry in persisted.entries {
|
||||||
if entry.sequence == 0 || !restored_sequences.insert(entry.sequence) {
|
if entry.sequence == 0 || !restored_sequences.insert(entry.sequence) {
|
||||||
@@ -1183,6 +1455,20 @@ fn write_snapshot(
|
|||||||
.collect(),
|
.collect(),
|
||||||
})
|
})
|
||||||
.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 =
|
let bytes =
|
||||||
serde_json::to_vec(&snapshot).map_err(|_| ConversationError::PersistenceUnavailable)?;
|
serde_json::to_vec(&snapshot).map_err(|_| ConversationError::PersistenceUnavailable)?;
|
||||||
|
|||||||
@@ -509,6 +509,79 @@ fn operator_metadata_delete_and_expire_never_require_content_access() {
|
|||||||
assert!(reasons.contains(&MemoryReason::OperatorExpired));
|
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)]
|
#[cfg(unix)]
|
||||||
#[test]
|
#[test]
|
||||||
fn persistence_uses_restrictive_unix_permissions() {
|
fn persistence_uses_restrictive_unix_permissions() {
|
||||||
|
|||||||
@@ -1228,8 +1228,9 @@ async fn process_batch(
|
|||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
let outcome = if delivered == parts.len() {
|
let outcome = if delivered == parts.len() {
|
||||||
let _ = conversation.append(
|
let _ = conversation.append_if_session(
|
||||||
conversation_key,
|
conversation_key,
|
||||||
|
update.session_id.as_str(),
|
||||||
MemoryRecord::agent_visible_response(safe.clone()),
|
MemoryRecord::agent_visible_response(safe.clone()),
|
||||||
);
|
);
|
||||||
DeliveryOutcome::Succeeded
|
DeliveryOutcome::Succeeded
|
||||||
@@ -1781,6 +1782,7 @@ pub struct PolicyLlmResponder {
|
|||||||
runtime: Arc<mentra::Runtime>,
|
runtime: Arc<mentra::Runtime>,
|
||||||
agents: Mutex<BTreeMap<String, MentraAgentHandle>>,
|
agents: Mutex<BTreeMap<String, MentraAgentHandle>>,
|
||||||
memory_admin: Arc<crate::memory_admin::MentraMemoryAdmin>,
|
memory_admin: Arc<crate::memory_admin::MentraMemoryAdmin>,
|
||||||
|
conversation_mappings: Arc<crate::memory_admin::MentraConversationMappings>,
|
||||||
active_tools: Arc<Mutex<BTreeMap<String, ActiveMentraRequest>>>,
|
active_tools: Arc<Mutex<BTreeMap<String, ActiveMentraRequest>>>,
|
||||||
gateway: Arc<crate::policy::PolicyGateway>,
|
gateway: Arc<crate::policy::PolicyGateway>,
|
||||||
backend: Arc<dyn crate::backend::AuthorizedToolBackend>,
|
backend: Arc<dyn crate::backend::AuthorizedToolBackend>,
|
||||||
@@ -1789,6 +1791,24 @@ pub struct PolicyLlmResponder {
|
|||||||
storage_path: std::path::PathBuf,
|
storage_path: std::path::PathBuf,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||||
|
pub struct MentraTranscriptEntry {
|
||||||
|
pub entry_id: String,
|
||||||
|
pub parent_id: Option<String>,
|
||||||
|
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<MentraTranscriptEntry>,
|
||||||
|
pub busy: bool,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
struct MentraAgentHandle {
|
struct MentraAgentHandle {
|
||||||
id: String,
|
id: String,
|
||||||
@@ -1800,6 +1820,7 @@ struct ActiveMentraRequest {
|
|||||||
executor: Arc<crate::policy::PolicyToolExecutor>,
|
executor: Arc<crate::policy::PolicyToolExecutor>,
|
||||||
cancellation: CancellationToken,
|
cancellation: CancellationToken,
|
||||||
mentra_cancellation: mentra::runtime::CancellationToken,
|
mentra_cancellation: mentra::runtime::CancellationToken,
|
||||||
|
session_id: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
struct AutonomousApprovalReviewer {
|
struct AutonomousApprovalReviewer {
|
||||||
@@ -2009,6 +2030,12 @@ impl PolicyLlmResponder {
|
|||||||
)
|
)
|
||||||
.map_err(InteractionError::Persistence)?,
|
.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()
|
let mut builder = mentra::Runtime::builder()
|
||||||
.with_runtime_identifier(MENTRA_RUNTIME_IDENTIFIER)
|
.with_runtime_identifier(MENTRA_RUNTIME_IDENTIFIER)
|
||||||
.with_store(store)
|
.with_store(store)
|
||||||
@@ -2047,6 +2074,7 @@ impl PolicyLlmResponder {
|
|||||||
runtime: Arc::new(runtime),
|
runtime: Arc::new(runtime),
|
||||||
agents: Mutex::new(agents),
|
agents: Mutex::new(agents),
|
||||||
memory_admin,
|
memory_admin,
|
||||||
|
conversation_mappings,
|
||||||
active_tools,
|
active_tools,
|
||||||
gateway,
|
gateway,
|
||||||
backend,
|
backend,
|
||||||
@@ -2072,6 +2100,7 @@ impl PolicyLlmResponder {
|
|||||||
.register(&request.sender_id.to_string(), &key, &handle.id)
|
.register(&request.sender_id.to_string(), &key, &handle.id)
|
||||||
.map_err(|_| InteractionModelError::Failed)?;
|
.map_err(|_| InteractionModelError::Failed)?;
|
||||||
}
|
}
|
||||||
|
self.register_conversation(request, &key, &handle.id)?;
|
||||||
return Ok(Arc::clone(&handle.agent));
|
return Ok(Arc::clone(&handle.agent));
|
||||||
}
|
}
|
||||||
let tools = self.gateway.tools_for(context, (self.now)());
|
let tools = self.gateway.tools_for(context, (self.now)());
|
||||||
@@ -2118,13 +2147,146 @@ impl PolicyLlmResponder {
|
|||||||
.register(&request.sender_id.to_string(), &key, &id)
|
.register(&request.sender_id.to_string(), &key, &id)
|
||||||
.map_err(|_| InteractionModelError::Failed)?;
|
.map_err(|_| InteractionModelError::Failed)?;
|
||||||
}
|
}
|
||||||
|
self.register_conversation(request, &key, &id)?;
|
||||||
Ok(agent)
|
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]
|
#[must_use]
|
||||||
pub fn memory_admin(&self) -> Arc<crate::memory_admin::MentraMemoryAdmin> {
|
pub fn memory_admin(&self) -> Arc<crate::memory_admin::MentraMemoryAdmin> {
|
||||||
Arc::clone(&self.memory_admin)
|
Arc::clone(&self.memory_admin)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn conversation_transcript(
|
||||||
|
&self,
|
||||||
|
session_id: &str,
|
||||||
|
offset: usize,
|
||||||
|
limit: usize,
|
||||||
|
) -> Result<MentraTranscriptMetadata, String> {
|
||||||
|
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<bool, String> {
|
||||||
|
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 {
|
pub(crate) fn mentra_agent_name(request: &ResponseRequest) -> String {
|
||||||
@@ -2196,6 +2358,7 @@ impl InteractionResponder for PolicyLlmResponder {
|
|||||||
executor,
|
executor,
|
||||||
cancellation: cancellation.clone(),
|
cancellation: cancellation.clone(),
|
||||||
mentra_cancellation: mentra_cancellation.clone(),
|
mentra_cancellation: mentra_cancellation.clone(),
|
||||||
|
session_id: request.session_id.as_str().to_owned(),
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
let cancellation_bridge = {
|
let cancellation_bridge = {
|
||||||
|
|||||||
@@ -96,27 +96,31 @@ pub use control_plane::{
|
|||||||
AuditEventView, CONTROL_PROTOCOL_VERSION, ControlContext, ControlError, ControlErrorBody,
|
AuditEventView, CONTROL_PROTOCOL_VERSION, ControlContext, ControlError, ControlErrorBody,
|
||||||
ControlErrorCode, ControlEvent, ControlEventKind, ControlFuture, ControlLimits, ControlPayload,
|
ControlErrorCode, ControlEvent, ControlEventKind, ControlFuture, ControlLimits, ControlPayload,
|
||||||
ControlPlane, ControlRequest, ControlRequestEnvelope, ControlResponseEnvelope, ControlRole,
|
ControlPlane, ControlRequest, ControlRequestEnvelope, ControlResponseEnvelope, ControlRole,
|
||||||
ControlSubscription, ControlTarget, ConversationChannelView, HealthView,
|
ControlSubscription, ControlTarget, ConversationChannelView, ConversationDetailView,
|
||||||
InProcessControlClient, MemoryAgentView, MemoryCursorView, MemoryDetailView, MemoryFilterView,
|
ConversationEntryView, ConversationFilterView, ConversationMetadataView,
|
||||||
MemoryHealthView, MemoryKindView, MemoryPageView, MemoryRecordView, MemorySortView, Page,
|
ConversationStatusView, HealthView, InProcessControlClient, MemoryAgentView, MemoryCursorView,
|
||||||
PageRequest, PendingApprovalView, RuntimeView, ScheduledJobView, SessionMetadataView,
|
MemoryDetailView, MemoryFilterView, MemoryHealthView, MemoryKindView, MemoryPageView,
|
||||||
TcpControlClient, TcpControlConfig, TcpControlServer,
|
MemoryRecordView, MemorySortView, Page, PageRequest, PendingApprovalView, RuntimeView,
|
||||||
|
ScheduledJobView, SessionMetadataView, TcpControlClient, TcpControlConfig, TcpControlServer,
|
||||||
|
TranscriptEntryView,
|
||||||
};
|
};
|
||||||
pub use control_runtime::{AgentControlTarget, RuntimeControlCommand};
|
pub use control_runtime::{AgentControlTarget, RuntimeControlCommand};
|
||||||
pub use conversation::{
|
pub use conversation::{
|
||||||
ConversationChannel, ConversationClock, ConversationContext, ConversationError,
|
ConversationChannel, ConversationClock, ConversationContext, ConversationDetailProjection,
|
||||||
ConversationKey, ConversationLimits, ConversationMetadata, ConversationPersistence,
|
ConversationEntryProjection, ConversationError, ConversationKey, ConversationLimits,
|
||||||
ConversationStore, MemoryEvent, MemoryReason, MemoryRecord, MemoryTrust, MemoryUpdate,
|
ConversationMetadata, ConversationPersistence, ConversationStatus, ConversationStore,
|
||||||
SystemConversationClock,
|
MemoryEvent, MemoryReason, MemoryRecord, MemoryTrust, MemoryUpdate,
|
||||||
|
OperatorConversationMetadata, SystemConversationClock,
|
||||||
};
|
};
|
||||||
pub use interaction::{
|
pub use interaction::{
|
||||||
DeliveryFuture, DeliveryOutcome, ImDialogKind, InboundInteraction, InboundSource,
|
DeliveryFuture, DeliveryOutcome, ImDialogKind, InboundInteraction, InboundSource,
|
||||||
InferenceOutcome, InteractionChannel, InteractionCoordinator, InteractionDeliveryError,
|
InferenceOutcome, InteractionChannel, InteractionCoordinator, InteractionDeliveryError,
|
||||||
InteractionError, InteractionHandle, InteractionIngress, InteractionIntent,
|
InteractionError, InteractionHandle, InteractionIngress, InteractionIntent,
|
||||||
InteractionModelError, InteractionObservation, InteractionOrigin, InteractionPacingError,
|
InteractionModelError, InteractionObservation, InteractionOrigin, InteractionPacingError,
|
||||||
InteractionResponder, InteractionSettings, InteractionSink, OutboundInteraction, PacerFuture,
|
InteractionResponder, InteractionSettings, InteractionSink, MentraTranscriptEntry,
|
||||||
PolicyLlmResponder, ResponderFuture, ResponsePacer, ResponseRequest, SuppressionReason,
|
MentraTranscriptMetadata, OutboundInteraction, PacerFuture, PolicyLlmResponder,
|
||||||
VisibleResponse, split_utf8,
|
ResponderFuture, ResponsePacer, ResponseRequest, SuppressionReason, VisibleResponse,
|
||||||
|
split_utf8,
|
||||||
};
|
};
|
||||||
pub use landmarks::{
|
pub use landmarks::{
|
||||||
LANDMARK_CREATE_TOOL, LANDMARK_LIST_TOOL, LANDMARK_SCHEDULE_TOOL, LANDMARK_STATUS_TOOL,
|
LANDMARK_CREATE_TOOL, LANDMARK_LIST_TOOL, LANDMARK_SCHEDULE_TOOL, LANDMARK_STATUS_TOOL,
|
||||||
@@ -133,7 +137,9 @@ pub use landmarks::{
|
|||||||
pub use llm::{
|
pub use llm::{
|
||||||
CompletionMessage, ContentPart, ImageDetail, LlmClient, LlmError, ToolDefinition, ToolSchema,
|
CompletionMessage, ContentPart, ImageDetail, LlmClient, LlmError, ToolDefinition, ToolSchema,
|
||||||
};
|
};
|
||||||
pub use memory_admin::{MemoryAgentMapping, MentraMemoryAdmin};
|
pub use memory_admin::{
|
||||||
|
ConversationAgentMapping, MemoryAgentMapping, MentraConversationMappings, MentraMemoryAdmin,
|
||||||
|
};
|
||||||
pub use observability::{
|
pub use observability::{
|
||||||
AgentMetrics, CorrelationIds, DiagnosticDirection, EventDraft, EventFamily, EventOrigin,
|
AgentMetrics, CorrelationIds, DiagnosticDirection, EventDraft, EventFamily, EventOrigin,
|
||||||
EventSeverity, EventSubscription, JournalConfig, LatencyMetrics, MetricsSnapshot,
|
EventSeverity, EventSubscription, JournalConfig, LatencyMetrics, MetricsSnapshot,
|
||||||
|
|||||||
@@ -416,6 +416,7 @@ async fn run_live(
|
|||||||
control_target.attach_build_control(live.build_control.clone());
|
control_target.attach_build_control(live.build_control.clone());
|
||||||
control_target.attach_vision_control(live.vision.clone());
|
control_target.attach_vision_control(live.vision.clone());
|
||||||
control_target.attach_memory_control(live.memory_admin.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());
|
control_target.update_session(handle.status());
|
||||||
let erased_target: Arc<dyn ControlTarget> = control_target.clone();
|
let erased_target: Arc<dyn ControlTarget> = control_target.clone();
|
||||||
let (control_plane, integrated_client, control_server) = match config.mode {
|
let (control_plane, integrated_client, control_server) = match config.mode {
|
||||||
@@ -734,6 +735,7 @@ struct LiveInteractions {
|
|||||||
audit: Arc<metacrate_grid_agent::MemoryPolicyAudit>,
|
audit: Arc<metacrate_grid_agent::MemoryPolicyAudit>,
|
||||||
observability: Arc<metacrate_grid_agent::Observability>,
|
observability: Arc<metacrate_grid_agent::Observability>,
|
||||||
memory_admin: Arc<metacrate_grid_agent::MentraMemoryAdmin>,
|
memory_admin: Arc<metacrate_grid_agent::MentraMemoryAdmin>,
|
||||||
|
conversation_admin: Arc<metacrate_grid_agent::PolicyLlmResponder>,
|
||||||
_appearance_recovery: libremetaverse_types::compat::Subscription,
|
_appearance_recovery: libremetaverse_types::compat::Subscription,
|
||||||
_landmark_intake: metacrate_grid_agent::LibremetaverseLandmarkIntake,
|
_landmark_intake: metacrate_grid_agent::LibremetaverseLandmarkIntake,
|
||||||
landmark_roaming: metacrate_grid_agent::LandmarkRoamingHandle,
|
landmark_roaming: metacrate_grid_agent::LandmarkRoamingHandle,
|
||||||
@@ -906,6 +908,7 @@ async fn start_live_interactions(
|
|||||||
config.storage_path.join("mentra"),
|
config.storage_path.join("mentra"),
|
||||||
)?);
|
)?);
|
||||||
let memory_admin = responder.memory_admin();
|
let memory_admin = responder.memory_admin();
|
||||||
|
let conversation_admin = responder.clone();
|
||||||
let vision_limits = config.vision;
|
let vision_limits = config.vision;
|
||||||
let scene_source = Arc::new(LibremetaverseSceneSource::new(owner, vision_limits));
|
let scene_source = Arc::new(LibremetaverseSceneSource::new(owner, vision_limits));
|
||||||
scene_source.start_prefetch();
|
scene_source.start_prefetch();
|
||||||
@@ -940,6 +943,7 @@ async fn start_live_interactions(
|
|||||||
audit,
|
audit,
|
||||||
observability,
|
observability,
|
||||||
memory_admin,
|
memory_admin,
|
||||||
|
conversation_admin,
|
||||||
_appearance_recovery: appearance_recovery,
|
_appearance_recovery: appearance_recovery,
|
||||||
_landmark_intake: landmark_intake,
|
_landmark_intake: landmark_intake,
|
||||||
landmark_roaming,
|
landmark_roaming,
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ use std::path::{Path, PathBuf};
|
|||||||
use std::sync::{Mutex, MutexGuard};
|
use std::sync::{Mutex, MutexGuard};
|
||||||
|
|
||||||
const MAX_MAPPINGS: usize = 4_096;
|
const MAX_MAPPINGS: usize = 4_096;
|
||||||
|
const MAX_MAPPING_BYTES: u64 = 8 * 1024 * 1024;
|
||||||
|
|
||||||
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
|
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
|
||||||
pub struct MemoryAgentMapping {
|
pub struct MemoryAgentMapping {
|
||||||
@@ -21,6 +22,100 @@ pub struct MemoryAgentMapping {
|
|||||||
pub mentra_agent_id: String,
|
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<BTreeMap<String, ConversationAgentMapping>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl MentraConversationMappings {
|
||||||
|
pub fn new(path: PathBuf) -> Result<Self, String> {
|
||||||
|
let values: Vec<ConversationAgentMapping> = 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<ConversationAgentMapping> {
|
||||||
|
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<bool, String> {
|
||||||
|
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 {
|
pub struct MentraMemoryAdmin {
|
||||||
store: HybridRuntimeStore,
|
store: HybridRuntimeStore,
|
||||||
mappings_path: PathBuf,
|
mappings_path: PathBuf,
|
||||||
@@ -160,21 +255,7 @@ impl MentraMemoryAdmin {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn load_mappings(path: &Path) -> Result<BTreeMap<String, MemoryAgentMapping>, String> {
|
fn load_mappings(path: &Path) -> Result<BTreeMap<String, MemoryAgentMapping>, String> {
|
||||||
let bytes = match std::fs::read(path) {
|
let values: Vec<MemoryAgentMapping> = load_values(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<MemoryAgentMapping> =
|
|
||||||
serde_json::from_slice(&bytes).map_err(|e| e.to_string())?;
|
|
||||||
if values.len() > MAX_MAPPINGS || values.iter().any(|mapping| !valid_mapping(mapping)) {
|
if values.len() > MAX_MAPPINGS || values.iter().any(|mapping| !valid_mapping(mapping)) {
|
||||||
return Err("invalid or excessive memory-agent mappings".to_owned());
|
return Err("invalid or excessive memory-agent mappings".to_owned());
|
||||||
}
|
}
|
||||||
@@ -190,6 +271,33 @@ fn load_mappings(path: &Path) -> Result<BTreeMap<String, MemoryAgentMapping>, St
|
|||||||
Ok(mappings)
|
Ok(mappings)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn load_values<T: for<'de> Deserialize<'de>>(path: &Path) -> Result<Vec<T>, 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<Vec<u8>> {
|
||||||
|
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 {
|
fn valid_mapping(mapping: &MemoryAgentMapping) -> bool {
|
||||||
[
|
[
|
||||||
mapping.avatar_id.as_str(),
|
mapping.avatar_id.as_str(),
|
||||||
@@ -203,6 +311,13 @@ fn valid_mapping(mapping: &MemoryAgentMapping) -> bool {
|
|||||||
fn persist_mappings<'a>(
|
fn persist_mappings<'a>(
|
||||||
path: &Path,
|
path: &Path,
|
||||||
mappings: impl Iterator<Item = &'a MemoryAgentMapping>,
|
mappings: impl Iterator<Item = &'a MemoryAgentMapping>,
|
||||||
|
) -> Result<(), String> {
|
||||||
|
persist_values(path, mappings)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn persist_values<'a, T: Serialize + 'a>(
|
||||||
|
path: &Path,
|
||||||
|
mappings: impl Iterator<Item = &'a T>,
|
||||||
) -> Result<(), String> {
|
) -> Result<(), String> {
|
||||||
if let Some(parent) = path.parent() {
|
if let Some(parent) = path.parent() {
|
||||||
std::fs::create_dir_all(parent).map_err(|error| error.to_string())?;
|
std::fs::create_dir_all(parent).map_err(|error| error.to_string())?;
|
||||||
@@ -229,6 +344,20 @@ fn persist_mappings<'a>(
|
|||||||
Ok(())
|
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<T>(value: &Mutex<T>) -> MutexGuard<'_, T> {
|
fn lock<T>(value: &Mutex<T>) -> MutexGuard<'_, T> {
|
||||||
value
|
value
|
||||||
.lock()
|
.lock()
|
||||||
@@ -271,6 +400,29 @@ mod tests {
|
|||||||
let _ = std::fs::remove_dir_all(root);
|
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 {
|
fn record(id: &str, agent_id: &str) -> MemoryRecord {
|
||||||
MemoryRecord {
|
MemoryRecord {
|
||||||
record_id: id.to_owned(),
|
record_id: id.to_owned(),
|
||||||
|
|||||||
@@ -7,7 +7,8 @@
|
|||||||
|
|
||||||
use crate::control_plane::{
|
use crate::control_plane::{
|
||||||
AuditEventView, CONTROL_PROTOCOL_VERSION, ControlEvent, ControlEventKind, ControlPayload,
|
AuditEventView, CONTROL_PROTOCOL_VERSION, ControlEvent, ControlEventKind, ControlPayload,
|
||||||
ControlRequest, ControlRequestEnvelope, ControlResponseEnvelope, HealthView,
|
ControlRequest, ControlRequestEnvelope, ControlResponseEnvelope, ConversationDetailView,
|
||||||
|
ConversationFilterView, ConversationMetadataView, ConversationStatusView, HealthView,
|
||||||
InProcessControlClient, MemoryAgentView, MemoryCursorView, MemoryDetailView, MemoryFilterView,
|
InProcessControlClient, MemoryAgentView, MemoryCursorView, MemoryDetailView, MemoryFilterView,
|
||||||
MemoryHealthView, MemoryKindView, MemoryRecordView, MemorySortView, Page, PageRequest,
|
MemoryHealthView, MemoryKindView, MemoryRecordView, MemorySortView, Page, PageRequest,
|
||||||
PendingApprovalView, RuntimeView, ScheduledJobView, SessionMetadataView, TcpControlClient,
|
PendingApprovalView, RuntimeView, ScheduledJobView, SessionMetadataView, TcpControlClient,
|
||||||
@@ -339,6 +340,8 @@ pub struct OperatorSnapshot {
|
|||||||
pub runtime: Option<RuntimeView>,
|
pub runtime: Option<RuntimeView>,
|
||||||
pub metrics: Option<MetricsSnapshot>,
|
pub metrics: Option<MetricsSnapshot>,
|
||||||
pub sessions: Vec<SessionMetadataView>,
|
pub sessions: Vec<SessionMetadataView>,
|
||||||
|
pub conversations: Vec<ConversationMetadataView>,
|
||||||
|
pub conversation_detail: Option<ConversationDetailView>,
|
||||||
pub schedules: Vec<ScheduledJobView>,
|
pub schedules: Vec<ScheduledJobView>,
|
||||||
pub approvals: Vec<PendingApprovalView>,
|
pub approvals: Vec<PendingApprovalView>,
|
||||||
pub memory_health: Option<MemoryHealthView>,
|
pub memory_health: Option<MemoryHealthView>,
|
||||||
@@ -358,6 +361,8 @@ impl Default for OperatorSnapshot {
|
|||||||
runtime: None,
|
runtime: None,
|
||||||
metrics: None,
|
metrics: None,
|
||||||
sessions: Vec::new(),
|
sessions: Vec::new(),
|
||||||
|
conversations: Vec::new(),
|
||||||
|
conversation_detail: None,
|
||||||
schedules: Vec::new(),
|
schedules: Vec::new(),
|
||||||
approvals: Vec::new(),
|
approvals: Vec::new(),
|
||||||
memory_health: None,
|
memory_health: None,
|
||||||
@@ -397,6 +402,9 @@ pub enum OperatorCommand {
|
|||||||
avatar_id: String,
|
avatar_id: String,
|
||||||
direct_im: bool,
|
direct_im: bool,
|
||||||
},
|
},
|
||||||
|
DeleteConversation {
|
||||||
|
session_id: String,
|
||||||
|
},
|
||||||
ToggleSchedule {
|
ToggleSchedule {
|
||||||
job_id: String,
|
job_id: String,
|
||||||
enabled: bool,
|
enabled: bool,
|
||||||
@@ -432,6 +440,9 @@ impl OperatorCommand {
|
|||||||
crate::control_plane::ConversationChannelView::PublicChat
|
crate::control_plane::ConversationChannelView::PublicChat
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
Self::DeleteConversation { session_id } => ControlRequest::DeleteConversation {
|
||||||
|
session_id: session_id.clone(),
|
||||||
|
},
|
||||||
Self::ToggleSchedule { job_id, enabled } => ControlRequest::SetRoamingJob {
|
Self::ToggleSchedule { job_id, enabled } => ControlRequest::SetRoamingJob {
|
||||||
job_id: job_id.clone(),
|
job_id: job_id.clone(),
|
||||||
enabled: *enabled,
|
enabled: *enabled,
|
||||||
@@ -456,7 +467,9 @@ impl OperatorCommand {
|
|||||||
| Self::Reconnect
|
| Self::Reconnect
|
||||||
| Self::ExpireSession { .. }
|
| Self::ExpireSession { .. }
|
||||||
| Self::ToggleSchedule { .. } => CommandConfirmation::Normal,
|
| 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,
|
FilterBackspace,
|
||||||
FilterCommit,
|
FilterCommit,
|
||||||
FilterCancel,
|
FilterCancel,
|
||||||
|
ConversationLoad,
|
||||||
|
ConversationClose,
|
||||||
|
ConversationNextPage,
|
||||||
|
ConversationPreviousPage,
|
||||||
MemoryLoad,
|
MemoryLoad,
|
||||||
MemoryNextAgent,
|
MemoryNextAgent,
|
||||||
MemoryNextPage,
|
MemoryNextPage,
|
||||||
@@ -653,6 +670,10 @@ pub struct OperatorTui {
|
|||||||
memory_current_offset: u64,
|
memory_current_offset: u64,
|
||||||
memory_next_offset: Option<u64>,
|
memory_next_offset: Option<u64>,
|
||||||
memory_refresh_pending: bool,
|
memory_refresh_pending: bool,
|
||||||
|
conversation_cursors: Vec<u64>,
|
||||||
|
conversation_next_cursor: Option<u64>,
|
||||||
|
conversation_detail_cursor: u64,
|
||||||
|
conversation_detail_next: Option<u64>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for OperatorTui {
|
impl Default for OperatorTui {
|
||||||
@@ -683,6 +704,10 @@ impl Default for OperatorTui {
|
|||||||
memory_current_offset: 0,
|
memory_current_offset: 0,
|
||||||
memory_next_offset: None,
|
memory_next_offset: None,
|
||||||
memory_refresh_pending: false,
|
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(" ");
|
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 {
|
fn visible_rows(&self) -> usize {
|
||||||
match self.screen {
|
match self.screen {
|
||||||
OperatorScreen::Sessions => self.snapshot.sessions.len(),
|
OperatorScreen::Sessions => self.snapshot.conversations.len(),
|
||||||
OperatorScreen::Memories => self.snapshot.memories.len(),
|
OperatorScreen::Memories => self.snapshot.memories.len(),
|
||||||
OperatorScreen::Roaming => self.snapshot.schedules.len(),
|
OperatorScreen::Roaming => self.snapshot.schedules.len(),
|
||||||
OperatorScreen::Approvals => self.snapshot.approvals.len(),
|
OperatorScreen::Approvals => self.snapshot.approvals.len(),
|
||||||
@@ -801,6 +862,15 @@ impl OperatorTui {
|
|||||||
record_id: record.record_id.clone(),
|
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 {
|
match key {
|
||||||
'p' => Some(OperatorCommand::Pause),
|
'p' => Some(OperatorCommand::Pause),
|
||||||
'u' => Some(OperatorCommand::Resume),
|
'u' => Some(OperatorCommand::Resume),
|
||||||
@@ -822,11 +892,12 @@ impl OperatorTui {
|
|||||||
}),
|
}),
|
||||||
'e' => self
|
'e' => self
|
||||||
.snapshot
|
.snapshot
|
||||||
.sessions
|
.conversations
|
||||||
.get(self.selected[Self::screen_index(OperatorScreen::Sessions)])
|
.get(self.selected[Self::screen_index(OperatorScreen::Sessions)])
|
||||||
.map(|session| OperatorCommand::ExpireSession {
|
.map(|session| OperatorCommand::ExpireSession {
|
||||||
avatar_id: session.avatar_id.clone(),
|
avatar_id: session.avatar_id.clone(),
|
||||||
direct_im: session.channel == "direct_im",
|
direct_im: session.channel
|
||||||
|
== crate::control_plane::ConversationChannelView::DirectIm,
|
||||||
}),
|
}),
|
||||||
't' => self
|
't' => self
|
||||||
.snapshot
|
.snapshot
|
||||||
@@ -853,6 +924,9 @@ impl OperatorTui {
|
|||||||
self.scroll = 0;
|
self.scroll = 0;
|
||||||
if self.screen == OperatorScreen::Memories {
|
if self.screen == OperatorScreen::Memories {
|
||||||
TuiAction::Request(ControlRequest::ListMemoryAgents { page: page() })
|
TuiAction::Request(ControlRequest::ListMemoryAgents { page: page() })
|
||||||
|
} else if self.screen == OperatorScreen::Sessions {
|
||||||
|
self.conversation_cursors.clear();
|
||||||
|
TuiAction::Request(self.conversation_request())
|
||||||
} else {
|
} else {
|
||||||
TuiAction::None
|
TuiAction::None
|
||||||
}
|
}
|
||||||
@@ -867,17 +941,32 @@ impl OperatorTui {
|
|||||||
self.scroll = 0;
|
self.scroll = 0;
|
||||||
if self.screen == OperatorScreen::Memories {
|
if self.screen == OperatorScreen::Memories {
|
||||||
TuiAction::Request(ControlRequest::ListMemoryAgents { page: page() })
|
TuiAction::Request(ControlRequest::ListMemoryAgents { page: page() })
|
||||||
|
} else if self.screen == OperatorScreen::Sessions {
|
||||||
|
self.conversation_cursors.clear();
|
||||||
|
TuiAction::Request(self.conversation_request())
|
||||||
} else {
|
} else {
|
||||||
TuiAction::None
|
TuiAction::None
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
TuiInput::ScrollUp => {
|
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);
|
let index = Self::screen_index(self.screen);
|
||||||
self.selected[index] = self.selected[index].saturating_sub(1);
|
self.selected[index] = self.selected[index].saturating_sub(1);
|
||||||
self.scroll = self.selected[index];
|
self.scroll = self.selected[index];
|
||||||
TuiAction::None
|
TuiAction::None
|
||||||
}
|
}
|
||||||
TuiInput::ScrollDown => {
|
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);
|
let index = Self::screen_index(self.screen);
|
||||||
self.selected[index] = self.selected[index]
|
self.selected[index] = self.selected[index]
|
||||||
.saturating_add(1)
|
.saturating_add(1)
|
||||||
@@ -945,6 +1034,8 @@ impl OperatorTui {
|
|||||||
self.filter_editing = true;
|
self.filter_editing = true;
|
||||||
self.status = if self.screen == OperatorScreen::Memories {
|
self.status = if self.screen == OperatorScreen::Memories {
|
||||||
"memory search/filter: type query, Enter applies, Esc clears".into()
|
"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 {
|
} else {
|
||||||
"event/audit filter: type query, Enter applies, Esc clears".into()
|
"event/audit filter: type query, Enter applies, Esc clears".into()
|
||||||
};
|
};
|
||||||
@@ -993,6 +1084,69 @@ impl OperatorTui {
|
|||||||
TuiAction::None
|
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 => {
|
TuiInput::MemoryLoad => {
|
||||||
if let (Some(agent), Some(record)) = (
|
if let (Some(agent), Some(record)) = (
|
||||||
self.snapshot.memory_agents.get(self.memory_agent),
|
self.snapshot.memory_agents.get(self.memory_agent),
|
||||||
@@ -1135,7 +1289,7 @@ impl OperatorTui {
|
|||||||
ControlRequest::Runtime,
|
ControlRequest::Runtime,
|
||||||
ControlRequest::Metrics,
|
ControlRequest::Metrics,
|
||||||
ControlRequest::MemoryHealth,
|
ControlRequest::MemoryHealth,
|
||||||
ControlRequest::ListSessions { page: page() },
|
self.conversation_request(),
|
||||||
ControlRequest::ListScheduledJobs { page: page() },
|
ControlRequest::ListScheduledJobs { page: page() },
|
||||||
ControlRequest::ListPendingApprovals { page: page() },
|
ControlRequest::ListPendingApprovals { page: page() },
|
||||||
ControlRequest::ListAuditEvents { page: page() },
|
ControlRequest::ListAuditEvents { page: page() },
|
||||||
@@ -1269,6 +1423,26 @@ impl OperatorTui {
|
|||||||
.get(self.selected[index])
|
.get(self.selected[index])
|
||||||
.map(|item| item.session_id.clone());
|
.map(|item| item.session_id.clone());
|
||||||
self.snapshot.sessions = items;
|
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
|
self.selected[index] = identity
|
||||||
.and_then(|id| {
|
.and_then(|id| {
|
||||||
self.snapshot
|
self.snapshot
|
||||||
@@ -1279,6 +1453,39 @@ impl OperatorTui {
|
|||||||
.unwrap_or(0)
|
.unwrap_or(0)
|
||||||
.min(self.snapshot.sessions.len().saturating_sub(1));
|
.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, .. }) => {
|
ControlPayload::ScheduledJobs(Page { items, .. }) => {
|
||||||
let index = Self::screen_index(OperatorScreen::Roaming);
|
let index = Self::screen_index(OperatorScreen::Roaming);
|
||||||
let identity = self
|
let identity = self
|
||||||
@@ -1701,7 +1908,7 @@ impl OperatorTui {
|
|||||||
fn draw_sessions(&self, frame: &mut Frame<'_>, area: Rect, color: bool) {
|
fn draw_sessions(&self, frame: &mut Frame<'_>, area: Rect, color: bool) {
|
||||||
let rows = self
|
let rows = self
|
||||||
.snapshot
|
.snapshot
|
||||||
.sessions
|
.conversations
|
||||||
.iter()
|
.iter()
|
||||||
.enumerate()
|
.enumerate()
|
||||||
.skip(self.selected_row().saturating_sub(5))
|
.skip(self.selected_row().saturating_sub(5))
|
||||||
@@ -1712,24 +1919,32 @@ impl OperatorTui {
|
|||||||
vec![
|
vec![
|
||||||
item.session_id.clone(),
|
item.session_id.clone(),
|
||||||
item.avatar_id.clone(),
|
item.avatar_id.clone(),
|
||||||
item.channel.clone(),
|
format!("{:?}", item.channel).to_lowercase(),
|
||||||
item.turns.to_string(),
|
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(
|
let detail = self.snapshot.conversations.get(self.selected_row()).map_or_else(
|
||||||
|| vec![Line::raw("No active sessions")],
|
|| vec![Line::raw("No conversations")],
|
||||||
|item| {
|
|item| {
|
||||||
vec![
|
let mut lines = vec![
|
||||||
Line::raw(format!("session: {}", item.session_id)),
|
Line::raw(format!("session: {}", item.session_id)),
|
||||||
Line::raw(format!("avatar: {}", item.avatar_id)),
|
Line::raw(format!("avatar: {}", item.avatar_id)),
|
||||||
Line::raw(format!(
|
Line::raw(format!(
|
||||||
"created: {} last active: {}",
|
"status: {:?} compacted={} created: {} last: {}",
|
||||||
item.created_unix_millis, item.last_active_unix_millis
|
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(
|
Self::draw_table_detail(
|
||||||
@@ -1737,7 +1952,7 @@ impl OperatorTui {
|
|||||||
area,
|
area,
|
||||||
color,
|
color,
|
||||||
"Sessions",
|
"Sessions",
|
||||||
["Session", "Avatar", "Channel", "Turns", "Bytes"],
|
["Session", "Avatar", "Channel", "Turns", "Bytes/status"],
|
||||||
rows,
|
rows,
|
||||||
detail,
|
detail,
|
||||||
);
|
);
|
||||||
@@ -2935,6 +3150,7 @@ async fn worker_send(
|
|||||||
.map_err(|error| TuiError(format!("{:?}: {}", error.code, error.message)))
|
.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<TuiInput> {
|
fn map_event(app: &OperatorTui, value: &event::Event) -> Option<TuiInput> {
|
||||||
let event::Event::Key(key) = value else {
|
let event::Event::Key(key) = value else {
|
||||||
return matches!(value, event::Event::Resize(_, _)).then_some(TuiInput::Refresh);
|
return matches!(value, event::Event::Resize(_, _)).then_some(TuiInput::Refresh);
|
||||||
@@ -2980,6 +3196,12 @@ fn map_event(app: &OperatorTui, value: &event::Event) -> Option<TuiInput> {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
match key.code {
|
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::Tab | event::KeyCode::Right => Some(TuiInput::NextScreen),
|
||||||
event::KeyCode::BackTab | event::KeyCode::Left => Some(TuiInput::PreviousScreen),
|
event::KeyCode::BackTab | event::KeyCode::Left => Some(TuiInput::PreviousScreen),
|
||||||
event::KeyCode::Up => Some(TuiInput::ScrollUp),
|
event::KeyCode::Up => Some(TuiInput::ScrollUp),
|
||||||
@@ -2987,7 +3209,7 @@ fn map_event(app: &OperatorTui, value: &event::Event) -> Option<TuiInput> {
|
|||||||
event::KeyCode::Char('/')
|
event::KeyCode::Char('/')
|
||||||
if matches!(
|
if matches!(
|
||||||
app.screen,
|
app.screen,
|
||||||
OperatorScreen::Timeline | OperatorScreen::Memories
|
OperatorScreen::Timeline | OperatorScreen::Memories | OperatorScreen::Sessions
|
||||||
) =>
|
) =>
|
||||||
{
|
{
|
||||||
Some(TuiInput::BeginFilter)
|
Some(TuiInput::BeginFilter)
|
||||||
@@ -2995,15 +3217,24 @@ fn map_event(app: &OperatorTui, value: &event::Event) -> Option<TuiInput> {
|
|||||||
event::KeyCode::Enter if app.screen == OperatorScreen::Memories => {
|
event::KeyCode::Enter if app.screen == OperatorScreen::Memories => {
|
||||||
Some(TuiInput::MemoryLoad)
|
Some(TuiInput::MemoryLoad)
|
||||||
}
|
}
|
||||||
|
event::KeyCode::Enter if app.screen == OperatorScreen::Sessions => {
|
||||||
|
Some(TuiInput::ConversationLoad)
|
||||||
|
}
|
||||||
event::KeyCode::Char('g') if app.screen == OperatorScreen::Memories => {
|
event::KeyCode::Char('g') if app.screen == OperatorScreen::Memories => {
|
||||||
Some(TuiInput::MemoryNextAgent)
|
Some(TuiInput::MemoryNextAgent)
|
||||||
}
|
}
|
||||||
event::KeyCode::Char('n') if app.screen == OperatorScreen::Memories => {
|
event::KeyCode::Char('n') if app.screen == OperatorScreen::Memories => {
|
||||||
Some(TuiInput::MemoryNextPage)
|
Some(TuiInput::MemoryNextPage)
|
||||||
}
|
}
|
||||||
|
event::KeyCode::Char('n') if app.screen == OperatorScreen::Sessions => {
|
||||||
|
Some(TuiInput::ConversationNextPage)
|
||||||
|
}
|
||||||
event::KeyCode::Char('b') if app.screen == OperatorScreen::Memories => {
|
event::KeyCode::Char('b') if app.screen == OperatorScreen::Memories => {
|
||||||
Some(TuiInput::MemoryPreviousPage)
|
Some(TuiInput::MemoryPreviousPage)
|
||||||
}
|
}
|
||||||
|
event::KeyCode::Char('b') if app.screen == OperatorScreen::Sessions => {
|
||||||
|
Some(TuiInput::ConversationPreviousPage)
|
||||||
|
}
|
||||||
event::KeyCode::Char('k') if app.screen == OperatorScreen::Memories => {
|
event::KeyCode::Char('k') if app.screen == OperatorScreen::Memories => {
|
||||||
Some(TuiInput::MemoryCycleKind)
|
Some(TuiInput::MemoryCycleKind)
|
||||||
}
|
}
|
||||||
@@ -3088,6 +3319,43 @@ mod ratatui_private_tests {
|
|||||||
app.command_shortcut('e'),
|
app.command_shortcut('e'),
|
||||||
Some(OperatorCommand::ExpireSession { avatar_id, .. }) if avatar_id == "avatar-b"
|
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 {
|
app.apply(ControlPayload::Sessions(Page {
|
||||||
items: (0..20).map(|index| session(&index.to_string())).collect(),
|
items: (0..20).map(|index| session(&index.to_string())).collect(),
|
||||||
|
|||||||
8
vendor/mentra/src/agent.rs
vendored
8
vendor/mentra/src/agent.rs
vendored
@@ -415,6 +415,14 @@ impl Agent {
|
|||||||
self.memory.transcript()
|
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.
|
/// The transcript entry the next turn will continue from.
|
||||||
pub fn leaf(&self) -> Option<&crate::transcript::EntryId> {
|
pub fn leaf(&self) -> Option<&crate::transcript::EntryId> {
|
||||||
self.transcript().leaf()
|
self.transcript().leaf()
|
||||||
|
|||||||
9
vendor/mentra/src/memory/journal/ops.rs
vendored
9
vendor/mentra/src/memory/journal/ops.rs
vendored
@@ -165,6 +165,15 @@ impl AgentMemory {
|
|||||||
&self.history_cache
|
&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 {
|
pub fn revision(&self) -> u64 {
|
||||||
self.state.revision
|
self.state.revision
|
||||||
}
|
}
|
||||||
|
|||||||
37
vendor/mentra/src/memory/journal/tests.rs
vendored
37
vendor/mentra/src/memory/journal/tests.rs
vendored
@@ -5,6 +5,7 @@ use serde_json::json;
|
|||||||
use crate::{
|
use crate::{
|
||||||
AgentTranscript, ContentBlock, Message, TranscriptKind,
|
AgentTranscript, ContentBlock, Message, TranscriptKind,
|
||||||
memory::journal::{AgentMemory, AgentMemoryState, CompactionOutcome, PendingTurnState},
|
memory::journal::{AgentMemory, AgentMemoryState, CompactionOutcome, PendingTurnState},
|
||||||
|
memory::{MemoryRecord, MemoryRecordKind, MemoryStore},
|
||||||
runtime::VolatileRuntimeStore,
|
runtime::VolatileRuntimeStore,
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -42,6 +43,42 @@ fn begin_run_commit_and_finish_persist_memory_state() {
|
|||||||
assert!(memory.resumable_user_message().is_none());
|
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]
|
#[test]
|
||||||
fn rollback_and_compaction_update_memory_state() {
|
fn rollback_and_compaction_update_memory_state() {
|
||||||
let store = Arc::new(VolatileRuntimeStore::new());
|
let store = Arc::new(VolatileRuntimeStore::new());
|
||||||
|
|||||||
Reference in New Issue
Block a user