// 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: `/oplog/bot-.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, }, /// 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, 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 { 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, pub aborted: bool, pub started: Vec, } /// 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, } /// What a log says, read from start to end. #[derive(Clone, Debug, Default, PartialEq)] pub struct Reduced { pub open: Option, pub pending: Vec, } /// 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 = None; let mut pending: Vec = vec![]; for r in recs { let mine = |o: &Option, 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_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, /// 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, } /// What `recover` found; `close` writes it down once the chat is saved. pub struct Recovery { pub interrupted: Vec, /// Did the chats change (save them before `close`). pub changed: bool, closing: Vec, } 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 { 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, 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 = 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::().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 = 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::>().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![("c2", Replay::Never)]); assert_eq!(o.started.len(), 2); assert_eq!(r.pending.iter().map(|p| p.id.as_str()).collect::>(), 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!["早", "幫我訂餐廳"]); 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()); } }