AnyBot/src/oplog.rs

// 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 },
    ToolFinished { op: String, call_id: String, status: String },
    /// 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 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 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 },
            Record::ToolFinished { op: "t1".into(), call_id: "c1".into(), status: "done".into() },
            Record::ToolStarted { op: "t1".into(), call_id: "c2".into(), name: "send_sms".into(), args: "{}".into(), replay: Replay::Never },
            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() },
        ];
        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 });
        log.append(&Record::ToolStarted { op: "t1".into(), call_id: "c2".into(), name: "send_sms".into(), args: r#"{"to":"媽媽","text":"晚點到"}"#.into(), replay: Replay::Never });
        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 });
        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());
    }
}