wip: sqlite audit trail
This commit is contained in:
AnyBot Dev
daniel wang
parent
d490b9d94d
commit
38707ccd72
|
|
@ -53,6 +53,7 @@ mod teams;
|
|||
mod theme;
|
||||
mod tools;
|
||||
mod triggers;
|
||||
mod store;
|
||||
mod vault;
|
||||
mod vault_store;
|
||||
mod vnc;
|
||||
|
|
|
|||
10
src/oplog.rs
10
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
|
||||
|
|
|
|||
|
|
@ -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<Connection>,
|
||||
}
|
||||
|
||||
static STORE: OnceLock<Option<Store>> = 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<Store> {
|
||||
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<i64>>(2)?,
|
||||
r.get::<_, String>(3)?,
|
||||
r.get::<_, Option<String>>(4)?,
|
||||
r.get::<_, Option<String>>(5)?,
|
||||
r.get::<_, Option<String>>(6)?,
|
||||
))
|
||||
})
|
||||
.map(|it| it.filter_map(Result::ok).collect::<Vec<_>>())
|
||||
.unwrap_or_default();
|
||||
drop(st);
|
||||
|
||||
#[derive(Default)]
|
||||
struct Turn {
|
||||
op: String,
|
||||
at: i64,
|
||||
prompt: String,
|
||||
outcome: Option<String>,
|
||||
steps: Vec<serde_json::Value>,
|
||||
}
|
||||
let mut turns: Vec<Turn> = Vec::new();
|
||||
let pos = |turns: &mut Vec<Turn>, 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<serde_json::Value> = 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");
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue