//! Matrix SDK session handling, controlled sync ingestion, and event policy.
use std::{
path::{Path, PathBuf},
sync::Arc,
time::Duration,
};
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use matrix_sdk::{
Client,
config::SyncSettings,
ruma::{
OwnedEventId, OwnedRoomId, OwnedTransactionId, UInt,
events::{
relation::Reply,
room::message::{
Relation, RoomMessageEventContent, RoomMessageEventContentWithoutRelation,
},
},
},
};
use serde_json::Value;
use sha2::{Digest, Sha256};
use tokio::sync::{Mutex, RwLock};
use tracing::debug;
use crate::{config::MatrixConfig, errors::MatrixError, models::InvocationCandidate};
/// Holds the Matrix events returned from one controlled sync request.
#[derive(Clone, Debug)]
pub struct SyncBatch {
/// Matrix token to use as `since` for the next sync request.
pub next_batch: String,
/// Joined-room updates, in their server-provided event order.
pub rooms: Vec<SyncRoom>,
}
/// Holds one room's timeline events from a controlled sync request.
#[derive(Clone, Debug)]
pub struct SyncRoom {
/// Room ID containing these events.
pub room_id: String,
/// Raw Matrix timeline events supplied by the SDK.
pub events: Vec<Value>,
}
/// Reports whether a durable candidate is an accepted invocation.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Classification {
/// The event invokes Lisa and should receive one reply.
Accepted,
/// The event must not invoke Lisa, with a stable diagnostic reason.
Ignored(&'static str),
}
/// Wraps the Matrix SDK while keeping Lisa's sync cursor authoritative.
#[derive(Clone, Debug)]
pub struct MatrixClient {
client: Arc<RwLock<Client>>,
config: MatrixConfig,
matrix_store_path: Arc<PathBuf>,
reconnect_lock: Arc<Mutex<()>>,
}
impl MatrixClient {
/// Builds an unencrypted Matrix SDK client and creates Lisa's session.
pub async fn connect(
config: MatrixConfig,
matrix_store_path: &Path,
) -> Result<Self, MatrixError> {
let matrix_store_path = Arc::new(matrix_store_path.to_owned());
let client = Self::login(&config, &matrix_store_path).await?;
Ok(Self {
client: Arc::new(RwLock::new(client)),
config,
matrix_store_path,
reconnect_lock: Arc::new(Mutex::new(())),
})
}
/// Recreates Lisa's Matrix session after a terminal token failure.
pub async fn reconnect(&self) -> Result<(), MatrixError> {
let _guard = self.reconnect_lock.lock().await;
let client = Self::login(&self.config, &self.matrix_store_path).await?;
*self.client.write().await = client;
Ok(())
}
async fn login(config: &MatrixConfig, matrix_store_path: &Path) -> Result<Client, MatrixError> {
let client = Client::builder()
.homeserver_url(config.homeserver_url.as_str())
.sqlite_store(matrix_store_path, None)
.handle_refresh_tokens()
.build()
.await
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
client
.matrix_auth()
.login_username(&config.user_id, config.password.expose())
.device_id(&config.device_id)
.initial_device_display_name(&config.device_display_name)
.request_refresh_token()
.send()
.await
.map_err(loginError)?;
let identity = client
.whoami()
.await
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
if identity.user_id.as_str() != config.user_id
|| identity
.device_id
.as_deref()
.is_none_or(|device_id| device_id.as_str() != config.device_id)
{
return Err(MatrixError::IdentityMismatch);
}
Ok(client)
}
async fn activeClient(&self) -> Client {
self.client.read().await.clone()
}
/// Executes exactly one explicit-token Matrix sync request.
pub async fn syncOnce(&self, since: Option<String>) -> Result<SyncBatch, MatrixError> {
let client = self.activeClient().await;
let mut settings =
SyncSettings::new().timeout(Duration::from_millis(self.config.sync_timeout_ms));
if let Some(token) = since {
settings = settings.token(token);
}
let response = client.sync_once(settings).await?;
let mut rooms = Vec::new();
for (room_id, update) in response.rooms.joined {
if !self.config.allowed_room_ids.contains(room_id.as_str()) {
debug!(
room_id = %room_id,
timeline_event_count = update.timeline.events.len(),
"matrix_sync_room_not_allowed"
);
continue;
}
debug!(
room_id = %room_id,
timeline_event_count = update.timeline.events.len(),
"matrix_sync_room_allowed"
);
let events = update
.timeline
.events
.into_iter()
.filter_map(|event| serde_json::from_str(event.kind.raw().json().get()).ok())
.collect();
rooms.push(SyncRoom {
room_id: room_id.to_string(),
events,
});
}
Ok(SyncBatch {
next_batch: response.next_batch,
rooms,
})
}
/// Returns whether a reply target was sent by Lisa in the same room.
pub async fn replyTargetIsLisa(
&self,
room_id: &str,
event_id: &str,
) -> Result<bool, MatrixError> {
let client = self.activeClient().await;
let room_id: OwnedRoomId = room_id
.parse::<OwnedRoomId>()
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
let event_id: OwnedEventId = event_id
.parse::<OwnedEventId>()
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
let Some(room) = client.get_room(&room_id) else {
return Ok(false);
};
let event = room.event(&event_id, None).await?;
let raw: Value = serde_json::from_str(event.kind.raw().json().get())
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
Ok(raw.get("sender").and_then(Value::as_str) == Some(&self.config.user_id))
}
/// Sends the exact persisted test-reply payload with its deterministic transaction ID.
pub async fn sendPersistedReply(
&self,
room_id: &str,
response_json: &str,
transaction_id: &str,
) -> Result<String, MatrixError> {
let client = self.activeClient().await;
let room_id: OwnedRoomId = room_id
.parse::<OwnedRoomId>()
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
let transaction_id: OwnedTransactionId = transaction_id.into();
let Some(room) = client.get_room(&room_id) else {
return Err(MatrixError::InvalidResponse(
"configured room is unavailable".to_owned(),
));
};
let content: RoomMessageEventContent = serde_json::from_str(response_json)
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
let response = room
.send(content)
.with_transaction_id(transaction_id)
.await?;
Ok(response.response.event_id.to_string())
}
/// Fetches only the bounded history preceding an invocation event.
pub async fn contextBefore(
&self,
room_id: &str,
event_id: &str,
maximum_events: u32,
) -> Result<Vec<Value>, MatrixError> {
let client = self.activeClient().await;
let room_id: OwnedRoomId = room_id
.parse::<OwnedRoomId>()
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
let event_id: OwnedEventId = event_id
.parse::<OwnedEventId>()
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
let Some(room) = client.get_room(&room_id) else {
return Err(MatrixError::InvalidResponse(
"configured room is unavailable".to_owned(),
));
};
let limit = UInt::new(u64::from(maximum_events))
.ok_or_else(|| MatrixError::InvalidResponse("invalid context limit".to_owned()))?;
let response = room
.event_with_context(&event_id, true, limit, None)
.await?;
Ok(response
.events_before
.into_iter()
.filter_map(|event| serde_json::from_str(event.kind.raw().json().get()).ok())
.collect())
}
/// Updates Lisa's typing notice for one configured joined room.
pub async fn setTyping(&self, room_id: &str, typing: bool) -> Result<(), MatrixError> {
let client = self.activeClient().await;
let room_id: OwnedRoomId = room_id
.parse::<OwnedRoomId>()
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
let Some(room) = client.get_room(&room_id) else {
return Err(MatrixError::InvalidResponse(
"configured room is unavailable".to_owned(),
));
};
room.typing_notice(typing).await?;
Ok(())
}
}
fn loginError(error: matrix_sdk::Error) -> MatrixError {
if let matrix_sdk::Error::Http(http_error) = &error
&& matches!(
http_error.client_api_error_kind(),
Some(matrix_sdk::ruma::api::error::ErrorKind::Forbidden)
)
{
MatrixError::InvalidCredentials
}
else
{
MatrixError::Sdk(error)
}
}
/// Returns a deterministic transaction ID for an invocation's single reply.
#[must_use]
pub fn transactionId(room_id: &str, event_id: &str) -> String {
let mut digest = Sha256::new();
digest.update(room_id.as_bytes());
digest.update([0]);
digest.update(event_id.as_bytes());
let encoded = URL_SAFE_NO_PAD.encode(digest.finalize());
format!("lisa_{}", &encoded[..32])
}
/// Determines whether a raw event is cheap enough to durably retain as a candidate.
#[must_use]
pub fn isPotentialCandidate(event: &Value, config: &MatrixConfig) -> bool {
if !basic_text_event(event, config) {
return false;
}
replyTarget(event).is_some() || event_mentions_lisa(event, config)
}
/// Classifies a retained candidate after reply-target verification.
#[must_use]
pub fn classifyEvent(
event: &Value,
room_id: &str,
config: &MatrixConfig,
reply_target_is_lisa: bool,
) -> Classification {
if !config.allowed_room_ids.contains(room_id) {
return Classification::Ignored("room_not_allowed");
}
if !basic_text_event(event, config) {
return Classification::Ignored("not_text_message");
}
if event_mentions_lisa(event, config) || (replyTarget(event).is_some() && reply_target_is_lisa)
{
Classification::Accepted
} else {
Classification::Ignored("not_a_lisa_invocation")
}
}
/// Creates an invocation candidate from a previously filtered raw Matrix event.
pub fn candidateFromEvent(event: &Value, room_id: &str) -> Option<InvocationCandidate> {
let event_id = event.get("event_id")?.as_str()?.to_owned();
let sender_id = event.get("sender")?.as_str()?.to_owned();
let received_at_ms = event
.get("origin_server_ts")
.and_then(Value::as_u64)
.unwrap_or_default();
Some(InvocationCandidate {
matrix_txn_id: transactionId(room_id, &event_id),
event_id,
room_id: room_id.to_owned(),
sender_id,
received_at_ms,
raw_event_json: event.to_string(),
})
}
/// Extracts a rich-reply target event ID from an event.
#[must_use]
pub fn replyTarget(event: &Value) -> Option<&str> {
event
.pointer("/content/m.relates_to/m.in_reply_to/event_id")?
.as_str()
}
/// Builds the persisted Phase 1 response payload for a valid invocation.
pub fn testReplyPayload(invocation_event_id: &str) -> Result<String, MatrixError> {
let event_id: OwnedEventId = invocation_event_id
.parse::<OwnedEventId>()
.map_err(|error| MatrixError::InvalidResponse(error.to_string()))?;
let mut content = RoomMessageEventContent::text_plain("Lisa's Matrix foundation is online.");
content.relates_to = Some(Relation::<RoomMessageEventContentWithoutRelation>::Reply(
Reply::with_event_id(event_id),
));
serde_json::to_string(&content).map_err(|error| MatrixError::InvalidResponse(error.to_string()))
}
fn basic_text_event(event: &Value, config: &MatrixConfig) -> bool {
event.get("type").and_then(Value::as_str) == Some("m.room.message")
&& event.pointer("/content/msgtype").and_then(Value::as_str) == Some("m.text")
&& event.get("sender").and_then(Value::as_str) != Some(&config.user_id)
&& event.pointer("/unsigned/redacted_because").is_none()
&& event
.pointer("/content/m.relates_to/rel_type")
.and_then(Value::as_str)
!= Some("m.replace")
&& event
.pointer("/content/body")
.and_then(Value::as_str)
.is_some()
}
fn event_mentions_lisa(event: &Value, config: &MatrixConfig) -> bool {
let structured = event
.pointer("/content/m.mentions/user_ids")
.and_then(Value::as_array)
.is_some_and(|users| {
users
.iter()
.any(|user| user.as_str() == Some(&config.user_id))
});
structured
|| event
.pointer("/content/body")
.and_then(Value::as_str)
.is_some_and(|body| alias_matches(&strip_reply_fallback(body), &config.mention_aliases))
}
fn strip_reply_fallback(body: &str) -> String {
if !body.starts_with('>') {
return body.to_owned();
}
body.split_once("\n\n")
.map_or_else(|| body.to_owned(), |(_, reply)| reply.to_owned())
}
fn alias_matches(body: &str, aliases: &[String]) -> bool {
let normalized_body = body.to_lowercase();
aliases.iter().any(|alias| {
let alias = alias.to_lowercase();
normalized_body.match_indices(&alias).any(|(index, _)| {
let before = normalized_body[..index].chars().next_back();
let after = normalized_body[index + alias.len()..].chars().next();
before.is_none_or(|character| !is_word(character))
&& after.is_none_or(|character| !is_word(character))
})
})
}
fn is_word(character: char) -> bool {
character.is_alphanumeric() || character == '_'
}
#[cfg(test)]
mod tests {
use serde_json::{Value, json};
use super::{
Classification, classifyEvent, isPotentialCandidate, testReplyPayload, transactionId,
};
use crate::config::AppConfig;
fn config() -> crate::config::MatrixConfig {
let raw = r#"
[matrix]
homeserver_url = "https://matrix.example.test"
user_id = "@lisa:example.test"
device_id = "LISABOT"
device_display_name = "Lisa bot"
password = "env:TOKEN"
allowed_room_ids = ["!room:example.test"]
mention_aliases = ["Lisa", "@Lisa"]
sync_timeout_ms = 30000
sync_timeline_limit = 100
gap_page_limit = 20
[runtime]
database_path = "state.sqlite3"
matrix_store_path = "matrix_sdk"
max_concurrent_rooms = 1
invocation_timeout_seconds = 10
shutdown_grace_seconds = 10
log_level = "INFO"
[llm]
base_url = "http://llm.example.test/v1/"
api_key = "env:TOKEN"
model = "operator-model"
request_timeout_seconds = 90
max_output_tokens = 700
max_request_characters = 120000
max_response_characters = 12000
max_iterations = 8
max_tool_calls = 6
temperature = 0.7
[persona]
path = "persona.md"
generic_error = "Sorry [{error_code}]"
overload_error = "Too much"
[context]
max_events = 40
max_characters = 24000
max_event_characters = 4000
max_reply_depth = 2
[search]
endpoint = "https://search.example.test/search"
max_results = 8
timeout_seconds = 15
[web]
connect_timeout_seconds = 5
read_timeout_seconds = 15
total_timeout_seconds = 20
max_redirects = 5
max_download_bytes = 2000000
max_tool_output_characters = 30000
max_total_artifact_bytes = 4000000
pandoc_timeout_seconds = 10
user_agent = "LisaBot/1.0"
"#;
AppConfig::from_toml_with_env(raw, |_| Some("token".to_owned()))
.unwrap()
.matrix
}
fn message(body: &str) -> Value {
json!({
"event_id": "$event:example.test",
"sender": "@human:example.test",
"type": "m.room.message",
"origin_server_ts": 1,
"content": {"msgtype": "m.text", "body": body}
})
}
#[test]
fn accepts_structured_and_bounded_alias_mentions() {
let config = config();
let event = message("hello Lisa");
assert!(isPotentialCandidate(&event, &config));
event.as_object().unwrap();
assert_eq!(
classifyEvent(&event, "!other:example.test", &config, false),
Classification::Ignored("room_not_allowed")
);
let mut event = message("hello LISA");
assert_eq!(
classifyEvent(&event, "!room:example.test", &config, false),
Classification::Accepted
);
event
.pointer_mut("/content/body")
.unwrap()
.clone_from(&json!("lisabet is not a mention"));
assert!(!isPotentialCandidate(&event, &config));
}
#[test]
fn ignores_reply_fallback_and_generates_stable_transaction_id() {
let config = config();
let event = message("> <@lisa:example.test> Lisa\n>\n> quoted\n\nordinary reply");
assert!(!isPotentialCandidate(&event, &config));
assert_eq!(
transactionId("!room:example.test", "$event:example.test"),
transactionId("!room:example.test", "$event:example.test")
);
assert_ne!(
transactionId("!other:example.test", "$event:example.test"),
transactionId("!room:example.test", "$event:example.test")
);
assert!(
testReplyPayload("$event:example.test")
.unwrap()
.contains("m.in_reply_to")
);
}
}