Merge p4/sqlite: SQLite audit trail, queryable history (P4 step 1)

mobile/index.html

@@ -1318,7 +1318,7 @@ function tabBody(b, d) {
   if (S.tab === "audit") {
     const a = S.audit[b.id];
     if (a === undefined) { loadAudit(b.id); return `<div class="note" style="padding:14px 16px">載入中…</div>`; }
-    if (!a.turns.length) return `<div class="empty">現在沒有進行中的工作。稽核時間軸記的是**這回合**做了什麼：每次問模型、每個工具的參數與結果、你插入的話、停止。回合結束後只留最近的，聊天記錄本身才是長久的。</div>`;
+    if (!a.turns.length) return `<div class="empty">現在沒有進行中的工作。稽核時間軸記的是**這回合**做了什麼：每次問模型、每個工具的參數與結果、你插入的話、停止。這裡記的是最近 30 天的每一回合；更早的故事在聊天記錄裡。</div>`;
     const OUT = {completed: "已結束", aborted: "被你停止", failed: "失敗"};
     const ST = {done: "完成", running: "進行中…", waiting: "等核准", denied: "被拒絕", failed: "失敗"};
     const TI = {computer: "zap", menu_click: "zap", fill_form: "zap", wait_for: "clock", run_cmd: "zap", web_fetch: "globe", browser: "globe"};

src/main.rs

@@ -53,6 +53,7 @@ mod teams;
 mod theme;
 mod tools;
 mod triggers;
+mod store;
 mod vault;
 mod vault_store;
 mod vnc;

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

src/store.rs

@@ -0,0 +1,246 @@
+//! 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 p = path();
+    if let Some(d) = p.parent() {
+        let _ = std::fs::create_dir_all(d); // (the data folder may not exist yet on first run)
+    }
+    let c = Connection::open(p).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 bot = bot as i64;
+    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.
+        // Times ride on "now" — the trail prunes anything older than a month.)
+        let bot = 4242;
+        let t0 = crate::model::now_ms() - 60_000;
+        let at = |d| t0 + d;
+        record(bot, &Record::OperationStarted { op: "op1".into(), at: at(0), source_leaf: 0, prompt: Some(Prompt { at: at(0), 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: at(10) });
+        record(bot, &Record::ToolFinished { op: "op1".into(), call_id: "c1".into(), status: "done".into(), at: at(20) });
+        record(bot, &Record::QueueEnqueued { op: Some("op1".into()), id: "q1".into(), text: "快一點".into(), at: at(11) });
+        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: at(100), 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: at(110) });
+
+        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");
+    }
+}