daniel.w
/
AnyBot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

多台電腦（大腦、手腳、螢幕）、一行加電腦、文件與 make #2

Merged
daniel.w merged 9 commits from daniel.w/multi-device into main 2026-11-04 14:40:38 +00:00
Conversation 0 Commits 9 Files Changed 46 +321 -54
4 changed files with 321 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a70d29fa51 - Show all commits

19
relay/src/main.rs
View File

@ -33,6 +33,8 @@ const ICON: &str = r##"<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 96 9
const HOST_POLL_SECS: u64 = 25; const HOST_POLL_SECS: u64 = 25;
/// How long a phone's request waits for the AnyBot to answer (its own long-poll is 20 s). /// How long a phone's request waits for the AnyBot to answer (its own long-poll is 20 s).
const FORWARD_SECS: u64 = 35; const FORWARD_SECS: u64 = 35;
/// How long a request may wait to be picked up before the host counts as gone.
const PICKUP_SECS: u64 = 8;
/// A host counts as online when it asked within this many seconds. /// A host counts as online when it asked within this many seconds.
const ONLINE_SECS: u64 = 60; const ONLINE_SECS: u64 = 60;
/// Sessions last this long without use. /// Sessions last this long without use.
@ -229,7 +231,20 @@ impl Relay {
let job = json!({"id": id, "method": method, "path": path, "query": query, "body": base64::engine::general_purpose::STANDARD.encode(body), "user": who.0, "role": who.1}); let job = json!({"id": id, "method": method, "path": path, "query": query, "body": base64::engine::general_purpose::STANDARD.encode(body), "user": who.0, "role": who.1});
self.queues.lock().unwrap().entry(host_id.to_string()).or_default().push_back(job); self.queues.lock().unwrap().entry(host_id.to_string()).or_default().push_back(job);
self.arrived.notify_all(); self.arrived.notify_all();
let r = rx.recv_timeout(Duration::from_secs(FORWARD_SECS)); // A running AnyBot always has a worker waiting, so it picks a request up within a moment.
// Still in the queue after a few seconds means it went away (closed, asleep, offline).
let mut r = rx.recv_timeout(Duration::from_secs(PICKUP_SECS));
if r.is_err() {
let queued = self.queues.lock().unwrap().get(host_id).map(|q| q.iter().any(|j| j["id"].as_str() == Some(id.as_str()))).unwrap_or(false);
if queued {
self.waiting.lock().unwrap().remove(&id);
if let Some(q) = self.queues.lock().unwrap().get_mut(host_id) {
q.retain(|j| j["id"].as_str() != Some(id.as_str()));
}
return Resp::error(503, "那台 AnyBot 目前離線（電腦可能關機、睡眠、斷網，或 AnyBot 沒開）");
}
r = rx.recv_timeout(Duration::from_secs(FORWARD_SECS - PICKUP_SECS));
}
self.waiting.lock().unwrap().remove(&id); self.waiting.lock().unwrap().remove(&id);
// Not picked up in time: drop it so it doesn't run late. // Not picked up in time: drop it so it doesn't run late.
if let Some(q) = self.queues.lock().unwrap().get_mut(host_id) { if let Some(q) = self.queues.lock().unwrap().get_mut(host_id) {
@ -249,7 +264,7 @@ impl Relay {
} }
resp resp
} }
Err(_) => Resp::error(504, "那台 AnyBot 沒有及時回應"), Err(_) => Resp::error(504, "那台 AnyBot 收到了，但太久沒有回應（可能太忙，或網路很慢）"),
} }
} }

290
src/app/mirror_ui.rs
View File

@ -167,6 +167,66 @@ pub(super) fn diff_ops(base: &Value, cur: &Value) -> Vec<Value> {
ops ops
} }
/// Replay edit operations on settings JSON: a screen keeps its own not-yet-confirmed edits on
/// top of what the brain last sent, so typing isn't undone by the brain's slightly older copy.
pub(super) fn apply_ops(cfg: &mut Value, ops: &[Value]) {
let id = |x: &Value| x["id"].as_u64().unwrap_or(0);
for op in ops {
let target = op["id"].as_u64().unwrap_or(0);
match op["op"].as_str().unwrap_or("") {
"set" => {
if let Some(k) = op["key"].as_str() {
cfg[k] = op["value"].clone();
}
}
"bot_add" => {
if let Some(a) = cfg["bots"].as_array_mut() {
if !a.iter().any(|b| id(b) == id(&op["bot"])) {
a.push(op["bot"].clone());
}
}
}
"bot_del" => {
if let Some(a) = cfg["bots"].as_array_mut() {
a.retain(|b| id(b) != target);
}
}
"bot_set" | "bot_read" => {
if let Some(b) = cfg["bots"].as_array_mut().and_then(|a| a.iter_mut().find(|b| id(b) == target)) {
if op["op"] == "bot_read" {
b["unread"] = json!(false);
} else if let Some(k) = op["key"].as_str() {
b[k] = op["value"].clone();
}
}
}
"bot_order" => {
let ids: Vec<u64> = op["ids"].as_array().into_iter().flatten().filter_map(Value::as_u64).collect();
if let Some(a) = cfg["bots"].as_array_mut() {
a.sort_by_key(|b| ids.iter().position(|i| *i == id(b)).unwrap_or(usize::MAX));
}
}
_ => {}
}
}
}
/// A schedule edited on a screen keeps what the brain knows about runs (when it last ran, when it
/// runs next), so an older copy can't make a routine run twice.
fn keep_run_times(new: &mut Value, old: &Value) {
for r in new.as_array_mut().into_iter().flatten() {
let Some(o) = old.as_array().and_then(|a| a.iter().find(|o| o["id"] == r["id"])) else { continue };
let last = |x: &Value| x["last_run"].as_i64();
if last(o) > last(r) || r["last_run"].is_null() {
r["last_run"] = o["last_run"].clone();
}
let same = ["schedule", "trigger", "paused"].iter().all(|k| r[*k] == o[*k]);
if same {
r["next_run"] = o["next_run"].clone();
}
}
}
// ---------- the brain's side ---------- // ---------- the brain's side ----------
/// What the brain last sent each screen. /// What the brain last sent each screen.
@ -176,6 +236,8 @@ pub struct Screens {
/// The settings fingerprint, refreshed now and then while something changes. /// The settings fingerprint, refreshed now and then while something changes.
pub(super) cfg_hash: u64, pub(super) cfg_hash: u64,
cfg_at: Option<Instant>, cfg_at: Option<Instant>,
/// The last edit (by number) each screen sent, so it knows which of its edits are in.
acked: HashMap<String, u64>,
} }
#[derive(Default)] #[derive(Default)]
@ -194,6 +256,14 @@ pub enum Incoming {
Offline(String), Offline(String),
} }
/// What a screen sends to the brain, one at a time and in order.
enum Out {
Edit { seq: u64, ops: Vec<Value> },
Act(Value),
Send { bot: u64, text: String, atts: Vec<crate::attach::Attachment> },
Move { ops: Vec<Value>, n: usize },
}
#[derive(Default)] #[derive(Default)]
pub struct Mirror { pub struct Mirror {
/// This window shows a brain (instead of its own bots). /// This window shows a brain (instead of its own bots).
@ -218,6 +288,16 @@ pub struct Mirror {
pub(super) leave: bool, pub(super) leave: bool,
pub(super) moving: Arc<AtomicBool>, pub(super) moving: Arc<AtomicBool>,
pub(super) move_note: Arc<Mutex<String>>, pub(super) move_note: Arc<Mutex<String>>,
/// Requests to the brain, sent in order by one worker.
outbox: Option<Sender<Out>>,
/// Edits sent but not yet confirmed (by number).
pending: Arc<Mutex<Vec<(u64, Vec<Value>)>>>,
seq: u64,
/// Something didn't reach the brain: ask for everything again.
resync: Arc<AtomicBool>,
/// Why the brain can't be reached (shown above the chat), and since when.
pub(super) reason: String,
down_since: Option<Instant>,
} }
fn cache_path() -> PathBuf { fn cache_path() -> PathBuf {
@ -271,7 +351,8 @@ impl App {
} }
seen.at = Some(Instant::now()); seen.at = Some(Instant::now());
seen.version = version; seen.version = version;
let mut out = json!({"version": version, "brain": self.state.remote.host_name}); let acked = self.scr.acked.get(client).copied().unwrap_or(0);
let mut out = json!({"version": version, "brain": self.state.remote.host_name, "edit_seq": acked});
let mut c = config_json(&mut self.state); let mut c = config_json(&mut self.state);
redact(&mut c); redact(&mut c);
let ch = hash_value(&c); let ch = hash_value(&c);
@ -320,6 +401,7 @@ impl App {
/// A screen's edits. /// A screen's edits.
pub(super) fn brain_edit(&mut self, body: &Value) -> Value { pub(super) fn brain_edit(&mut self, body: &Value) -> Value {
let mut applied = 0; let mut applied = 0;
let total = body["ops"].as_array().map(|a| a.len()).unwrap_or(0);
for op in body["ops"].as_array().into_iter().flatten() { for op in body["ops"].as_array().into_iter().flatten() {
let ok = match op["op"].as_str().unwrap_or("") { let ok = match op["op"].as_str().unwrap_or("") {
"set" => self.brain_set_top(op["key"].as_str().unwrap_or(""), op["value"].clone()), "set" => self.brain_set_top(op["key"].as_str().unwrap_or(""), op["value"].clone()),
@ -367,6 +449,15 @@ impl App {
applied += 1; applied += 1;
} }
} }
if let (Some(client), Some(seq)) = (body["client"].as_str(), body["seq"].as_u64()) {
self.scr.acked.insert(client.to_string(), seq);
// Something was refused: send that screen the settings again, so it drops it.
if applied < total {
if let Some(s) = self.scr.seen.get_mut(client) {
s.config = 0;
}
}
}
if applied > 0 { if applied > 0 {
self.mark(); self.mark();
// Screens hear of it at once. // Screens hear of it at once.
@ -436,6 +527,10 @@ impl App {
let msgs = std::mem::take(&mut b.messages); let msgs = std::mem::take(&mut b.messages);
let threads = std::mem::take(&mut b.threads); let threads = std::mem::take(&mut b.threads);
let mut v = serde_json::to_value(&*b).unwrap_or(Value::Null); let mut v = serde_json::to_value(&*b).unwrap_or(Value::Null);
let mut value = value;
if key == "routines" {
keep_run_times(&mut value, &v["routines"]);
}
v[key] = value; v[key] = value;
let ok = match serde_json::from_value::<Bot>(v) { let ok = match serde_json::from_value::<Bot>(v) {
Ok(nb) => { Ok(nb) => {
@ -482,8 +577,12 @@ impl App {
self.mir.on = true; self.mir.on = true;
self.mir.online = false; self.mir.online = false;
self.mir.version.store(0, Ordering::SeqCst); self.mir.version.store(0, Ordering::SeqCst);
self.mir.pending.lock().unwrap().clear();
self.mir.reason.clear();
self.mir.down_since = None;
*self.mir.status.lock().unwrap() = "正在連到大腦…".into(); *self.mir.status.lock().unwrap() = "正在連到大腦…".into();
self.mirror_worker(); self.mirror_worker();
self.mirror_outbox();
} }
/// Back to this computer's own bots. /// Back to this computer's own bots.
@ -491,6 +590,7 @@ impl App {
self.mir.gen.fetch_add(1, Ordering::SeqCst); self.mir.gen.fetch_add(1, Ordering::SeqCst);
self.mir.on = false; self.mir.on = false;
self.mir.online = false; self.mir.online = false;
self.mir.outbox = None;
self.streams.clear(); self.streams.clear();
self.approvals.clear(); self.approvals.clear();
if let Some(mut local) = self.mir.local.take() { if let Some(mut local) = self.mir.local.take() {
@ -536,12 +636,13 @@ impl App {
let (tx, rx) = channel(); let (tx, rx) = channel();
self.mir.rx = Some(rx); self.mir.rx = Some(rx);
self.mir.tx = Some(tx.clone()); self.mir.tx = Some(tx.clone());
let (g, version, client, s) = (self.mir.gen.clone(), self.mir.version.clone(), self.mir.client.clone(), self.state.remote.clone()); let (g, version, client, s, resync) = (self.mir.gen.clone(), self.mir.version.clone(), self.mir.client.clone(), self.state.remote.clone(), self.mir.resync.clone());
let repaint = self.repaint_fn(); let repaint = self.repaint_fn();
std::thread::spawn(move || { std::thread::spawn(move || {
let mut backoff = 1u64; let mut backoff = 1u64;
while g.load(Ordering::SeqCst) == gen { while g.load(Ordering::SeqCst) == gen {
match remote::sync_pull(&s, version.load(Ordering::SeqCst), &client) { let since = if resync.swap(false, Ordering::SeqCst) { 0 } else { version.load(Ordering::SeqCst) };
match remote::sync_pull(&s, since, &client) {
Ok(v) => { Ok(v) => {
backoff = 1; backoff = 1;
if g.load(Ordering::SeqCst) != gen { if g.load(Ordering::SeqCst) != gen {
@ -601,15 +702,21 @@ impl App {
Incoming::Offline(e) => { Incoming::Offline(e) => {
self.mir.online = false; self.mir.online = false;
self.mir.base = Value::Null; self.mir.base = Value::Null;
*self.mir.status.lock().unwrap() = format!("連不到大腦「{}」：{e}。現在只能看，連上後會自動恢復。", self.state.remote.brain_name); self.mir.down_since.get_or_insert_with(Instant::now);
*self.mir.status.lock().unwrap() = format!("連不到大腦：{e}");
self.mir.reason = e;
} }
Incoming::Pull(v) => { Incoming::Pull(v) => {
if let Some(e) = v["error"].as_str() { if let Some(e) = v["error"].as_str() {
self.mir.online = false; self.mir.online = false;
self.mir.down_since.get_or_insert_with(Instant::now);
*self.mir.status.lock().unwrap() = e.to_string(); *self.mir.status.lock().unwrap() = e.to_string();
self.mir.reason = e.to_string();
continue; continue;
} }
self.mir.online = true; self.mir.online = true;
self.mir.down_since = None;
self.mir.reason.clear();
*self.mir.status.lock().unwrap() = format!("已連到大腦「{}」", v["brain"].as_str().unwrap_or(&self.state.remote.brain_name)); *self.mir.status.lock().unwrap() = format!("已連到大腦「{}」", v["brain"].as_str().unwrap_or(&self.state.remote.brain_name));
self.mirror_apply(v); self.mirror_apply(v);
} }
@ -618,8 +725,21 @@ impl App {
} }
fn mirror_apply(&mut self, v: Value) { fn mirror_apply(&mut self, v: Value) {
// Edits the brain has taken in are no longer "on the way".
let acked = v["edit_seq"].as_u64().unwrap_or(0);
self.mir.pending.lock().unwrap().retain(|(s, _)| *s > acked);
if v["config"].is_object() { if v["config"].is_object() {
let cfg = v["config"].clone(); // What the brain will have once the edits on the way arrive, and on top of that what
// the person changed here and hasn't sent yet.
let inflight: Vec<Value> = self.mir.pending.lock().unwrap().iter().flat_map(|(_, o)| o.clone()).collect();
let unsaved = if self.mir.base.is_object() { diff_ops(&self.mir.base, &config_json(&mut self.state)) } else { vec![] };
let mut base = v["config"].clone();
apply_ops(&mut base, &inflight);
let mut cfg = base.clone();
apply_ops(&mut cfg, &unsaved);
if !unsaved.is_empty() {
self.dirty = true;
}
let local = std::mem::replace(&mut self.state, State::first_run()); let local = std::mem::replace(&mut self.state, State::first_run());
match serde_json::from_value::<State>(cfg.clone()) { match serde_json::from_value::<State>(cfg.clone()) {
Ok(mut st) => { Ok(mut st) => {
@ -633,7 +753,7 @@ impl App {
} }
self.state = st; self.state = st;
self.keep_local_fields(&local); self.keep_local_fields(&local);
self.mir.base = cfg; self.mir.base = base;
} }
Err(e) => { Err(e) => {
self.state = local; self.state = local;
@ -738,10 +858,64 @@ impl App {
return; return;
} }
self.mir.base = cur; self.mir.base = cur;
let (s, err) = (self.state.remote.clone(), self.mir.error.clone()); self.mir.seq += 1;
let seq = self.mir.seq;
self.mir.pending.lock().unwrap().push((seq, ops.clone()));
self.mirror_out(Out::Edit { seq, ops });
}
fn mirror_out(&mut self, o: Out) {
if let Some(tx) = &self.mir.outbox {
let _ = tx.send(o);
}
}
/// One worker sends everything to the brain in order (a rename before the message that uses
/// it, two quick edits in the order they were made).
fn mirror_outbox(&mut self) {
let (tx, rx) = channel::<Out>();
self.mir.outbox = Some(tx);
let (s, err, client, pending, resync) = (self.state.remote.clone(), self.mir.error.clone(), self.mir.client.clone(), self.mir.pending.clone(), self.mir.resync.clone());
let (note, busy) = (self.mir.move_note.clone(), self.mir.moving.clone());
let repaint = self.repaint_fn();
std::thread::spawn(move || { std::thread::spawn(move || {
if let Err(e) = remote::sync_post(&s, "edit", &json!({"ops": ops})) { for o in rx {
*err.lock().unwrap() = format!("改動沒有存到大腦：{e}"); match o {
Out::Edit { seq, ops } => {
if let Err(e) = remote::sync_post(&s, "edit", &json!({"ops": ops, "client": client, "seq": seq})) {
*err.lock().unwrap() = format!("剛才的改動沒有存到大腦：{e}");
pending.lock().unwrap().retain(|(n, _)| *n != seq);
resync.store(true, Ordering::SeqCst);
}
}
Out::Act(body) => {
if let Err(e) = remote::sync_post(&s, "act", &body) {
*err.lock().unwrap() = format!("大腦沒有收到：{e}");
}
}
Out::Send { bot, text, atts } => {
let mut paths = vec![];
for a in &atts {
match std::fs::read(&a.path).map_err(|e| format!("讀不到這個檔案（{e}）")).and_then(|b| remote::sync_upload(&s, bot, &a.name, &b)) {
Ok(p) => paths.push(p),
Err(e) => *err.lock().unwrap() = format!("「{}」沒有傳到大腦：{e}", a.name),
}
}
if let Err(e) = remote::sync_post(&s, "act", &json!({"bot": bot, "action": "send", "text": text, "attachments": paths})) {
*err.lock().unwrap() = format!("訊息沒有送到大腦：{e}");
// Take back the message shown ahead of time.
resync.store(true, Ordering::SeqCst);
}
}
Out::Move { ops, n } => {
*note.lock().unwrap() = match remote::sync_post(&s, "edit", &json!({"ops": ops})) {
Ok(v) => format!("已把 {n} 個 Bot 搬到大腦（{} 項）", v["applied"].as_u64().unwrap_or(0)),
Err(e) => format!("搬移失敗：{e}"),
};
busy.store(false, Ordering::SeqCst);
}
}
repaint();
} }
}); });
} }
@ -752,12 +926,7 @@ impl App {
*self.mir.error.lock().unwrap() = "連不到大腦，現在只能看。".into(); *self.mir.error.lock().unwrap() = "連不到大腦，現在只能看。".into();
return; return;
} }
let (s, err) = (self.state.remote.clone(), self.mir.error.clone()); self.mirror_out(Out::Act(body));
std::thread::spawn(move || {
if let Err(e) = remote::sync_post(&s, "act", &body) {
*err.lock().unwrap() = format!("大腦沒有收到：{e}");
}
});
} }
/// Send a message through the brain: files go up first, then the text. /// Send a message through the brain: files go up first, then the text.
@ -772,21 +941,11 @@ impl App {
b.messages.push(Message { attachments: atts.clone(), ..Message::user(&text) }); b.messages.push(Message { attachments: atts.clone(), ..Message::user(&text) });
} }
} }
let (s, err) = (self.state.remote.clone(), self.mir.error.clone()); // Settings changed just before go first.
std::thread::spawn(move || { if self.dirty {
let mut paths = vec![]; self.mirror_save();
for a in &atts {
match std::fs::read(&a.path).map_err(|e| e.to_string()).and_then(|b| remote::sync_upload(&s, bot, &a.name, &b)) {
Ok(p) => paths.push(p),
Err(e) => {
*err.lock().unwrap() = format!("「{}」沒有傳到大腦：{e}", a.name);
} }
} self.mirror_out(Out::Send { bot, text, atts });
}
if let Err(e) = remote::sync_post(&s, "act", &json!({"bot": bot, "action": "send", "text": text, "attachments": paths})) {
*err.lock().unwrap() = format!("訊息沒有送到大腦：{e}");
}
});
true true
} }
@ -807,15 +966,9 @@ impl App {
for k in ["providers", "skills", "mcp_servers"] { for k in ["providers", "skills", "mcp_servers"] {
ops.push(json!({"op": "merge", "key": k, "value": top[k]})); ops.push(json!({"op": "merge", "key": k, "value": top[k]}));
} }
let (s, note, busy, n) = (self.state.remote.clone(), self.mir.move_note.clone(), self.mir.moving.clone(), local.bots.len()); *self.mir.move_note.lock().unwrap() = "搬移中…".into();
*note.lock().unwrap() = "搬移中…".into(); let n = local.bots.len();
std::thread::spawn(move || { self.mirror_out(Out::Move { ops, n });
*note.lock().unwrap() = match remote::sync_post(&s, "edit", &json!({"ops": ops})) {
Ok(v) => format!("已把 {n} 個 Bot 搬到大腦（{} 項）", v["applied"].as_u64().unwrap_or(0)),
Err(e) => format!("搬移失敗：{e}"),
};
busy.store(false, Ordering::SeqCst);
});
} }
/// The bar above the chat while showing a brain. /// The bar above the chat while showing a brain.
@ -831,11 +984,25 @@ impl App {
if self.mir.online { if self.mir.online {
return; return;
} }
let status = self.mir.status.lock().unwrap().clone(); // A blip (the account server restarting) isn't worth a warning.
let connecting = self.mir.down_since.is_none();
let (title, reason, tail) = match self.mir.down_since {
None => (format!("正在連到大腦「{}」…", self.state.remote.brain_name), String::new(), String::new()),
Some(t) if t.elapsed().as_secs() < 4 => return,
Some(_) => (format!("連不到大腦「{}」", self.state.remote.brain_name), self.mir.reason.clone(), t("之前的對話還能看，但現在不能傳訊息。連回來後會自動恢復。")),
};
ui.ctx().request_repaint_after(std::time::Duration::from_secs(1));
ui.vertical_centered(|ui| { ui.vertical_centered(|ui| {
ui.set_max_width(TRANSCRIPT_MAX_WIDTH.min(ui.available_width() - 48.0)); ui.set_max_width(TRANSCRIPT_MAX_WIDTH.min(ui.available_width() - 48.0));
Frame::new().fill(p.danger.gamma_multiply(0.08)).corner_radius(8).inner_margin(Margin::symmetric(12, 8)).show(ui, |ui| { Frame::new().fill(if connecting { p.hover } else { p.danger.gamma_multiply(0.08) }).corner_radius(8).inner_margin(Margin::symmetric(12, 8)).show(ui, |ui| {
ui.add(egui::Label::new(RichText::new(status).size(FONT_SM).color(p.text2)).wrap()); ui.vertical(|ui| {
ui.label(RichText::new(title).size(FONT_SM).strong().color(p.text));
for line in [reason, tail] {
if !line.is_empty() {
ui.add(egui::Label::new(RichText::new(line).size(FONT_SM).color(p.text2)).wrap());
}
}
});
}); });
}); });
ui.add_space(6.0); ui.add_space(6.0);
@ -879,4 +1046,45 @@ mod tests {
unredact(&mut ch, &old["channels"]); unredact(&mut ch, &old["channels"]);
assert_eq!(ch[0]["token"], "t"); assert_eq!(ch[0]["token"], "t");
} }
#[test]
fn edits_replay_on_a_newer_copy() {
let base = json!({"folders": ["a"], "bots": [{"id": 1, "name": "A", "unread": true}, {"id": 2, "name": "B"}]});
let cur = json!({"folders": ["a", "b"], "bots": [{"id": 2, "name": "B"}, {"id": 1, "name": "A2", "unread": false}, {"id": 3, "name": "C"}]});
let ops = diff_ops(&base, &cur);
let mut again = base.clone();
apply_ops(&mut again, &ops);
assert_eq!(again, cur);
// The brain meanwhile renamed bot 2: the screen's own edit stays, the brain's shows too.
let mut newer = json!({"folders": ["a"], "bots": [{"id": 1, "name": "A", "unread": true}, {"id": 2, "name": "B from brain"}]});
apply_ops(&mut newer, &ops);
assert_eq!(newer["bots"][1]["name"], "A2");
assert_eq!(newer["bots"][0]["name"], "B from brain");
assert_eq!(newer["folders"], json!(["a", "b"]));
}
#[test]
fn an_old_copy_cant_make_a_routine_run_twice() {
let brain = json!([{"id": 7, "schedule": "daily", "trigger": null, "paused": false, "last_run": 200, "next_run": 300}]);
let mut edit = json!([{"id": 7, "title": "new", "schedule": "daily", "trigger": null, "paused": false, "last_run": 100, "next_run": 150}]);
keep_run_times(&mut edit, &brain);
assert_eq!(edit[0]["last_run"], 200);
assert_eq!(edit[0]["next_run"], 300);
assert_eq!(edit[0]["title"], "new");
// A new schedule gets worked out again.
let mut moved = json!([{"id": 7, "schedule": "weekly", "trigger": null, "paused": false, "last_run": 100, "next_run": null}]);
keep_run_times(&mut moved, &brain);
assert_eq!(moved[0]["last_run"], 200);
assert!(moved[0]["next_run"].is_null());
}
#[test]
fn failures_read_as_words_not_codes() {
let w = remote::why_refused(502, &Value::Null, "大腦");
assert!(!w.contains("502") && w.contains("帳號伺服器"));
let off = remote::why_refused(503, &json!({"error": "那台 AnyBot 目前離線（電腦關機、睡眠，或 AnyBot 沒開）"}), "大腦");
assert!(off.starts_with("大腦目前離線"));
assert!(remote::why_refused(401, &Value::Null, "大腦").contains("重新登入"));
assert!(remote::why_unreachable(&ureq::Error::HostNotFound).contains("網路"));
}
} }

2
src/app/remote_ui.rs
View File

@ -152,7 +152,7 @@ impl App {
for b in &self.state.bots { for b in &self.state.bots {
(b.id, &b.name, &b.color, b.unread, b.messages.len()).hash(&mut h); (b.id, &b.name, &b.color, b.unread, b.messages.len()).hash(&mut h);
if let Some(m) = b.messages.last() { if let Some(m) = b.messages.last() {
(m.text.len(), m.tool_calls.len(), m.error).hash(&mut h); (m.text.len(), m.thinking.len(), m.tool_calls.len(), m.error).hash(&mut h);
for c in &m.tool_calls { for c in &m.tool_calls {
(c.status as u8).hash(&mut h); (c.status as u8).hash(&mut h);
} }

62
src/remote.rs
View File

@ -174,7 +174,7 @@ impl Hub {
if let Some(r) = self.repaint.lock().unwrap().as_ref() { if let Some(r) = self.repaint.lock().unwrap().as_ref() {
r(); r();
} }
rx.recv_timeout(Duration::from_secs(10)).map_err(|_| "大腦沒有及時回應".to_string()) rx.recv_timeout(Duration::from_secs(10)).map_err(|_| "大腦太忙，沒有及時回應".to_string())
} }
pub fn version(&self) -> u64 { pub fn version(&self) -> u64 {
self.view.lock().unwrap().version self.view.lock().unwrap().version
@ -646,17 +646,56 @@ fn agent(secs: u64) -> ureq::Agent {
ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(secs))).http_status_as_error(false).build().into() ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(secs))).http_status_as_error(false).build().into()
} }
/// Why a request never got an answer, in plain words (never an error code).
pub fn why_unreachable(e: &ureq::Error) -> String {
use std::io::ErrorKind as K;
use ureq::Error as E;
let text = e.to_string().to_lowercase();
let w = match e {
E::HostNotFound => "找不到伺服器：這台電腦可能沒連上網路，或伺服器網址打錯了",
E::ConnectionFailed => "伺服器沒有回應：可能沒開機，或網路把連線擋住了",
E::Timeout(_) => "等太久沒有回應：網路很慢或斷斷續續",
E::Io(io) => match io.kind() {
K::ConnectionRefused => "伺服器沒有開（連線被拒絕）",
K::ConnectionReset | K::ConnectionAborted | K::BrokenPipe | K::UnexpectedEof => "連線到一半斷掉了",
K::TimedOut => "等太久沒有回應：網路很慢或斷斷續續",
K::NetworkUnreachable | K::HostUnreachable | K::NetworkDown | K::AddrNotAvailable => "這台電腦沒有連上網路",
_ => "網路連線出了問題",
},
_ if text.contains("tls") || text.contains("certificate") => "和伺服器的安全連線失敗（伺服器的憑證有問題）",
_ if text.contains("dns") || text.contains("resolve") => "找不到伺服器：這台電腦可能沒連上網路，或伺服器網址打錯了",
_ => "網路連線出了問題",
};
w.to_string()
}
/// An answer that wasn't a yes, in plain words. `whom` names the other side (大腦, 那台電腦…).
pub fn why_refused(st: u16, v: &Value, whom: &str) -> String {
if let Some(e) = v["error"].as_str().filter(|e| !e.is_empty()) {
return e.replace("那台 AnyBot ", whom).replace("那台 AnyBot", whom);
}
match st {
401 | 403 => "這台電腦的帳號登入失效了，請到「設定 › 手機與帳號」重新登入".into(),
404 => format!("{whom}不在這個帳號裡了（可能被移除，或兩邊 AnyBot 版本不同）"),
413 => "檔案太大，帳號伺服器不收".into(),
502 | 503 => "帳號伺服器暫時無法使用（可能正在重新啟動），稍後會自動再試".into(),
504 => format!("{whom}沒有及時回應（可能太忙或網路很慢）"),
500..=599 => "帳號伺服器出了問題，稍後會自動再試".into(),
_ => format!("{whom}的回應看不懂，請把兩邊的 AnyBot 都更新到最新版"),
}
}
pub fn relay_post(url: &str, bearer: &str, body: &Value) -> Result<Value, String> { pub fn relay_post(url: &str, bearer: &str, body: &Value) -> Result<Value, String> {
let mut r = agent(30).post(url).header("Content-Type", "application/json"); let mut r = agent(30).post(url).header("Content-Type", "application/json");
if !bearer.is_empty() { if !bearer.is_empty() {
r = r.header("Authorization", &format!("Bearer {bearer}")); r = r.header("Authorization", &format!("Bearer {bearer}"));
} }
let mut resp = r.send(body.to_string().as_bytes()).map_err(|e| format!("連不到帳號伺服器：{e}"))?; let mut resp = r.send(body.to_string().as_bytes()).map_err(|e| format!("連不到帳號伺服器：{}", why_unreachable(&e)))?;
let st = resp.status().as_u16(); let st = resp.status().as_u16();
let text = resp.body_mut().read_to_string().unwrap_or_default(); let text = resp.body_mut().read_to_string().unwrap_or_default();
let v: Value = serde_json::from_str(&text).unwrap_or(Value::Null); let v: Value = serde_json::from_str(&text).unwrap_or(Value::Null);
if !(200..300).contains(&st) { if !(200..300).contains(&st) {
return Err(v["error"].as_str().map(String::from).unwrap_or(format!("HTTP {st}"))); return Err(why_refused(st, &v, "帳號伺服器"));
} }
Ok(v) Ok(v)
} }
@ -725,7 +764,7 @@ fn relay_worker(url: &str, token: &str, gen: u64) {
let mut resp = match r { let mut resp = match r {
Ok(r) => r, Ok(r) => r,
Err(e) => { Err(e) => {
hub().set_relay_status(&format!("連不上帳號伺服器：{e}")); hub().set_relay_status(&format!("連不上帳號伺服器：{}", why_unreachable(&e)));
std::thread::sleep(Duration::from_secs(backoff)); std::thread::sleep(Duration::from_secs(backoff));
backoff = (backoff * 2).min(30); backoff = (backoff * 2).min(30);
continue; continue;
@ -743,7 +782,9 @@ fn relay_worker(url: &str, token: &str, gen: u64) {
} }
let text = resp.body_mut().read_to_string().unwrap_or_default(); let text = resp.body_mut().read_to_string().unwrap_or_default();
let Ok(job) = serde_json::from_str::<Value>(&text) else { let Ok(job) = serde_json::from_str::<Value>(&text) else {
hub().set_relay_status(&format!("連不上帳號伺服器：{}", why_refused(st, &Value::Null, "帳號伺服器")));
std::thread::sleep(Duration::from_secs(backoff)); std::thread::sleep(Duration::from_secs(backoff));
backoff = (backoff * 2).min(30);
continue; continue;
}; };
hub().set_relay_status("已連線"); hub().set_relay_status("已連線");
@ -831,12 +872,12 @@ pub fn run_on_device(host_id: &str, tool: &str, args: &str, who: &str) -> Result
.header("Authorization", &format!("Bearer {}", s.host_token)) .header("Authorization", &format!("Bearer {}", s.host_token))
.header("Content-Type", "application/json") .header("Content-Type", "application/json")
.send(json!({"tool": tool, "args": args, "who": who}).to_string().as_bytes()) .send(json!({"tool": tool, "args": args, "who": who}).to_string().as_bytes())
.map_err(|e| format!("連不到那台電腦：{e}"))?; .map_err(|e| format!("連不到帳號伺服器，所以也連不到那台電腦：{}", why_unreachable(&e)))?;
let st = resp.status().as_u16(); let st = resp.status().as_u16();
let text = resp.body_mut().with_config().limit(40 * 1024 * 1024).read_to_string().unwrap_or_default(); let text = resp.body_mut().with_config().limit(40 * 1024 * 1024).read_to_string().unwrap_or_default();
let v: Value = serde_json::from_str(&text).unwrap_or(Value::Null); let v: Value = serde_json::from_str(&text).unwrap_or(Value::Null);
if !(200..300).contains(&st) { if !(200..300).contains(&st) {
return Err(v["error"].as_str().map(String::from).unwrap_or(format!("那台電腦沒有回應（HTTP {st}）"))); return Err(why_refused(st, &v, "那台電腦"));
} }
Ok(v) Ok(v)
} }
@ -854,12 +895,15 @@ fn brain_url(s: &Settings, rest: &str) -> Result<String, String> {
} }
fn read_reply(r: Result<ureq::http::Response<ureq::Body>, ureq::Error>) -> Result<Value, String> { fn read_reply(r: Result<ureq::http::Response<ureq::Body>, ureq::Error>) -> Result<Value, String> {
let mut resp = r.map_err(|e| format!("連不到帳號伺服器：{e}"))?; let mut resp = r.map_err(|e| format!("連不到帳號伺服器：{}", why_unreachable(&e)))?;
let st = resp.status().as_u16(); let st = resp.status().as_u16();
let text = resp.body_mut().with_config().limit(200 * 1024 * 1024).read_to_string().unwrap_or_default(); let text = resp.body_mut().with_config().limit(200 * 1024 * 1024).read_to_string().map_err(|_| "和大腦的連線到一半斷掉了".to_string())?;
let v: Value = serde_json::from_str(&text).unwrap_or(Value::Null); let v: Value = serde_json::from_str(&text).unwrap_or(Value::Null);
if !(200..300).contains(&st) { if !(200..300).contains(&st) {
return Err(v["error"].as_str().map(String::from).unwrap_or(format!("大腦沒有回應（HTTP {st}）"))); return Err(why_refused(st, &v, "大腦"));
}
if v.is_null() {
return Err("大腦的回應看不懂，請把兩邊的 AnyBot 都更新到最新版".into());
} }
Ok(v) Ok(v)
} }