807 lines
38 KiB
Rust
807 lines
38 KiB
Rust
// Each bot's work log, after pi's durable-harness design (harness-v2, Part II), cut down: before
|
||
// something happens a line says it is about to (a turn starting, a model call, a step about to
|
||
// run, a message typed mid-turn), and after, a line says how it ended. Every line is on disk
|
||
// (synced) before the thing it announces starts.
|
||
//
|
||
// Nothing reads it while AnyBot runs. It is for the next start after AnyBot was killed mid-turn:
|
||
// which turn was cut off, which steps were running then (and whether one may have done its thing
|
||
// anyway — a text sent, a site published), and which messages the person typed that never
|
||
// reached the bot. The chat in state.json is saved every couple of seconds; this is the part
|
||
// that can't wait.
|
||
//
|
||
// One file per bot: `<data folder>/oplog/bot-<id>.jsonl`. When a turn ends, the file shrinks to
|
||
// the typed messages still waiting (usually nothing).
|
||
|
||
use serde::{Deserialize, Serialize};
|
||
use std::io::Write;
|
||
use std::path::PathBuf;
|
||
use std::sync::Mutex;
|
||
|
||
use crate::model::{Message, Role, State, ToolStatus};
|
||
|
||
/// May this step run again after AnyBot was killed while it ran? Looking can; anything that
|
||
/// reaches outside (a text, a click, a file written, a site published) never runs twice on
|
||
/// its own.
|
||
#[derive(Clone, Copy, Serialize, Deserialize, Debug, PartialEq, Eq)]
|
||
#[serde(rename_all = "snake_case")]
|
||
pub enum Replay {
|
||
Safe,
|
||
Never,
|
||
}
|
||
|
||
#[derive(Clone, Copy, Serialize, Deserialize, Debug, PartialEq, Eq)]
|
||
#[serde(rename_all = "snake_case")]
|
||
pub enum Outcome {
|
||
Completed,
|
||
Aborted,
|
||
Failed,
|
||
}
|
||
|
||
/// The person's message a turn answers (kept so a message typed just before a crash, before the
|
||
/// chat was saved, isn't lost).
|
||
#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
|
||
pub struct Prompt {
|
||
pub at: i64,
|
||
pub text: String,
|
||
}
|
||
|
||
#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
|
||
#[serde(tag = "type", rename_all = "snake_case")]
|
||
pub enum Record {
|
||
/// A turn starts. `source_leaf`: the time of the newest message it starts from.
|
||
OperationStarted {
|
||
op: String,
|
||
at: i64,
|
||
source_leaf: i64,
|
||
#[serde(default)]
|
||
prompt: Option<Prompt>,
|
||
},
|
||
/// A model call is about to be made (`step`: assistant | compaction), the n-th for this step.
|
||
StepAttempt { op: String, step: String, attempt: u32 },
|
||
/// A step passed its checks and is about to run, with the arguments it runs with.
|
||
ToolStarted {
|
||
op: String,
|
||
call_id: String,
|
||
name: String,
|
||
args: String,
|
||
replay: Replay,
|
||
/// Wall time, so the audit timeline can show when each step ran (old logs: 0).
|
||
#[serde(default)]
|
||
at: i64,
|
||
},
|
||
ToolFinished {
|
||
op: String,
|
||
call_id: String,
|
||
status: String,
|
||
#[serde(default)]
|
||
at: i64,
|
||
},
|
||
/// The person typed this while the bot worked; it waits for the bot's next step.
|
||
QueueEnqueued {
|
||
#[serde(default)]
|
||
op: Option<String>,
|
||
id: String,
|
||
text: String,
|
||
at: i64,
|
||
},
|
||
/// …and reached the bot (or became the next turn's message).
|
||
QueueConsumed { id: String },
|
||
/// …or was thrown away (the person stopped the turn).
|
||
QueueCancelled { id: String },
|
||
/// The person pressed stop.
|
||
AbortRequested { op: String },
|
||
OperationFinished { op: String, outcome: Outcome },
|
||
}
|
||
|
||
/// One writer at a time, in this process (the window and every turn's thread write here).
|
||
static WRITE: Mutex<()> = Mutex::new(());
|
||
|
||
pub fn dir() -> PathBuf {
|
||
// Tests never touch the person's data folder.
|
||
if cfg!(test) {
|
||
return std::env::temp_dir().join(format!("anybot-oplog-test-{}", std::process::id()));
|
||
}
|
||
State::path().parent().map(|p| p.join("oplog")).unwrap_or_else(|| std::env::temp_dir().join("anybot-oplog"))
|
||
}
|
||
|
||
/// One bot's log.
|
||
#[derive(Clone, Debug)]
|
||
pub struct OpLog {
|
||
path: PathBuf,
|
||
}
|
||
|
||
impl OpLog {
|
||
pub fn for_bot(bot_id: u64) -> Self {
|
||
Self::in_dir(&dir(), bot_id)
|
||
}
|
||
|
||
fn in_dir(d: &std::path::Path, bot_id: u64) -> Self {
|
||
OpLog { path: d.join(format!("bot-{bot_id}.jsonl")) }
|
||
}
|
||
|
||
#[cfg(test)]
|
||
pub fn at(path: PathBuf) -> Self {
|
||
OpLog { path }
|
||
}
|
||
|
||
/// Add one line and wait until it is on disk.
|
||
pub fn append(&self, r: &Record) {
|
||
let _g = WRITE.lock().unwrap_or_else(|e| e.into_inner());
|
||
self.append_locked(r);
|
||
}
|
||
|
||
fn append_locked(&self, r: &Record) {
|
||
let Ok(line) = serde_json::to_string(r) else { return };
|
||
if let Some(d) = self.path.parent() {
|
||
let _ = std::fs::create_dir_all(d);
|
||
}
|
||
let written = std::fs::OpenOptions::new().create(true).append(true).open(&self.path).and_then(|mut f| {
|
||
f.write_all(format!("{line}\n").as_bytes())?;
|
||
f.sync_data()
|
||
});
|
||
if let Err(e) = written {
|
||
eprintln!("AnyBot：寫不進工作紀錄 {}：{e}", self.path.display());
|
||
}
|
||
}
|
||
|
||
/// Every line this version can read. A line cut short (power lost while writing it) is the
|
||
/// last one and is skipped.
|
||
pub fn read(&self) -> Vec<Record> {
|
||
std::fs::read_to_string(&self.path).map(|s| s.lines().filter_map(|l| serde_json::from_str(l).ok()).collect()).unwrap_or_default()
|
||
}
|
||
|
||
/// A person-readable timeline: turns (newest last), each with its prompt, outcome and
|
||
/// steps — model attempts merged, tools with the arguments they ran with and their status,
|
||
/// messages typed while the bot worked. This is what the phone's 稽核時間軸 shows; the
|
||
/// log itself stays the machine format the recovery code reads.
|
||
pub fn timeline(recs: &[Record]) -> serde_json::Value {
|
||
#[derive(Default)]
|
||
struct Turn {
|
||
op: String,
|
||
at: i64,
|
||
prompt: String,
|
||
outcome: Option<&'static str>,
|
||
steps: Vec<serde_json::Value>,
|
||
}
|
||
// (A plain fn, not a closure: a closure returning a borrow of its argument can't tie
|
||
// the two lifetimes together.)
|
||
fn find<'a>(op: &str, turns: &'a mut [Turn]) -> Option<&'a mut Turn> {
|
||
turns.iter_mut().rev().find(|t| t.op == op)
|
||
}
|
||
let mut turns: Vec<Turn> = Vec::new();
|
||
for r in recs {
|
||
match r {
|
||
Record::OperationStarted { op, at, prompt, .. } => turns.push(Turn {
|
||
op: op.clone(),
|
||
at: *at,
|
||
prompt: prompt.as_ref().map(|p| p.text.clone()).unwrap_or_default(),
|
||
..Default::default()
|
||
}),
|
||
Record::StepAttempt { op, step, .. } => {
|
||
if let Some(t) = find(op, &mut turns) {
|
||
// Retry the same step? One line, with the attempts counted.
|
||
match t.steps.last_mut().filter(|s| s["type"] == "model" && s["step"] == *step) {
|
||
Some(last) => {
|
||
let n = last["attempts"].as_u64().unwrap_or(1) + 1;
|
||
last["attempts"] = serde_json::json!(n);
|
||
}
|
||
None => t.steps.push(serde_json::json!({"type": "model", "step": step, "attempts": 1})),
|
||
}
|
||
}
|
||
}
|
||
Record::ToolStarted { op, call_id, name, args, replay, at } => {
|
||
if let Some(t) = find(op, &mut turns) {
|
||
let short: String = args.chars().take(240).collect();
|
||
t.steps.push(serde_json::json!({"type": "tool", "call_id": call_id, "name": name, "args": short, "replay": replay, "status": "running", "at": at}));
|
||
}
|
||
}
|
||
Record::ToolFinished { op, call_id, status, at } => {
|
||
if let Some(t) = find(op, &mut turns) {
|
||
if let Some(s) = t.steps.iter_mut().rev().find(|s| s["type"] == "tool" && s["call_id"] == *call_id) {
|
||
s["status"] = serde_json::json!(status);
|
||
s["end_at"] = serde_json::json!(at);
|
||
}
|
||
}
|
||
}
|
||
Record::QueueEnqueued { op, id, text, at } => {
|
||
// The turn it was typed into; a log without that turn (cut short) lands
|
||
// in the newest one — the phone only needs to show it somewhere in order.
|
||
let i = op.as_ref().and_then(|o| turns.iter().rposition(|t| t.op == *o)).or_else(|| turns.len().checked_sub(1));
|
||
if let Some(i) = i {
|
||
turns[i].steps.push(serde_json::json!({"type": "queue", "id": id, "text": text, "at": at, "state": "waiting"}));
|
||
}
|
||
}
|
||
Record::QueueConsumed { id } | Record::QueueCancelled { id } => {
|
||
let state = if matches!(r, Record::QueueConsumed { .. }) { "consumed" } else { "cancelled" };
|
||
for t in turns.iter_mut().rev() {
|
||
if let Some(s) = t.steps.iter_mut().rev().find(|s| s["id"] == *id) {
|
||
s["state"] = serde_json::json!(state);
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
Record::AbortRequested { op } => {
|
||
if let Some(t) = find(op, &mut turns) {
|
||
t.steps.push(serde_json::json!({"type": "abort"}));
|
||
}
|
||
}
|
||
Record::OperationFinished { op, outcome } => {
|
||
if let Some(t) = find(op, &mut turns) {
|
||
t.outcome = Some(match outcome {
|
||
Outcome::Completed => "completed",
|
||
Outcome::Aborted => "aborted",
|
||
Outcome::Failed => "failed",
|
||
});
|
||
}
|
||
}
|
||
}
|
||
}
|
||
let steps = |t: &Turn| -> Vec<serde_json::Value> { t.steps.iter().map(|s| { let mut v = s.clone(); v.as_object_mut().map(|o| o.remove("call_id")); v }).collect() };
|
||
let keep = turns.len().min(20);
|
||
let turns: Vec<serde_json::Value> = turns[turns.len() - keep..]
|
||
.iter()
|
||
.map(|t| {
|
||
serde_json::json!({"op": t.op, "at": t.at, "prompt": t.prompt, "outcome": t.outcome, "steps": steps(t)})
|
||
})
|
||
.collect();
|
||
serde_json::json!({"turns": turns})
|
||
}
|
||
|
||
/// A turn ended: its last line, then the file keeps only the typed messages still waiting.
|
||
pub fn finish(&self, op: &str, outcome: Outcome) {
|
||
let _g = WRITE.lock().unwrap_or_else(|e| e.into_inner());
|
||
self.append_locked(&Record::OperationFinished { op: op.to_string(), outcome });
|
||
self.shrink_locked();
|
||
}
|
||
|
||
/// Lines saying these typed messages reached a chat, then shrink (used after a restart).
|
||
fn consumed_and_shrink(&self, ids: &[String]) {
|
||
let _g = WRITE.lock().unwrap_or_else(|e| e.into_inner());
|
||
for id in ids {
|
||
self.append_locked(&Record::QueueConsumed { id: id.clone() });
|
||
}
|
||
self.shrink_locked();
|
||
}
|
||
|
||
/// With no turn open, everything but the messages still waiting is history: drop it.
|
||
fn shrink_locked(&self) {
|
||
let r = reduce(&self.read());
|
||
if r.open.is_some() {
|
||
return;
|
||
}
|
||
if r.pending.is_empty() {
|
||
let _ = std::fs::remove_file(&self.path);
|
||
return;
|
||
}
|
||
let body: String = r
|
||
.pending
|
||
.iter()
|
||
.filter_map(|p| serde_json::to_string(&Record::QueueEnqueued { op: p.op.clone(), id: p.id.clone(), text: p.text.clone(), at: p.at }).ok())
|
||
.map(|l| format!("{l}\n"))
|
||
.collect();
|
||
let tmp = self.path.with_extension("jsonl.tmp");
|
||
let ok = std::fs::File::create(&tmp).and_then(|mut f| {
|
||
f.write_all(body.as_bytes())?;
|
||
f.sync_all()
|
||
});
|
||
if ok.is_ok() {
|
||
let _ = std::fs::rename(&tmp, &self.path);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// A step the open turn started.
|
||
#[derive(Clone, Debug, PartialEq)]
|
||
pub struct Started {
|
||
pub call_id: String,
|
||
pub name: String,
|
||
pub args: String,
|
||
pub replay: Replay,
|
||
/// Its end was written down (its result may still not be in the saved chat).
|
||
pub finished: bool,
|
||
}
|
||
|
||
/// The turn that was open when AnyBot stopped.
|
||
#[derive(Clone, Debug, PartialEq)]
|
||
pub struct OpenOp {
|
||
pub op: String,
|
||
pub prompt: Option<Prompt>,
|
||
pub aborted: bool,
|
||
pub started: Vec<Started>,
|
||
}
|
||
|
||
/// A typed message that never reached the bot.
|
||
#[derive(Clone, Debug, PartialEq)]
|
||
pub struct Pending {
|
||
pub id: String,
|
||
pub text: String,
|
||
pub at: i64,
|
||
pub op: Option<String>,
|
||
}
|
||
|
||
/// What a log says, read from start to end.
|
||
#[derive(Clone, Debug, Default, PartialEq)]
|
||
pub struct Reduced {
|
||
pub open: Option<OpenOp>,
|
||
pub pending: Vec<Pending>,
|
||
}
|
||
|
||
/// Fold the lines into where things stand. A turn started while another was still open means
|
||
/// the earlier one died without its last line; the newer one is what counts.
|
||
pub fn reduce(recs: &[Record]) -> Reduced {
|
||
let mut open: Option<OpenOp> = None;
|
||
let mut pending: Vec<Pending> = vec![];
|
||
for r in recs {
|
||
let mine = |o: &Option<OpenOp>, op: &str| o.as_ref().map(|o| o.op == op).unwrap_or(false);
|
||
match r {
|
||
Record::OperationStarted { op, prompt, .. } => open = Some(OpenOp { op: op.clone(), prompt: prompt.clone(), aborted: false, started: vec![] }),
|
||
Record::ToolStarted { op, call_id, name, args, replay, .. } if mine(&open, op) => {
|
||
if let Some(o) = open.as_mut() {
|
||
o.started.retain(|s| &s.call_id != call_id);
|
||
o.started.push(Started { call_id: call_id.clone(), name: name.clone(), args: args.clone(), replay: *replay, finished: false });
|
||
}
|
||
}
|
||
Record::ToolFinished { op, call_id, .. } if mine(&open, op) => {
|
||
if let Some(o) = open.as_mut() {
|
||
for s in o.started.iter_mut().filter(|s| &s.call_id == call_id) {
|
||
s.finished = true;
|
||
}
|
||
}
|
||
}
|
||
Record::AbortRequested { op } if mine(&open, op) => {
|
||
if let Some(o) = open.as_mut() {
|
||
o.aborted = true;
|
||
}
|
||
}
|
||
Record::OperationFinished { op, .. } if mine(&open, op) => open = None,
|
||
Record::QueueEnqueued { op, id, text, at } => {
|
||
if !pending.iter().any(|p| &p.id == id) {
|
||
pending.push(Pending { id: id.clone(), text: text.clone(), at: *at, op: op.clone() });
|
||
}
|
||
}
|
||
Record::QueueConsumed { id } | Record::QueueCancelled { id } => pending.retain(|p| &p.id != id),
|
||
_ => {}
|
||
}
|
||
}
|
||
Reduced { open, pending }
|
||
}
|
||
|
||
/// Steps that only look: running one again changes nothing. Everything else (and every tool not
|
||
/// named here: MCP connectors, new tools) is `Never`.
|
||
pub fn replay_of(name: &str) -> Replay {
|
||
const LOOKS: &[&str] = &[
|
||
"computer_use_screenshot",
|
||
"computer_use_zoom",
|
||
"computer_use_show",
|
||
"computer_use_check_permissions",
|
||
"browser_read",
|
||
"browser_screenshot",
|
||
"browser_profiles",
|
||
"browser_wait",
|
||
"browser_tabs",
|
||
"web_fetch",
|
||
"read_file",
|
||
"list_dir",
|
||
"search",
|
||
"read_terminal",
|
||
"vault_list",
|
||
"view_demo_step",
|
||
"list_bots",
|
||
"list_computers",
|
||
"list_routines",
|
||
"list_pages",
|
||
"read_page",
|
||
"mac_read_mail",
|
||
"mac_read_notes",
|
||
"mac_mail_accounts",
|
||
"ios_read_reminders",
|
||
"mobile_read_calendar",
|
||
"mobile_read_contacts",
|
||
];
|
||
if LOOKS.contains(&name) {
|
||
Replay::Safe
|
||
} else {
|
||
Replay::Never
|
||
}
|
||
}
|
||
|
||
/// What a tool result says when AnyBot was killed while a step that reaches outside ran.
|
||
pub const MAYBE_RAN: &str = "（AnyBot 在執行這個步驟時被關掉了，它可能已經執行了。重做之前先確認結果（例如看畫面、讀檔案或問使用者），不要重複做一次。）";
|
||
/// …and when the step only looked.
|
||
pub const CUT_OFF: &str = "（這個步驟被中止了，沒有執行完。）";
|
||
|
||
/// A turn cut off by AnyBot stopping, as found on the next start.
|
||
#[derive(Clone, Debug)]
|
||
pub struct Interrupted {
|
||
pub bot_id: u64,
|
||
pub op: String,
|
||
/// The person had pressed stop.
|
||
pub aborted: bool,
|
||
/// Steps that may have done their thing (how the chat names them).
|
||
pub maybe_ran: Vec<String>,
|
||
/// Typed messages put back into the chat.
|
||
pub restored: usize,
|
||
}
|
||
|
||
/// What one bot's log gets once the chats are saved: the cut-off turn's last line, and the typed
|
||
/// messages now in the chat.
|
||
struct Closing {
|
||
log: OpLog,
|
||
finish: Option<(String, Outcome)>,
|
||
consumed: Vec<String>,
|
||
}
|
||
|
||
/// What `recover` found; `close` writes it down once the chat is saved.
|
||
pub struct Recovery {
|
||
pub interrupted: Vec<Interrupted>,
|
||
/// Did the chats change (save them before `close`).
|
||
pub changed: bool,
|
||
closing: Vec<Closing>,
|
||
}
|
||
|
||
impl Recovery {
|
||
/// After the chats are saved: each cut-off turn gets its last line, and the typed messages
|
||
/// put back are marked as delivered. In this order a crash in between only repeats the
|
||
/// recovery (which finds everything already in the chat and changes nothing).
|
||
pub fn close(self) -> Vec<Interrupted> {
|
||
for c in &self.closing {
|
||
if let Some((op, outcome)) = &c.finish {
|
||
c.log.finish(op, *outcome);
|
||
}
|
||
c.log.consumed_and_shrink(&c.consumed);
|
||
}
|
||
self.interrupted
|
||
}
|
||
}
|
||
|
||
/// A new message at its time in the chat (after everything older).
|
||
fn insert_by_time(msgs: &mut Vec<Message>, m: Message) {
|
||
let i = msgs.iter().rposition(|x| x.at <= m.at).map(|i| i + 1).unwrap_or(0);
|
||
msgs.insert(i, m);
|
||
}
|
||
|
||
/// On start, before anything runs: read every bot's log and put the chats right.
|
||
/// - The message a cut-off turn answered is put back if the chat lost it.
|
||
/// - Steps that were running: one that only looked shows as cut off; one that reaches outside
|
||
/// shows as "may have run" (a tool result the bot reads, and a line the person sees) — never
|
||
/// as "not allowed", which is what a stop looked like before.
|
||
/// - Messages typed mid-turn that never reached the bot go back into the chat, so the next turn
|
||
/// reads them.
|
||
///
|
||
/// - A turn that wasn't stopped by the person ends with a 「繼續」 card: what it was doing, the
|
||
/// steps that may have run, the messages put back. Nothing starts again until the person
|
||
/// presses it (App::resume_turn); the steps that may have run stay closed as such.
|
||
pub fn recover(state: &mut State) -> Recovery {
|
||
recover_in(&dir(), state)
|
||
}
|
||
|
||
fn recover_in(d: &std::path::Path, state: &mut State) -> Recovery {
|
||
let mut out = Recovery { interrupted: vec![], changed: false, closing: vec![] };
|
||
let ids: Vec<u64> = state.bots.iter().map(|b| b.id).collect();
|
||
// Logs of bots that no longer exist: nothing to put back.
|
||
if let Ok(rd) = std::fs::read_dir(d) {
|
||
for e in rd.flatten() {
|
||
let name = e.file_name().to_string_lossy().to_string();
|
||
let id = name.strip_prefix("bot-").and_then(|r| r.strip_suffix(".jsonl")).and_then(|n| n.parse::<u64>().ok());
|
||
if id.map(|id| !ids.contains(&id)).unwrap_or(false) {
|
||
let _ = std::fs::remove_file(e.path());
|
||
}
|
||
}
|
||
}
|
||
for id in ids {
|
||
let log = OpLog::in_dir(d, id);
|
||
let r = reduce(&log.read());
|
||
if r.open.is_none() && r.pending.is_empty() {
|
||
continue;
|
||
}
|
||
let Some(b) = state.bot_mut(id) else { continue };
|
||
let mut finish = None;
|
||
let mut maybe_ran = vec![];
|
||
if let Some(o) = &r.open {
|
||
if let Some(p) = &o.prompt {
|
||
if !b.messages.iter().any(|m| m.role == Role::User && m.at == p.at) {
|
||
insert_by_time(&mut b.messages, Message { at: p.at, ..Message::user(&p.text) });
|
||
out.changed = true;
|
||
}
|
||
}
|
||
// Every step that started but whose result never reached the saved chat (a result in
|
||
// the chat is the step's whole outcome; without one, it was cut off — or it finished
|
||
// and the chat wasn't saved after).
|
||
let result_of = |msgs: &[Message], id: &str| msgs.iter().find(|m| m.role == Role::Tool && m.tool_call_id == id).map(|m| m.text.clone());
|
||
// (A result already saying "may have run" is one an earlier pass of this put there.)
|
||
for s in o.started.iter().filter(|s| s.replay == Replay::Never) {
|
||
if result_of(&b.messages, &s.call_id).map(|t| t == MAYBE_RAN).unwrap_or(true) {
|
||
maybe_ran.push(format!("{} {}", crate::remote::tool_label(&s.name), crate::remote::tool_detail(&s.name, &s.args)).trim().to_string());
|
||
}
|
||
}
|
||
let missing: Vec<Started> = o.started.iter().filter(|s| result_of(&b.messages, &s.call_id).is_none()).cloned().collect();
|
||
for s in &missing {
|
||
let Some(mi) = b.messages.iter().rposition(|m| m.role == Role::Assistant && m.tool_calls.iter().any(|c| c.id == s.call_id)) else { continue };
|
||
for c in b.messages[mi].tool_calls.iter_mut().filter(|c| c.id == s.call_id) {
|
||
if c.status != ToolStatus::Failed {
|
||
c.status = ToolStatus::Failed;
|
||
out.changed = true;
|
||
}
|
||
}
|
||
if !b.messages.iter().any(|m| m.role == Role::Tool && m.tool_call_id == s.call_id) {
|
||
let mut at = mi + 1;
|
||
while at < b.messages.len() && b.messages[at].role == Role::Tool {
|
||
at += 1;
|
||
}
|
||
let text = if s.replay == Replay::Never { MAYBE_RAN } else { CUT_OFF };
|
||
b.messages.insert(at, Message { at: b.messages[mi].at, ..Message::tool_result(&s.call_id, &s.name, text) });
|
||
out.changed = true;
|
||
}
|
||
}
|
||
finish = Some((o.op.clone(), if o.aborted { Outcome::Aborted } else { Outcome::Failed }));
|
||
}
|
||
// Typed messages that never reached the bot go back where they were typed: the next
|
||
// turn (「繼續」, or whatever the person says next) reads them.
|
||
let mut consumed = vec![];
|
||
for p in &r.pending {
|
||
if !b.messages.iter().any(|m| m.role == Role::User && m.at == p.at && m.text == p.text) {
|
||
insert_by_time(&mut b.messages, Message { at: p.at, ..Message::user(&p.text) });
|
||
out.changed = true;
|
||
}
|
||
consumed.push(p.id.clone());
|
||
}
|
||
if let Some(o) = &r.open {
|
||
// A turn the person stopped stays stopped; any other gets the 「繼續」 card, in place
|
||
// of the "interrupted, press 重新產生" line the load left on an empty reply.
|
||
if !o.aborted {
|
||
let from = o.prompt.as_ref().and_then(|p| b.messages.iter().position(|m| m.role == Role::User && m.at == p.at)).unwrap_or(0);
|
||
let before = b.messages.len();
|
||
let mut i = 0;
|
||
b.messages.retain(|m| {
|
||
i += 1;
|
||
!(i - 1 > from && m.role == Role::Assistant && m.tool_calls.is_empty() && m.text == crate::model::INTERRUPTED_NOTE)
|
||
});
|
||
out.changed |= b.messages.len() != before;
|
||
let card = card_text(o.prompt.as_ref(), &maybe_ran, r.pending.len());
|
||
if !b.messages.iter().any(|m| m.interrupted && m.text == card) {
|
||
b.messages.push(Message { error: true, interrupted: true, ..Message::assistant(&card, "") });
|
||
out.changed = true;
|
||
}
|
||
}
|
||
out.interrupted.push(Interrupted { bot_id: id, op: o.op.clone(), aborted: o.aborted, maybe_ran, restored: r.pending.len() });
|
||
}
|
||
out.closing.push(Closing { log, finish, consumed });
|
||
}
|
||
out
|
||
}
|
||
|
||
/// What the 「繼續」 card says: what was being done, which steps may have run anyway (they are
|
||
/// closed as such and never run again by themselves), and the typed messages put back.
|
||
fn card_text(prompt: Option<&Prompt>, maybe_ran: &[String], restored: usize) -> String {
|
||
let mut s = String::from("AnyBot 在工作時被關掉了，這個回合沒有做完。");
|
||
if let Some(p) = prompt {
|
||
let line: String = p.text.lines().next().unwrap_or("").chars().take(40).collect();
|
||
let more = if p.text.chars().count() > line.chars().count() { "…" } else { "" };
|
||
s.push_str(&format!("\n正在做：「{line}{more}」"));
|
||
}
|
||
if !maybe_ran.is_empty() {
|
||
s.push_str("\n那時正在執行的這些步驟可能已經完成了（繼續時不會再做一次）：\n");
|
||
s.push_str(&maybe_ran.iter().map(|d| format!("・{d}")).collect::<Vec<_>>().join("\n"));
|
||
}
|
||
if restored > 0 {
|
||
s.push_str(&format!("\n它工作時你傳的 {restored} 則訊息已經放回聊天，繼續時它會讀到。"));
|
||
}
|
||
s.push_str("\n按「繼續」從中斷的地方接著做。");
|
||
s
|
||
}
|
||
|
||
/// An id for a typed message (unique in this process and across restarts).
|
||
pub fn new_id() -> String {
|
||
static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
|
||
format!("q-{}-{}", crate::model::now_ms(), N.fetch_add(1, std::sync::atomic::Ordering::Relaxed))
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
fn scratch(name: &str) -> OpLog {
|
||
let p = std::env::temp_dir().join(format!("anybot-oplog-{name}-{}.jsonl", std::process::id()));
|
||
let _ = std::fs::remove_file(&p);
|
||
OpLog::at(p)
|
||
}
|
||
|
||
#[test]
|
||
fn the_timeline_reads_like_a_story() {
|
||
use serde_json::json;
|
||
let recs = vec![
|
||
Record::OperationStarted { op: "t1".into(), at: 100, source_leaf: 0, prompt: Some(Prompt { at: 99, text: "幫我叫車".into() }) },
|
||
Record::StepAttempt { op: "t1".into(), step: "assistant".into(), attempt: 1 },
|
||
Record::StepAttempt { op: "t1".into(), step: "assistant".into(), attempt: 2 },
|
||
Record::ToolStarted { op: "t1".into(), call_id: "c1".into(), name: "computer".into(), args: "{\"action\":\"click\"}".into(), replay: Replay::Never, at: 110 },
|
||
Record::QueueEnqueued { op: Some("t1".into()), id: "q1".into(), text: "快一點".into(), at: 111 },
|
||
Record::ToolFinished { op: "t1".into(), call_id: "c1".into(), status: "done".into(), at: 120 },
|
||
Record::QueueConsumed { id: "q1".into() },
|
||
Record::OperationFinished { op: "t1".into(), outcome: Outcome::Completed },
|
||
// A second turn, running: no finish; its tool has no end.
|
||
Record::OperationStarted { op: "t2".into(), at: 200, source_leaf: 100, prompt: None },
|
||
Record::ToolStarted { op: "t2".into(), call_id: "c9".into(), name: "run_cmd".into(), args: "{}".into(), replay: Replay::Safe, at: 210 },
|
||
];
|
||
let v = OpLog::timeline(&recs);
|
||
assert_eq!(v["turns"].as_array().unwrap().len(), 2);
|
||
let t1 = &v["turns"][0];
|
||
assert_eq!(t1["prompt"], "幫我叫車");
|
||
assert_eq!(t1["outcome"], "completed");
|
||
// (Two model attempts are one line, counted.)
|
||
assert_eq!(t1["steps"][0], json!({"type": "model", "step": "assistant", "attempts": 2}));
|
||
assert_eq!(t1["steps"][1]["name"], "computer");
|
||
assert_eq!(t1["steps"][1]["status"], "done", "the finished tool shows its status");
|
||
assert_eq!(t1["steps"][1].get("call_id"), None, "call ids are for matching, not for people");
|
||
assert_eq!(t1["steps"][2]["state"], "consumed");
|
||
let t2 = &v["turns"][1];
|
||
assert_eq!(t2["outcome"], json!(null), "a turn without its last line has no outcome");
|
||
assert_eq!(t2["steps"][0]["status"], "running", "what is running now reads as running");
|
||
}
|
||
|
||
#[test]
|
||
fn a_finished_turn_leaves_only_waiting_messages() {
|
||
let log = scratch("finish");
|
||
log.append(&Record::OperationStarted { op: "t1".into(), at: 1, source_leaf: 1, prompt: None });
|
||
log.append(&Record::StepAttempt { op: "t1".into(), step: "assistant".into(), attempt: 1 });
|
||
log.append(&Record::QueueEnqueued { op: Some("t1".into()), id: "q1".into(), text: "順便".into(), at: 2 });
|
||
log.append(&Record::QueueEnqueued { op: Some("t1".into()), id: "q2".into(), text: "還有".into(), at: 3 });
|
||
log.append(&Record::QueueConsumed { id: "q1".into() });
|
||
log.finish("t1", Outcome::Completed);
|
||
assert_eq!(log.read(), vec![Record::QueueEnqueued { op: Some("t1".into()), id: "q2".into(), text: "還有".into(), at: 3 }]);
|
||
log.append(&Record::QueueConsumed { id: "q2".into() });
|
||
log.finish("t2", Outcome::Completed);
|
||
assert!(!log.path.exists(), "nothing waiting: no file");
|
||
}
|
||
|
||
#[test]
|
||
fn a_torn_last_line_is_skipped() {
|
||
let log = scratch("torn");
|
||
log.append(&Record::OperationStarted { op: "t1".into(), at: 1, source_leaf: 1, prompt: None });
|
||
let mut f = std::fs::OpenOptions::new().append(true).open(&log.path).unwrap();
|
||
f.write_all(b"{\"type\":\"tool_star").unwrap();
|
||
let r = reduce(&log.read());
|
||
assert_eq!(r.open.map(|o| o.op), Some("t1".to_string()));
|
||
}
|
||
|
||
#[test]
|
||
fn the_reduction_finds_what_was_running_and_what_was_waiting() {
|
||
let recs = vec![
|
||
Record::OperationStarted { op: "t0".into(), at: 1, source_leaf: 1, prompt: None },
|
||
Record::OperationStarted { op: "t1".into(), at: 2, source_leaf: 2, prompt: Some(Prompt { at: 2, text: "傳簡訊".into() }) },
|
||
Record::ToolStarted { op: "t1".into(), call_id: "c1".into(), name: "list_dir".into(), args: "{}".into(), replay: Replay::Safe, at: 0 },
|
||
Record::ToolFinished { op: "t1".into(), call_id: "c1".into(), status: "done".into(), at: 0 },
|
||
Record::ToolStarted { op: "t1".into(), call_id: "c2".into(), name: "send_sms".into(), args: "{}".into(), replay: Replay::Never, at: 0 },
|
||
Record::QueueEnqueued { op: Some("t1".into()), id: "q1".into(), text: "等等".into(), at: 5 },
|
||
Record::QueueEnqueued { op: Some("t1".into()), id: "q2".into(), text: "算了".into(), at: 6 },
|
||
Record::QueueCancelled { id: "q2".into() },
|
||
// A stale line for a turn that isn't the open one changes nothing.
|
||
Record::ToolFinished { op: "t0".into(), call_id: "c2".into(), status: "done".into(), at: 0 },
|
||
];
|
||
let r = reduce(&recs);
|
||
let o = r.open.unwrap();
|
||
assert_eq!(o.op, "t1");
|
||
assert_eq!(o.started.iter().filter(|s| !s.finished).map(|s| (s.call_id.as_str(), s.replay)).collect::<Vec<_>>(), vec![("c2", Replay::Never)]);
|
||
assert_eq!(o.started.len(), 2);
|
||
assert_eq!(r.pending.iter().map(|p| p.id.as_str()).collect::<Vec<_>>(), vec!["q1"]);
|
||
let mut done = recs.clone();
|
||
done.push(Record::AbortRequested { op: "t1".into() });
|
||
assert!(reduce(&done).open.unwrap().aborted);
|
||
done.push(Record::OperationFinished { op: "t1".into(), outcome: Outcome::Aborted });
|
||
assert!(reduce(&done).open.is_none());
|
||
}
|
||
|
||
#[test]
|
||
fn looking_may_run_again_and_reaching_out_never() {
|
||
assert_eq!(replay_of("computer_use_screenshot"), Replay::Safe);
|
||
assert_eq!(replay_of("read_file"), Replay::Safe);
|
||
for t in ["send_sms", "post_to_x", "publish_site", "browser_fill_card", "mac_send_mail", "shell", "edit_file", "computer_use_click", "computer_use_typing", "mcp__github__create_issue"] {
|
||
assert_eq!(replay_of(t), Replay::Never, "{t}");
|
||
}
|
||
}
|
||
|
||
/// After a crash: the lost message comes back, a text that may have gone out says so (not
|
||
/// "not allowed"), a look shows as cut off, a typed message is put back, and the log closes.
|
||
/// A second pass (a crash before the log closed) changes nothing.
|
||
/// A state with one empty bot (id `id`), and a folder of logs of its own.
|
||
fn one_bot(id: u64, name: &str) -> (State, PathBuf) {
|
||
let mut st = State::samples();
|
||
st.bots.truncate(1);
|
||
st.bots[0].id = id;
|
||
st.bots[0].messages.clear();
|
||
let d = std::env::temp_dir().join(format!("anybot-oplog-{name}-{}", std::process::id()));
|
||
let _ = std::fs::remove_dir_all(&d);
|
||
(st, d)
|
||
}
|
||
|
||
#[test]
|
||
fn a_restart_puts_the_chat_right() {
|
||
let (mut st, d) = one_bot(77, "restart");
|
||
let b = &mut st.bots[0];
|
||
b.messages.push(Message { at: 10, ..Message::user("先看資料夾再傳簡訊給媽媽") });
|
||
let mut a = Message { at: 11, ..Message::assistant("", "test") };
|
||
a.tool_calls = vec![
|
||
crate::model::ToolCall { id: "c1".into(), name: "list_dir".into(), args: "{}".into(), status: ToolStatus::Running },
|
||
crate::model::ToolCall { id: "c2".into(), name: "send_sms".into(), args: r#"{"to":"媽媽","text":"晚點到"}"#.into(), status: ToolStatus::Running },
|
||
];
|
||
b.messages.push(a);
|
||
// The next step's empty reply bubble (load turns it into "press 重新產生").
|
||
b.messages.push(Message { at: 13, ..Message::assistant("", "test") });
|
||
// What load() does first: running steps look denied.
|
||
b.repair_interrupted();
|
||
let log = OpLog::in_dir(&d, 77);
|
||
log.append(&Record::OperationStarted { op: "t1".into(), at: 10, source_leaf: 10, prompt: Some(Prompt { at: 10, text: "先看資料夾再傳簡訊給媽媽".into() }) });
|
||
log.append(&Record::ToolStarted { op: "t1".into(), call_id: "c1".into(), name: "list_dir".into(), args: "{}".into(), replay: Replay::Safe, at: 0 });
|
||
log.append(&Record::ToolStarted { op: "t1".into(), call_id: "c2".into(), name: "send_sms".into(), args: r#"{"to":"媽媽","text":"晚點到"}"#.into(), replay: Replay::Never, at: 0 });
|
||
log.append(&Record::QueueEnqueued { op: Some("t1".into()), id: "q1".into(), text: "記得說謝謝".into(), at: 12 });
|
||
// And a log for a bot that was deleted.
|
||
OpLog::in_dir(&d, 999).append(&Record::OperationStarted { op: "x".into(), at: 1, source_leaf: 1, prompt: None });
|
||
|
||
let rec = recover_in(&d, &mut st);
|
||
assert!(rec.changed);
|
||
let found = rec.close();
|
||
assert_eq!(found.len(), 1);
|
||
assert_eq!((found[0].bot_id, found[0].restored, found[0].maybe_ran.len()), (77, 1, 1));
|
||
let m = &st.bots[0].messages;
|
||
let calls = &m[1].tool_calls;
|
||
assert!(calls.iter().all(|c| c.status == ToolStatus::Failed), "{calls:?}");
|
||
let result = |id: &str| m.iter().find(|x| x.role == Role::Tool && x.tool_call_id == id).map(|x| x.text.clone()).unwrap_or_default();
|
||
assert_eq!(result("c1"), CUT_OFF);
|
||
assert_eq!(result("c2"), MAYBE_RAN);
|
||
assert!(m.iter().any(|x| x.role == Role::User && x.text == "記得說謝謝"));
|
||
// The 「繼續」 card comes last: what was being done, the text that may have gone out, the
|
||
// message put back; the "press 重新產生" line it replaces is gone.
|
||
let card = m.last().unwrap();
|
||
assert!(card.interrupted && card.error, "{card:?}");
|
||
assert!(card.text.contains("先看資料夾再傳簡訊給媽媽") && card.text.contains("可能已經完成") && card.text.contains("1 則訊息"), "{}", card.text);
|
||
assert!(!m.iter().any(|x| x.text == crate::model::INTERRUPTED_NOTE));
|
||
// The model is told the same (a result per call, nothing "not allowed").
|
||
let sent = crate::providers::repair_history(m);
|
||
assert!(!sent.iter().any(|x| x.text.contains("沒有允許")));
|
||
assert!(!OpLog::in_dir(&d, 77).path.exists() && !OpLog::in_dir(&d, 999).path.exists());
|
||
|
||
// A crash after the chat was saved but before the log closed: the same lines again.
|
||
let before = st.bots[0].messages.len();
|
||
let log = OpLog::in_dir(&d, 77);
|
||
log.append(&Record::OperationStarted { op: "t1".into(), at: 10, source_leaf: 10, prompt: Some(Prompt { at: 10, text: "先看資料夾再傳簡訊給媽媽".into() }) });
|
||
log.append(&Record::ToolStarted { op: "t1".into(), call_id: "c2".into(), name: "send_sms".into(), args: r#"{"to":"媽媽","text":"晚點到"}"#.into(), replay: Replay::Never, at: 0 });
|
||
log.append(&Record::QueueEnqueued { op: Some("t1".into()), id: "q1".into(), text: "記得說謝謝".into(), at: 12 });
|
||
let again = recover_in(&d, &mut st);
|
||
assert!(!again.changed);
|
||
again.close();
|
||
assert_eq!(st.bots[0].messages.len(), before);
|
||
}
|
||
|
||
/// The message a turn answered survives a crash that came before the chat was saved.
|
||
#[test]
|
||
fn a_message_the_chat_lost_comes_back() {
|
||
let (mut st, d) = one_bot(78, "lost");
|
||
st.bots[0].messages.push(Message { at: 5, ..Message::user("早") });
|
||
let log = OpLog::in_dir(&d, 78);
|
||
log.append(&Record::OperationStarted { op: "t9".into(), at: 20, source_leaf: 20, prompt: Some(Prompt { at: 20, text: "幫我訂餐廳".into() }) });
|
||
let rec = recover_in(&d, &mut st);
|
||
assert!(rec.changed);
|
||
let found = rec.close();
|
||
assert!(found[0].maybe_ran.is_empty() && !found[0].aborted);
|
||
let m = &st.bots[0].messages;
|
||
assert_eq!(m.iter().filter(|x| x.role == Role::User).map(|x| x.text.as_str()).collect::<Vec<_>>(), vec!["早", "幫我訂餐廳"]);
|
||
assert!(m.last().unwrap().interrupted);
|
||
}
|
||
|
||
/// A turn the person had stopped (the stop is on disk, the thread never wound down) ends as
|
||
/// stopped: no 「繼續」 card, and the messages they typed before stopping stay dropped.
|
||
#[test]
|
||
fn a_stopped_turn_stays_stopped() {
|
||
let (mut st, d) = one_bot(79, "stopped");
|
||
st.bots[0].messages.push(Message { at: 5, ..Message::user("整理照片") });
|
||
let log = OpLog::in_dir(&d, 79);
|
||
log.append(&Record::OperationStarted { op: "t1".into(), at: 5, source_leaf: 5, prompt: Some(Prompt { at: 5, text: "整理照片".into() }) });
|
||
log.append(&Record::QueueEnqueued { op: Some("t1".into()), id: "q1".into(), text: "先不要刪".into(), at: 6 });
|
||
log.append(&Record::AbortRequested { op: "t1".into() });
|
||
log.append(&Record::QueueCancelled { id: "q1".into() });
|
||
let found = recover_in(&d, &mut st).close();
|
||
assert!(found[0].aborted);
|
||
assert!(!st.bots[0].messages.iter().any(|m| m.interrupted || m.text == "先不要刪"));
|
||
assert!(!log.path.exists());
|
||
}
|
||
}
|