//! Foundation application lifecycle and exactly-once test-response flow.
use std::{
fs,
sync::Arc,
time::{SystemTime, UNIX_EPOCH},
};
use serde_json::Value;
use tokio::{
sync::{Semaphore, mpsc},
task::{JoinHandle, JoinSet},
time::{Duration, sleep, timeout},
};
use tokio_util::sync::CancellationToken;
use tracing::{info, warn};
use crate::{
config::{AppConfig, ERROR_CODE_PLACEHOLDER},
context::{buildContext, renderContext},
errors::AppError,
llm::{LlmClient, ModelReply, appendToolResult, initialMessages, systemPrompt},
matrix::{
Classification, MatrixClient, SyncBatch, candidateFromEvent, classifyEvent,
isPotentialCandidate, replyTarget,
},
rendering::{renderReply, renderReplyWithSources},
state::StateStore,
tools::{ToolRegistry, ToolScope},
};
/// Coordinates controlled sync, durable state, and the Phase 1 test response.
#[derive(Clone)]
pub struct LisaApp {
config: AppConfig,
state: StateStore,
matrix: MatrixClient,
llm: LlmClient,
tools: ToolRegistry,
system_prompt: Arc<str>,
}
impl LisaApp {
/// Connects Lisa after local configuration and state have been initialized.
pub async fn connect(config: AppConfig) -> Result<Self, AppError> {
let metadata =
fs::metadata(&config.persona.path).map_err(crate::errors::ConfigurationError::Load)?;
if !metadata.is_file() || metadata.len() > 128 * 1024 {
return Err(crate::errors::ConfigurationError::InvalidValue {
key: "persona.path",
reason: "must be a regular file below 128 KiB".to_owned(),
}
.into());
}
let persona = fs::read_to_string(&config.persona.path)
.map_err(crate::errors::ConfigurationError::Load)?;
if persona.trim().is_empty() {
return Err(crate::errors::ConfigurationError::InvalidValue {
key: "persona.path",
reason: "must not be empty".to_owned(),
}
.into());
}
let state = StateStore::open(&config.runtime.database_path)?;
let matrix =
MatrixClient::connect(config.matrix.clone(), &config.runtime.matrix_store_path).await?;
let llm = LlmClient::new(config.llm.clone())?;
let tools = ToolRegistry::new(config.search.clone(), config.web.clone())
.map_err(crate::errors::LlmError::Request)?;
Ok(Self {
config,
state,
matrix,
llm,
tools,
system_prompt: Arc::from(systemPrompt(&persona)),
})
}
/// Establishes the first-sync baseline or resumes safely persisted work.
pub async fn initialize(&self) -> Result<(), AppError> {
let sync_state = self.state.syncState().await?;
if !sync_state.initialized {
let baseline = self.matrix.syncOnce(None).await?;
self.state
.recordBaseline(baseline.next_batch, now_ms())
.await?;
info!("matrix_sync_baseline_recorded");
}
self.state.requeueUnfinished(now_ms()).await?;
self.deliverReadyReplies().await?;
Ok(())
}
/// Performs one controlled sync request and durably records its candidates.
pub async fn pollOnce(&self) -> Result<usize, AppError> {
let sync_state = self.state.syncState().await?;
let Some(token) = sync_state.next_batch else {
return Ok(0);
};
let batch = self.matrix.syncOnce(Some(token)).await?;
self.ingestSyncBatch(batch).await
}
/// Runs Lisa until an operating-system shutdown signal is received.
pub async fn run(&self) -> Result<(), AppError> {
self.initialize().await?;
let semaphore = Arc::new(Semaphore::new(
self.config.runtime.max_concurrent_rooms as usize,
));
let cancellation = CancellationToken::new();
let mut room_senders = Vec::new();
let mut workers = JoinSet::new();
for room_id in &self.config.matrix.allowed_room_ids {
let (sender, receiver) = mpsc::channel(1);
let app = self.clone();
let room_id = room_id.clone();
let semaphore = Arc::clone(&semaphore);
let worker_cancellation = cancellation.clone();
workers.spawn(async move {
app.runRoomWorker(room_id, receiver, semaphore, worker_cancellation).await;
});
room_senders.push(sender);
}
for sender in &room_senders {
let _ = sender.try_send(());
}
loop
{
tokio::select!
{
shutdown = waitForShutdown() =>
{
if let Err(error) = shutdown
{
warn!(error = %error, "shutdown_signal_failed");
}
break;
}
result = self.pollOnce() =>
{
let added = match result
{
Ok(added) => added,
Err(error) if error.needsMatrixReauthentication() =>
{
warn!("matrix_session_reauthentication_started");
self.matrix.reconnect().await?;
info!("matrix_session_reauthentication_complete");
continue;
}
Err(error) => return Err(error),
};
if added > 0
{
info!(candidate_count = added, "matrix_candidates_ingested");
}
for sender in &room_senders
{
let _ = sender.try_send(());
}
self.deliverReadyReplies().await?;
}
}
}
cancellation.cancel();
drop(room_senders);
for room_id in &self.config.matrix.allowed_room_ids
{
let _ = self.matrix.setTyping(room_id, false).await;
}
let grace = Duration::from_secs(self.config.runtime.shutdown_grace_seconds);
if timeout(grace, async { while workers.join_next().await.is_some() {} }).await.is_err()
{
workers.abort_all();
}
self.state.releaseProcessing(now_ms()).await?;
info!("lisa_shutdown_complete");
Ok(())
}
async fn runRoomWorker(
self,
room_id: String,
mut notifications: mpsc::Receiver<()>,
semaphore: Arc<Semaphore>,
cancellation: CancellationToken,
) {
loop
{
let notification = tokio::select!
{
() = cancellation.cancelled() => return,
notification = notifications.recv() => notification,
};
if notification.is_none()
{
return;
}
let Ok(permit) = Arc::clone(&semaphore).acquire_owned().await else {
return;
};
if let Err(error) = self.processRoom(&room_id).await {
warn!(error = %error, "room_worker_failed");
}
drop(permit);
}
}
/// Ingests a parsed SDK sync response in one durable state transition.
pub async fn ingestSyncBatch(&self, batch: SyncBatch) -> Result<usize, AppError> {
let mut candidates = Vec::new();
let mut room_cursors = Vec::new();
for room in batch.rooms {
let mut last_event_id = None;
for event in room.events {
last_event_id = event
.get("event_id")
.and_then(Value::as_str)
.map(str::to_owned)
.or(last_event_id);
if isPotentialCandidate(&event, &self.config.matrix)
&& let Some(candidate) = candidateFromEvent(&event, &room.room_id)
{
candidates.push(candidate);
}
}
room_cursors.push((room.room_id, last_event_id));
}
Ok(self
.state
.ingestBatch(batch.next_batch, room_cursors, candidates, now_ms())
.await?)
}
/// Processes queued candidates serially for a caller-selected room.
pub async fn processRoom(&self, room_id: &str) -> Result<(), AppError> {
let lease_until = now_ms() + self.config.runtime.invocation_timeout_seconds * 1_000;
while let Some(invocation) = self
.state
.claimNext(room_id.to_owned(), now_ms(), lease_until)
.await?
{
let Some(raw_event) = invocation.raw_event_json.as_deref() else {
self.state
.markIgnored(
invocation.event_id,
"missing_raw_event".to_owned(),
now_ms(),
)
.await?;
continue;
};
let event: Value = serde_json::from_str(raw_event)
.map_err(|error| crate::errors::MatrixError::InvalidResponse(error.to_string()))?;
let reply_target_is_lisa = if let Some(target) = replyTarget(&event) {
self.state
.isBotEvent(room_id.to_owned(), target.to_owned())
.await?
|| self.matrix.replyTargetIsLisa(room_id, target).await?
} else {
false
};
match classifyEvent(&event, room_id, &self.config.matrix, reply_target_is_lisa) {
Classification::Accepted => {
let typing = TypingLease::start(self.matrix.clone(), room_id.to_owned()).await;
let payload = self
.answerInvocation(&invocation, &event)
.await
.unwrap_or_else(|error| {
let error_code = error.userVisibleCode();
warn!(error = %error, error_code, "invocation_using_fallback");
self.fallbackPayload(&invocation, &event, error_code)
});
typing.stop().await;
self.state
.storeReply(invocation.event_id, payload, now_ms())
.await?;
self.deliverReadyReplies().await?;
}
Classification::Ignored(reason) => {
self.state
.markIgnored(invocation.event_id, reason.to_owned(), now_ms())
.await?;
}
}
}
Ok(())
}
async fn answerInvocation(
&self,
invocation: &crate::models::InvocationRecord,
event: &Value,
) -> Result<String, AppError> {
let history = self
.matrix
.contextBefore(
&invocation.room_id,
&invocation.event_id,
self.config.context.max_events,
)
.await?;
let messages = buildContext(&invocation.room_id, event, history, &self.config.context);
let context = renderContext(&messages);
let mut messages = initialMessages(&self.system_prompt, &context);
let schemas = ToolRegistry::schemas();
let mut scope = ToolScope::new(&self.config.web);
let mut tool_call_count: u32 = 0;
for _ in 0..self.config.llm.max_iterations
{
match self.llm.completeTurn(&messages, &schemas).await?
{
ModelReply::Final(answer) =>
{
let sources = scope.sources.select(&answer.source_urls)
.ok_or(crate::errors::LlmError::Protocol)?;
if !scope.sources.is_empty() && sources.is_empty()
{
return Err(crate::errors::LlmError::Protocol.into());
}
let excerpt = event.pointer("/content/body")
.and_then(Value::as_str).unwrap_or_default();
return Ok(renderReplyWithSources(
&invocation.event_id,
&invocation.sender_id,
excerpt,
&answer.markdown,
&sources,
self.config.llm.max_response_characters,
)?);
}
ModelReply::ToolCalls(calls) =>
{
let call_count = u32::try_from(calls.len())
.map_err(|_| crate::errors::LlmError::Protocol)?;
tool_call_count = tool_call_count.saturating_add(call_count);
if calls.is_empty() || tool_call_count > self.config.llm.max_tool_calls
{
return Err(crate::errors::LlmError::Protocol.into());
}
for call in calls
{
let result = self.tools.execute(
&call.name,
call.arguments.clone(),
&mut scope,
).await;
appendToolResult(&mut messages, &call, &result);
}
}
}
}
Err(crate::errors::LlmError::Protocol.into())
}
fn fallbackPayload(
&self,
invocation: &crate::models::InvocationRecord,
event: &Value,
error_code: &str,
) -> String {
let excerpt = event
.pointer("/content/body")
.and_then(Value::as_str)
.unwrap_or_default();
renderReply(
&invocation.event_id,
&invocation.sender_id,
excerpt,
&self
.config
.persona
.generic_error
.replace(ERROR_CODE_PLACEHOLDER, error_code),
self.config.llm.max_response_characters,
)
.expect("validated configured fallback must render")
}
/// Retries fixed persisted payloads with their original transaction IDs.
pub async fn deliverReadyReplies(&self) -> Result<(), AppError> {
for invocation in self.state.readyToSend().await? {
let Some(response_json) = invocation.response_json.as_deref() else {
warn!(event_id = %invocation.event_id, "ready_reply_missing_payload");
continue;
};
match self
.matrix
.sendPersistedReply(
&invocation.room_id,
response_json,
&invocation.matrix_txn_id,
)
.await
{
Ok(sent_event_id) => {
self.state
.recordSent(invocation.event_id, sent_event_id, now_ms())
.await?;
info!("matrix_test_reply_sent");
}
Err(error) => {
warn!(error = %error, "matrix_reply_send_deferred");
}
}
}
Ok(())
}
}
struct TypingLease {
cancellation: CancellationToken,
refresh_task: JoinHandle<()>,
matrix: MatrixClient,
room_id: String,
}
impl TypingLease {
async fn start(matrix: MatrixClient, room_id: String) -> Self {
if let Err(error) = matrix.setTyping(&room_id, true).await {
warn!(error = %error, "typing_start_failed");
}
let cancellation = CancellationToken::new();
let refresh_matrix = matrix.clone();
let refresh_room_id = room_id.clone();
let refresh_cancellation = cancellation.clone();
let refresh_task = tokio::spawn(async move {
loop {
tokio::select! {
() = refresh_cancellation.cancelled() => return,
() = sleep(Duration::from_secs(3)) => {
if let Err(error) = refresh_matrix.setTyping(&refresh_room_id, true).await {
warn!(error = %error, "typing_refresh_failed");
}
}
}
}
});
Self {
cancellation,
refresh_task,
matrix,
room_id,
}
}
async fn stop(self) {
self.cancellation.cancel();
let _ = self.refresh_task.await;
if let Err(error) = self.matrix.setTyping(&self.room_id, false).await {
warn!(error = %error, "typing_stop_failed");
}
}
}
fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis()
.try_into()
.unwrap_or(u64::MAX)
}
async fn waitForShutdown() -> Result<(), std::io::Error>
{
#[cfg(unix)]
{
let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
tokio::select!
{
result = tokio::signal::ctrl_c() => result,
_ = terminate.recv() => Ok(()),
}
}
#[cfg(not(unix))]
{
tokio::signal::ctrl_c().await
}
}