1383 lines
60 KiB
Rust
1383 lines
60 KiB
Rust
// 大腦: 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::<String>().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()
|
||
}
|
||
|
||
/// What names an item in a list: its id, else its name, else (plain values) the value itself.
|
||
fn item_key(x: &Value) -> Option<Value> {
|
||
match x {
|
||
Value::Object(o) => o.get("id").filter(|v| !v.is_null()).or_else(|| o.get("name").filter(|v| v.is_string())).cloned(),
|
||
Value::Array(_) | Value::Null => None,
|
||
v => Some(v.clone()),
|
||
}
|
||
}
|
||
|
||
/// Every item has its own key, and no two share one.
|
||
fn keyed(a: &[Value]) -> bool {
|
||
let mut seen = std::collections::HashSet::new();
|
||
a.iter().all(|x| item_key(x).map(|k| seen.insert(k.to_string())).unwrap_or(false))
|
||
}
|
||
|
||
/// The smallest change from `old` to `new`: lists of named things and maps change item by item,
|
||
/// so two computers editing different items of one list don't undo each other.
|
||
pub(super) fn patch_of(old: &Value, new: &Value) -> Value {
|
||
match (old, new) {
|
||
(Value::Object(o), Value::Object(n)) => {
|
||
let put: serde_json::Map<String, Value> = n.iter().filter(|(k, v)| o.get(*k) != Some(*v)).map(|(k, v)| (k.clone(), v.clone())).collect();
|
||
let del: Vec<&String> = o.keys().filter(|k| !n.contains_key(*k)).collect();
|
||
json!({"map": {"put": put, "del": del}})
|
||
}
|
||
(Value::Array(o), Value::Array(n)) if keyed(o) && keyed(n) => {
|
||
let key = |x: &Value| item_key(x).unwrap_or(Value::Null);
|
||
let put: Vec<&Value> = n.iter().filter(|x| !o.iter().any(|y| key(y) == key(x) && y == *x)).collect();
|
||
let del: Vec<Value> = o.iter().map(key).filter(|k| !n.iter().any(|x| key(x) == *k)).collect();
|
||
let common_old: Vec<Value> = o.iter().map(key).filter(|k| n.iter().any(|x| key(x) == *k)).collect();
|
||
let common_new: Vec<Value> = n.iter().map(key).filter(|k| o.iter().any(|x| key(x) == *k)).collect();
|
||
let mut p = json!({"list": {"put": put, "del": del}});
|
||
if common_old != common_new {
|
||
p["list"]["order"] = Value::Array(n.iter().map(key).collect());
|
||
}
|
||
p
|
||
}
|
||
_ => json!({"value": new}),
|
||
}
|
||
}
|
||
|
||
/// Apply a change made by `patch_of` to whatever the value is now.
|
||
pub(super) fn apply_patch(cur: &mut Value, p: &Value) {
|
||
if let Some(v) = p.get("value") {
|
||
*cur = v.clone();
|
||
} else if let Some(m) = p.get("map") {
|
||
if !cur.is_object() {
|
||
*cur = json!({});
|
||
}
|
||
let o = cur.as_object_mut().expect("object");
|
||
for (k, v) in m["put"].as_object().into_iter().flatten() {
|
||
o.insert(k.clone(), v.clone());
|
||
}
|
||
for k in m["del"].as_array().into_iter().flatten().filter_map(Value::as_str) {
|
||
o.remove(k);
|
||
}
|
||
} else if let Some(l) = p.get("list") {
|
||
if !cur.is_array() {
|
||
*cur = json!([]);
|
||
}
|
||
let a = cur.as_array_mut().expect("array");
|
||
let key = |x: &Value| item_key(x).unwrap_or(Value::Null);
|
||
for item in l["put"].as_array().into_iter().flatten() {
|
||
match a.iter_mut().find(|x| key(x) == key(item)) {
|
||
Some(x) => *x = item.clone(),
|
||
None => a.push(item.clone()),
|
||
}
|
||
}
|
||
let del: Vec<Value> = l["del"].as_array().cloned().unwrap_or_default();
|
||
a.retain(|x| !del.contains(&key(x)));
|
||
if let Some(order) = l["order"].as_array() {
|
||
a.sort_by_key(|x| order.iter().position(|k| *k == key(x)).unwrap_or(usize::MAX));
|
||
}
|
||
}
|
||
}
|
||
|
||
/// The person's edits since `base`, as operations the brain applies one by one.
|
||
pub(super) fn diff_ops(base: &Value, cur: &Value) -> Vec<Value> {
|
||
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) {
|
||
let p = patch_of(b.get(k).unwrap_or(&Value::Null), v);
|
||
ops.push(if p.get("value").is_some() { json!({"op": "set", "key": k, "value": v}) } else { json!({"op": "patch", "key": k, "patch": p}) });
|
||
}
|
||
}
|
||
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) {
|
||
let p = patch_of(ob.get(k).unwrap_or(&Value::Null), v);
|
||
ops.push(if p.get("value").is_some() { json!({"op": "bot_set", "id": id(nb), "key": k, "value": v}) } else { json!({"op": "bot_patch", "id": id(nb), "key": k, "patch": p}) });
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
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<Value>| a.iter().map(id).collect::<Vec<_>>();
|
||
let (oo, no) = (order(old), order(new));
|
||
let kept: Vec<u64> = oo.iter().copied().filter(|i| no.contains(i)).collect();
|
||
let kept_new: Vec<u64> = no.iter().copied().filter(|i| oo.contains(i)).collect();
|
||
if kept != kept_new {
|
||
ops.push(json!({"op": "bot_order", "ids": no}));
|
||
}
|
||
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();
|
||
}
|
||
}
|
||
"patch" => {
|
||
if let Some(k) = op["key"].as_str() {
|
||
apply_patch(&mut cfg[k], &op["patch"]);
|
||
}
|
||
}
|
||
"bot_patch" => {
|
||
if let (Some(b), Some(k)) = (cfg["bots"].as_array_mut().and_then(|a| a.iter_mut().find(|b| id(b) == target)), op["key"].as_str()) {
|
||
apply_patch(&mut b[k], &op["patch"]);
|
||
}
|
||
}
|
||
"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 ----------
|
||
|
||
/// What the brain last sent each screen.
|
||
#[derive(Default)]
|
||
pub struct Screens {
|
||
seen: HashMap<String, Seen>,
|
||
/// The settings fingerprint, refreshed now and then while something changes.
|
||
pub(super) cfg_hash: u64,
|
||
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)]
|
||
struct Seen {
|
||
version: u64,
|
||
config: u64,
|
||
bots: HashMap<u64, Vec<u64>>,
|
||
threads: HashMap<u64, u64>,
|
||
at: Option<Instant>,
|
||
}
|
||
|
||
// ---------- the screen's side ----------
|
||
|
||
pub enum Incoming {
|
||
Pull(Value),
|
||
Offline(String),
|
||
/// A file from the brain arrived (its local copy's path).
|
||
File(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)]
|
||
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<Mutex<String>>,
|
||
rx: Option<Receiver<Incoming>>,
|
||
tx: Option<Sender<Incoming>>,
|
||
gen: Arc<AtomicU64>,
|
||
version: Arc<AtomicU64>,
|
||
client: String,
|
||
/// This computer's own data, kept aside while it shows the brain.
|
||
local: Option<State>,
|
||
started: bool,
|
||
/// A request to the brain failed (shown above the message box).
|
||
pub(super) error: Arc<Mutex<String>>,
|
||
/// 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<AtomicBool>,
|
||
/// Just signed in: copy this computer's bots to the brain once it answers.
|
||
pub(super) move_after_join: bool,
|
||
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>,
|
||
/// Files from the brain being fetched (by local path).
|
||
fetching: std::collections::HashSet<String>,
|
||
/// 加一台電腦: the line to run on the new computer (or why there is none), and when it was made.
|
||
pub(super) pair: Arc<Mutex<(String, bool, Option<Instant>)>>,
|
||
/// The bots' own computer on the brain: {label, running, available, busy}.
|
||
pub(super) box_state: Value,
|
||
/// Each bot's computers on the brain, by bot id: {names: [{name, kind}], current, missing}.
|
||
pub(super) computers: HashMap<u64, Value>,
|
||
}
|
||
|
||
/// What a computer is, in a word a screen can show next to its name.
|
||
pub(super) fn computer_kind(t: &crate::tools::Target, r: &crate::tools::Roles) -> &'static str {
|
||
use crate::tools::Place;
|
||
if t.name == r.own {
|
||
"own"
|
||
} else if r.person.as_deref() == Some(t.name.as_str()) {
|
||
"person"
|
||
} else {
|
||
match t.place {
|
||
Place::Local { .. } if r.server => "server",
|
||
Place::Local { .. } => "this",
|
||
Place::Device { .. } => "device",
|
||
_ => "ssh",
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Where a screen keeps its copy of a file the brain has.
|
||
fn local_copy(brain_path: &str, name: &str) -> PathBuf {
|
||
let h = crate::httpd::sha256_hex(brain_path);
|
||
let name: String = name.chars().filter(|c| !matches!(c, '/' | '\\' | '\0' | ':')).take(80).collect();
|
||
State::path().with_file_name("brain-files").join(format!("{}-{name}", &h[..16]))
|
||
}
|
||
|
||
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::Models(name) => {
|
||
let kind = serde_json::from_value::<crate::providers::ProviderKind>(json!(name)).ok();
|
||
match kind {
|
||
Some(kind) => {
|
||
let ps = self.state.provider(kind);
|
||
let (base, key) = (crate::providers::resolve_base(kind, &ps.base_url), crate::providers::resolve_key(kind, &ps.api_key));
|
||
let reply = r.reply;
|
||
// The list can take a while; the app doesn't wait for it.
|
||
std::thread::spawn(move || {
|
||
let _ = reply.send(match crate::providers::list_models(kind, &base, &key) {
|
||
Ok(models) => json!({"models": models}),
|
||
Err(e) => json!({"error": e}),
|
||
});
|
||
});
|
||
continue;
|
||
}
|
||
None => json!({"error": "大腦不認得這個模型服務（兩邊 AnyBot 版本不同？）"}),
|
||
}
|
||
}
|
||
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 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);
|
||
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<u64> = 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::<serde_json::Map<_, _>>().into();
|
||
out["approvals"] = self
|
||
.approvals
|
||
.iter()
|
||
.map(|(id, a)| {
|
||
let kind = match a.kind {
|
||
ApprovalKind::Action => "action",
|
||
ApprovalKind::EnableTools => "tools",
|
||
ApprovalKind::EnableThisComputer => "local",
|
||
ApprovalKind::Handoff => "handoff",
|
||
};
|
||
json!({"bot": id, "call_id": a.call_id, "tool": a.tool, "reason": a.reason, "grant": a.grant, "kind": kind})
|
||
})
|
||
.collect();
|
||
// The bots' computer lives here; a screen shows its state, not its own computer's.
|
||
let run = crate::cloudbox::run_state();
|
||
out["box"] = json!({"label": run.label(), "running": run.is_running(), "available": crate::cloudbox::available(), "busy": crate::cloudbox::is_busy()});
|
||
// Which computers each bot has (only the brain knows the account's other computers).
|
||
out["computers"] = self
|
||
.state
|
||
.bots
|
||
.iter()
|
||
.map(|b| {
|
||
let t = self.targets_for(b);
|
||
let r = self.roles_for(b, &t);
|
||
let (current, missing) = Self::chosen_computer(b, &t, &r);
|
||
let names: Vec<Value> = t.iter().map(|x| json!({"name": x.name, "kind": computer_kind(x, &r)})).collect();
|
||
(b.id.to_string(), json!({"names": names, "current": current, "missing": missing}))
|
||
})
|
||
.collect::<serde_json::Map<_, _>>()
|
||
.into();
|
||
out
|
||
}
|
||
|
||
/// A screen's edits.
|
||
pub(super) fn brain_edit(&mut self, body: &Value) -> Value {
|
||
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() {
|
||
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::<Bot>(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);
|
||
self.state.fix_primary();
|
||
true
|
||
}
|
||
Err(_) => false,
|
||
}
|
||
}
|
||
"bot_del" => {
|
||
let id = op["id"].as_u64().unwrap_or(0);
|
||
if self.state.bot(id).is_none() {
|
||
false
|
||
} else {
|
||
self.stop(id);
|
||
self.state.bots.retain(|b| b.id != id);
|
||
self.state.fix_primary();
|
||
true
|
||
}
|
||
}
|
||
"merge" => self.brain_merge_top(op["key"].as_str().unwrap_or(""), &op["value"]),
|
||
"patch" => {
|
||
let key = op["key"].as_str().unwrap_or("");
|
||
let mut v = self.top_value(key);
|
||
apply_patch(&mut v, &op["patch"]);
|
||
self.brain_set_top(key, v)
|
||
}
|
||
"bot_patch" => {
|
||
let (id, key) = (op["id"].as_u64().unwrap_or(0), op["key"].as_str().unwrap_or(""));
|
||
match self.state.bot(id).map(|b| serde_json::to_value(b).ok().map(|v| v[key].clone()).unwrap_or(Value::Null)) {
|
||
Some(mut v) => {
|
||
let old = v.clone();
|
||
apply_patch(&mut v, &op["patch"]);
|
||
unredact(&mut v, &old);
|
||
self.brain_set_bot(id, key, v)
|
||
}
|
||
None => false,
|
||
}
|
||
}
|
||
"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<u64> = 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 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 {
|
||
self.mark();
|
||
// Screens hear of it at once.
|
||
self.scr.cfg_at = None;
|
||
self.brain_watch();
|
||
}
|
||
json!({"ok": true, "applied": applied})
|
||
}
|
||
|
||
/// One top-level setting as JSON (with the brain's real keys).
|
||
fn top_value(&mut self, key: &str) -> Value {
|
||
let bots = std::mem::take(&mut self.state.bots);
|
||
let v = serde_json::to_value(&self.state).map(|t| t[key].clone()).unwrap_or(Value::Null);
|
||
self.state.bots = bots;
|
||
v
|
||
}
|
||
|
||
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::<State>(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);
|
||
let mut value = value;
|
||
if key == "routines" {
|
||
keep_run_times(&mut value, &v["routines"]);
|
||
}
|
||
v[key] = value;
|
||
let ok = match serde_json::from_value::<Bot>(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<u64> = 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::<State>(&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;
|
||
// The bots' computer is the brain's now; this one's box isn't kept awake.
|
||
crate::cloudbox::set_brain(false);
|
||
self.mir.online = false;
|
||
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.mirror_worker();
|
||
self.mirror_outbox();
|
||
crate::tunnel::hello(&self.state.remote.brain_id, self.state.remote.device_link);
|
||
}
|
||
|
||
/// Back to this computer's own bots.
|
||
fn mirror_leave(&mut self) {
|
||
self.mir.gen.fetch_add(1, Ordering::SeqCst);
|
||
self.mir.on = false;
|
||
crate::cloudbox::set_brain(true);
|
||
self.mir.online = false;
|
||
self.mir.outbox = None;
|
||
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();
|
||
remote.hands_for.clear();
|
||
remote.hands_for_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.remote.hands_for.clear();
|
||
self.state.remote.hands_for_name.clear();
|
||
}
|
||
self.state.save();
|
||
self.dirty = false;
|
||
crate::tunnel::hello("", self.state.remote.device_link);
|
||
}
|
||
|
||
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, 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();
|
||
std::thread::spawn(move || {
|
||
let mut backoff = 1u64;
|
||
while g.load(Ordering::SeqCst) == gen {
|
||
let since = if resync.swap(false, Ordering::SeqCst) { 0 } else { version.load(Ordering::SeqCst) };
|
||
match remote::sync_pull(&s, since, &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<dyn Fn() + Send + Sync> {
|
||
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();
|
||
}
|
||
// This computer's hands are for that brain only.
|
||
self.state.remote.hands_for = id.clone();
|
||
self.state.remote.hands_for_name = name.clone();
|
||
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.down_since.get_or_insert_with(Instant::now);
|
||
*self.mir.status.lock().unwrap() = format!("連不到大腦：{e}");
|
||
self.mir.reason = e;
|
||
}
|
||
Incoming::File(path) => {
|
||
self.mir.fetching.remove(&path);
|
||
self.att.thumbs.remove(&path);
|
||
}
|
||
Incoming::Pull(v) => {
|
||
if let Some(e) = v["error"].as_str() {
|
||
self.mir.online = false;
|
||
self.mir.down_since.get_or_insert_with(Instant::now);
|
||
*self.mir.status.lock().unwrap() = e.to_string();
|
||
self.mir.reason = e.to_string();
|
||
continue;
|
||
}
|
||
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.mirror_apply(v);
|
||
if std::mem::take(&mut self.mir.move_after_join) {
|
||
self.mirror_move_local();
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
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() {
|
||
// 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());
|
||
match serde_json::from_value::<State>(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 = base;
|
||
}
|
||
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 mut msgs: Vec<Message> = serde_json::from_value(d["messages"].clone()).unwrap_or_default();
|
||
self.mirror_files(&mut msgs);
|
||
if let Some(b) = self.state.bot_mut(id) {
|
||
// The brain sends what changed from message `from` on, sure this copy has the
|
||
// ones before. A copy that is shorter (cleared here, or an answer lost on the
|
||
// way) would get them appended in the wrong place — the same message twice and
|
||
// half-written replies piling up. Ask for the whole chat again instead.
|
||
if b.messages.len() < from {
|
||
self.mir.resync.store(true, Ordering::SeqCst);
|
||
continue;
|
||
}
|
||
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<u64, (Option<String>, 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,
|
||
"handoff" => ApprovalKind::Handoff,
|
||
_ => 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);
|
||
}
|
||
}
|
||
}
|
||
if v["box"].is_object() {
|
||
self.mir.box_state = v["box"].clone();
|
||
}
|
||
if let Some(c) = v["computers"].as_object() {
|
||
self.mir.computers = c.iter().filter_map(|(k, x)| Some((k.parse().ok()?, x.clone()))).collect();
|
||
}
|
||
let keep: Vec<u64> = 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();
|
||
}
|
||
|
||
/// Files in the brain's chats live on the brain: show local copies, fetched once.
|
||
fn mirror_files(&mut self, msgs: &mut [Message]) {
|
||
for a in msgs.iter_mut().flat_map(|m| m.attachments.iter_mut()) {
|
||
if std::path::Path::new(&a.path).exists() {
|
||
continue;
|
||
}
|
||
let local = local_copy(&a.path, &a.name);
|
||
let lp = local.display().to_string();
|
||
let brain_path = std::mem::replace(&mut a.path, lp.clone());
|
||
if local.exists() || !self.mir.fetching.insert(lp.clone()) {
|
||
continue;
|
||
}
|
||
let (s, tx, repaint) = (self.state.remote.clone(), self.mir.tx.clone(), self.repaint_fn());
|
||
std::thread::spawn(move || {
|
||
if let Ok(bytes) = remote::sync_file(&s, &brain_path) {
|
||
let _ = std::fs::create_dir_all(local.parent().unwrap_or(std::path::Path::new(".")));
|
||
let tmp = local.with_extension("part");
|
||
if std::fs::write(&tmp, bytes).is_ok() {
|
||
let _ = std::fs::rename(&tmp, &local);
|
||
}
|
||
}
|
||
if let Some(tx) = tx {
|
||
let _ = tx.send(Incoming::File(lp));
|
||
}
|
||
repaint();
|
||
});
|
||
}
|
||
}
|
||
|
||
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;
|
||
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 || {
|
||
for o in rx {
|
||
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(_) if n == 0 => "已同步：大腦上已經有這台的 Bot；模型金鑰、技能和連接器有缺的已補上".into(),
|
||
Ok(_) => format!("已把 {n} 個 Bot（連同聊天）搬到大腦"),
|
||
Err(e) => format!("搬移失敗：{e}"),
|
||
};
|
||
busy.store(false, Ordering::SeqCst);
|
||
}
|
||
}
|
||
repaint();
|
||
}
|
||
});
|
||
}
|
||
|
||
/// 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;
|
||
}
|
||
self.mirror_out(Out::Act(body));
|
||
}
|
||
|
||
/// 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<crate::attach::Attachment>) -> 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) });
|
||
}
|
||
}
|
||
// Settings changed just before go first.
|
||
if self.dirty {
|
||
self.mirror_save();
|
||
}
|
||
self.mirror_out(Out::Send { bot, text, atts });
|
||
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<Value> = vec![];
|
||
// Skip what the brain already has: the same starter bot never used here, or a bot moved
|
||
// before (same name, same chat).
|
||
let last = |b: &Bot| b.messages.last().map(|m| (m.at, m.text.clone()));
|
||
let moving: Vec<&Bot> = local
|
||
.bots
|
||
.iter()
|
||
.filter(|b| !self.state.bots.iter().any(|o| o.name == b.name && (b.messages.is_empty() || (o.messages.len() == b.messages.len() && last(o) == last(b)))))
|
||
.collect();
|
||
for b in moving.iter().copied() {
|
||
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 n = moving.len();
|
||
*self.mir.move_note.lock().unwrap() = "搬移中…".into();
|
||
self.mirror_out(Out::Move { ops, n });
|
||
}
|
||
|
||
/// 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;
|
||
}
|
||
// 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.set_max_width(TRANSCRIPT_MAX_WIDTH.min(ui.available_width() - 48.0));
|
||
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.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);
|
||
}
|
||
}
|
||
|
||
#[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"] == "patch" && o["key"] == "folders" && o["patch"]["list"]["put"] == json!(["b"])));
|
||
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");
|
||
}
|
||
|
||
#[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("網路"));
|
||
}
|
||
|
||
#[test]
|
||
fn two_computers_edit_one_list_without_undoing_each_other() {
|
||
let base = json!({"skills": [{"name": "a", "on": true}, {"name": "b", "on": true}], "providers": {"xai": {"model": "1"}, "openai": {"model": "1"}}, "bots": [{"id": 1, "memories": ["x", "y"]}]});
|
||
// Laptop turns skill a off, changes xai, forgets memory x.
|
||
let mut laptop = base.clone();
|
||
laptop["skills"][0]["on"] = json!(false);
|
||
laptop["providers"]["xai"]["model"] = json!("2");
|
||
laptop["bots"][0]["memories"] = json!(["y"]);
|
||
// Meanwhile the desk computer turned skill b off, changed openai, and the bot learned z.
|
||
let mut brain = base.clone();
|
||
brain["skills"][1]["on"] = json!(false);
|
||
brain["skills"].as_array_mut().unwrap().push(json!({"name": "c", "on": true}));
|
||
brain["providers"]["openai"]["model"] = json!("3");
|
||
brain["bots"][0]["memories"] = json!(["x", "y", "z"]);
|
||
let ops = diff_ops(&base, &laptop);
|
||
assert!(ops.iter().all(|o| o["op"] != "set" || o["key"] != "skills"), "{ops:?}");
|
||
apply_ops(&mut brain, &ops);
|
||
assert_eq!(brain["skills"], json!([{"name": "a", "on": false}, {"name": "b", "on": false}, {"name": "c", "on": true}]));
|
||
assert_eq!(brain["providers"], json!({"xai": {"model": "2"}, "openai": {"model": "3"}}));
|
||
assert_eq!(brain["bots"][0]["memories"], json!(["y", "z"]));
|
||
}
|
||
|
||
#[test]
|
||
fn lists_without_names_are_sent_whole() {
|
||
let p = patch_of(&json!([[1], [2]]), &json!([[2]]));
|
||
assert_eq!(p, json!({"value": [[2]]}));
|
||
let mut cur = json!([{"id": 1}, {"id": 2}]);
|
||
apply_patch(&mut cur, &patch_of(&json!([{"id": 1}, {"id": 2}]), &json!([{"id": 2}, {"id": 1}])));
|
||
assert_eq!(cur, json!([{"id": 2}, {"id": 1}]));
|
||
}
|
||
}
|