From 38707ccd721bb22ab0c2f6ac08ec5a06fee91584 Mon Sep 17 00:00:00 2001 From: AnyBot Dev Date: Mon, 9 Nov 2026 18:00:56 +0800 Subject: [PATCH] wip: sqlite audit trail --- src/main.rs | 1 + src/oplog.rs | 10 ++- src/store.rs | 238 +++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 247 insertions(+), 2 deletions(-) create mode 100644 src/store.rs diff --git a/src/main.rs b/src/main.rs index e78ee57..f95e681 100644 --- a/src/main.rs +++ b/src/main.rs @@ -53,6 +53,7 @@ mod teams; mod theme; mod tools; mod triggers; +mod store; mod vault; mod vault_store; mod vnc; diff --git a/src/oplog.rs b/src/oplog.rs index fae6478..a6e828e 100644 --- a/src/oplog.rs +++ b/src/oplog.rs @@ -108,6 +108,7 @@ pub fn dir() -> PathBuf { #[derive(Clone, Debug)] pub struct OpLog { path: PathBuf, + bot: u64, } impl OpLog { @@ -116,12 +117,14 @@ impl OpLog { } fn in_dir(d: &std::path::Path, bot_id: u64) -> Self { - OpLog { path: d.join(format!("bot-{bot_id}.jsonl")) } + OpLog { path: d.join(format!("bot-{bot_id}.jsonl")), bot: bot_id } } #[cfg(test)] pub fn at(path: PathBuf) -> Self { - OpLog { path } + // The bot id rides in the file name (audit mirroring needs it; tests: 0 is fine). + let bot = path.file_stem().and_then(|s| s.to_str()).and_then(|s| s.strip_prefix("bot-")).and_then(|s| s.parse().ok()).unwrap_or(0); + OpLog { path, bot } } /// Add one line and wait until it is on disk. @@ -142,6 +145,9 @@ impl OpLog { if let Err(e) = written { eprintln!("AnyBot：寫不進工作紀錄 {}：{e}", self.path.display()); } + // The same line, mirrored into the queryable trail (SQLite). Best-effort: an + // audit that never breaks a turn. + crate::store::record(self.bot, r); } /// Every line this version can read. A line cut short (power lost while writing it) is the diff --git a/src/store.rs b/src/store.rs new file mode 100644 index 0000000..0d764cf --- /dev/null +++ b/src/store.rs @@ -0,0 +1,238 @@ +//! The SQLite layer: a queryable audit trail beside the JSON files. +//! +//! `state.json` stays the app's truth (one struct, written atomically; every layer of +//! the app reads it as a struct, not as rows). What JSON files cannot do is *ask +//! questions*: "what has this bot done in the last week?" The audit trail therefore +//! lives in SQLite (bundled, WAL, one file in the data folder) and keeps what the +//! per-turn oplog drops — turns finished days ago, answerable from the phone. The +//! memory/RAG layer lands here too. +//! +//! It is deliberately side-car: every write is best-effort, an audit that never +//! breaks a turn. If the database cannot be opened, the log of the turn in flight +//! (oplog) still works and the timeline page just shows less. + +use crate::oplog::{Outcome, Record}; +use rusqlite::{params, Connection}; +use std::path::PathBuf; +use std::sync::{Mutex, OnceLock}; + +/// How far back the timeline answers. Rows older than this are deleted on open. +const KEEP_DAYS: i64 = 30; + +pub struct Store { + conn: Mutex, +} + +static STORE: OnceLock> = OnceLock::new(); + +fn path() -> PathBuf { + crate::oplog::dir().join("audit.db") +} + +/// Open (create if needed) the data folder's audit database. Failure means "no +/// history", never "no bot". +fn open() -> Option { + let c = Connection::open(path()).ok()?; + c.pragma_update(None, "journal_mode", "WAL").ok()?; + c.pragma_update(None, "synchronous", "NORMAL").ok()?; + c.execute_batch( + "CREATE TABLE IF NOT EXISTS audit( + id INTEGER PRIMARY KEY, + bot INTEGER NOT NULL, + op TEXT NOT NULL, + at INTEGER NOT NULL, + end_at INTEGER, + kind TEXT NOT NULL, + name TEXT, + detail TEXT, + status TEXT, + call_id TEXT + ); + CREATE INDEX IF NOT EXISTS audit_bot_at ON audit(bot, at DESC); + CREATE INDEX IF NOT EXISTS audit_call ON audit(bot, op, call_id);", + ) + .ok()?; + let cutoff = crate::model::now_ms() - KEEP_DAYS * 86_400_000; + let _ = c.execute("DELETE FROM audit WHERE at < ?1", params![cutoff]); + Some(Store { conn: Mutex::new(c) }) +} + +fn store() -> Option<&'static Store> { + STORE.get_or_init(open).as_ref() +} + +/// One record of the oplog, mirrored into the trail. Best-effort by contract. +pub fn record(bot: u64, r: &Record) { + let bot = bot as i64; // (rusqlite speaks i64; ids stay far below 2^63.) + let Some(s) = store() else { return }; + let Ok(c) = s.conn.lock() else { return }; + let now = crate::model::now_ms(); + match r { + Record::OperationStarted { op, at, prompt, .. } => { + let _ = c.execute( + "INSERT INTO audit(bot,op,at,kind,detail) VALUES(?1,?2,?3,'turn',?4)", + params![bot, op, at, prompt.as_ref().map(|p| p.text.clone())], + ); + } + Record::StepAttempt { op, step, .. } => { + let _ = c.execute("INSERT INTO audit(bot,op,at,kind,name) VALUES(?1,?2,?3,'model',?4)", params![bot, op, now, step]); + } + Record::ToolStarted { op, call_id, name, args, at, .. } => { + let _ = c.execute( + "INSERT INTO audit(bot,op,at,kind,name,detail,status,call_id) VALUES(?1,?2,?3,'tool',?4,?5,'running',?6)", + params![bot, op, at, name, args, call_id], + ); + } + Record::ToolFinished { op, call_id, status, at } => { + let _ = c.execute( + "UPDATE audit SET status=?1, end_at=?2 WHERE id=(SELECT id FROM audit WHERE bot=?3 AND op=?4 AND call_id=?5 AND kind='tool' ORDER BY id DESC LIMIT 1)", + params![status, at, bot, op, call_id], + ); + } + Record::QueueEnqueued { op, id, text, at } => { + let _ = c.execute( + "INSERT INTO audit(bot,op,at,kind,detail,status,call_id) VALUES(?1,?2,?3,'queue',?4,'waiting',?5)", + params![bot, op.as_deref().unwrap_or(""), at, text, id], + ); + } + Record::QueueConsumed { id } | Record::QueueCancelled { id } => { + let state = if matches!(r, Record::QueueConsumed { .. }) { "consumed" } else { "cancelled" }; + let _ = c.execute( + "UPDATE audit SET status=?1 WHERE id=(SELECT id FROM audit WHERE bot=?2 AND call_id=?3 AND kind='queue' ORDER BY id DESC LIMIT 1)", + params![state, bot, id], + ); + } + Record::AbortRequested { op } => { + let _ = c.execute("INSERT INTO audit(bot,op,at,kind) VALUES(?1,?2,?3,'abort')", params![bot, op, now]); + } + Record::OperationFinished { op, outcome } => { + let name = match outcome { + Outcome::Completed => "completed", + Outcome::Aborted => "aborted", + Outcome::Failed => "failed", + }; + let _ = c.execute( + "UPDATE audit SET status=?1 WHERE id=(SELECT id FROM audit WHERE bot=?2 AND op=?3 AND kind='turn' ORDER BY id DESC LIMIT 1)", + params![name, bot, op], + ); + } + } +} + +/// The phone's 稽核時間軸: finished turns stay here, so the page answers for +/// KEEP_DAYS, not just for the turn running now. Newest turn first; the last 20 turns. +pub fn turns(bot: u64) -> serde_json::Value { + let Some(s) = store() else { return serde_json::json!({"turns": []}) }; + let Ok(c) = s.conn.lock() else { return serde_json::json!({"turns": []}) }; + let cutoff = crate::model::now_ms() - KEEP_DAYS * 86_400_000; + let Ok(mut st) = c.prepare( + "SELECT op,at,end_at,kind,name,detail,status,call_id FROM audit WHERE bot=?1 AND at>?2 ORDER BY id ASC", + ) else { + return serde_json::json!({"turns": []}) + }; + let rows = st + .query_map(params![bot, cutoff], |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, i64>(1)?, + r.get::<_, Option>(2)?, + r.get::<_, String>(3)?, + r.get::<_, Option>(4)?, + r.get::<_, Option>(5)?, + r.get::<_, Option>(6)?, + )) + }) + .map(|it| it.filter_map(Result::ok).collect::>()) + .unwrap_or_default(); + drop(st); + + #[derive(Default)] + struct Turn { + op: String, + at: i64, + prompt: String, + outcome: Option, + steps: Vec, + } + let mut turns: Vec = Vec::new(); + let pos = |turns: &mut Vec, op: &str| -> usize { + turns.iter().rposition(|t| t.op == op).unwrap_or_else(|| { + turns.push(Turn { op: op.to_string(), at: 0, prompt: String::new(), outcome: None, steps: Vec::new() }); + turns.len() - 1 + }) + }; + for (op, at, end_at, kind, name, detail, status) in rows { + let i = pos(&mut turns, &op); + match kind.as_str() { + "turn" => { + turns[i].at = at; + turns[i].prompt = detail.unwrap_or_default(); + turns[i].outcome = status; + } + "model" => { + // Consecutive attempts at the same step read as one line, counted. + match turns[i].steps.last_mut().filter(|s| s["type"] == "model" && s["step"] == *name.as_deref().unwrap_or("")) { + Some(last) => last["attempts"] = serde_json::json!(last["attempts"].as_u64().unwrap_or(1) + 1), + None => turns[i].steps.push(serde_json::json!({"type": "model", "step": name.unwrap_or_default(), "attempts": 1})), + } + } + "tool" => { + // The row carries its end status; args stay as they ran. + let at_tool = at; + match turns[i].steps.iter_mut().rev().find(|s| s["type"] == "tool" && s["name"] == *name.as_deref().unwrap_or("") && s["status"] == "running") { + Some(s) => { + s["status"] = serde_json::json!(status.as_deref().unwrap_or("running")); + s["end_at"] = serde_json::json!(end_at); + } + None => turns[i].steps.push(serde_json::json!({"type": "tool", "name": name.unwrap_or_default(), "args": detail.unwrap_or_default(), "status": status.unwrap_or_else(|| "running".into()), "at": at_tool})), + } + } + "queue" => turns[i].steps.push(serde_json::json!({"type": "queue", "text": detail.unwrap_or_default(), "at": at, "state": status.unwrap_or_else(|| "waiting".into())})), + "abort" => turns[i].steps.push(serde_json::json!({"type": "abort"})), + _ => {} + } + } + let keep = turns.len().min(20); + let out: Vec = turns[turns.len() - keep..] + .iter() + .rev() + .map(|t| serde_json::json!({"op": t.op, "at": t.at, "prompt": t.prompt, "outcome": t.outcome, "steps": t.steps})) + .collect(); + serde_json::json!({"turns": out}) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::oplog::{Prompt, Replay}; + + #[test] + fn the_trail_keeps_the_story_after_the_turn_log_cleans_up() { + // (Tests share the temp dir of this process; one database, one bot id.) + let bot = 4242; + record(bot, &Record::OperationStarted { op: "op1".into(), at: 100, source_leaf: 0, prompt: Some(Prompt { at: 99, text: "幫我叫車".into() }) }); + record(bot, &Record::StepAttempt { op: "op1".into(), step: "assistant".into(), attempt: 1 }); + record(bot, &Record::StepAttempt { op: "op1".into(), step: "assistant".into(), attempt: 2 }); + record(bot, &Record::ToolStarted { op: "op1".into(), call_id: "c1".into(), name: "computer".into(), args: "{\"action\":\"click\"}".into(), replay: Replay::Never, at: 110 }); + record(bot, &Record::ToolFinished { op: "op1".into(), call_id: "c1".into(), status: "done".into(), at: 120 }); + record(bot, &Record::QueueEnqueued { op: Some("op1".into()), id: "q1".into(), text: "快一點".into(), at: 111 }); + record(bot, &Record::QueueConsumed { id: "q1".into() }); + record(bot, &Record::OperationFinished { op: "op1".into(), outcome: Outcome::Completed }); + record(bot, &Record::OperationStarted { op: "op2".into(), at: 200, source_leaf: 100, prompt: None }); + record(bot, &Record::ToolStarted { op: "op2".into(), call_id: "c9".into(), name: "run_cmd".into(), args: "{}".into(), replay: Replay::Safe, at: 210 }); + + let v = turns(bot); + let ts = v["turns"].as_array().unwrap(); + assert_eq!(ts.len(), 2, "both turns are in the trail"); + // (Newest first: the phone puts the running turn at the top.) + assert_eq!(ts[0]["op"], "op2"); + assert_eq!(ts[0]["outcome"], serde_json::Value::Null); + assert_eq!(ts[0]["steps"][0]["status"], "running"); + assert_eq!(ts[1]["prompt"], "幫我叫車"); + assert_eq!(ts[1]["outcome"], "completed"); + assert_eq!(ts[1]["steps"][0], serde_json::json!({"type": "model", "step": "assistant", "attempts": 2})); + assert_eq!(ts[1]["steps"][1]["name"], "computer"); + assert_eq!(ts[1]["steps"][1]["status"], "done", "the finished tool answers with its end status"); + assert_eq!(ts[1]["steps"][2]["state"], "consumed"); + } +}