studio/sol
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: per-user auto-memory with ResponseContext

Browse Source 
Three memory channels: hidden tool (sol.memory.set/get in scripts),
pre-response injection (relevant memories loaded into system prompt),
and post-response extraction (ministral-3b extracts facts after each
response). User isolation enforced at Rust level — user_id derived
from Matrix sender, never from script arguments.

New modules: context (ResponseContext), memory (schema, store, extractor).
ResponseContext threaded through responder → tools → script runtime.
OpenSearch index sol_user_memory created on startup alongside archive.
This commit is contained in:
Sienna Meridian Satterwhite
2026-03-21 15:51:31 +00:00
parent 4dc20bee23
commit 4949e70ecc
23 changed files with 4494 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
src/main.rs
View File

@@ -1,7 +1,9 @@
mod archive;
mod brain;
mod config;
mod context;
mod matrix_utils;
mod memory;
mod sync;
mod tools;
@@ -18,7 +20,8 @@ use url::Url;
use archive::indexer::Indexer;
use archive::schema::create_index_if_not_exists;
use brain::conversation::ConversationManager;
use brain::conversation::{ContextMessage, ConversationManager};
use memory::schema::create_index_if_not_exists as create_memory_index;
use brain::evaluator::Evaluator;
use brain::personality::Personality;
use brain::responder::Responder;
@@ -93,8 +96,9 @@ async fn main() -> anyhow::Result<()> {
.build()?;
let os_client = OpenSearch::new(os_transport);
// Ensure index exists
// Ensure indices exist
create_index_if_not_exists(&os_client, &config.opensearch.index).await?;
create_memory_index(&os_client, &config.opensearch.memory_index).await?;
// Initialize Mistral client
let mistral_client = mistralai_client::v1::client::Client::new(
@@ -107,22 +111,32 @@ async fn main() -> anyhow::Result<()> {
// Build components
let personality = Arc::new(Personality::new(system_prompt));
let conversations = Arc::new(Mutex::new(ConversationManager::new(
config.behavior.room_context_window,
config.behavior.dm_context_window,
)));
// Backfill conversation context from archive before starting
if config.behavior.backfill_on_join {
info!("Backfilling conversation context from archive...");
if let Err(e) = backfill_conversations(&os_client, &config, &conversations).await {
error!("Backfill failed (non-fatal): {e}");
}
}
let tool_registry = Arc::new(ToolRegistry::new(
os_client.clone(),
matrix_client.clone(),
config.clone(),
));
let indexer = Arc::new(Indexer::new(os_client, config.clone()));
let indexer = Arc::new(Indexer::new(os_client.clone(), config.clone()));
let evaluator = Arc::new(Evaluator::new(config.clone()));
let responder = Arc::new(Responder::new(
config.clone(),
personality,
tool_registry,
os_client.clone(),
));
let conversations = Arc::new(Mutex::new(ConversationManager::new(
config.behavior.room_context_window,
config.behavior.dm_context_window,
)));
// Start background flush task
let _flush_handle = indexer.start_flush_task();
@@ -135,8 +149,17 @@ async fn main() -> anyhow::Result<()> {
responder,
conversations,
mistral,
opensearch: os_client,
last_response: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),
responding_in: Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())),
});
// Backfill reactions from Matrix room timelines
info!("Backfilling reactions from room timelines...");
if let Err(e) = backfill_reactions(&matrix_client, &state.indexer).await {
error!("Reaction backfill failed (non-fatal): {e}");
}
// Start sync loop in background
let sync_client = matrix_client.clone();
let sync_state = state.clone();
@@ -158,3 +181,160 @@ async fn main() -> anyhow::Result<()> {
info!("Sol has shut down");
Ok(())
}
/// Backfill conversation context from the OpenSearch archive.
///
/// Queries the most recent messages per room and seeds the ConversationManager
/// so Sol has context surviving restarts.
async fn backfill_conversations(
os_client: &OpenSearch,
config: &Config,
conversations: &Arc<Mutex<ConversationManager>>,
) -> anyhow::Result<()> {
use serde_json::json;
let window = config.behavior.room_context_window.max(config.behavior.dm_context_window);
let index = &config.opensearch.index;
// Get all distinct rooms
let agg_body = json!({
"size": 0,
"aggs": {
"rooms": {
"terms": { "field": "room_id", "size": 500 }
}
}
});
let response = os_client
.search(opensearch::SearchParts::Index(&[index]))
.body(agg_body)
.send()
.await?;
let body: serde_json::Value = response.json().await?;
let buckets = body["aggregations"]["rooms"]["buckets"]
.as_array()
.cloned()
.unwrap_or_default();
let mut total = 0;
for bucket in &buckets {
let room_id = bucket["key"].as_str().unwrap_or("");
if room_id.is_empty() {
continue;
}
// Fetch recent messages for this room
let query = json!({
"size": window,
"sort": [{ "timestamp": "asc" }],
"query": {
"bool": {
"filter": [
{ "term": { "room_id": room_id } },
{ "term": { "redacted": false } }
]
}
},
"_source": ["sender_name", "sender", "content", "timestamp"]
});
let resp = os_client
.search(opensearch::SearchParts::Index(&[index]))
.body(query)
.send()
.await?;
let data: serde_json::Value = resp.json().await?;
let hits = data["hits"]["hits"].as_array().cloned().unwrap_or_default();
if hits.is_empty() {
continue;
}
let mut convs = conversations.lock().await;
for hit in &hits {
let src = &hit["_source"];
let sender = src["sender_name"]
.as_str()
.or_else(|| src["sender"].as_str())
.unwrap_or("unknown");
let content = src["content"].as_str().unwrap_or("");
let timestamp = src["timestamp"].as_i64().unwrap_or(0);
convs.add_message(
room_id,
false, // we don't know if it's a DM from the archive, use group window
ContextMessage {
sender: sender.to_string(),
content: content.to_string(),
timestamp,
},
);
total += 1;
}
}
info!(rooms = buckets.len(), messages = total, "Backfill complete");
Ok(())
}
/// Backfill reactions from Matrix room timelines into the archive.
///
/// For each joined room, fetches recent timeline events and indexes any
/// m.reaction events that aren't already in the archive.
async fn backfill_reactions(
client: &Client,
indexer: &Arc<Indexer>,
) -> anyhow::Result<()> {
use matrix_sdk::room::MessagesOptions;
use ruma::events::AnySyncTimelineEvent;
use ruma::uint;
let rooms = client.joined_rooms();
let mut total = 0;
for room in &rooms {
let room_id = room.room_id().to_string();
// Fetch recent messages (backwards from now)
let mut options = MessagesOptions::backward();
options.limit = uint!(500);
let messages = match room.messages(options).await {
Ok(m) => m,
Err(e) => {
error!(room = room_id.as_str(), "Failed to fetch timeline for reaction backfill: {e}");
continue;
}
};
for event in &messages.chunk {
let Ok(deserialized) = event.raw().deserialize() else {
continue;
};
if let AnySyncTimelineEvent::MessageLike(
ruma::events::AnySyncMessageLikeEvent::Reaction(reaction_event),
) = deserialized
{
let original = match reaction_event {
ruma::events::SyncMessageLikeEvent::Original(ref o) => o,
_ => continue,
};
let target_event_id = original.content.relates_to.event_id.to_string();
let sender = original.sender.to_string();
let emoji = &original.content.relates_to.key;
let timestamp: i64 = original.origin_server_ts.0.into();
indexer.add_reaction(&target_event_id, &sender, emoji, timestamp).await;
total += 1;
}
}
}
info!(reactions = total, rooms = rooms.len(), "Reaction backfill complete");
Ok(())
}