From 2f603c23623787926403d63cd4c83a59803ca02f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E6=80=A7=E9=A9=8A?= Date: Mon, 9 Nov 2026 00:49:12 +0800 Subject: [PATCH] =?UTF-8?q?A=20turn=20cut=20off=20by=20a=20crash=20leaves?= =?UTF-8?q?=20a=20=E3=80=8C=E7=B9=BC=E7=BA=8C=E3=80=8D=20card,=20and=20not?= =?UTF-8?q?hing=20it=20may=20have=20done=20runs=20twice?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each bot keeps a work log (oplog/bot-.jsonl, after pi's durable-harness design, cut down): before something happens a line says it is about to — the turn, each model call, each step once its checks passed (with the arguments it runs with and whether it may run again), each message typed mid-turn — synced to disk first. When the turn ends the file shrinks to the messages still waiting, usually nothing. On the next start after AnyBot was killed mid-turn, the log puts the chat right before anything runs: the message the turn answered comes back if the chat lost it; a step that was running and reaches outside (a text, a click, a file, a published site) shows as 「可能已經執行了」, both to the person and in the result the bot reads — no longer as "not allowed"; a step that only looked shows as cut off; messages typed mid-turn that never reached the bot go back into the chat. The chat is saved, then the log closes the turn. A turn the person didn't stop ends with a 「繼續」 card (what it was doing, what may have run, the messages put back); it goes on only when they press it — in the window, on a screen of the brain, or on the phone — from where its steps stand, never running a step that may have run again. Pressing stop is on disk at once, and a stopped turn stays stopped. Also in the turn loop: - 自動審核 that doesn't answer (or answers something that isn't a verdict) asks the person; in 直接操作 mode it used to let the step through unchecked. - "Too long for this model" folds older messages into a summary and asks the same step again, once per thing the person said, instead of ending the turn. - Folding keeps the person's chat whole: the summary goes in at the fold and the model reads from the newest summary on (it used to delete what it folded). - A step moved to the chosen computer runs there, but the chat keeps the call as the model wrote it; the result says where it ran. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Uirha5F5gJt93t3yEx3XcS --- docs/architecture.md | 1 + mobile/index.html | 10 +- src/agent.rs | 440 +++++++++++++++++++++++++--- src/app.rs | 154 ++++++++-- src/app/chat_ui.rs | 17 ++ src/app/remote_ui.rs | 13 +- src/main.rs | 1 + src/model.rs | 55 +++- src/oplog.rs | 661 +++++++++++++++++++++++++++++++++++++++++++ src/providers.rs | 31 +- src/remote.rs | 6 +- 11 files changed, 1316 insertions(+), 73 deletions(-) create mode 100644 src/oplog.rs diff --git a/docs/architecture.md b/docs/architecture.md index bb16680..3cbb952 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -66,6 +66,7 @@ AnyBot 是一個 Rust 程式（`anybot`），用 egui/eframe 畫介面。同一 | `brain-copy.json` | 螢幕電腦保留的大腦副本（離線時還能看） | | `brain-files/` | 螢幕電腦從大腦下載的附件 | | `image-notes.json` | 看圖小幫手對每張圖的描述（不用重看） | +| `oplog/bot-.jsonl` | 每個 Bot 的工作紀錄：回合開始、每次問模型、每個步驟開始和結束、做事時傳的訊息。每行先寫進磁碟才做那件事；AnyBot 在做事時被關掉，下次打開靠它把聊天修好，並留一張「繼續」卡片（`src/oplog.rs`）。回合結束就清空 | 寫入時先寫暫存檔再改名，不會寫一半。 diff --git a/mobile/index.html b/mobile/index.html index 765265f..69ede9f 100644 --- a/mobile/index.html +++ b/mobile/index.html @@ -1019,7 +1019,7 @@ function renderMsgs() { // is done; while it works, the newest three stay in view. const turns = []; let curT = null; S.msgs.forEach((m, i) => { - if (m.role === "user" || m.summary) { curT = null; return; } + if (m.role === "user" || m.summary || m.interrupted) { curT = null; return; } if (!curT) { curT = {first: i, last: i, key: String(m.at || i), steps: 0}; turns.push(curT); } curT.last = i; curT.steps += (m.tools || []).filter((t) => !t.page && !t.card).length; }); @@ -1036,6 +1036,13 @@ function renderMsgs() { h += `
${ic("history", 15)} 較早的對話已整理成摘要 ${ic("chevron-down", 14)}
${esc(m.text)}
`; return; } + // Where AnyBot stopping cut a turn off: what it was doing, what may have run, and 「繼續」 + // while it is the newest thing in the chat. + if (m.interrupted) { + const go = m === lastMsg && !busy && owner() ? `` : ""; + h += `
${esc(m.text)}
${go}
`; + return; + } if (m.role === "user") { const tag = m.wake ? `
${ic("alarm-clock", 12)} ${esc(m.via || "例行任務")}
` : ""; const q = String(m.text || "").match(/^((?:>.*\n?)+)\n*([\s\S]*)$/); @@ -1084,6 +1091,7 @@ $("msgs").addEventListener("click", async (e) => { const c = e.target.closest("[data-copy]"); if (c) { copyText(S.msgs[+c.dataset.copy].text); return; } if (e.target.closest("[data-regen]")) { regen(); return; } if (e.target.closest("[data-resend]")) { act("reply", {bot: S.open}); return; } + if (e.target.closest("[data-resume]")) { act("resume", {bot: S.open}, "繼續中…"); return; } const mo = e.target.closest("[data-more]"); if (mo) { msgMenu(+mo.dataset.more); return; } const r = e.target.closest("[data-react]"); if (r) { act("react", {bot: S.open, index: +r.dataset.i, emoji: r.dataset.react}, false).then(() => loadChat()); return; } const sg = e.target.closest("[data-sugg]"); diff --git a/src/agent.rs b/src/agent.rs index 63935aa..64bc8a2 100644 --- a/src/agent.rs +++ b/src/agent.rs @@ -10,7 +10,7 @@ use std::time::Duration; use crate::appctl; use crate::computer; -use crate::model::{Message, Role, ToolStatus}; +use crate::model::{Message, Role, ToolCall, ToolStatus}; use crate::providers::{self, AppCall, Approval, ApprovalAnswer, ApprovalKind, ChatRequest, StreamEvent}; use crate::tools::{self, Decision}; use crate::web; @@ -56,13 +56,69 @@ pub struct Job { /// The bot's name, for the pop-up text. pub via_name: String, /// Messages the person sends while the turn runs (they steer it; no new turn). - pub steer: Arc>>, + pub steer: Arc>>, /// 自動審核 rules, when on. pub review: Option, /// 示範教學 this bot has learned (for view_demo_step). pub demos: Vec, /// The skills there are (a longer job is looked back on against them; see learn). pub skills: Vec, + /// This turn's id: its lines in the bot's work log carry it. + pub op: String, + /// The bot's work log (see oplog.rs); None where nothing is kept (tests, the demo). + pub log: Option, +} + +impl Job { + /// Write a line to the work log (on disk before this returns). + fn note(&self, r: crate::oplog::Record) { + if let Some(l) = &self.log { + l.append(&r); + } + } +} + +/// A message the person typed while the bot works: its words, when it was typed, and the id its +/// work-log lines carry (so one typed just before a crash isn't lost). +#[derive(Clone, Debug)] +pub struct Queued { + pub id: String, + pub text: String, + pub at: i64, +} + +impl Queued { + pub fn new(text: impl Into) -> Self { + Queued { id: crate::oplog::new_id(), text: text.into(), at: crate::model::now_ms() } + } +} + +/// A turn's last line in the work log, written once: explicitly (before Done / Error reach the +/// window), else when the turn's thread ends any other way — stopped (aborted), or a panic +/// (failed). +struct Closing { + log: Option, + op: String, + cancel: Arc, + written: bool, +} + +impl Closing { + fn close(&mut self, outcome: crate::oplog::Outcome) { + if !self.written { + self.written = true; + if let Some(l) = &self.log { + l.finish(&self.op, outcome); + } + } + } +} + +impl Drop for Closing { + fn drop(&mut self) { + let o = if self.cancel.load(Ordering::Relaxed) { crate::oplog::Outcome::Aborted } else { crate::oplog::Outcome::Failed }; + self.close(o); + } } #[derive(Clone, Default)] @@ -77,8 +133,14 @@ pub enum ReviewVerdict { Allow(String), Ask(String), NoMatch, + /// The reviewer didn't answer, or answered something that isn't a verdict: nobody checked the + /// rules, so this never counts as "no rule matched" (a check that breaks, blocks). + Failed, } +/// Said in the question when the reviewer couldn't check a step. +pub const REVIEW_FAILED_NOTE: &str = "（自動審核沒有回應，所以先問你）"; + /// Ask the model whether an action matches the person's rules. "Ask first" wins on conflict /// (NFRGrQ: 規則衝突時，「先詢問」優先); built-in safety checks have already run before this. fn review_action(job: &Job, review: &Review, action: &str, cancel: &AtomicBool) -> ReviewVerdict { @@ -92,8 +154,13 @@ fn review_action(job: &Job, review: &Review, action: &str, cancel: &AtomicBool) tools: vec![], ..job.req.clone() }; - let Ok(out) = providers::run_step(&req, cancel, &|_| {}) else { return ReviewVerdict::NoMatch }; - let text = out.text.trim(); + let Ok(out) = providers::run_step(&req, cancel, &|_| {}) else { return ReviewVerdict::Failed }; + parse_verdict(&out.text) +} + +/// The reviewer's one line of JSON. Anything but a clear ask / allow / none is a failed review. +fn parse_verdict(text: &str) -> ReviewVerdict { + let text = text.trim(); // A truncated answer can put the last "}" before the first "{" — slicing then panics. let json_part = text.find('{').and_then(|a| text.rfind('}').filter(|b| *b >= a).map(|b| &text[a..=b])).unwrap_or(text); let v: serde_json::Value = serde_json::from_str(json_part).unwrap_or_default(); @@ -101,7 +168,8 @@ fn review_action(job: &Job, review: &Review, action: &str, cancel: &AtomicBool) match v["verdict"].as_str() { Some("ask") => ReviewVerdict::Ask(rule), Some("allow") => ReviewVerdict::Allow(rule), - _ => ReviewVerdict::NoMatch, + Some("none") => ReviewVerdict::NoMatch, + _ => ReviewVerdict::Failed, } } @@ -136,12 +204,14 @@ fn learn(job: &Job, history: &[Message], cancel: &AtomicBool, send: &dyn Fn(Stre } } -/// Take what the person typed mid-turn and add it to the chat as steering. +/// Take what the person typed mid-turn and add it to the chat as steering. (The window marks it +/// delivered in the work log once the chat holding it is saved; the message keeps the time it was +/// typed, which is how a restart tells it is already there.) fn drain_steer(job: &Job, history: &mut Vec, send: &dyn Fn(StreamEvent)) -> bool { - let queued: Vec = std::mem::take(&mut *job.steer.lock().unwrap()); - for t in &queued { - let m = Message { steer: true, ..Message::user(t) }; - send(StreamEvent::Steered(m.clone())); + let queued: Vec = std::mem::take(&mut *job.steer.lock().unwrap()); + for q in &queued { + let m = Message { steer: true, at: q.at, ..Message::user(&q.text) }; + send(StreamEvent::Steered(m.clone(), q.id.clone())); history.push(m); } !queued.is_empty() @@ -603,6 +673,7 @@ fn maybe_compact(job: &Job, history: &mut Vec, percent: u64, cancel: &A send(StreamEvent::Alive); } }; + job.note(crate::oplog::Record::StepAttempt { op: job.op.clone(), step: "compaction".into(), attempt: 1 }); let out = providers::run_step(&req, cancel, &quiet).ok()?; if out.text.trim().is_empty() || cancel.load(Ordering::Relaxed) { return None; @@ -645,11 +716,24 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< let mut tried: Vec = Vec::new(); let mut local_on = job.local_execution; let mut browsers = UsedBrowsers::new(&job.via_name, bot_id); + // The turn's first line in the work log, and (whatever happens) its last. + job.note(crate::oplog::Record::OperationStarted { + op: job.op.clone(), + at: crate::model::now_ms(), + source_leaf: history.last().map(|m| m.at).unwrap_or(0), + prompt: history.last().filter(|m| m.role == Role::User).map(|m| crate::oplog::Prompt { at: m.at, text: m.text.clone() }), + }); + let mut closing = Closing { log: job.log.clone(), op: job.op.clone(), cancel: cancel.clone(), written: false }; + // Folding the chat because the model said it was too long: once per thing the person + // said (a second time means folding didn't help). + let mut overflow_folded = false; maybe_compact(&job, &mut history, COMPACT_THRESHOLD_PERCENT, &cancel, &send); for step in 0..MAX_STEPS { if step > 0 { maybe_compact(&job, &mut history, COMPACT_THRESHOLD_PERCENT + COMPACT_TOLERANCE_PERCENT, &cancel, &send); - drain_steer(&job, &mut history, &send); + if drain_steer(&job, &mut history, &send) { + overflow_folded = false; + } check_in(&mut history, step); if let Some(note) = refresh_own(&mut job, bot_id) { if let Some(m) = history.last_mut().filter(|m| m.role == Role::Tool) { @@ -658,25 +742,44 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< } send(StreamEvent::NextStep(job.via.clone())); } - let req = ChatRequest { history: history.clone(), ..job.req.clone() }; - let out = match providers::run_step(&req, &cancel, &send) { - Ok(o) => o, - Err(e) => { - send(StreamEvent::Error(e)); - return; + let mut attempt = 0; + let out = loop { + attempt += 1; + job.note(crate::oplog::Record::StepAttempt { op: job.op.clone(), step: "assistant".into(), attempt }); + let req = ChatRequest { history: history.clone(), ..job.req.clone() }; + match providers::run_step(&req, &cancel, &send) { + Ok(o) => break o, + // Too long for the model: that answer is thrown away, older messages are + // folded into a summary (however full the chat looks by our estimate), and + // the same step is asked again. + Err(e) if providers::is_context_overflow(&e) && !overflow_folded && !cancel.load(Ordering::Relaxed) => { + overflow_folded = true; + if maybe_compact(&job, &mut history, 0, &cancel, &send).is_none() { + closing.close(crate::oplog::Outcome::Failed); + send(StreamEvent::Error(e)); + return; + } + } + Err(e) => { + let e = if providers::is_context_overflow(&e) && overflow_folded { format!("{e}\n（已經把較早的對話整理成摘要再試過一次，還是太長。）") } else { e }; + closing.close(crate::oplog::Outcome::Failed); + send(StreamEvent::Error(e)); + return; + } } }; if cancel.load(Ordering::Relaxed) { return; } - let mut out = out; - // Steps moved back to the chosen computer: what the bot is told with each result. + // Steps moved back to the chosen computer: what the bot is told with each result, and + // the arguments they run with. The chat keeps the call as the model wrote it. let mut moved: std::collections::HashMap = std::collections::HashMap::new(); - for c in out.tool_calls.iter_mut() { + let mut runs_with: std::collections::HashMap = std::collections::HashMap::new(); + for c in out.tool_calls.iter() { let (args, note) = keep_to_chosen(&job, &history, &c.name, &c.args); - c.args = args; if let Some(n) = note { moved.insert(c.id.clone(), n); + runs_with.insert(c.id.clone(), args); } } let msg = Message { @@ -695,6 +798,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< if calls.is_empty() { // Something typed while it answered keeps the same turn going. if job.steer.lock().unwrap().is_empty() { + closing.close(crate::oplog::Outcome::Completed); send(StreamEvent::Done); browsers.close_tabs(); learn(&job, &history, &cancel, &send); @@ -702,13 +806,15 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< } continue; } - // (Each call's computer was already kept to the chosen one, see keep_to_chosen.) - for call in calls { + for asked in calls { + // What runs: the call as asked, on the chosen computer (see keep_to_chosen). + let call = ToolCall { args: runs_with.get(&asked.id).cloned().unwrap_or_else(|| asked.args.clone()), ..asked.clone() }; // 鬼打牆: the same screen or browser step again and again (a click on the same // empty spot ten times). The third time it runs with a warning; after that it isn't - // run at all, and the bot is told to change course. + // run at all, and the bot is told to change course. (Counted as the model asked, + // which is how the chat keeps them.) let mut repeat_note = String::new(); - if let Some(key) = step_key(&call.name, &call.args) { + if let Some(key) = step_key(&asked.name, &asked.args) { let times = tried.iter().rev().take(10).filter(|k| **k == key).count() + 1; tried.push(key); if times >= 4 { @@ -726,7 +832,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< // The same look again and again with the same answer (a blank tab read eleven // times in a row): the third time with a warning, after that not at all. if repeat_note.is_empty() { - let same = same_answers(&history, &call.name, &call.args); + let same = same_answers(&history, &asked.name, &asked.args); if same >= 3 { let text = format!("沒有執行：同一個工具、同樣的參數你已經做了 {same} 次，每次結果都一模一樣，再做也不會不同。換一個做法（例如用 browser_open 打開要看的網址），或直接告訴使用者你看到了什麼。"); send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Failed)); @@ -742,15 +848,22 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< // Every step ends the same way: its status, then its result (with the notes) and // any picture, shown and kept. let finish = |history: &mut Vec, output: &str, status: ToolStatus, image: Option| { + job.note(crate::oplog::Record::ToolFinished { op: job.op.clone(), call_id: call.id.clone(), status: format!("{status:?}").to_lowercase() }); send(StreamEvent::ToolStatus(call.id.clone(), status)); let mut result = Message::tool_result(&call.id, &call.name, &format!("{repeat_note}{}", noted(&moved, &call.id, output))); result.images.extend(image); send(StreamEvent::ToolResult(result.clone())); history.push(result); }; + // About to run, every check passed: the work log hears it first, with the + // arguments it runs with and whether it may run again after a crash. + let started = || { + job.note(crate::oplog::Record::ToolStarted { op: job.op.clone(), call_id: call.id.clone(), name: call.name.clone(), args: call.args.clone(), replay: crate::oplog::replay_of(&call.name) }); + send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + }; if appctl::is_app_tool(&call.name) { // App actions run on the window's side, which also asks for deletions. - send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + started(); let (rtx, rrx) = channel(); send(StreamEvent::AppAction(AppCall { call_id: call.id.clone(), name: call.name.clone(), args: call.args.clone(), reply: rtx })); let (output, ok) = loop { @@ -768,7 +881,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< } // The person's own recording, shown back to the bot: nothing to ask. if call.name == "view_demo_step" { - send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + started(); let (output, ok, img) = crate::teach::view_step(&job.demos, &call.args); finish(&mut history, &output, if ok { ToolStatus::Done } else { ToolStatus::Failed }, img); continue; @@ -777,7 +890,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< // …and neither does AnyBot's own browser (like Grok Bot's own browser). let at = web::is_web_tool(&call.name).then(|| browser_place(&job, &call.name, &call.args)); if let Some((place, on)) = at.filter(|(place, _)| call.name == "web_fetch" || bots_browser(place, &call.name, &call.args)) { - send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + started(); if call.name != "web_fetch" { let profile = tools::args_of(&call.args).get("profile").and_then(|p| p.as_str()).unwrap_or("").trim().to_string(); browsers.add(match &place { @@ -874,7 +987,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< } // 傳檔案: put a file the bot made into the chat as a card the person can open. if call.name == "send_file" { - send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + started(); let path = tools::args_of(&call.args)["path"].as_str().unwrap_or("").to_string(); let local = match target_for(&job, &call.args) { // Files ride to the chat only from this bot's own folder on the box. @@ -918,7 +1031,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< } // 檔案預覽 / Bot 做的頁面: open what the bot made on the person's screen. if call.name == "open_for_person" { - send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + started(); let v = tools::args_of(&call.args); let path = v["path"].as_str().unwrap_or("").to_string(); let on_box = match target_for(&job, &call.args) { @@ -979,7 +1092,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< let (output, status) = if answer == ApprovalAnswer::Reject { ("使用者沒有允許。請不要再試，改問使用者。".to_string(), ToolStatus::Denied) } else { - send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + started(); match crate::phone::run(&call.name, &call.args) { (t, true) => (t, ToolStatus::Done), (t, false) => (t, ToolStatus::Failed), @@ -1011,7 +1124,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< let (output, status) = if !allowed { ("使用者沒有允許發布。".to_string(), ToolStatus::Denied) } else { - send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + started(); let local = match target_for(&job, &call.args) { Ok(tools::Target { place: tools::Place::Box { dir }, .. }) => crate::cloudbox::copy_out(&if path.starts_with('/') { path.clone() } else { format!("{dir}/{path}") }), Ok(tools::Target { place: tools::Place::Sealed { root }, .. }) => Ok(tools::resolve(&root, &path)), @@ -1072,7 +1185,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< if !allowed { ("使用者沒有允許使用保險庫的這個項目。請改問使用者。".to_string(), ToolStatus::Denied) } else { - send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + started(); let r = match plan { Plan::Login(pl) => crate::vault::fill_login(&pl, tools::args_of(&call.args)["submit"].as_bool().unwrap_or(false)), Plan::Card => crate::vault::fill_card(&call.args), @@ -1109,7 +1222,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< return; } let (output, status, img) = if allowed { - send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + started(); let (o, ok, img) = crate::mcp::call(&call.name, &call.args); (o, if ok { ToolStatus::Done } else { ToolStatus::Failed }, img) } else { @@ -1268,6 +1381,8 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< }; match verdict { ReviewVerdict::Ask(rule) => Decision::Ask { reason: format!("{reason}（符合你的「一律詢問」規則：{rule}）"), grant: String::new() }, + // Nobody could check the person's "always ask" rules: ask, never go ahead. + ReviewVerdict::Failed => Decision::Ask { reason: format!("{reason}{REVIEW_FAILED_NOTE}"), grant: String::new() }, // (No note on the result: the model doesn't need it, and it was on every step.) _ => { let _ = grant; @@ -1282,6 +1397,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< } ReviewVerdict::Ask(rule) => Decision::Ask { reason: format!("{reason}（符合你的「一律詢問」規則：{rule}）"), grant: String::new() }, ReviewVerdict::NoMatch => Decision::Ask { reason, grant }, + ReviewVerdict::Failed => Decision::Ask { reason: format!("{reason}{REVIEW_FAILED_NOTE}"), grant }, }, (d, _) => d, }; @@ -1318,7 +1434,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< let mut image = None; let (output, status) = match (approved, &target) { (Ok(()), Ok(t)) => { - send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); + started(); let ids = tools::RunIds { conversation: bot_id.to_string(), request: call.id.clone() }; let (out, ok) = if is_mac && on_device.is_none() { crate::macapps::run(&call.name, &call.args) @@ -1370,6 +1486,7 @@ pub fn spawn(job: Job, bot_id: u64, tx: Sender<(u64, StreamEvent)>, cancel: Arc< } send(StreamEvent::NextStep(job.via.clone())); send(StreamEvent::StepDone(Message::assistant(&format!("（這次已經用了 {MAX_STEPS} 個步驟，先停在這裡。要繼續的話，回覆「繼續」。）"), &job.via))); + closing.close(crate::oplog::Outcome::Completed); send(StreamEvent::Done); }); } @@ -1400,6 +1517,8 @@ mod tests { review: None, demos: vec![], skills: vec![], + op: String::new(), + log: None, } } @@ -1982,6 +2101,8 @@ echo 'done送出了' > done.html; (setsid pyt review: None, demos: vec![], skills: vec![], + op: String::new(), + log: None, }; let (tx, rx) = channel(); let cancel = Arc::new(AtomicBool::new(false)); @@ -2037,7 +2158,7 @@ echo 'done送出了' > done.html; (setsid pyt let _ = s.write_all(body.as_bytes()); } }); - let steer = Arc::new(Mutex::new(vec!["順便用英文".to_string()])); + let steer = Arc::new(Mutex::new(vec![Queued::new("順便用英文")])); let job = Job { req: ChatRequest { kind: ProviderKind::Custom, base_url: format!("http://127.0.0.1:{port}/v1"), api_key: String::new(), model: "m".into(), system: String::new(), history: vec![Message::user("打招呼")], tools: vec![], session: String::new() }, computers: vec![], @@ -2055,13 +2176,15 @@ echo 'done送出了' > done.html; (setsid pyt review: None, demos: vec![], skills: vec![], + op: String::new(), + log: None, }; let (tx, rx) = channel(); spawn(job, 1, tx, Arc::new(AtomicBool::new(false)), eframe::egui::Context::default()); let (mut steered, mut done, mut texts) = (0, 0, vec![]); loop { match rx.recv_timeout(Duration::from_secs(10)).expect("loop stalled") { - (_, StreamEvent::Steered(m)) => { + (_, StreamEvent::Steered(m, _)) => { assert!(m.steer); steered += 1; } @@ -2134,6 +2257,8 @@ echo 'done送出了' > done.html; (setsid pyt review: None, demos: vec![], skills: vec![], + op: String::new(), + log: None, }; let (tx, rx) = channel(); spawn(job, 1, tx, Arc::new(AtomicBool::new(false)), eframe::egui::Context::default()); @@ -2274,6 +2399,8 @@ echo 'done送出了' > done.html; (setsid pyt review: None, demos: vec![], skills: vec![], + op: String::new(), + log: None, }; let (tx, rx) = channel(); spawn(job, 1, tx, Arc::new(AtomicBool::new(false)), eframe::egui::Context::default()); @@ -2295,4 +2422,241 @@ echo 'done送出了' > done.html; (setsid pyt assert!(w.join("made.txt").exists()); assert!(results[1].contains("一律不允許"), "{results:?}"); } + + /// A made-up model that answers each request with the next (HTTP status, body) and keeps + /// what each request said. + fn fake_http(replies: Vec<(u16, String)>) -> (u16, Arc>>) { + use std::io::{Read, Write}; + let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = l.local_addr().unwrap().port(); + let seen = Arc::new(Mutex::new(vec![])); + let seen2 = seen.clone(); + std::thread::spawn(move || { + for (status, body) in replies { + let Ok((mut s, _)) = l.accept() else { return }; + let mut req = Vec::new(); + let mut buf = [0u8; 8192]; + loop { + let n = s.read(&mut buf).unwrap_or(0); + if n == 0 { + break; + } + req.extend_from_slice(&buf[..n]); + let t = String::from_utf8_lossy(&req).to_string(); + if let Some(end) = t.find("\r\n\r\n") { + let len = t[..end].lines().find_map(|l| l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::().unwrap_or(0))).unwrap_or(0); + if req.len() >= end + 4 + len { + break; + } + } + } + seen2.lock().unwrap().push(String::from_utf8_lossy(&req).to_string()); + let head = if status == 200 { "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\n".to_string() } else { format!("HTTP/1.1 {status} Error\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len()) }; + let _ = s.write_all(head.as_bytes()); + let _ = s.write_all(body.as_bytes()); + } + }); + (port, seen) + } + + fn sse_text(t: &str) -> (u16, String) { + (200, format!("data: {}\n\ndata: [DONE]\n\n", serde_json::json!({"choices": [{"delta": {"content": t}}]}))) + } + + fn sse_call(id: &str, name: &str, args: serde_json::Value) -> (u16, String) { + (200, format!("data: {}\n\ndata: [DONE]\n\n", serde_json::json!({"choices": [{"delta": {"tool_calls": [{"index": 0, "id": id, "function": {"name": name, "arguments": args.to_string()}}]}}]}))) + } + + const OVERFLOW_BODY: &str = r#"{"error":{"message":"This model's maximum context length is 8192 tokens. However, your messages resulted in 9000 tokens."}}"#; + + /// A long chat: enough older messages for a fold to have something to fold. + fn long_chat() -> Vec { + let mut h = vec![]; + for i in 0..6 { + h.push(Message { at: i * 2 + 1, ..Message::user(&format!("第 {i} 個問題")) }); + h.push(Message { at: i * 2 + 2, ..Message::assistant(&format!("第 {i} 個回答"), "test") }); + } + h.push(Message { at: 100, ..Message::user("現在呢？") }); + h + } + + /// Run a job to its end, collecting what came back. Approvals get `answer`. + fn drive(job: Job, answer: ApprovalAnswer) -> (Vec, Vec) { + let (tx, rx) = channel(); + spawn(job, 1, tx, Arc::new(AtomicBool::new(false)), eframe::egui::Context::default()); + let mut evs = vec![]; + let mut asked = vec![]; + loop { + let (_, ev) = rx.recv_timeout(Duration::from_secs(15)).expect("loop stalled"); + match ev { + StreamEvent::NeedApproval(a) => { + asked.push(a.reason.clone()); + let _ = a.reply.send(answer.clone()); + } + StreamEvent::Done | StreamEvent::Error(_) => { + evs.push(ev); + break; + } + ev => evs.push(ev), + } + } + (evs, asked) + } + + #[test] + fn a_verdict_is_clear_or_the_review_failed() { + assert!(matches!(parse_verdict(r#"{"verdict":"ask","rule":"刪除前先問"}"#), ReviewVerdict::Ask(r) if r == "刪除前先問")); + assert!(matches!(parse_verdict(r#"好的 {"verdict":"allow","rule":"讀檔"}"#), ReviewVerdict::Allow(_))); + assert!(matches!(parse_verdict(r#"{"verdict":"none","rule":""}"#), ReviewVerdict::NoMatch)); + for broken in ["", "嗯…讓我想想", r#"{"verdict":"maybe"}"#, r#"} {"#] { + assert!(matches!(parse_verdict(broken), ReviewVerdict::Failed), "{broken}"); + } + } + + /// 直接操作 with an 「一律詢問」 rule: when the reviewer doesn't give a verdict, the step is + /// asked about, never run unchecked. + #[test] + fn a_review_that_fails_asks_instead_of_allowing() { + let (port, _) = fake_http(vec![ + sse_call("c1", "shell", serde_json::json!({"command": "echo hi > made.txt", "computer": crate::model::THIS_COMPUTER_NAME})), + sse_text("嗯…讓我想想"), + sse_text("好，我不做了"), + ]); + let w = std::env::temp_dir().join(format!("anybot-review-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&w); + std::fs::create_dir_all(&w).unwrap(); + let job = Job { + req: ChatRequest { kind: ProviderKind::Custom, base_url: format!("http://127.0.0.1:{port}/v1"), api_key: String::new(), model: "m".into(), system: String::new(), history: vec![Message::user("做一個檔案")], tools: tools::specs(), session: String::new() }, + computers: vec![tools::Target { name: crate::model::THIS_COMPUTER_NAME.into(), place: tools::Place::Local { workdir: w.clone() } }], + default_computer: crate::model::THIS_COMPUTER_NAME.into(), + roles: tools::Roles { own: crate::model::BOT_COMPUTER_NAME.into(), person: Some(crate::model::THIS_COMPUTER_NAME.into()), server: false }, + direct: Arc::new(AtomicBool::new(true)), + review: Some(Review { allow: vec![], ask: vec!["刪除或修改任何檔案前先問我".into()] }), + ..test_job() + }; + let (evs, asked) = drive(job, ApprovalAnswer::Reject); + assert_eq!(asked.len(), 1, "{asked:?}"); + assert!(asked[0].contains(REVIEW_FAILED_NOTE), "{}", asked[0]); + assert!(!w.join("made.txt").exists()); + assert!(matches!(evs.last(), Some(StreamEvent::Done))); + } + + /// The model says the chat is too long: older messages are folded into a summary and the same + /// step is asked again (with the summary), instead of the turn failing. + #[test] + fn an_overflow_folds_the_chat_and_asks_again() { + let (port, seen) = fake_http(vec![(400, OVERFLOW_BODY.into()), sse_text("摘要：問過六個問題"), sse_text("好了")]); + let job = Job { req: ChatRequest { kind: ProviderKind::Custom, base_url: format!("http://127.0.0.1:{port}/v1"), api_key: String::new(), model: "m".into(), system: String::new(), history: long_chat(), tools: vec![], session: String::new() }, ..test_job() }; + let (evs, _) = drive(job, ApprovalAnswer::Reject); + assert!(matches!(evs.last(), Some(StreamEvent::Done)), "{:?}", evs.iter().filter_map(|e| if let StreamEvent::Error(e) = e { Some(e.clone()) } else { None }).collect::>()); + assert!(evs.iter().any(|e| matches!(e, StreamEvent::Compacted(..)))); + let texts: Vec = evs.iter().filter_map(|e| if let StreamEvent::StepDone(m) = e { Some(m.text.clone()) } else { None }).collect(); + assert_eq!(texts, vec!["好了"]); + let seen = seen.lock().unwrap(); + assert_eq!(seen.len(), 3); + assert!(seen[2].contains("摘要：問過六個問題") && !seen[2].contains("第 0 個問題"), "{}", seen[2]); + } + + /// Still too long after folding once: the turn ends with a clear error (no loop of folds). + #[test] + fn a_second_overflow_ends_the_turn() { + let (port, seen) = fake_http(vec![(400, OVERFLOW_BODY.into()), sse_text("摘要"), (400, OVERFLOW_BODY.into()), sse_text("不該問到這裡")]); + let job = Job { req: ChatRequest { kind: ProviderKind::Custom, base_url: format!("http://127.0.0.1:{port}/v1"), api_key: String::new(), model: "m".into(), system: String::new(), history: long_chat(), tools: vec![], session: String::new() }, ..test_job() }; + let (evs, _) = drive(job, ApprovalAnswer::Reject); + match evs.last() { + Some(StreamEvent::Error(e)) => assert!(providers::is_context_overflow(e) && e.contains("還是太長"), "{e}"), + _ => panic!("expected an error"), + } + assert_eq!(seen.lock().unwrap().len(), 3); + } + + /// The work log while a turn runs: the turn and its model call are written first; a step + /// waiting for the person isn't "started"; once allowed it is (with the arguments it runs + /// with), then finished; a finished turn leaves no file. A stopped turn ends as stopped. + #[test] + fn the_work_log_follows_the_turn() { + use crate::oplog::{reduce, OpLog}; + let w = std::env::temp_dir().join(format!("anybot-oplog-turn-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&w); + std::fs::create_dir_all(&w).unwrap(); + std::fs::write(w.join("a.txt"), "hi").unwrap(); + let path = w.join("log.jsonl"); + for stop in [false, true] { + let (port, _) = fake_http(vec![sse_call("c1", "shell", serde_json::json!({"command": "echo hi > b.txt"})), sse_text("做好了")]); + let cancel = Arc::new(AtomicBool::new(false)); + let job = Job { + req: ChatRequest { kind: ProviderKind::Custom, base_url: format!("http://127.0.0.1:{port}/v1"), api_key: String::new(), model: "m".into(), system: String::new(), history: vec![Message { at: 42, ..Message::user("做 b.txt") }], tools: tools::specs(), session: String::new() }, + computers: vec![tools::Target { name: crate::model::THIS_COMPUTER_NAME.into(), place: tools::Place::Local { workdir: w.clone() } }], + default_computer: crate::model::THIS_COMPUTER_NAME.into(), + roles: tools::Roles { own: crate::model::BOT_COMPUTER_NAME.into(), person: Some(crate::model::THIS_COMPUTER_NAME.into()), server: false }, + local_execution: true, + op: "turn-test".into(), + log: Some(OpLog::at(path.clone())), + ..test_job() + }; + let read = || reduce(&OpLog::at(path.clone()).read()); + let raw = || std::fs::read_to_string(&path).unwrap_or_default(); + let (tx, rx) = channel(); + spawn(job, 1, tx, cancel.clone(), eframe::egui::Context::default()); + loop { + match rx.recv_timeout(Duration::from_secs(15)).expect("loop stalled") { + (_, StreamEvent::NeedApproval(a)) => { + let open = read().open.expect("the turn is open"); + assert_eq!((open.op.as_str(), open.prompt.map(|p| p.at)), ("turn-test", Some(42))); + assert!(open.started.is_empty(), "waiting for the person is not started"); + assert!(raw().contains("\"step_attempt\"")); + if stop { + cancel.store(true, Ordering::Relaxed); + drop(a); + break; + } + let _ = a.reply.send(ApprovalAnswer::Once); + } + (_, StreamEvent::ToolResult(_)) => { + let open = read().open.expect("the turn is still open"); + assert_eq!(open.started.len(), 1); + assert!(open.started[0].finished && open.started[0].replay == crate::oplog::Replay::Never); + assert!(raw().contains("echo hi > b.txt")); + } + (_, StreamEvent::Done) => break, + (_, StreamEvent::Error(e)) => panic!("{e}"), + _ => {} + } + } + if stop { + // The thread winds down on its own; its last line says stopped. + let until = std::time::Instant::now() + Duration::from_secs(5); + while path.exists() && std::time::Instant::now() < until { + std::thread::sleep(Duration::from_millis(50)); + } + } + assert!(!path.exists(), "a closed turn leaves nothing behind: {}", raw()); + } + } + + /// A step moved to the chosen computer runs there, but the chat keeps the call as the model + /// wrote it; the result says where it ran. + #[test] + fn a_moved_step_is_kept_as_the_model_asked() { + let (port, _) = fake_http(vec![sse_call("c1", "list_dir", serde_json::json!({"computer": "王小明的Mac mini"})), sse_text("好")]); + let w = std::env::temp_dir().join(format!("anybot-moved-{}", std::process::id())); + std::fs::create_dir_all(&w).unwrap(); + std::fs::write(w.join("here.txt"), "hi").unwrap(); + let job = Job { + req: ChatRequest { kind: ProviderKind::Custom, base_url: format!("http://127.0.0.1:{port}/v1"), api_key: String::new(), model: "m".into(), system: String::new(), history: vec![Message::user("有什麼檔案？")], tools: tools::specs(), session: String::new() }, + computers: vec![ + tools::Target { name: crate::model::THIS_COMPUTER_NAME.into(), place: tools::Place::Local { workdir: w.clone() } }, + tools::Target { name: "王小明的Mac mini".into(), place: tools::Place::Device { host: "m".into(), dir: String::new() } }, + ], + default_computer: crate::model::THIS_COMPUTER_NAME.into(), + roles: tools::Roles { own: crate::model::BOT_COMPUTER_NAME.into(), person: Some(crate::model::THIS_COMPUTER_NAME.into()), server: false }, + local_execution: true, + ..test_job() + }; + let (evs, _) = drive(job, ApprovalAnswer::Once); + let asked = evs.iter().find_map(|e| if let StreamEvent::StepDone(m) = e { m.tool_calls.first().cloned() } else { None }).unwrap(); + assert!(asked.args.contains("王小明的Mac mini"), "{}", asked.args); + let result = evs.iter().find_map(|e| if let StreamEvent::ToolResult(m) = e { Some(m.text.clone()) } else { None }).unwrap(); + assert!(result.contains("這一步你指定了「王小明的Mac mini」") && result.contains("here.txt"), "{result}"); + } } diff --git a/src/app.rs b/src/app.rs index 40734ea..74d7f96 100644 --- a/src/app.rs +++ b/src/app.rs @@ -114,7 +114,14 @@ pub struct App { /// App actions (e.g. deleting a routine) waiting for that yes/no. app_pending: HashMap, /// Messages that arrived as a turn ended: they start the next turn. - pending_turns: Vec<(u64, Vec)>, + pending_turns: Vec<(u64, Vec)>, + /// Typed messages now in a chat (bot, id): the work log hears they were delivered once the + /// chat holding them is saved. + delivered: Vec<(u64, String)>, + /// Save at the next frame (a typed message just entered a chat), not up to 2 s later. + save_now: bool, + /// Turns that AnyBot stopping cut off, found at start (see resume_interrupted). + interrupted: Vec, last_tick: Instant, /// The routines list for this bot is open. show_routines: Option, @@ -218,7 +225,19 @@ impl App { pub fn new(ctx: &egui::Context, screenshot: Option) -> Self { // ANYBOT_SHOT_REAL=1: a screenshot of the real data (and connections), not the demo. let demo = screenshot.is_some() && std::env::var("ANYBOT_SHOT_REAL").is_err(); - let state = if demo { State::demo() } else { State::load() }; + let mut state = if demo { State::demo() } else { State::load() }; + // After a crash: the work log puts the chats right (what was running, what was typed), + // the chats are saved, then the log closes the cut-off turns. Not for a screenshot run of + // the real data: that is a second copy next to the one that owns the log. + let interrupted = if demo || screenshot.is_some() { + vec![] + } else { + let found = crate::oplog::recover(&mut state); + if found.changed { + state.save(); + } + found.close() + }; crate::routines::set_person_timezone(&state.timezone); let (tx, rx) = channel(); let mut app = App { @@ -235,6 +254,9 @@ impl App { approvals: HashMap::new(), app_pending: HashMap::new(), pending_turns: vec![], + delivered: vec![], + save_now: false, + interrupted, last_tick: Instant::now(), show_routines: None, new_remote: Default::default(), @@ -330,7 +352,7 @@ impl App { produced: true, status: None, paused: true, - steer: Arc::new(Mutex::new(vec!["2026-06 的照片先不要動".into()])), + steer: Arc::new(Mutex::new(vec![agent::Queued::new("2026-06 的照片先不要動")])), direct: Arc::new(AtomicBool::new(false)), turn_id: "demo".into(), }); @@ -671,9 +693,9 @@ impl App { self.typed_on = Some(String::new()); // While it works, a new message steers the running turn (the original's `turnSteer`: // it doesn't count as a new turn) and reaches the model at its next step. - if let Some(l) = self.streams.get(&bot_id) { + if self.streams.contains_key(&bot_id) { let (extra, _) = crate::attach::for_model(&attachments); - l.steer.lock().unwrap().push(format!("{}{extra}", text.trim())); + self.queue_steer(bot_id, format!("{}{extra}", text.trim())); return; } if let Some(b) = self.state.bot_mut(bot_id) { @@ -682,6 +704,17 @@ impl App { self.start_reply(ctx, bot_id); } + /// A message for a bot that is working: it waits for the bot's next step. The work log has it + /// (on disk) before it is queued, so a crash before the bot reads it doesn't lose it. + fn queue_steer(&self, bot_id: u64, text: String) { + let Some(l) = self.streams.get(&bot_id) else { return }; + let q = agent::Queued::new(text); + if !self.demo { + crate::oplog::OpLog::for_bot(bot_id).append(&crate::oplog::Record::QueueEnqueued { op: Some(l.turn_id.clone()), id: q.id.clone(), text: q.text.clone(), at: q.at }); + } + l.steer.lock().unwrap().push(q); + } + fn start_reply(&mut self, ctx: &egui::Context, bot_id: u64) { if self.mir.on { self.mirror_act(serde_json::json!({"bot": bot_id, "action": "reply"})); @@ -816,7 +849,7 @@ impl App { let steer = Arc::new(Mutex::new(Vec::new())); let turn_id = format!("turn-{bot_id}-{}", chrono::Utc::now().timestamp_millis()); let direct = Arc::new(AtomicBool::new(self.direct_on())); - self.streams.insert(bot_id, Live { cancel: cancel.clone(), started: Instant::now(), last: Instant::now(), produced: false, status: None, paused: false, steer: steer.clone(), turn_id, direct: direct.clone() }); + self.streams.insert(bot_id, Live { cancel: cancel.clone(), started: Instant::now(), last: Instant::now(), produced: false, status: None, paused: false, steer: steer.clone(), turn_id: turn_id.clone(), direct: direct.clone() }); self.stick_bottom = true; let review = self.state.auto_review.then(|| agent::Review { allow: self.state.review_allow_rules.clone(), ask: self.state.review_ask_rules.clone() }); let job = agent::Job { @@ -842,6 +875,8 @@ impl App { review, demos: b.demos.clone(), skills: self.state.skills.clone(), + op: turn_id, + log: (!self.demo).then(|| crate::oplog::OpLog::for_bot(bot_id)), }; agent::spawn(job, bot_id, self.tx.clone(), cancel, ctx.clone()); // (Its eyes start reading while it works; see face_pose.) @@ -1095,7 +1130,7 @@ impl App { } } // Messages that arrived too late to steer start the next turn. - let leftover: Vec = std::mem::take(&mut *l.steer.lock().unwrap()); + let leftover: Vec = std::mem::take(&mut *l.steer.lock().unwrap()); if !leftover.is_empty() { self.pending_turns.push((bot_id, leftover)); } @@ -1106,8 +1141,8 @@ impl App { /// Wake a bot with a message: a new turn when idle, steering when it is already working. fn wake(&mut self, ctx: &egui::Context, bot_id: u64, msg: Message) { - if let Some(l) = self.streams.get(&bot_id) { - l.steer.lock().unwrap().push(msg.text); + if self.streams.contains_key(&bot_id) { + self.queue_steer(bot_id, msg.text); return; } let selected = self.state.selected; @@ -1120,12 +1155,48 @@ impl App { self.start_reply(ctx, bot_id); } + /// Turns cut off by AnyBot stopping (found at start, chats already put right by + /// oplog::recover): what happens to them next. The policy: nothing goes on by itself, not + /// even a turn that was only looking. Each chat got a 「繼續」 card, and the turn goes on only + /// when the person presses it (resume_turn). This only notes them in the log. + fn resume_interrupted(&mut self, _ctx: &egui::Context) { + for t in std::mem::take(&mut self.interrupted) { + eprintln!( + "AnyBot：Bot {} 的回合 {} 在 AnyBot 關掉時被中斷了（{}；可能已執行的步驟 {} 個；放回聊天的訊息 {} 則）", + t.bot_id, + t.op, + if t.aborted { "已按停止" } else { "沒有按停止" }, + t.maybe_ran.len(), + t.restored + ); + } + } + + /// 「繼續」 on the card a crash left (here, on a screen of this brain, or on the phone): the + /// cut-off turn goes on from where its steps stand. A step that may have run is closed as + /// such (its result says so) and never runs again by itself; messages typed while it worked + /// were put back in the chat and are read now. Only the newest card, while nothing came + /// after it (a new message already carries the chat on). + pub(super) fn resume_turn(&mut self, ctx: &egui::Context, bot_id: u64) { + if self.mir.on { + self.mirror_act(serde_json::json!({"bot": bot_id, "action": "resume"})); + return; + } + if self.streams.contains_key(&bot_id) { + return; + } + if self.state.bot(bot_id).and_then(|b| b.messages.last()).map(|m| m.interrupted).unwrap_or(false) { + self.start_reply(ctx, bot_id); + } + } + /// Run due routines (a routine firing is a "wake" turn). fn tick_routines(&mut self, ctx: &egui::Context) { if self.demo || self.last_tick.elapsed() < std::time::Duration::from_secs(1) { return; } self.last_tick = Instant::now(); + self.resume_interrupted(ctx); let now = crate::routines::now(); let mut fire = vec![]; for b in self.state.bots.iter_mut() { @@ -1155,8 +1226,11 @@ impl App { } for (id, texts) in std::mem::take(&mut self.pending_turns) { if let Some(b) = self.state.bot_mut(id) { - for t in texts { - b.messages.push(Message::user(&t)); + for q in texts { + // (At the time it was typed: how a restart tells it is already in the chat.) + b.messages.push(Message { at: q.at, ..Message::user(&q.text) }); + self.delivered.push((id, q.id)); + self.save_now = true; } } if self.streams.contains_key(&id) { @@ -1214,7 +1288,16 @@ impl App { if let Some(live) = self.streams.get(&bot_id) { live.cancel.store(true, Ordering::Relaxed); // Stopping keeps what was typed but not yet read: it becomes the next turn only if sent again. - live.steer.lock().unwrap().clear(); + let dropped: Vec = std::mem::take(&mut *live.steer.lock().unwrap()); + // The stop is on disk at once (a crash before the turn's thread winds down still ends + // it as stopped), and the dropped messages won't come back after a restart. + if !self.demo { + let log = crate::oplog::OpLog::for_bot(bot_id); + log.append(&crate::oplog::Record::AbortRequested { op: live.turn_id.clone() }); + for q in dropped { + log.append(&crate::oplog::Record::QueueCancelled { id: q.id }); + } + } self.settle(bot_id, TurnOutcome::Cancelled); if let Some(b) = self.state.bot_mut(bot_id) { for c in b.messages.iter_mut().rev().take(6).flat_map(|m| m.tool_calls.iter_mut()) { @@ -1463,15 +1546,15 @@ impl App { let Some(b) = self.state.bot_mut(id) else { continue }; match ev { StreamEvent::Alive | StreamEvent::Retrying(_) | StreamEvent::NeedApproval(_) | StreamEvent::AppAction(_) | StreamEvent::React(_) | StreamEvent::Attach(_) => continue, - StreamEvent::Steered(m) => b.messages.push(m), + StreamEvent::Steered(m, qid) => { + b.messages.push(m); + // Delivered, once this chat is on disk (saved at the next frame). + self.delivered.push((id, qid)); + self.save_now = true; + } StreamEvent::NextStep(via) => b.messages.push(Message::assistant("", &via)), StreamEvent::ToolResult(m) => b.messages.push(m), - StreamEvent::Compacted(n, summary, boundary) => { - // The model's history can be shorter than the window's chat (group chats - // filter); land the cut at the real first-kept message, not the raw index. - let cut = b.messages.iter().position(|m| m.at == boundary && m.role == Role::User).unwrap_or(n.min(b.messages.len())); - b.messages.splice(..cut, [summary]); - } + StreamEvent::Compacted(n, summary, boundary) => crate::model::fold_in(&mut b.messages, n, summary, boundary), StreamEvent::ToolStatus(call_id, st) => { if let Some(c) = b.messages.iter_mut().rev().flat_map(|m| m.tool_calls.iter_mut()).find(|c| c.id == call_id) { c.status = st; @@ -2549,7 +2632,7 @@ impl App { }); }); ui.add_space(4.0); - let queued: Vec = self.streams.get(&bot.id).map(|l| l.steer.lock().unwrap().clone()).unwrap_or_default(); + let queued: Vec = self.streams.get(&bot.id).map(|l| l.steer.lock().unwrap().iter().map(|q| q.text.clone()).collect()).unwrap_or_default(); let notice = self.notice.clone(); ui.vertical_centered(|ui| { if let Some(n) = notice.filter(|_| queued.is_empty()) { @@ -3502,13 +3585,17 @@ impl App { } fn save_if_dirty(&mut self) { - if self.dirty && !self.demo && (self.streams.is_empty() || self.last_save.elapsed().as_secs() >= 2) { + if self.dirty && !self.demo && (self.streams.is_empty() || self.save_now || self.last_save.elapsed().as_secs() >= 2) { if self.mir.on { self.mirror_save(); - } else { - self.state.save(); + } else if self.state.save() { + // Typed messages now on disk in their chats: the work log can let them go. + for (bot, id) in std::mem::take(&mut self.delivered) { + crate::oplog::OpLog::for_bot(bot).append(&crate::oplog::Record::QueueConsumed { id }); + } } self.dirty = false; + self.save_now = false; self.last_save = Instant::now(); } } @@ -3888,7 +3975,7 @@ struct Live { /// Waiting for an approval or a running tool: the stall watchdog stays quiet. paused: bool, /// What the person typed during this turn, waiting for the next step. - steer: Arc>>, + steer: Arc>>, turn_id: String, /// 直接操作 for this turn (switched on mid-turn by "全部允許"). direct: Arc, @@ -3929,7 +4016,9 @@ pub fn message_max_width(container: f32) -> f32 { fn group_history(messages: &[Message], speaker: Option, st: &State) -> Vec { let mut out = vec![]; let mut skip_tool_ids: std::collections::HashSet = Default::default(); - for m in messages { + // The model reads from the newest summary on (the chat itself keeps what was folded), and + // never the 「繼續」 cards a crash left (they are for the person). + for m in messages[crate::model::model_view_start(messages)..].iter().filter(|m| !m.interrupted) { let mut m = m.clone(); if !m.attachments.is_empty() { let (extra, images) = crate::attach::for_model(&m.attachments); @@ -4408,6 +4497,21 @@ fn dialog_footer(ui: &mut egui::Ui, p: Palette, body: impl FnOnce(&mut egui::Ui) #[cfg(test)] mod tests { + /// The model reads a chat from its newest summary on, and never the 「繼續」 card. + #[test] + fn the_model_reads_from_the_fold_and_skips_the_card() { + let st = State::demo(); + let msgs = vec![ + Message::user("很久以前的問題"), + Message::assistant("很久以前的回答", "x"), + Message { summary: true, ..Message::assistant("摘要：問過以前的事", "x") }, + Message::user("最近的問題"), + Message { interrupted: true, error: true, ..Message::assistant("AnyBot 在工作時被關掉了…", "") }, + ]; + let h = group_history(&msgs, None, &st); + assert_eq!(h.iter().map(|m| m.text.as_str()).collect::>(), vec!["摘要：問過以前的事", "最近的問題"]); + } + #[test] fn group_members_see_each_other_as_labelled_words() { let st = State::demo(); diff --git a/src/app/chat_ui.rs b/src/app/chat_ui.rs index d28e3a3..fc9704b 100644 --- a/src/app/chat_ui.rs +++ b/src/app/chat_ui.rs @@ -172,6 +172,23 @@ impl App { }); ui.add_space(10.0); } + // Where AnyBot stopping cut a turn off: what it was doing and what may have run, + // with 「繼續」 while it is the newest thing in the chat. + _ if m.interrupted => { + let can_go_on = last && !self.streams.contains_key(&bot.id); + let mut go_on = false; + Frame::new().stroke(Stroke::new(1.0, p.border)).corner_radius(14).inner_margin(Margin::symmetric(12, 10)).show(ui, |ui| { + ui.label(RichText::new(&m.text).size(13.0).color(p.text2)); + if can_go_on { + ui.add_space(6.0); + go_on = ui.button(RichText::new(t("繼續")).size(13.0)).clicked(); + } + }); + ui.add_space(10.0); + if go_on { + self.resume_turn(ctx, bot.id); + } + } _ => self.bot_message(ui, ctx, bot, m, last, idx), } }); diff --git a/src/app/remote_ui.rs b/src/app/remote_ui.rs index bb25ca7..326dead 100644 --- a/src/app/remote_ui.rs +++ b/src/app/remote_ui.rs @@ -291,7 +291,7 @@ impl App { "threads": b.threads.iter().enumerate().rev().take(20).map(|(i, (title, msgs))| json!({"index": i, "title": title, "count": msgs.len(), "at": msgs.last().map(|m| m.at).unwrap_or(0)})).collect::>(), // The sidebar's sections, the colour by name (the editor), what is queued while it works. "folder": b.folder, "color_key": b.color, - "queued": live.map(|l| l.steer.lock().unwrap().clone()).unwrap_or_default(), + "queued": live.map(|l| l.steer.lock().unwrap().iter().map(|q| q.text.clone()).collect::>()).unwrap_or_default(), })); let start = b.messages.len().saturating_sub(remote::PHONE_TAIL); // Each message with its place in the chat (表情回應), its thinking and, on each step, @@ -375,9 +375,9 @@ impl App { Some(crate::attach::Attachment { image: crate::attach::is_image_name(&name), name, path: f.display().to_string(), bytes: meta.len(), on_box }) }) .collect(); - if let Some(l) = self.streams.get(&bot) { + if self.streams.contains_key(&bot) { let (extra, _) = crate::attach::for_model(&atts); - l.steer.lock().unwrap().push(format!("{text}{extra}")); + self.queue_steer(bot, format!("{text}{extra}")); return; } if let Some(b) = self.state.bot_mut(bot) { @@ -414,6 +414,7 @@ impl App { self.start_reply(ctx, bot); } } + Cmd::Resume { bot } => self.resume_turn(ctx, bot), Cmd::RestoreThread { bot, index } => { self.stop(bot); if let Some(b) = self.state.bot_mut(bot) { @@ -949,7 +950,7 @@ impl App { // model and computer, a new chat, regenerate) plus looking and small chat things; making, // changing or deleting bots, routines, memories, pages and settings stays the owner's. let admin_ok = match action { - "detail" | "settings" | "pages" | "restart_bot" | "react" | "reply" => true, + "detail" | "settings" | "pages" | "restart_bot" | "react" | "reply" | "resume" => true, "bot" => b["key"].as_str() == Some("unread"), "box" => matches!(b["op"].as_str(), Some("wake" | "disk")), _ => false, @@ -1108,6 +1109,10 @@ impl App { self.remote_cmd(ctx, Cmd::Reply { bot }); done("重新傳送中…") } + "resume" => { + self.remote_cmd(ctx, Cmd::Resume { bot }); + done("繼續中…") + } "clear" => { self.remote_cmd(ctx, Cmd::Clear { bot }); done("已清除聊天記錄") diff --git a/src/main.rs b/src/main.rs index a7f8a67..3fafa0c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -34,6 +34,7 @@ mod mcp; mod memory; mod model; mod notify; +mod oplog; mod phone; mod policy; mod preview; diff --git a/src/model.rs b/src/model.rs index 536b5cb..8ab244f 100644 --- a/src/model.rs +++ b/src/model.rs @@ -80,12 +80,37 @@ pub struct Message { /// When it was written (ms since 1970; 0 for messages from before this was kept). #[serde(default, skip_serializing_if = "is_zero_i64")] pub at: i64, + /// A card where AnyBot stopping cut a turn off (put there at the next start, see + /// oplog::recover), with 「繼續」 to go on from there. Never sent to the model. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub interrupted: bool, } fn is_zero_i64(n: &i64) -> bool { *n == 0 } +/// Where the model's view of a chat starts: the newest summary (folding keeps the person's chat +/// whole; the model reads from the fold on). +pub fn model_view_start(msgs: &[Message]) -> usize { + msgs.iter().rposition(|m| m.summary).unwrap_or(0) +} + +/// A summary the turn made (it folded the first `n` messages of its view; `boundary` is the time +/// of the first one it kept) goes into the chat at the fold. Nothing is removed: the person's +/// chat stays whole and the model reads from the newest summary on. The model's view can be +/// shorter than the chat (group chats filter), so the fold lands at the real first-kept message, +/// not the raw index. +pub fn fold_in(msgs: &mut Vec, n: usize, summary: Message, boundary: i64) { + let from = model_view_start(msgs); + let cut = msgs.iter().position(|m| m.at == boundary && m.role == Role::User).filter(|&i| i > from).unwrap_or((from + n).min(msgs.len())); + // Pictures in what was folded never go to the model again (and aren't saved anyway). + for m in &mut msgs[..cut] { + m.images.clear(); + } + msgs.insert(cut, summary); +} + pub fn now_ms() -> i64 { chrono::Utc::now().timestamp_millis() } @@ -455,7 +480,8 @@ impl State { s } - pub fn save(&self) { + /// Write everything to disk; true once it is there. + pub fn save(&self) -> bool { let p = Self::path(); if let Some(dir) = p.parent() { let _ = std::fs::create_dir_all(dir); @@ -467,10 +493,11 @@ impl State { if let Ok(mut f) = std::fs::File::create(&tmp) { use std::io::Write; if f.write_all(s.as_bytes()).and_then(|_| f.sync_all()).is_ok() { - let _ = std::fs::rename(&tmp, &p); + return std::fs::rename(&tmp, &p).is_ok(); } } } + false } pub fn provider(&self, kind: ProviderKind) -> ProviderSettings { @@ -765,6 +792,30 @@ pub fn bot_computer_dir(bot_id: u64) -> PathBuf { mod start_tests { use super::*; + /// Folding keeps the person's chat whole: each summary goes in at its fold, and the model's + /// view starts at the newest one. + #[test] + fn folding_keeps_every_message() { + let mut chat: Vec = (0..10).map(|i| Message { at: i + 1, ..if i % 2 == 0 { Message::user(&format!("問 {i}")) } else { Message::assistant(&format!("答 {i}"), "x") } }).collect(); + let sum = |t: &str, at| Message { summary: true, at, ..Message::assistant(t, "x") }; + // The turn folded the first 4 messages; the first it kept is 「問 4」 (at 5). + fold_in(&mut chat, 4, sum("摘要一", 100), 5); + assert_eq!(chat.len(), 11); + assert!(chat[4].summary && chat[5].text == "問 4"); + assert_eq!(model_view_start(&chat), 4); + // Again later: the model's view started at 「摘要一」; it folded 3 of it (the summary, + // 問 4, 答 5) and kept from 「問 6」 (at 7). + fold_in(&mut chat, 3, sum("摘要二", 200), 7); + assert_eq!(chat.len(), 12); + assert_eq!(chat.iter().filter(|m| m.summary).count(), 2); + assert_eq!(model_view_start(&chat), 7); + assert_eq!(chat[8].text, "問 6"); + assert_eq!(chat.iter().filter(|m| !m.summary).count(), 10, "nothing the person saw was removed"); + // A boundary the chat can't find: the fold lands by count within the model's view. + fold_in(&mut chat, 2, sum("摘要三", 300), 9999); + assert!(chat[9].summary && chat[9].text == "摘要三"); + } + #[test] fn a_new_install_has_no_bots() { let mut s = State::first_run(); diff --git a/src/oplog.rs b/src/oplog.rs new file mode 100644 index 0000000..aaa82df --- /dev/null +++ b/src/oplog.rs @@ -0,0 +1,661 @@ +// 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()); + } +} diff --git a/src/providers.rs b/src/providers.rs index f331df6..63110a4 100644 --- a/src/providers.rs +++ b/src/providers.rs @@ -192,7 +192,8 @@ pub enum StreamEvent { /// `boundary` the timestamp of the first kept message (the window's own copy may be longer). Compacted(usize, Message, i64), /// A message the person sent mid-turn was handed to the model (add it to the chat). - Steered(Message), + /// (With the id of the typed message, for the work log.) + Steered(Message, String), /// Do something in the app (routines, bots, settings) and answer on the sender. AppAction(AppCall), /// 表情回應: the bot put an emoji on the person's latest message. @@ -343,6 +344,10 @@ fn page_words(body: &str) -> String { /// waited out and tried again (it came through raw English, as the reply, and was never retried). fn stream_failure(msg: &str) -> Failure { let low = msg.to_lowercase(); + // The same overflow, reported inside the stream instead of as an HTTP 400. + if ["prompt is too long", "maximum context", "context length", "context window", "too many tokens"].iter().any(|k| low.contains(k)) { + return Failure::fatal(format!("{CONTEXT_OVERFLOW}請清除聊天記錄或換一個模型。\n（{msg}）")); + } if ["capacity", "overloaded", "rate limit", "too many requests", "try again", "temporarily", "unavailable", "timeout", "timed out"].iter().any(|k| low.contains(k)) { Failure { msg: format!("模型供應商現在太忙，正在等一下再試。\n（{msg}）"), retryable: true, retry_after_ms: Some(BUSY_RETRY_MS) } } else { @@ -353,6 +358,16 @@ fn stream_failure(msg: &str) -> Failure { /// Wait before asking a busy model again (estimated; it says "a few minutes", 1–4 s was too soon). const BUSY_RETRY_MS: u64 = 8_000; +/// How a "the conversation no longer fits this model" failure starts (SAND-E0404 +/// contextWindowOverflow). The turn loop recognizes it by this and folds older messages into a +/// summary before asking once more, instead of giving up. +pub const CONTEXT_OVERFLOW: &str = "對話太長了，超過這個模型能讀的長度。"; + +/// The request didn't fit the model's context (a retry as-is would fail the same way). +pub fn is_context_overflow(msg: &str) -> bool { + msg.starts_with(CONTEXT_OVERFLOW) +} + fn http_error(status: u16, body: &str, retry_after: Option) -> Failure { let detail = serde_json::from_str::(body) .ok() @@ -373,7 +388,7 @@ fn http_error(status: u16, body: &str, retry_after: Option) -> Failure { let lower = detail.to_lowercase(); // SAND-E0404 contextWindowOverflow: a retry would repeat it. if (status == 400 || status == 413) && ["context", "too long", "too many tokens", "maximum context", "prompt is too long"].iter().any(|k| lower.contains(k)) { - return Failure::fatal(format!("對話太長了，超過這個模型能讀的長度。請清除聊天記錄或換一個模型。\n（{detail}）")); + return Failure::fatal(format!("{CONTEXT_OVERFLOW}請清除聊天記錄或換一個模型。\n（{detail}）")); } if status == 400 && lower.contains("tool") && (lower.contains("support") || lower.contains("not available")) { return Failure::fatal(format!("這個模型不支援工具。請換一個支援工具的模型，或在 Bot 設定裡關掉「電腦操作」。\n（{detail}）")); @@ -1302,6 +1317,18 @@ mod tests { assert!(!http_error(400, "{\"error\":{\"message\":\"prompt is too long\"}}", None).retryable); } + /// A conversation too long for the model is told apart from other failures (the turn folds + /// it into a summary and asks again), whether it comes as an HTTP 400 or inside the stream. + #[test] + fn an_overflow_is_recognized_as_one() { + assert!(is_context_overflow(&http_error(400, "{\"error\":{\"message\":\"This model's maximum context length is 128000 tokens\"}}", None).msg)); + assert!(is_context_overflow(&http_error(413, "{\"error\":{\"message\":\"prompt is too long: 210000 tokens > 200000 maximum\"}}", None).msg)); + assert!(is_context_overflow(&stream_failure("prompt is too long: 210000 tokens > 200000 maximum").msg)); + assert!(!is_context_overflow(&http_error(401, "{}", None).msg)); + assert!(!is_context_overflow(&http_error(400, "{\"error\":{\"message\":\"tools are not supported\"}}", None).msg)); + assert!(!is_context_overflow(&stream_failure("overloaded").msg)); + } + #[test] fn openai_style_stream() { let body = "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"hm\"}}]}\n\ndata: {\"choices\":[{\"delta\":{\"content\":\"你好\"}}]}\n\ndata: {\"choices\":[{\"delta\":{\"content\":\"！\"}}]}\n\ndata: [DONE]\n\n"; diff --git a/src/remote.rs b/src/remote.rs index f1ef2ca..2694e13 100644 --- a/src/remote.rs +++ b/src/remote.rs @@ -128,6 +128,8 @@ pub enum Cmd { /// Answer the last message again (重新產生) / reply to what is there (重新傳送). Regenerate { bot: u64 }, Reply { bot: u64 }, + /// 「繼續」 on the card a crash left: the cut-off turn goes on. + Resume { bot: u64 }, /// Bring back an earlier conversation (過去的對話). RestoreThread { bot: u64, index: usize }, /// 清除聊天記錄 from a screen of this brain (the chat lives here). @@ -260,7 +262,7 @@ pub fn tool_label(name: &str) -> &'static str { } } -fn tool_detail(name: &str, args: &str) -> String { +pub(crate) fn tool_detail(name: &str, args: &str) -> String { let d = if crate::appctl::is_app_tool(name) { crate::appctl::describe(name, args) } else if crate::computer::is_computer_tool(name) { @@ -306,6 +308,7 @@ pub fn message_json(m: &crate::model::Message, names: &HashMap, pag "via": m.via, "error": m.error, "summary": m.summary, + "interrupted": m.interrupted, "wake": m.wake, "author": if m.author == 0 { String::new() } else { names.get(&m.author).cloned().unwrap_or_default() }, "reactions": m.reactions, @@ -548,6 +551,7 @@ pub fn handle(req: Req, via_relay: bool) -> Resp { "new" => Cmd::NewThread { bot }, "regenerate" => Cmd::Regenerate { bot }, "reply" => Cmd::Reply { bot }, + "resume" => Cmd::Resume { bot }, "restore" => Cmd::RestoreThread { bot, index: b["index"].as_u64().unwrap_or(0) as usize }, "clear" => Cmd::Clear { bot }, _ => return Resp::error(404, "沒有這個動作"),