From 5841d797be70456897a43bc2c4295c1e06304607 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E6=80=A7=E9=A9=8A?= Date: Wed, 4 Nov 2026 17:46:43 +0800 Subject: [PATCH] One brain, many screens: other AnyBots show and edit the brain's bots MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Brain answers /api/sync (per-screen deltas of settings and messages), /api/sync/edit (small edit operations) and /api/sync/act (chat actions). - Relay forwards host-to-host sync requests within one account (role mirror). - Desktop: choose a brain in 手機與帳號; its own bots are kept aside; offline shows the last copy read-only and reconnects by itself. - Keys and tokens stay on the brain (redacted and restored on edits). - Browser tools can run on another computer of the account (computer arg). - anybot --serve: ANYBOT_DEVICE_LINK=1 and ANYBOT_BRAIN=名稱. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Uirha3oEtWeHh58zA7ErHa --- relay/src/main.rs | 10 + src/agent.rs | 19 +- src/app.rs | 103 ++++- src/app/mirror_ui.rs | 882 +++++++++++++++++++++++++++++++++++++++++++ src/app/remote_ui.rs | 88 ++++- src/remote.rs | 146 ++++++- src/web.rs | 25 +- 7 files changed, 1254 insertions(+), 19 deletions(-) create mode 100644 src/app/mirror_ui.rs diff --git a/relay/src/main.rs b/relay/src/main.rs index 98823e9..aa14d66 100644 --- a/relay/src/main.rs +++ b/relay/src/main.rs @@ -499,6 +499,16 @@ impl Relay { } self.forward(target, "POST", "/api/device/run", &HashMap::new(), &req.body, (&me.user, "device")) } + // 大腦: one AnyBot of the account shows (and edits) another one's bots and chats. + (m, ["relay", "hosts", target, "api", "sync", rest @ ..]) => { + let Some(me) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") }; + let same = self.data.lock().unwrap().hosts.iter().any(|h| h.id == *target && h.user == me.user); + if !same || *target == me.id { + return Resp::error(403, "不是同一個帳號的電腦"); + } + let path = if rest.is_empty() { "/api/sync".to_string() } else { format!("/api/sync/{}", rest.join("/")) }; + self.forward(target, m, &path, &req.query, &req.body, (&me.user, "mirror")) + } // The phone, through to one of the account's AnyBots. (m, ["h", host, rest @ ..]) => { let Some(user) = self.session(&req) else { return Resp::error(401, "請先登入") }; diff --git a/src/agent.rs b/src/agent.rs index 3a7caff..59231fe 100644 --- a/src/agent.rs +++ b/src/agent.rs @@ -114,6 +114,16 @@ fn target_for(job: &Job, args: &str) -> Result { }) } +/// The other computer (裝置連結) a call names, if it names one. +fn device_named(job: &Job, args: &str) -> Option { + let v = tools::args_of(args); + let n = v.get("computer").and_then(|c| c.as_str()).map(str::trim).filter(|c| !c.is_empty())?; + job.computers.iter().find(|t| t.name == n).and_then(|t| match &t.place { + tools::Place::Device { host } => Some(host.clone()), + _ => None, + }) +} + /// Rough token count: about 4 Latin characters or 1 Chinese character per token (estimated). /// Which screen a computer_use call is for: the one named, else the bot's default when it has a /// screen, else the bot's own desktop when there is one, else this Mac. @@ -321,7 +331,14 @@ 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). if call.name == "web_fetch" || (web::is_web_tool(&call.name) && !web::uses_this_computer(&call.name, &call.args)) { send(StreamEvent::ToolStatus(call.id.clone(), ToolStatus::Running)); - let (output, ok, img) = web::run_full(&call.name, &call.args, &cancel); + // 裝置連結: the AnyBot browser of another computer (its own logins). + let (output, ok, img) = match device_named(&job, &call.args).filter(|_| call.name != "web_fetch") { + Some(host) => match crate::remote::run_on_device(&host, &call.name, &call.args, &job.via_name) { + Ok(v) => (v["text"].as_str().unwrap_or("").to_string(), v["ok"].as_bool().unwrap_or(false), v["image"].as_str().map(String::from)), + Err(e) => (e, false, None), + }, + None => web::run_full(&call.name, &call.args, &cancel), + }; send(StreamEvent::ToolStatus(call.id.clone(), if ok { ToolStatus::Done } else { ToolStatus::Failed })); let mut result = Message::tool_result(&call.id, &call.name, &output); result.images.extend(img); diff --git a/src/app.rs b/src/app.rs index 2cec2ec..f46266d 100644 --- a/src/app.rs +++ b/src/app.rs @@ -33,6 +33,7 @@ mod sidebar_ui; mod menu_ui; mod vault_ui; mod markdown_ui; +mod mirror_ui; // Sidebar rules from index.eager-app: defaults `FT={expandedWidth:280}` (@912025), // clamp `jT(..,240,400)` (@914394), and the drag rule in `QJe` (@969523): @@ -197,11 +198,16 @@ pub struct App { group_menu_for: Option, /// When the launch greeting started. splash_at: Option, + /// 大腦: this window shows another AnyBot's bots (a screen), or answers screens (the brain). + mir: mirror_ui::Mirror, + scr: mirror_ui::Screens, + ctx_handle: Option, } impl App { pub fn new(ctx: &egui::Context, screenshot: Option) -> Self { - let demo = screenshot.is_some(); + // 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 (tx, rx) = channel(); let mut app = App { @@ -279,6 +285,9 @@ impl App { renaming: None, group_menu_for: None, splash_at: None, + mir: Default::default(), + scr: Default::default(), + ctx_handle: None, }; crate::i18n::set(&app.state.language); app.computers_start(); @@ -593,8 +602,14 @@ impl App { let quoted: String = q.lines().take(8).map(|l| format!("> {l}\n")).collect(); text = format!("{quoted}\n{}", text.trim()); } - self.drafts.insert(bot_id, String::new()); let attachments = self.take_attachments(bot_id); + if self.mir.on { + if self.mirror_send(bot_id, text.trim().to_string(), attachments) { + self.drafts.insert(bot_id, String::new()); + } + return; + } + self.drafts.insert(bot_id, 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) { @@ -609,6 +624,10 @@ impl App { } 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"})); + return; + } let Some(conv) = self.state.bot(bot_id).cloned() else { return }; // 群組聊天: the next member answers, with its own model, instructions and computers. let speaker = if conv.members.is_empty() { @@ -642,7 +661,8 @@ impl App { let default_computer = if names.contains(&b.default_computer) { b.default_computer.clone() } else { crate::model::BOT_COMPUTER_NAME.to_string() }; let mut tool_specs = appctl::specs(); tool_specs.extend(tools::specs_for(&names, &default_computer)); - tool_specs.extend(web::specs()); + let devices: Vec = computers.iter().filter(|t| matches!(t.place, tools::Place::Device { .. })).map(|t| t.name.clone()).collect(); + tool_specs.extend(web::specs_for(&devices)); // Screen tools: the bot's own desktop (when it has one) and this Mac. let mut screens = vec![]; if crate::cloudbox::available() { @@ -873,6 +893,12 @@ impl App { } fn stop(&mut self, bot_id: u64) { + if self.mir.on { + if self.streams.contains_key(&bot_id) || self.approvals.contains_key(&bot_id) { + self.mirror_act(serde_json::json!({"bot": bot_id, "action": "stop"})); + } + return; + } self.approvals.remove(&bot_id); if let Some(call) = self.app_pending.remove(&bot_id) { let _ = call.reply.send(("已停止".into(), false)); @@ -961,6 +987,17 @@ impl App { } fn answer(&mut self, ctx: &egui::Context, bot_id: u64, answer: ApprovalAnswer) { + if self.mir.on { + let a = match answer { + ApprovalAnswer::Once => "once", + ApprovalAnswer::Always(_) => "always", + ApprovalAnswer::Reject => "reject", + }; + self.approvals.remove(&bot_id); + self.popup_hidden.remove(&bot_id); + self.mirror_act(serde_json::json!({"bot": bot_id, "action": "answer", "answer": a})); + return; + } if self.want_direct.replace(false) { self.enable_direct_mode(); } @@ -1004,6 +1041,10 @@ impl App { } fn regenerate(&mut self, ctx: &egui::Context, bot_id: u64) { + if self.mir.on { + self.mirror_act(serde_json::json!({"bot": bot_id, "action": "regenerate"})); + return; + } if self.streams.contains_key(&bot_id) { return; } @@ -1448,7 +1489,9 @@ impl App { // 討論串: put this conversation aside and start fresh; earlier ones can come back. if menu_ui::item(ui, p, Icon::NewChat, "開始新對話").clicked() { self.stop(id); - if let Some(b) = self.state.bot_mut(id) { + if self.mir.on { + self.mirror_act(serde_json::json!({"bot": id, "action": "new"})); + } else if let Some(b) = self.state.bot_mut(id) { if !b.messages.is_empty() { let title = b.messages.iter().find(|m| m.role == Role::User).map(|m| first_line(&m.text, 24)).unwrap_or_else(|| "對話".into()); let msgs = std::mem::take(&mut b.messages); @@ -1465,7 +1508,9 @@ impl App { for (i, title) in threads.iter().enumerate() { if menu_ui::item_text(ui, p, None, title).clicked() { self.stop(id); - if let Some(b) = self.state.bot_mut(id) { + if self.mir.on { + self.mirror_act(serde_json::json!({"bot": id, "action": "restore", "index": i})); + } else if let Some(b) = self.state.bot_mut(id) { let (_, msgs) = b.threads.remove(i); if !b.messages.is_empty() { let cur_title = b.messages.iter().find(|m| m.role == Role::User).map(|m| first_line(&m.text, 24)).unwrap_or_else(|| "對話".into()); @@ -1564,6 +1609,7 @@ impl App { return; }; self.header(ui, &bot); + self.mirror_banner(ui); // The composer grows with a long draft: keep room for last frame's height. let h_id = egui::Id::unique("composer-height"); let composer_h: f32 = ui.data(|d| d.get_temp(h_id)).unwrap_or(116.0); @@ -1964,7 +2010,7 @@ impl App { }); let right_w = 28.0 * 2.0 + 8.0 + 4.0; let text_w = (ui.available_width() - right_w).max(60.0); - let hint = t("傳訊息給 {name}").replace("{name}", &bot.name); + let hint = if self.mir.on && !self.mir.online { t("連不到大腦，現在只能看") } else { t("傳訊息給 {name}").replace("{name}", &bot.name) }; ui.allocate_ui_with_layout(vec2(text_w, 28.0), Layout::top_down(Align::Min), |ui| { egui::ScrollArea::vertical().id_salt(("composer_scroll", bot.id)).max_height(160.0).show(ui, |ui| { ui.add_space(4.0); @@ -2957,11 +3003,17 @@ impl App { /// Everything that keeps bots working, without drawing (the window runs it every frame; /// `anybot --serve` runs it on its own). fn background(&mut self, ctx: &egui::Context) { + self.remote_start(ctx); + self.mirror_tick(); + // A screen of another AnyBot's brain: its bots work there, not here. + if self.mir.on { + self.remote_tick(ctx); + return; + } self.drain_events(ctx); if !self.demo { self.channels_poll(ctx); } - self.remote_start(ctx); self.remote_tick(ctx); self.tick_routines(ctx); self.watchdog(); @@ -2971,7 +3023,11 @@ impl App { fn save_if_dirty(&mut self) { if self.dirty && !self.demo && (self.streams.is_empty() || self.last_save.elapsed().as_secs() >= 2) { - self.state.save(); + if self.mir.on { + self.mirror_save(); + } else { + self.state.save(); + } self.dirty = false; self.last_save = Instant::now(); } @@ -3033,7 +3089,7 @@ impl App { } app.drain_events(ctx); app.watchdog(); - let _ = ctx.run_ui(Default::default(), |_| {}); + ctx.run_ui(Default::default(), |_| {}).textures_delta.clear(); std::thread::sleep(std::time::Duration::from_millis(100)); } let secs = t0.elapsed().as_secs(); @@ -3081,6 +3137,11 @@ impl App { app.dirty = true; } } + // 手腳: a computer the brain's bots may operate (screen, files, browser). + if std::env::var("ANYBOT_DEVICE_LINK").map(|v| v == "1").unwrap_or(false) && !app.state.remote.device_link { + app.state.remote.device_link = true; + app.dirty = true; + } app.serve_join(); app.background(ctx); if let Some(port) = crate::remote::lan_port() { @@ -3095,16 +3156,28 @@ impl App { eprintln!("AnyBot：還沒有設定手機密碼或帳號。用 ANYBOT_PHONE_PASSWORD=… 或 ANYBOT_ACCOUNT=網址|帳號 + ANYBOT_ACCOUNT_PASSWORD=… 啟動。"); } let mut last_try = Instant::now(); + let mut last_brain = Instant::now() - std::time::Duration::from_secs(60); loop { // The account server may not be up yet (or the account not made yet): keep trying. if app.state.remote.host_token.is_empty() && last_try.elapsed().as_secs() >= 30 { last_try = Instant::now(); app.serve_join(); } + // ANYBOT_BRAIN=名稱: show that AnyBot's bots instead of running its own. + if let Ok(want) = std::env::var("ANYBOT_BRAIN") { + if !want.trim().is_empty() && !app.state.remote.host_token.is_empty() && app.state.remote.brain_name != want && last_brain.elapsed().as_secs() >= 20 { + last_brain = Instant::now(); + if let Some((id, name, _)) = crate::remote::devices_online(&app.state.remote).into_iter().find(|(_, n, _)| n == want.trim()) { + eprintln!("AnyBot：用「{name}」當大腦，這台當手腳"); + app.mir.want = Some((id, name)); + app.state.remote.device_link = true; + } + } + } app.background(ctx); app.save_if_dirty(); // egui wants a frame now and then to forget old state; nothing is drawn. - let _ = ctx.run_ui(Default::default(), |_| {}); + ctx.run_ui(Default::default(), |_| {}).textures_delta.clear(); std::thread::sleep(std::time::Duration::from_millis(if app.streams.is_empty() { 250 } else { 60 })); } } @@ -3216,6 +3289,10 @@ impl eframe::App for App { fn on_exit(&mut self) { // Closing mid-reply ends that turn as cancelled, so it isn't left looking stuck. + // (On a screen of a brain the bots keep working there.) + if self.mir.on { + self.streams.clear(); + } let ids: Vec = self.streams.keys().copied().collect(); for id in ids { self.stop(id); @@ -3223,7 +3300,11 @@ impl eframe::App for App { tools::stop_all(); crate::cloudbox::shutdown(); if !self.demo { - self.state.save(); + if self.mir.on { + self.mirror_save(); + } else { + self.state.save(); + } } } } diff --git a/src/app/mirror_ui.rs b/src/app/mirror_ui.rs new file mode 100644 index 0000000..2760a2b --- /dev/null +++ b/src/app/mirror_ui.rs @@ -0,0 +1,882 @@ +// 大腦: one AnyBot of the account keeps the bots, chats, memory and schedules (the brain, usually +// `anybot --serve` next to the account server). Every other AnyBot signed in to the account shows +// the brain's bots in its own window and lends its hands (screen, files, browser) through 裝置連結. +// +// The brain side answers "what changed since version N?" per screen: the settings (bots without +// their chats) when they changed, and for each bot only the messages from the first one that +// changed. The screen side keeps a copy, sends the person's edits back as small operations, and +// turns chat actions (send, stop, approve…) into requests. When the brain can't be reached the +// window stays readable and waits. + +use super::*; +use crate::remote::{self, SyncKind}; +use serde_json::{json, Value}; +use std::hash::{Hash, Hasher}; +use std::sync::atomic::AtomicU64; + +/// Settings that belong to each computer, never to the brain. +const LOCAL_KEYS: &[&str] = &["remote", "theme", "accent", "sidebar_collapsed", "sidebar_width", "selected", "machine_label", "local_execution", "local_never", "messages_send", "messages_no_ask"]; +/// Bot fields the brain alone changes (its chat, and what a turn leaves behind). +const BOT_RUNTIME: &[&str] = &["id", "messages", "threads", "unread", "last_turn", "context_tokens"]; +/// Stands in for a key or token the brain doesn't hand out. +pub const REDACTED: &str = "••••（已設定，在大腦上）"; + +// ---------- shared ---------- + +/// The settings as JSON: everything but the chats (messages and earlier conversations). +pub(super) fn config_json(st: &mut State) -> Value { + let mut bots = std::mem::take(&mut st.bots); + let mut top = serde_json::to_value(&*st).unwrap_or(Value::Null); + let mut arr = Vec::with_capacity(bots.len()); + for b in bots.iter_mut() { + let m = std::mem::take(&mut b.messages); + let th = std::mem::take(&mut b.threads); + arr.push(serde_json::to_value(&*b).unwrap_or(Value::Null)); + b.messages = m; + b.threads = th; + } + st.bots = bots; + top["bots"] = Value::Array(arr); + top +} + +/// Keys and tokens stay on the brain. +fn redact(top: &mut Value) { + if let Some(o) = top.as_object_mut() { + o.remove("remote"); + } + let hide = |v: &mut Value| { + if v.as_str().map(|s| !s.is_empty()).unwrap_or(false) { + *v = Value::String(REDACTED.into()); + } + }; + if let Some(ps) = top["providers"].as_object_mut() { + for p in ps.values_mut() { + hide(&mut p["api_key"]); + } + } + for c in top["channels"].as_array_mut().into_iter().flatten() { + hide(&mut c["token"]); + } + for m in top["mcp_servers"].as_array_mut().into_iter().flatten() { + for k in ["env", "headers"] { + for v in m[k].as_object_mut().into_iter().flat_map(|o| o.values_mut()) { + hide(v); + } + } + } +} + +/// Put the brain's own keys back wherever an edit still carries the stand-in. +fn unredact(new: &mut Value, old: &Value) { + match new { + Value::String(s) if s == REDACTED => *new = old.clone(), + Value::Object(o) => { + for (k, v) in o.iter_mut() { + unredact(v, old.get(k).unwrap_or(&Value::Null)); + } + } + Value::Array(a) => { + for (i, v) in a.iter_mut().enumerate() { + let key = |x: &Value| x.get("id").cloned().filter(|x| !x.is_null()).or_else(|| x.get("name").cloned()); + let twin = old.as_array().and_then(|olds| match key(v) { + Some(k) => olds.iter().find(|o| key(o).as_ref() == Some(&k)), + None => olds.get(i), + }); + unredact(v, twin.unwrap_or(&Value::Null)); + } + } + _ => {} + } +} + +/// A cheap fingerprint of one message: enough to see it changed (streaming, a tool finishing…). +fn msg_sig(m: &Message) -> u64 { + let mut h = std::collections::hash_map::DefaultHasher::new(); + (m.role as u8, m.text.len(), m.thinking.len(), m.error, m.summary, m.images.len(), m.attachments.len(), m.at, &m.tool_call_id, &m.reactions).hash(&mut h); + m.text.chars().rev().take(48).collect::().hash(&mut h); + for c in &m.tool_calls { + (&c.id, c.args.len(), c.status as u8).hash(&mut h); + } + h.finish() +} + +fn threads_sig(b: &Bot) -> u64 { + let mut h = std::collections::hash_map::DefaultHasher::new(); + for (t, msgs) in &b.threads { + (t, msgs.len()).hash(&mut h); + } + h.finish() +} + +fn hash_value(v: &Value) -> u64 { + let mut h = std::collections::hash_map::DefaultHasher::new(); + v.to_string().hash(&mut h); + h.finish() +} + +/// The person's edits since `base`, as operations the brain applies one by one. +pub(super) fn diff_ops(base: &Value, cur: &Value) -> Vec { + let mut ops = vec![]; + let (Some(b), Some(c)) = (base.as_object(), cur.as_object()) else { return ops }; + for (k, v) in c { + if k == "bots" || LOCAL_KEYS.contains(&k.as_str()) { + continue; + } + if b.get(k) != Some(v) { + ops.push(json!({"op": "set", "key": k, "value": v})); + } + } + let id = |x: &Value| x["id"].as_u64().unwrap_or(0); + let empty = vec![]; + let old = base["bots"].as_array().unwrap_or(&empty); + let new = cur["bots"].as_array().unwrap_or(&empty); + for nb in new { + match old.iter().find(|o| id(o) == id(nb)) { + None => ops.push(json!({"op": "bot_add", "bot": nb})), + Some(ob) => { + for (k, v) in nb.as_object().into_iter().flatten() { + if k == "unread" { + if ob[k] == true && *v == false { + ops.push(json!({"op": "bot_read", "id": id(nb)})); + } + continue; + } + if BOT_RUNTIME.contains(&k.as_str()) { + continue; + } + if ob.get(k) != Some(v) { + ops.push(json!({"op": "bot_set", "id": id(nb), "key": k, "value": v})); + } + } + } + } + } + for ob in old { + if !new.iter().any(|n| id(n) == id(ob)) { + ops.push(json!({"op": "bot_del", "id": id(ob)})); + } + } + let order = |a: &Vec| a.iter().map(id).collect::>(); + let (oo, no) = (order(old), order(new)); + let kept: Vec = oo.iter().copied().filter(|i| no.contains(i)).collect(); + let kept_new: Vec = no.iter().copied().filter(|i| oo.contains(i)).collect(); + if kept != kept_new { + ops.push(json!({"op": "bot_order", "ids": no})); + } + ops +} + +// ---------- the brain's side ---------- + +/// What the brain last sent each screen. +#[derive(Default)] +pub struct Screens { + seen: HashMap, + /// The settings fingerprint, refreshed now and then while something changes. + pub(super) cfg_hash: u64, + cfg_at: Option, +} + +#[derive(Default)] +struct Seen { + version: u64, + config: u64, + bots: HashMap>, + threads: HashMap, + at: Option, +} + +// ---------- the screen's side ---------- + +pub enum Incoming { + Pull(Value), + Offline(String), +} + +#[derive(Default)] +pub struct Mirror { + /// This window shows a brain (instead of its own bots). + pub on: bool, + /// Heard from the brain lately. + pub online: bool, + /// The brain's settings as last received (edits are what differs from this). + pub(super) base: Value, + pub status: Arc>, + rx: Option>, + tx: Option>, + gen: Arc, + version: Arc, + client: String, + /// This computer's own data, kept aside while it shows the brain. + local: Option, + started: bool, + /// A request to the brain failed (shown above the message box). + pub(super) error: Arc>, + /// The person chose a brain in 設定 (applied on the next frame). + pub(super) want: Option<(String, String)>, + pub(super) leave: bool, + pub(super) moving: Arc, + pub(super) move_note: Arc>, +} + +fn cache_path() -> PathBuf { + State::path().with_file_name("brain-copy.json") +} + +impl App { + /// The brain: answer the screens' requests (on the app's thread, where the state is). + pub(super) fn brain_serve(&mut self) { + let reqs = remote::hub().take_sync(); + if reqs.is_empty() { + return; + } + for r in reqs { + let v = match r.kind { + SyncKind::Pull { client, since } => self.brain_pull(&client, since), + SyncKind::Edit(body) => { + if self.mir.on { + json!({"error": "這台 AnyBot 本身連著別的大腦"}) + } else { + self.brain_edit(&body) + } + } + }; + let _ = r.reply.send(v); + } + } + + /// Keep the settings fingerprint fresh (at most once a second) so screens hear of edits. + pub(super) fn brain_watch(&mut self) { + if self.mir.on || !self.dirty || self.scr.cfg_at.map(|t| t.elapsed().as_millis() < 1000).unwrap_or(false) { + return; + } + self.scr.cfg_at = Some(Instant::now()); + let mut c = config_json(&mut self.state); + redact(&mut c); + self.scr.cfg_hash = hash_value(&c); + } + + fn brain_pull(&mut self, client: &str, since: u64) -> Value { + if self.mir.on { + return json!({"error": "這台 AnyBot 本身連著別的大腦"}); + } + let version = remote::hub().version(); + // Forget screens not heard from in a day. + self.scr.seen.retain(|_, s| s.at.map(|t| t.elapsed().as_secs() < 86_400).unwrap_or(false)); + let seen = self.scr.seen.entry(client.to_string()).or_default(); + // A screen that missed an answer (or is new) gets everything again. + if seen.version != since || since == 0 { + *seen = Seen::default(); + } + seen.at = Some(Instant::now()); + seen.version = version; + let mut out = json!({"version": version, "brain": self.state.remote.host_name}); + let mut c = config_json(&mut self.state); + redact(&mut c); + let ch = hash_value(&c); + if seen.config != ch { + seen.config = ch; + out["config"] = c; + } + let mut bots = vec![]; + let mut threads = vec![]; + for b in &self.state.bots { + let sigs: Vec = b.messages.iter().map(msg_sig).collect(); + let old = seen.bots.get(&b.id); + let from = match old { + Some(o) => o.iter().zip(sigs.iter()).position(|(a, b)| a != b).unwrap_or(o.len().min(sigs.len())), + None => 0, + }; + if old.map(|o| o.len() != sigs.len() || from < sigs.len()).unwrap_or(true) { + bots.push(json!({"id": b.id, "from": from, "messages": &b.messages[from..]})); + } + seen.bots.insert(b.id, sigs); + let ts = threads_sig(b); + if seen.threads.get(&b.id) != Some(&ts) { + seen.threads.insert(b.id, ts); + threads.push(json!({"id": b.id, "threads": b.threads})); + } + } + seen.bots.retain(|id, _| self.state.bots.iter().any(|b| b.id == *id)); + out["bots"] = Value::Array(bots); + out["threads"] = Value::Array(threads); + out["live"] = self.streams.iter().map(|(id, l)| (id.to_string(), json!({"status": l.status, "paused": l.paused}))).collect::>().into(); + out["approvals"] = self + .approvals + .iter() + .map(|(id, a)| { + let kind = match a.kind { + ApprovalKind::Action => "action", + ApprovalKind::EnableTools => "tools", + ApprovalKind::EnableThisComputer => "local", + }; + json!({"bot": id, "call_id": a.call_id, "tool": a.tool, "reason": a.reason, "grant": a.grant, "kind": kind}) + }) + .collect(); + out + } + + /// A screen's edits. + pub(super) fn brain_edit(&mut self, body: &Value) -> Value { + let mut applied = 0; + for op in body["ops"].as_array().into_iter().flatten() { + let ok = match op["op"].as_str().unwrap_or("") { + "set" => self.brain_set_top(op["key"].as_str().unwrap_or(""), op["value"].clone()), + "bot_add" => { + let mut v = op["bot"].clone(); + match serde_json::from_value::(v.take()) { + Ok(mut b) => { + if b.id == 0 || self.state.bot(b.id).is_some() { + b.id = self.state.next_id.max(1); + } + self.state.next_id = self.state.next_id.max(b.id + 1); + self.state.bots.push(b); + true + } + Err(_) => false, + } + } + "bot_del" => { + let id = op["id"].as_u64().unwrap_or(0); + if id == self.state.primary_bot || self.state.bots.len() <= 1 { + false + } else { + self.stop(id); + self.state.bots.retain(|b| b.id != id); + true + } + } + "merge" => self.brain_merge_top(op["key"].as_str().unwrap_or(""), &op["value"]), + "bot_set" => self.brain_set_bot(op["id"].as_u64().unwrap_or(0), op["key"].as_str().unwrap_or(""), op["value"].clone()), + "bot_read" => match self.state.bot_mut(op["id"].as_u64().unwrap_or(0)) { + Some(b) => { + b.unread = false; + true + } + None => false, + }, + "bot_order" => { + let ids: Vec = op["ids"].as_array().into_iter().flatten().filter_map(Value::as_u64).collect(); + self.state.bots.sort_by_key(|b| ids.iter().position(|i| *i == b.id).unwrap_or(usize::MAX)); + true + } + _ => false, + }; + if ok { + applied += 1; + } + } + if applied > 0 { + self.mark(); + // Screens hear of it at once. + self.scr.cfg_at = None; + self.brain_watch(); + } + json!({"ok": true, "applied": applied}) + } + + fn brain_set_top(&mut self, key: &str, mut value: Value) -> bool { + if key.is_empty() || key == "bots" || LOCAL_KEYS.contains(&key) { + return false; + } + let bots = std::mem::take(&mut self.state.bots); + let mut top = serde_json::to_value(&self.state).unwrap_or(Value::Null); + unredact(&mut value, &top[key]); + // Counters never go back (a screen may be behind). + if key.starts_with("next_") { + value = json!(value.as_u64().unwrap_or(0).max(top[key].as_u64().unwrap_or(0))); + } + top[key] = value; + let ok = match serde_json::from_value::(top) { + Ok(mut st) => { + st.remote = std::mem::take(&mut self.state.remote); + self.state = st; + true + } + Err(_) => false, + }; + self.state.bots = bots; + ok + } + + /// 搬到大腦: add what the brain doesn't have yet (keys it lacks, skills and connectors by name). + fn brain_merge_top(&mut self, key: &str, value: &Value) -> bool { + let bots = std::mem::take(&mut self.state.bots); + let top = serde_json::to_value(&self.state).unwrap_or(Value::Null); + self.state.bots = bots; + let mut cur = top[key].clone(); + match (key, &mut cur) { + ("providers", Value::Object(o)) => { + for (k, v) in value.as_object().into_iter().flatten() { + let missing = o.get(k).map(|x| x["api_key"].as_str().unwrap_or("").is_empty()).unwrap_or(true); + if missing && v["api_key"].as_str().map(|s| !s.is_empty() && s != REDACTED).unwrap_or(false) { + o.insert(k.clone(), v.clone()); + } + } + } + ("skills" | "mcp_servers", Value::Array(a)) => { + for v in value.as_array().into_iter().flatten() { + if !a.iter().any(|x| x["name"] == v["name"]) { + a.push(v.clone()); + } + } + } + _ => return false, + } + self.brain_set_top(key, cur) + } + + fn brain_set_bot(&mut self, id: u64, key: &str, value: Value) -> bool { + if BOT_RUNTIME.contains(&key) { + return false; + } + let Some(i) = self.state.bots.iter().position(|b| b.id == id) else { return false }; + let b = &mut self.state.bots[i]; + let msgs = std::mem::take(&mut b.messages); + let threads = std::mem::take(&mut b.threads); + let mut v = serde_json::to_value(&*b).unwrap_or(Value::Null); + v[key] = value; + let ok = match serde_json::from_value::(v) { + Ok(nb) => { + *b = nb; + true + } + Err(_) => false, + }; + let b = &mut self.state.bots[i]; + b.messages = msgs; + b.threads = threads; + ok + } + + // ---------- the screen ---------- + + /// Once at start: a brain was chosen earlier, so show its copy. + pub(super) fn mirror_boot(&mut self) { + if self.mir.started || self.demo { + return; + } + self.mir.started = true; + self.mir.client = crate::httpd::random_hex(8); + if !self.state.remote.brain_id.is_empty() && !self.state.remote.host_token.is_empty() { + self.mirror_enter(); + } + } + + /// Start showing the brain: this computer's own bots are kept aside, untouched. + fn mirror_enter(&mut self) { + let ids: Vec = self.streams.keys().copied().collect(); + for id in ids { + self.stop(id); + } + self.approvals.clear(); + let local = self.state.clone(); + if let Some(copy) = std::fs::read_to_string(cache_path()).ok().and_then(|s| serde_json::from_str::(&s).ok()).filter(|s| !s.bots.is_empty()) { + self.state = copy; + self.keep_local_fields(&local); + // Edits wait for the brain's own answer; the copy may be old. + } + self.mir.base = Value::Null; + self.mir.local = Some(local); + self.mir.on = true; + self.mir.online = false; + self.mir.version.store(0, Ordering::SeqCst); + *self.mir.status.lock().unwrap() = "正在連到大腦…".into(); + self.mirror_worker(); + } + + /// Back to this computer's own bots. + fn mirror_leave(&mut self) { + self.mir.gen.fetch_add(1, Ordering::SeqCst); + self.mir.on = false; + self.mir.online = false; + self.streams.clear(); + self.approvals.clear(); + if let Some(mut local) = self.mir.local.take() { + let mut remote = self.state.remote.clone(); + remote.brain_id.clear(); + remote.brain_name.clear(); + local.remote = remote; + local.theme = self.state.theme; + local.accent = self.state.accent; + if local.bot(local.selected).is_none() { + local.selected = local.primary_bot; + } + self.state = local; + } else { + self.state.remote.brain_id.clear(); + self.state.remote.brain_name.clear(); + } + self.state.save(); + self.dirty = false; + } + + fn keep_local_fields(&mut self, from: &State) { + let st = &mut self.state; + st.remote = from.remote.clone(); + st.theme = from.theme; + st.accent = from.accent; + st.sidebar_collapsed = from.sidebar_collapsed; + st.sidebar_width = from.sidebar_width; + st.machine_label = from.machine_label.clone(); + st.local_execution = from.local_execution; + st.local_never = from.local_never; + st.messages_send = from.messages_send; + st.messages_no_ask = from.messages_no_ask; + if st.bot(from.selected).is_some() { + st.selected = from.selected; + } else if st.bot(st.selected).is_none() { + st.selected = st.primary_bot; + } + } + + fn mirror_worker(&mut self) { + let gen = self.mir.gen.fetch_add(1, Ordering::SeqCst) + 1; + let (tx, rx) = channel(); + self.mir.rx = Some(rx); + 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 repaint = self.repaint_fn(); + std::thread::spawn(move || { + let mut backoff = 1u64; + while g.load(Ordering::SeqCst) == gen { + match remote::sync_pull(&s, version.load(Ordering::SeqCst), &client) { + Ok(v) => { + backoff = 1; + if g.load(Ordering::SeqCst) != gen { + break; + } + version.store(v["version"].as_u64().unwrap_or(0), Ordering::SeqCst); + let _ = tx.send(Incoming::Pull(v)); + } + Err(e) => { + let _ = tx.send(Incoming::Offline(e)); + // Start over (with everything) once it is back. + version.store(0, Ordering::SeqCst); + std::thread::sleep(std::time::Duration::from_secs(backoff)); + backoff = (backoff * 2).min(20); + } + } + repaint(); + } + }); + } + + fn repaint_fn(&self) -> Arc { + match &self.ctx_handle { + Some(c) => { + let c = c.clone(); + Arc::new(move || c.request_repaint()) + } + None => Arc::new(|| {}), + } + } + + /// Every frame on a screen: take in what the brain sent. + pub(super) fn mirror_tick(&mut self) { + if let Some((id, name)) = self.mir.want.take() { + if self.mir.on { + self.mirror_leave(); + } + self.state.remote.brain_id = id; + self.state.remote.brain_name = name; + self.state.save(); + self.mirror_enter(); + } + if std::mem::take(&mut self.mir.leave) && self.mir.on { + self.mirror_leave(); + } + if !self.mir.on { + return; + } + let mut got = vec![]; + if let Some(rx) = &self.mir.rx { + while let Ok(m) = rx.try_recv() { + got.push(m); + } + } + for m in got { + match m { + Incoming::Offline(e) => { + self.mir.online = false; + self.mir.base = Value::Null; + *self.mir.status.lock().unwrap() = format!("連不到大腦「{}」：{e}。現在只能看，連上後會自動恢復。", self.state.remote.brain_name); + } + Incoming::Pull(v) => { + if let Some(e) = v["error"].as_str() { + self.mir.online = false; + *self.mir.status.lock().unwrap() = e.to_string(); + continue; + } + self.mir.online = true; + *self.mir.status.lock().unwrap() = format!("已連到大腦「{}」", v["brain"].as_str().unwrap_or(&self.state.remote.brain_name)); + self.mirror_apply(v); + } + } + } + } + + fn mirror_apply(&mut self, v: Value) { + if v["config"].is_object() { + let cfg = v["config"].clone(); + let local = std::mem::replace(&mut self.state, State::first_run()); + match serde_json::from_value::(cfg.clone()) { + Ok(mut st) => { + // Chats stay as they are (they come separately). + let mut old = local.bots.clone(); + for b in st.bots.iter_mut() { + if let Some(o) = old.iter_mut().find(|o| o.id == b.id) { + b.messages = std::mem::take(&mut o.messages); + b.threads = std::mem::take(&mut o.threads); + } + } + self.state = st; + self.keep_local_fields(&local); + self.mir.base = cfg; + } + Err(e) => { + self.state = local; + *self.mir.error.lock().unwrap() = format!("大腦的設定讀不出來（兩邊 AnyBot 版本不同？）：{e}"); + } + } + } + for d in v["bots"].as_array().into_iter().flatten() { + let id = d["id"].as_u64().unwrap_or(0); + let from = d["from"].as_u64().unwrap_or(0) as usize; + let msgs: Vec = serde_json::from_value(d["messages"].clone()).unwrap_or_default(); + if let Some(b) = self.state.bot_mut(id) { + b.messages.truncate(from); + b.messages.extend(msgs); + } + } + for d in v["threads"].as_array().into_iter().flatten() { + if let Some(b) = self.state.bot_mut(d["id"].as_u64().unwrap_or(0)) { + b.threads = serde_json::from_value(d["threads"].clone()).unwrap_or_default(); + } + } + // Who is working, and what waits for the person. + let live: HashMap, bool)> = v["live"].as_object().into_iter().flatten().filter_map(|(k, l)| Some((k.parse().ok()?, (l["status"].as_str().map(String::from), l["paused"].as_bool().unwrap_or(false))))).collect(); + self.streams.retain(|id, _| live.contains_key(id)); + for (id, (status, paused)) in live { + let l = self.streams.entry(id).or_insert_with(|| Live { + cancel: Arc::new(AtomicBool::new(false)), + started: Instant::now(), + last: Instant::now(), + produced: true, + status: None, + paused: false, + steer: Arc::default(), + turn_id: String::new(), + direct: Arc::default(), + }); + l.status = status; + l.paused = paused; + l.last = Instant::now(); + } + let mut waiting = HashMap::new(); + for a in v["approvals"].as_array().into_iter().flatten() { + let id = a["bot"].as_u64().unwrap_or(0); + let kind = match a["kind"].as_str().unwrap_or("") { + "tools" => ApprovalKind::EnableTools, + "local" => ApprovalKind::EnableThisComputer, + _ => ApprovalKind::Action, + }; + let (reply, _) = channel(); + let s = |k: &str| a[k].as_str().unwrap_or("").to_string(); + waiting.insert(id, Approval { call_id: s("call_id"), tool: s("tool"), reason: s("reason"), grant: s("grant"), reply, kind }); + } + let ctx = self.ctx_handle.clone(); + for (id, a) in waiting { + let new = self.approvals.get(&id).map(|o| o.call_id != a.call_id).unwrap_or(true); + self.approvals.insert(id, a); + if new { + if let Some(c) = &ctx { + self.on_new_approval(c, id); + } + } + } + let keep: Vec = v["approvals"].as_array().into_iter().flatten().filter_map(|a| a["bot"].as_u64()).collect(); + self.approvals.retain(|id, _| keep.contains(id)); + if self.state.bot(self.state.selected).is_none() { + self.state.selected = self.state.primary_bot; + } + self.mirror_save_copy(); + } + + fn mirror_save_copy(&mut self) { + if let Ok(s) = serde_json::to_string(&self.state) { + let p = cache_path(); + let tmp = p.with_extension("tmp"); + if std::fs::write(&tmp, s).is_ok() { + let _ = std::fs::rename(&tmp, &p); + } + } + } + + /// Saving on a screen: edits go to the brain; this computer's own file keeps its own bots. + pub(super) fn mirror_save(&mut self) { + if let Some(local) = self.mir.local.as_mut() { + local.remote = self.state.remote.clone(); + local.theme = self.state.theme; + local.accent = self.state.accent; + local.sidebar_collapsed = self.state.sidebar_collapsed; + local.sidebar_width = self.state.sidebar_width; + local.machine_label = self.state.machine_label.clone(); + local.local_execution = self.state.local_execution; + local.local_never = self.state.local_never; + local.messages_send = self.state.messages_send; + local.messages_no_ask = self.state.messages_no_ask; + local.save(); + } + if !self.mir.online || !self.mir.base.is_object() { + return; + } + let cur = config_json(&mut self.state); + let ops = diff_ops(&self.mir.base, &cur); + if ops.is_empty() { + return; + } + self.mir.base = cur; + let (s, err) = (self.state.remote.clone(), self.mir.error.clone()); + std::thread::spawn(move || { + if let Err(e) = remote::sync_post(&s, "edit", &json!({"ops": ops})) { + *err.lock().unwrap() = format!("改動沒有存到大腦：{e}"); + } + }); + } + + /// A chat action on a screen goes to the brain. + pub(super) fn mirror_act(&mut self, body: Value) { + if !self.mir.online { + *self.mir.error.lock().unwrap() = "連不到大腦，現在只能看。".into(); + return; + } + let (s, err) = (self.state.remote.clone(), self.mir.error.clone()); + 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. + pub(super) fn mirror_send(&mut self, bot: u64, text: String, atts: Vec) -> bool { + if !self.mir.online { + *self.mir.error.lock().unwrap() = "連不到大腦，訊息沒有送出。".into(); + return false; + } + // Shown at once; the brain's copy replaces it. + if !self.streams.contains_key(&bot) { + if let Some(b) = self.state.bot_mut(bot) { + b.messages.push(Message { attachments: atts.clone(), ..Message::user(&text) }); + } + } + let (s, err) = (self.state.remote.clone(), self.mir.error.clone()); + std::thread::spawn(move || { + let mut paths = vec![]; + 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); + } + } + } + if let Err(e) = remote::sync_post(&s, "act", &json!({"bot": bot, "action": "send", "text": text, "attachments": paths})) { + *err.lock().unwrap() = format!("訊息沒有送到大腦：{e}"); + } + }); + true + } + + /// 把這台的 Bot 搬到大腦: copy this computer's own bots (with their chats), keys, skills and + /// connectors to the brain, as new bots there. + pub(super) fn mirror_move_local(&mut self) { + let Some(local) = self.mir.local.clone() else { return }; + if self.mir.moving.swap(true, Ordering::SeqCst) { + return; + } + let mut ops: Vec = vec![]; + for b in &local.bots { + let mut v = serde_json::to_value(b).unwrap_or(Value::Null); + v["id"] = json!(0); + ops.push(json!({"op": "bot_add", "bot": v})); + } + let top = serde_json::to_value(&local).unwrap_or(Value::Null); + for k in ["providers", "skills", "mcp_servers"] { + 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()); + *note.lock().unwrap() = "搬移中…".into(); + std::thread::spawn(move || { + *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. + pub(super) fn mirror_banner(&mut self, ui: &mut egui::Ui) { + if !self.mir.on { + return; + } + let p = self.pal; + let err = std::mem::take(&mut *self.mir.error.lock().unwrap()); + if !err.is_empty() { + self.notice = Some(err); + } + if self.mir.online { + return; + } + let status = self.mir.status.lock().unwrap().clone(); + ui.vertical_centered(|ui| { + 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| { + ui.add(egui::Label::new(RichText::new(status).size(FONT_SM).color(p.text2)).wrap()); + }); + }); + ui.add_space(6.0); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn edits_become_small_operations() { + let base = json!({"theme": "light", "folders": ["a"], "bots": [{"id": 1, "name": "A", "unread": true}, {"id": 2, "name": "B"}]}); + let cur = json!({"theme": "dark", "folders": ["a", "b"], "bots": [{"id": 1, "name": "A2", "unread": false}, {"id": 3, "name": "C"}]}); + let ops = diff_ops(&base, &cur); + // theme is this computer's own; folders changed. + assert!(ops.iter().any(|o| o["op"] == "set" && o["key"] == "folders")); + assert!(!ops.iter().any(|o| o["key"] == "theme")); + assert!(ops.iter().any(|o| o["op"] == "bot_set" && o["key"] == "name" && o["value"] == "A2")); + assert!(ops.iter().any(|o| o["op"] == "bot_read" && o["id"] == 1)); + assert!(ops.iter().any(|o| o["op"] == "bot_add" && o["bot"]["id"] == 3)); + assert!(ops.iter().any(|o| o["op"] == "bot_del" && o["id"] == 2)); + } + + #[test] + fn keys_stay_on_the_brain() { + let mut v = json!({"remote": {"host_token": "x"}, "providers": {"xai": {"api_key": "sk-1"}, "openai": {"api_key": ""}}, "channels": [{"id": 4, "token": "t"}]}); + let old = v.clone(); + redact(&mut v); + assert!(v.get("remote").is_none()); + assert_eq!(v["providers"]["xai"]["api_key"], REDACTED); + assert_eq!(v["providers"]["openai"]["api_key"], ""); + assert_eq!(v["channels"][0]["token"], REDACTED); + // An edit that kept the stand-in keeps the real key. + let mut edit = v["providers"].clone(); + edit["openai"]["api_key"] = json!("sk-new"); + unredact(&mut edit, &old["providers"]); + assert_eq!(edit["xai"]["api_key"], "sk-1"); + assert_eq!(edit["openai"]["api_key"], "sk-new"); + let mut ch = v["channels"].clone(); + unredact(&mut ch, &old["channels"]); + assert_eq!(ch[0]["token"], "t"); + } +} diff --git a/src/app/remote_ui.rs b/src/app/remote_ui.rs index bd08465..b0aa019 100644 --- a/src/app/remote_ui.rs +++ b/src/app/remote_ui.rs @@ -51,9 +51,11 @@ impl App { return; } self.rm.started = true; + self.ctx_handle = Some(ctx.clone()); let c = ctx.clone(); remote::hub().set_repaint(Arc::new(move || c.request_repaint())); self.remote_apply_settings(); + self.mirror_boot(); } /// Hand changed settings to the server threads (and restart what depends on them). @@ -84,8 +86,14 @@ impl App { } self.remote_apply_settings(); for c in remote::hub().take_cmds() { + // A screen of another brain has no bots of its own to run phone requests on. + if self.mir.on && !matches!(c, Cmd::DeviceAdded(_) | Cmd::DeviceSeen(_)) { + continue; + } self.remote_cmd(ctx, c); } + self.brain_serve(); + self.brain_watch(); if let Some(rx) = &self.rm.join_rx { if let Ok(r) = rx.try_recv() { self.rm.join_rx = None; @@ -118,7 +126,7 @@ impl App { } // 自動備份: once a day while signed in. let s = &self.state.remote; - if s.auto_backup && !s.backup_key.is_empty() && !s.relay_url.is_empty() && chrono::Utc::now().timestamp() - s.last_backup > 86_400 && !self.rm.busy.load(Ordering::SeqCst) { + if !self.mir.on && s.auto_backup && !s.backup_key.is_empty() && !s.relay_url.is_empty() && chrono::Utc::now().timestamp() - s.last_backup > 86_400 && !self.rm.busy.load(Ordering::SeqCst) { self.remote_backup(); } // 裝置連結: keep the list of this account's computers fresh (once a minute). @@ -132,7 +140,7 @@ impl App { } }); } - if !self.rm.last_publish.map(|t| t.elapsed().as_millis() < 120).unwrap_or(false) { + if !self.mir.on && !self.rm.last_publish.map(|t| t.elapsed().as_millis() < 120).unwrap_or(false) { self.remote_publish(); } } @@ -140,6 +148,7 @@ impl App { fn remote_signature(&self) -> u64 { let mut h = std::collections::hash_map::DefaultHasher::new(); self.state.remote.host_name.hash(&mut h); + self.scr.cfg_hash.hash(&mut h); for b in &self.state.bots { (b.id, &b.name, &b.color, b.unread, b.messages.len()).hash(&mut h); if let Some(m) = b.messages.last() { @@ -214,6 +223,27 @@ impl App { self.start_reply(ctx, bot); } Cmd::Stop { bot } => self.stop(bot), + Cmd::Regenerate { bot } => self.regenerate(ctx, bot), + Cmd::Reply { bot } => { + if !self.streams.contains_key(&bot) && self.state.bot(bot).and_then(|b| b.messages.last()).map(|m| m.role == Role::User).unwrap_or(false) { + self.start_reply(ctx, bot); + } + } + Cmd::RestoreThread { bot, index } => { + self.stop(bot); + if let Some(b) = self.state.bot_mut(bot) { + if index < b.threads.len() { + let (_, msgs) = b.threads.remove(index); + if !b.messages.is_empty() { + let title = b.messages.iter().find(|m| m.role == Role::User).map(|m| first_line(&m.text, 24)).unwrap_or_else(|| "對話".into()); + let cur = std::mem::take(&mut b.messages); + b.threads.insert(0, (title, cur)); + } + b.messages = msgs; + } + } + self.mark(); + } Cmd::Answer { bot, answer } => { let grant = self.approvals.get(&bot).map(|a| a.grant.clone()).unwrap_or_default(); let a = match answer.as_str() { @@ -380,6 +410,60 @@ impl App { } }); + // ── 大腦 ── + if !self.state.remote.host_token.is_empty() { + setting_card(ui, p, |ui| { + ui.horizontal(|ui| { + ui.label(RichText::new(t("大腦")).size(FONT_BASE).strong()); + help_mark(ui, "帳號裡選一台 AnyBot 當大腦（通常是跟帳號伺服器放在一起、一直開著的那台）。所有 Bot、聊天、記憶和例行任務都放在大腦上，由它思考和排程。其他電腦打開 AnyBot 看到的都是同一份，同時也是大腦的手腳：大腦的 Bot 可以透過裝置連結操作它們。大腦連不到時，這裡還能看之前的對話，但不能傳訊息，連回來就恢復。", p); + }); + let s = &self.state.remote; + let status = self.mir.status.lock().unwrap().clone(); + if self.mir.on { + ui.horizontal(|ui| { + let (r, _) = ui.allocate_exact_size(vec2(10.0, 10.0), egui::Sense::hover()); + ui.painter().circle_filled(r.center(), 4.0, if self.mir.online { Color32::from_rgb(0x14, 0xc3, 0x70) } else { p.text3 }); + ui.label(RichText::new(status).size(13.0).color(p.text2)); + }); + ui.horizontal(|ui| { + let busy = self.mir.moving.load(Ordering::SeqCst); + if ui.add_enabled(self.mir.online && !busy, secondary_button("把這台的 Bot 搬過去")).on_hover_text(t("把這台電腦原本的 Bot（連同聊天）、模型金鑰、技能和連接器複製到大腦。這台的原始資料不會刪掉。")).clicked() { + self.mirror_move_local(); + } + if ui.add(secondary_button("改回用這台自己的 Bot")).clicked() { + self.mir.leave = true; + } + }); + let note = self.mir.move_note.lock().unwrap().clone(); + if !note.is_empty() { + ui.label(RichText::new(note).size(12.5).color(p.text2)); + } + } else { + ui.label(RichText::new(t("這台就是大腦：Bot 在這台電腦上思考和工作。")).size(13.0).color(p.text2)); + let me = s.host_id.clone(); + let peers: Vec<(String, String, bool)> = self.rm.peers.lock().unwrap().iter().filter(|(id, _, _)| !id.is_empty() && *id != me).cloned().collect(); + if peers.is_empty() { + hint(ui, "帳號裡還沒有其他 AnyBot。在伺服器上啟動 anybot --serve 並登入同一個帳號，它就會出現在這裡。", p); + } + for (id, name, online) in peers { + ui.horizontal(|ui| { + let (r, _) = ui.allocate_exact_size(vec2(10.0, 10.0), egui::Sense::hover()); + ui.painter().circle_filled(r.center(), 4.0, if online { Color32::from_rgb(0x14, 0xc3, 0x70) } else { p.text3 }); + ui.label(RichText::new(&name).size(13.5)); + ui.with_layout(Layout::right_to_left(Align::Center), |ui| { + if ui.add_enabled(online, secondary_button("改用它當大腦")).clicked() { + self.mir.want = Some((id.clone(), name.clone())); + // The brain's bots may use this computer (each action still asks). + self.state.remote.device_link = true; + changed = true; + } + }); + }); + } + } + }); + } + // ── 同一個網路 ── setting_card(ui, p, |ui| { setting_title(ui, "同一個網路直接連", "手機和這台電腦在同一個 Wi-Fi，或都裝了 Tailscale 時使用。", p); diff --git a/src/remote.rs b/src/remote.rs index ed25cc2..676415d 100644 --- a/src/remote.rs +++ b/src/remote.rs @@ -72,11 +72,17 @@ pub struct Settings { pub auto_backup: bool, #[serde(default)] pub last_backup: i64, + /// 大腦: the AnyBot of this account whose bots, chats and settings this one shows (its relay + /// host id). Empty = this AnyBot is its own brain. When set, this one is a screen and hands. + #[serde(default, skip_serializing_if = "String::is_empty")] + pub brain_id: String, + #[serde(default, skip_serializing_if = "String::is_empty")] + pub brain_name: String, } impl Default for Settings { fn default() -> Self { - Settings { lan: false, password: String::new(), devices: vec![], mac_from_phone: false, device_link: false, relay_url: String::new(), relay_user: String::new(), host_id: String::new(), host_token: String::new(), host_name: default_host_name(), backup_key: String::new(), auto_backup: true, last_backup: 0 } + Settings { lan: false, password: String::new(), devices: vec![], mac_from_phone: false, device_link: false, relay_url: String::new(), relay_user: String::new(), host_id: String::new(), host_token: String::new(), host_name: default_host_name(), backup_key: String::new(), auto_backup: true, last_backup: 0, brain_id: String::new(), brain_name: String::new() } } } @@ -103,6 +109,11 @@ pub enum Cmd { Answer { bot: u64, answer: String }, Read { bot: u64 }, NewThread { bot: u64 }, + /// Answer the last message again (重新產生) / reply to what is there (重新傳送). + Regenerate { bot: u64 }, + Reply { bot: u64 }, + /// Bring back an earlier conversation (過去的對話). + RestoreThread { bot: u64, index: usize }, /// A new phone signed in (to remember), or one was signed out. DeviceAdded(Device), DeviceSeen(String), @@ -126,11 +137,23 @@ pub struct Hub { settings: Mutex, fails: Mutex<(u32, Option)>, relay_status: Mutex, + sync: Mutex>, +} + +/// 大腦: another AnyBot of this account (a screen) asks for what changed, or sends its edits. +pub enum SyncKind { + Pull { client: String, since: u64 }, + Edit(Value), +} + +pub struct SyncReq { + pub kind: SyncKind, + pub reply: std::sync::mpsc::Sender, } pub fn hub() -> &'static Hub { static H: OnceLock = OnceLock::new(); - H.get_or_init(|| Hub { view: Mutex::new(View::default()), changed: Condvar::new(), cmds: Mutex::new(vec![]), repaint: Mutex::new(None), settings: Mutex::new(Settings::default()), fails: Mutex::new((0, None)), relay_status: Mutex::new(String::new()) }) + H.get_or_init(|| Hub { view: Mutex::new(View::default()), changed: Condvar::new(), cmds: Mutex::new(vec![]), repaint: Mutex::new(None), settings: Mutex::new(Settings::default()), fails: Mutex::new((0, None)), relay_status: Mutex::new(String::new()), sync: Mutex::new(vec![]) }) } static VERSION: AtomicU64 = AtomicU64::new(1); @@ -141,6 +164,21 @@ impl Hub { *self.view.lock().unwrap() = v; self.changed.notify_all(); } + pub fn take_sync(&self) -> Vec { + std::mem::take(&mut *self.sync.lock().unwrap()) + } + /// Hand a sync request to the app's thread and wait for its answer. + fn ask_sync(&self, kind: SyncKind) -> Result { + let (tx, rx) = std::sync::mpsc::channel(); + self.sync.lock().unwrap().push(SyncReq { kind, reply: tx }); + if let Some(r) = self.repaint.lock().unwrap().as_ref() { + r(); + } + rx.recv_timeout(Duration::from_secs(10)).map_err(|_| "大腦沒有及時回應".to_string()) + } + pub fn version(&self) -> u64 { + self.view.lock().unwrap().version + } pub fn take_cmds(&self) -> Vec { std::mem::take(&mut *self.cmds.lock().unwrap()) } @@ -328,6 +366,44 @@ pub fn handle(req: Req, via_relay: bool) -> Resp { ("GET", ["api", "screen", which]) => screen(which, &s), ("POST", ["api", "screen", which]) => screen_input(which, &req.json(), &s), ("POST", ["api", "device", "run"]) => device_run(&req.json(), &s, via_relay), + ("GET", ["api", "sync"]) => { + let since: u64 = req.query.get("since").and_then(|v| v.parse().ok()).unwrap_or(0); + let client = req.query.get("client").cloned().unwrap_or_default(); + if since > 0 { + let _ = hub().wait_newer(since, LONG_POLL_SECS); + } + match hub().ask_sync(SyncKind::Pull { client, since }) { + Ok(v) => Resp::json(&v), + Err(e) => Resp::error(503, &e), + } + } + ("POST", ["api", "sync", "edit"]) => match hub().ask_sync(SyncKind::Edit(req.json())) { + Ok(v) if v["error"].is_string() => Resp::error(400, v["error"].as_str().unwrap_or("")), + Ok(v) => Resp::json(&v), + Err(e) => Resp::error(503, &e), + }, + ("POST", ["api", "sync", "upload", id]) => upload(id.parse().unwrap_or(0), &req), + ("POST", ["api", "sync", "act"]) => { + let b = req.json(); + let bot = b["bot"].as_u64().unwrap_or(0); + let cmd = match b["action"].as_str().unwrap_or("") { + "send" => Cmd::Send { + bot, + text: b["text"].as_str().unwrap_or("").trim().to_string(), + attachments: b["attachments"].as_array().map(|a| a.iter().filter_map(|x| x.as_str().map(String::from)).collect()).unwrap_or_default(), + }, + "stop" => Cmd::Stop { bot }, + "answer" => Cmd::Answer { bot, answer: b["answer"].as_str().unwrap_or("reject").to_string() }, + "read" => Cmd::Read { bot }, + "new" => Cmd::NewThread { bot }, + "regenerate" => Cmd::Regenerate { bot }, + "reply" => Cmd::Reply { bot }, + "restore" => Cmd::RestoreThread { bot, index: b["index"].as_u64().unwrap_or(0) as usize }, + _ => return Resp::error(404, "沒有這個動作"), + }; + hub().push(cmd); + Resp::json(&json!({"ok": true})) + } ("POST", ["api", "logout"]) if !via_relay => { let h = httpd::sha256_hex(&req.bearer()); hub().push(Cmd::DeviceSeen(format!("-{h}"))); @@ -342,9 +418,12 @@ pub fn handle(req: Req, via_relay: bool) -> Resp { /// account) may run tools here. pub fn role_allows(role: &str, method: &str, path: &[&str]) -> bool { let device = matches!(path, ["api", "device", "run"]); + let sync = matches!(path, ["api", "sync", ..]); match role { "device" => device, - "owner" | "admin" => !device, + // 大腦: the owner's other AnyBots showing this one's bots. + "mirror" => sync, + "owner" | "admin" => !device && !sync, "member" => matches!((method, path), ("GET", ["api", "state"]) | ("GET", ["api", "bots", _]) | ("POST", ["api", "bots", _, "send" | "read" | "upload"])), _ => false, } @@ -484,6 +563,15 @@ fn device_run(v: &Value, s: &Settings, via_relay: bool) -> Resp { let (text, ok, img) = crate::computer::run_by(tool, args, who); return Resp::json(&json!({"text": text, "ok": ok, "image": img})); } + // The AnyBot browser here (its logins are this computer's). + if crate::web::is_web_tool(tool) && tool != "web_fetch" { + let mut v: Value = serde_json::from_str(args).unwrap_or(json!({})); + if let Some(o) = v.as_object_mut() { + o.remove("computer"); + } + let (text, ok, img) = crate::web::run_full(tool, &v.to_string(), &std::sync::atomic::AtomicBool::new(false)); + return Resp::json(&json!({"text": text, "ok": ok, "image": img})); + } if crate::macapps::is_mac_tool(tool) && cfg!(target_os = "macos") { let (text, ok) = crate::macapps::run(tool, args); return Resp::json(&json!({"text": text, "ok": ok})); @@ -753,6 +841,58 @@ pub fn run_on_device(host_id: &str, tool: &str, args: &str, who: &str) -> Result Ok(v) } +// ---------- 大腦: this AnyBot as a screen of another ---------- + +fn brain_url(s: &Settings, rest: &str) -> Result { + if s.relay_url.is_empty() || s.host_token.is_empty() { + return Err("沒有登入帳號".into()); + } + if s.brain_id.is_empty() { + return Err("沒有選大腦".into()); + } + Ok(format!("{}/relay/hosts/{}/api/sync{rest}", s.relay_url, s.brain_id)) +} + +fn read_reply(r: Result, ureq::Error>) -> Result { + let mut resp = r.map_err(|e| format!("連不到帳號伺服器：{e}"))?; + let st = resp.status().as_u16(); + let text = resp.body_mut().with_config().limit(200 * 1024 * 1024).read_to_string().unwrap_or_default(); + let v: Value = serde_json::from_str(&text).unwrap_or(Value::Null); + if !(200..300).contains(&st) { + return Err(v["error"].as_str().map(String::from).unwrap_or(format!("大腦沒有回應（HTTP {st}）"))); + } + Ok(v) +} + +/// What changed at the brain since `since` (waits up to LONG_POLL_SECS when nothing has). +pub fn sync_pull(s: &Settings, since: u64, client: &str) -> Result { + let url = brain_url(s, &format!("?since={since}&client={client}"))?; + read_reply(agent(LONG_POLL_SECS + 25).get(&url).header("Authorization", &format!("Bearer {}", s.host_token)).call()) +} + +/// Send edits ("edit") or a chat action ("act") to the brain. +pub fn sync_post(s: &Settings, what: &str, body: &Value) -> Result { + let url = brain_url(s, &format!("/{what}"))?; + read_reply(agent(60).post(&url).header("Authorization", &format!("Bearer {}", s.host_token)).header("Content-Type", "application/json").send(body.to_string().as_bytes())) +} + +/// A file for the next message, kept at the brain; gives back its path there. +pub fn sync_upload(s: &Settings, bot: u64, name: &str, bytes: &[u8]) -> Result { + let q: String = url_escape(name); + let url = brain_url(s, &format!("/upload/{bot}?name={q}"))?; + let v = read_reply(agent(120).post(&url).header("Authorization", &format!("Bearer {}", s.host_token)).header("Content-Type", "application/octet-stream").send(bytes))?; + v["path"].as_str().map(String::from).ok_or("大腦沒有收下檔案".into()) +} + +fn url_escape(s: &str) -> String { + s.bytes() + .map(|b| match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => (b as char).to_string(), + _ => format!("%{b:02X}"), + }) + .collect() +} + /// A notification on the person's phones (through the relay; phones that turned them on). pub fn notify(title: &str, body: &str, bot: u64) { let s = hub().settings(); diff --git a/src/web.rs b/src/web.rs index 80ab527..8315ed3 100644 --- a/src/web.rs +++ b/src/web.rs @@ -59,7 +59,28 @@ pub fn read_needs_permission(args: &str) -> bool { uses_person_browser(args) } -pub fn specs() -> Vec { +/// The browser tools, also when the browser is on another computer of the account (裝置連結): +/// then each takes a `computer`, since each computer's AnyBot browser has its own logins. +pub fn specs_for(devices: &[String]) -> Vec { + let local = crate::botbrowser::available(); + let mut v = specs_inner(local || !devices.is_empty()); + if !devices.is_empty() { + let mut names: Vec = devices.to_vec(); + if local { + names.insert(0, crate::model::THIS_COMPUTER_NAME.to_string()); + } + let desc = format!( + "Which computer's AnyBot browser to use. Each computer's browser has its own logins (a site logged in on one isn't on another). {}", + if local { format!("Leave out for {}.", crate::model::THIS_COMPUTER_NAME) } else { "This computer has no browser: always name one.".to_string() } + ); + for t in v.iter_mut().filter(|t| t["name"].as_str().map(|n| n.starts_with("browser_")).unwrap_or(false)) { + t["parameters"]["properties"]["computer"] = json!({"type": "string", "enum": names, "description": desc}); + } + } + v +} + +fn specs_inner(browser: bool) -> Vec { let (names, current) = if crate::botbrowser::available() { (crate::botbrowser::choices().join(", "), crate::botbrowser::current()) } else { @@ -76,7 +97,7 @@ pub fn specs() -> Vec { json!({"name": "browser_profiles", "description": "List, add or switch AnyBot browser profiles (設定檔/profile). Use when the person wants another profile or account: 'list' shows them, 'add' makes a new empty one (they log in once there) and switches to it, 'use' switches. Switching changes which profile browser_open/browser_read use.", "parameters": {"type": "object", "properties": {"action": {"type": "string", "enum": ["list", "add", "use"]}, "name": {"type": "string", "description": "Profile name for add/use."}}, "required": ["action"]}}), ]; - if crate::botbrowser::available() { + if browser { v.extend(crate::browser::specs(&profile)); } v