Changes
diff --git a/Cargo.toml b/Cargo.toml
index 11b8b65..544d604 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,7 +17,7 @@ serde_json = "1"
sha2 = "0.10"
thiserror = "2"
tempfile = "3"
-tokio = { version = "1", features = ["io-util", "macros", "process", "rt-multi-thread", "sync", "time"] }
+tokio = { version = "1", features = ["io-util", "macros", "process", "rt-multi-thread", "signal", "sync", "time"] }
tokio-util = "0.7"
toml = "0.8"
tracing = "0.1"
diff --git a/README.md b/README.md
index 5f93db9..6d48958 100644
--- a/README.md
+++ b/README.md
@@ -25,3 +25,6 @@ and displays a refreshed Matrix typing notice while an LLM response is pending.
`pandoc` must be installed on the host. Page content is held only in memory for
the current invocation; it never receives Matrix, LLM, or search credentials.
+
+See [operations guidance](docs/operations.md) for systemd, health checks,
+shutdown, backup, and upgrade procedures.
diff --git a/deploy/lisa.service b/deploy/lisa.service
new file mode 100644
index 0000000..f061734
--- /dev/null
+++ b/deploy/lisa.service
@@ -0,0 +1,23 @@
+[Unit]
+Description=Lisa Matrix bot
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=lisa
+Group=lisa
+EnvironmentFile=/etc/lisa/lisa.env
+ExecStart=/opt/lisa/bin/lisa --config /etc/lisa/config.toml
+Restart=on-failure
+RestartSec=5
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ReadWritePaths=/var/lib/lisa
+RestrictSUIDSGID=true
+LockPersonality=true
+
+[Install]
+WantedBy=multi-user.target
diff --git a/docs/operations.md b/docs/operations.md
new file mode 100644
index 0000000..d8b120e
--- /dev/null
+++ b/docs/operations.md
@@ -0,0 +1,41 @@
+# Operations
+
+Install `pandoc` and run Lisa as a dedicated `lisa` operating-system user. The
+user needs read access to the configuration and persona file, read/write access
+to `/var/lib/lisa`, and network access to Matrix, the LLM, SearXNG, DNS, and
+model-selected web destinations.
+
+Use [`deploy/lisa.service`](../deploy/lisa.service) as a systemd starting point.
+Keep credentials in `/etc/lisa/lisa.env` with restrictive permissions; do not
+place tokens in the TOML file or the systemd unit. The service restarts after
+temporary failures and uses its durable Matrix transaction IDs when retrying a
+reply.
+
+## Health and diagnostics
+
+Run the local, non-network health check:
+
+```sh
+lisa --config /etc/lisa/config.toml --health
+```
+
+It emits only queue counts and the age of the oldest unsent response. Routine
+logs omit room messages, prompts, downloaded pages, URLs with query strings,
+and credentials.
+
+## Shutdown, backup, and upgrade
+
+`SIGTERM` and `SIGINT` stop scheduling sync work, wait up to
+`shutdown_grace_seconds` for room workers, clear typing state, and return
+unfinished invocation leases to the candidate queue. Ready-to-send replies are
+not regenerated and remain safely retryable.
+
+For a consistent backup, stop Lisa and copy both `database_path` and
+`matrix_store_path`. Restore both together on rollback. The SQLite database is
+forward-migrated; use the saved snapshot rather than attempting a reverse
+migration.
+
+Before deploying against the production homeserver and LLM, verify the Matrix
+server's `/context`, filter, and transaction replay behavior, the LLM's
+OpenAI-compatible tool-call dialect, and SearXNG JSON output. These are the
+remaining operator-specific integration decisions from the architecture.
diff --git a/src/app.rs b/src/app.rs
index 0000640..730fab0 100644
--- a/src/app.rs
+++ b/src/app.rs
@@ -9,8 +9,8 @@ use std::{
use serde_json::Value;
use tokio::{
sync::{Semaphore, mpsc},
- task::JoinHandle,
- time::{Duration, sleep},
+ task::{JoinHandle, JoinSet},
+ time::{Duration, sleep, timeout},
};
use tokio_util::sync::CancellationToken;
use tracing::{info, warn};
@@ -102,45 +102,90 @@ impl LisaApp {
self.ingestSyncBatch(batch).await
}
- /// Runs the foundation service loop until the process is terminated.
+ /// 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);
- tokio::spawn(async move {
- app.runRoomWorker(room_id, receiver, semaphore).await;
+ 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 {
- let added = self.pollOnce().await?;
- if added > 0 {
- info!(candidate_count = added, "matrix_candidates_ingested");
- }
- 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 = result?;
+ if added > 0
+ {
+ info!(candidate_count = added, "matrix_candidates_ingested");
+ }
+ for sender in &room_senders
+ {
+ let _ = sender.try_send(());
+ }
+ self.deliverReadyReplies().await?;
+ }
}
- 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 receiver: mpsc::Receiver<()>,
+ mut notifications: mpsc::Receiver<()>,
semaphore: Arc<Semaphore>,
+ cancellation: CancellationToken,
) {
- while receiver.recv().await.is_some() {
+ 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;
};
@@ -400,3 +445,20 @@ fn now_ms() -> u64 {
.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
+ }
+}
diff --git a/src/main.rs b/src/main.rs
index c2bc305..21bd0de 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,19 +1,30 @@
//! Executable entry point for Lisa's Phase 1 foundation.
-use std::{env, path::PathBuf, process::ExitCode};
+use std::{
+ env,
+ path::PathBuf,
+ process::ExitCode,
+ time::{SystemTime, UNIX_EPOCH},
+};
-use lisa::{app::LisaApp, config::AppConfig, logging};
+use lisa::{app::LisaApp, config::AppConfig, logging, state::StateStore};
+
+struct Command
+{
+ config_path: PathBuf,
+ health: bool,
+}
#[tokio::main]
async fn main() -> ExitCode {
- let config_path = match config_path() {
- Ok(path) => path,
+ let command = match command() {
+ Ok(command) => command,
Err(error) => {
eprintln!("{error}");
return ExitCode::FAILURE;
}
};
- let config = match AppConfig::load(&config_path) {
+ let config = match AppConfig::load(&command.config_path) {
Ok(config) => config,
Err(error) => {
eprintln!("configuration error: {error}");
@@ -21,6 +32,10 @@ async fn main() -> ExitCode {
}
};
logging::init(&config.runtime.log_level);
+ if command.health
+ {
+ return health(&config).await;
+ }
let result = match LisaApp::connect(config).await {
Ok(app) => app.run().await,
Err(error) => Err(error),
@@ -34,11 +49,58 @@ async fn main() -> ExitCode {
}
}
-fn config_path() -> Result<PathBuf, String> {
+async fn health(config: &AppConfig) -> ExitCode
+{
+ let store = match StateStore::open(&config.runtime.database_path)
+ {
+ Ok(store) => store,
+ Err(error) =>
+ {
+ eprintln!("health check failed: {error}");
+ return ExitCode::FAILURE;
+ }
+ };
+ let now_ms = SystemTime::now().duration_since(UNIX_EPOCH)
+ .unwrap_or_default().as_millis().try_into().unwrap_or(u64::MAX);
+ match store.health(now_ms).await
+ {
+ Ok(snapshot) =>
+ {
+ println!(
+ "{{\"candidate_count\":{},\"ready_to_send_count\":{},\"oldest_ready_age_ms\":{}}}",
+ snapshot.candidate_count,
+ snapshot.ready_to_send_count,
+ snapshot.oldest_ready_age_ms.map_or_else(|| "null".to_owned(), |age| age.to_string()),
+ );
+ ExitCode::SUCCESS
+ }
+ Err(error) =>
+ {
+ eprintln!("health check failed: {error}");
+ ExitCode::FAILURE
+ }
+ }
+}
+
+fn command() -> Result<Command, String> {
let mut arguments = env::args_os().skip(1);
- match (arguments.next(), arguments.next()) {
- (None, None) => Ok(PathBuf::from("config.toml")),
- (Some(flag), Some(path)) if flag == "--config" => Ok(PathBuf::from(path)),
- _ => Err("usage: lisa [--config PATH]".to_owned()),
+ let mut config_path = PathBuf::from("config.toml");
+ let mut health = false;
+ while let Some(argument) = arguments.next()
+ {
+ if argument == "--health"
+ {
+ health = true;
+ }
+ else if argument == "--config"
+ {
+ config_path = arguments.next().map(PathBuf::from)
+ .ok_or_else(|| "--config requires a path".to_owned())?;
+ }
+ else
+ {
+ return Err("usage: lisa [--config PATH] [--health]".to_owned());
+ }
}
+ Ok(Command { config_path, health })
}
diff --git a/src/models.rs b/src/models.rs
index a9d3091..8b86b8a 100644
--- a/src/models.rs
+++ b/src/models.rs
@@ -114,3 +114,15 @@ pub struct SyncState {
/// Matrix next-batch token, if an initial sync has completed.
pub next_batch: Option<String>,
}
+
+/// Summarizes non-sensitive durable work for a local operator health check.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct HealthSnapshot
+{
+ /// Number of accepted candidates not currently leased by a worker.
+ pub candidate_count: u64,
+ /// Number of replies durably awaiting Matrix delivery.
+ pub ready_to_send_count: u64,
+ /// Age in milliseconds of the oldest reply awaiting delivery.
+ pub oldest_ready_age_ms: Option<u64>,
+}
diff --git a/src/state.rs b/src/state.rs
index 2a97345..68f7422 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -7,7 +7,7 @@ use tokio::sync::{mpsc, oneshot};
use crate::{
errors::StateStoreError,
- models::{InvocationCandidate, InvocationRecord, InvocationStatus, SyncState},
+ models::{HealthSnapshot, InvocationCandidate, InvocationRecord, InvocationStatus, SyncState},
};
/// Provides asynchronous access to Lisa's authoritative SQLite state.
@@ -64,6 +64,14 @@ enum StateCommand {
now_ms: u64,
response: oneshot::Sender<Result<usize, StateStoreError>>,
},
+ Health {
+ now_ms: u64,
+ response: oneshot::Sender<Result<HealthSnapshot, StateStoreError>>,
+ },
+ ReleaseProcessing {
+ now_ms: u64,
+ response: oneshot::Sender<Result<usize, StateStoreError>>,
+ },
}
impl StateStore {
@@ -213,6 +221,18 @@ impl StateStore {
.await
}
+ /// Returns non-sensitive local queue information for an operator health check.
+ pub async fn health(&self, now_ms: u64) -> Result<HealthSnapshot, StateStoreError>
+ {
+ self.request(|response| StateCommand::Health { now_ms, response }).await
+ }
+
+ /// Releases active processing leases during a controlled shutdown.
+ pub async fn releaseProcessing(&self, now_ms: u64) -> Result<usize, StateStoreError>
+ {
+ self.request(|response| StateCommand::ReleaseProcessing { now_ms, response }).await
+ }
+
async fn request<T, F>(&self, make_command: F) -> Result<T, StateStoreError>
where
F: FnOnce(oneshot::Sender<Result<T, StateStoreError>>) -> StateCommand,
@@ -290,6 +310,10 @@ fn handle_command(connection: &Connection, command: StateCommand) {
StateCommand::RequeueUnfinished { now_ms, response } => {
send(response, requeue_unfinished(connection, now_ms));
}
+ StateCommand::Health { now_ms, response } => send(response, health(connection, now_ms)),
+ StateCommand::ReleaseProcessing { now_ms, response } => {
+ send(response, release_processing(connection, now_ms));
+ }
}
}
@@ -573,6 +597,40 @@ fn requeue_unfinished(connection: &Connection, now_ms: u64) -> Result<usize, Sta
Ok(changed)
}
+fn health(connection: &Connection, now_ms: u64) -> Result<HealthSnapshot, StateStoreError>
+{
+ let candidate_count = connection.query_row(
+ "SELECT COUNT(*) FROM invocations WHERE status = 'candidate'",
+ [],
+ |row| row.get(0),
+ )?;
+ let oldest_ready: Option<u64> = connection.query_row(
+ "SELECT MIN(updated_at_ms) FROM invocations WHERE status = 'ready_to_send'",
+ [],
+ |row| row.get(0),
+ )?;
+ let ready_to_send_count = connection.query_row(
+ "SELECT COUNT(*) FROM invocations WHERE status = 'ready_to_send'",
+ [],
+ |row| row.get(0),
+ )?;
+ Ok(HealthSnapshot {
+ candidate_count,
+ ready_to_send_count,
+ oldest_ready_age_ms: oldest_ready.map(|timestamp| now_ms.saturating_sub(timestamp)),
+ })
+}
+
+fn release_processing(connection: &Connection, now_ms: u64) -> Result<usize, StateStoreError>
+{
+ let changed = connection.execute(
+ "UPDATE invocations SET status = 'candidate', available_at_ms = ?1,
+ updated_at_ms = ?1 WHERE status = 'processing'",
+ [now_ms],
+ )?;
+ Ok(changed)
+}
+
fn read_invocation(
transaction: &Transaction<'_>,
event_id: &str,
@@ -719,4 +777,16 @@ mod tests {
.is_some()
);
}
+
+ #[tokio::test]
+ async fn health_exposes_counts_without_invocation_content() {
+ let directory = tempdir().unwrap();
+ let store = StateStore::open(directory.path().join("state.sqlite3")).unwrap();
+ store.ingestBatch("s1".to_owned(), vec![], vec![candidate("$one")], 20)
+ .await.unwrap();
+ let health = store.health(30).await.unwrap();
+ assert_eq!(health.candidate_count, 1);
+ assert_eq!(health.ready_to_send_count, 0);
+ assert_eq!(health.oldest_ready_age_ms, None);
+ }
}