diff --git a/src/httpd.rs b/src/httpd.rs index 461b3e9..eac8782 100644 --- a/src/httpd.rs +++ b/src/httpd.rs @@ -90,23 +90,34 @@ fn reason(status: u16) -> &'static str { } fn decode(s: &str) -> String { + // Work on bytes only: slicing the &str at byte offsets can land inside a multi-byte + // character and panic (an unauthenticated request like "/%)…" used to kill its thread). let b = s.as_bytes(); let mut out = Vec::with_capacity(b.len()); + let hex = |c: u8| -> Option { + match c { + b'0'..=b'9' => Some(c - b'0'), + b'a'..=b'f' => Some(c - b'a' + 10), + b'A'..=b'F' => Some(c - b'A' + 10), + _ => None, + } + }; let mut i = 0; while i < b.len() { match b[i] { - b'+' => out.push(b' '), - b'%' if i + 2 < b.len() => { - if let Ok(v) = u8::from_str_radix(&s[i + 1..i + 3], 16) { - out.push(v); - i += 2; - } else { - out.push(b'%'); - } + b'+' => { + out.push(b' '); + i += 1; + } + b'%' if i + 2 < b.len() && hex(b[i + 1]).is_some() && hex(b[i + 2]).is_some() => { + out.push(hex(b[i + 1]).unwrap() * 16 + hex(b[i + 2]).unwrap()); + i += 3; + } + c => { + out.push(c); + i += 1; } - c => out.push(c), } - i += 1; } String::from_utf8_lossy(&out).to_string() } @@ -143,19 +154,61 @@ pub fn read_request(c: &mut TcpStream) -> Result { let method = fp.next().unwrap_or("").to_string(); let target = fp.next().unwrap_or("/").to_string(); let headers: HashMap = lines.filter_map(|l| l.split_once(':')).map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string())).collect(); - let len: usize = headers.get("content-length").and_then(|v| v.parse().ok()).unwrap_or(0); - if len > MAX_BODY { - return Err(413); - } + let chunked = headers.get("transfer-encoding").map(|v| v.to_ascii_lowercase().contains("chunked")).unwrap_or(false); let mut body = got[head_end + 4..].to_vec(); - while body.len() < len { - let n = c.read(&mut buf).map_err(|_| 400u16)?; - if n == 0 { - break; + if chunked { + // Some phone browsers send POST bodies in chunks (no Content-Length): read chunks + // until the zero-size terminator, reassembling the body. + let mut out: Vec = vec![]; + let mut pos = 0usize; + loop { + let size_end = loop { + if let Some(p) = body[pos..].windows(2).position(|w| w == b"\r\n") { + break pos + p; + } + if body.len() > MAX_BODY + 64 * 1024 { + return Err(413); + } + let n = c.read(&mut buf).map_err(|_| 400u16)?; + if n == 0 { + return Err(400); + } + body.extend_from_slice(&buf[..n]); + }; + let hex = String::from_utf8_lossy(&body[pos..size_end]).split(';').next().unwrap_or("0").trim().to_string(); + let size = usize::from_str_radix(&hex, 16).map_err(|_| 400u16)?; + if size == 0 { + break; + } + if out.len() + size > MAX_BODY { + return Err(413); + } + let end = size_end + 2 + size; + while body.len() < end + 2 { + let n = c.read(&mut buf).map_err(|_| 400u16)?; + if n == 0 { + return Err(400); + } + body.extend_from_slice(&buf[..n]); + } + out.extend_from_slice(&body[size_end + 2..end]); + pos = end + 2; } - body.extend_from_slice(&buf[..n]); + body = out; + } else { + let len: usize = headers.get("content-length").and_then(|v| v.parse().ok()).unwrap_or(0); + if len > MAX_BODY { + return Err(413); + } + while body.len() < len { + let n = c.read(&mut buf).map_err(|_| 400u16)?; + if n == 0 { + break; + } + body.extend_from_slice(&buf[..n]); + } + body.truncate(len); } - body.truncate(len); let (path, q) = target.split_once('?').unwrap_or((&target, "")); Ok(Req { method, path: decode(path), query: parse_query(q), headers, body, peer: c.peer_addr().map(|a| a.ip().to_string()).unwrap_or_default() }) } diff --git a/src/model.rs b/src/model.rs index edb9e45..b6e4633 100644 --- a/src/model.rs +++ b/src/model.rs @@ -407,10 +407,21 @@ impl State { } pub fn load() -> State { - let mut s = std::fs::read_to_string(Self::path()) - .ok() - .and_then(|s| serde_json::from_str::(&s).ok()) - .unwrap_or_else(State::first_run); + let p = Self::path(); + let raw = std::fs::read_to_string(&p); + let parsed = raw.as_ref().ok().and_then(|s| serde_json::from_str::(s).ok()); + let mut s = match parsed { + Some(s) => s, + None => { + // A damaged save must not vanish quietly: keep the bytes (so the person can + // recover them), then start fresh. Silently wiping left no trace before. + if let Ok(raw) = raw.as_ref() { + let damaged = p.with_extension(format!("json.damaged-{}", chrono::Utc::now().timestamp())); + let _ = std::fs::write(&damaged, raw); + } + State::first_run() + } + }; s.drop_untouched_samples(); for b in s.bots.iter_mut() { b.repair_interrupted(); @@ -426,8 +437,13 @@ impl State { } if let Ok(s) = serde_json::to_string_pretty(self) { let tmp = p.with_extension("json.tmp"); - if std::fs::write(&tmp, s).is_ok() { - let _ = std::fs::rename(&tmp, &p); + // Flush the bytes before renaming, so a power cut cannot leave a renamed-but-empty + // file where load() would find it (and treat it as a first run). + if let Ok(mut f) = std::fs::File::create(&tmp) { + use std::io::Write; + if f.write_all(s.as_bytes()).and_then(|_| f.sync_all()).is_ok() { + let _ = std::fs::rename(&tmp, &p); + } } } } @@ -483,12 +499,15 @@ impl State { pub fn take_in_bots(&mut self, other: &State) -> usize { let last = |b: &Bot| b.messages.last().map(|m| (m.at, m.text.clone())); let mut n = 0; + let mut reid: Vec<(u64, u64)> = vec![]; for b in &other.bots { if self.bots.iter().any(|o| o.name == b.name && (b.messages.is_empty() || (o.messages.len() == b.messages.len() && last(o) == last(b)))) { continue; } let mut b = b.clone(); + let old_id = b.id; b.id = self.next_id.max(1); + reid.push((old_id, b.id)); self.next_id = b.id + 1; for r in b.routines.iter_mut() { r.id = self.next_routine_id.max(1); @@ -497,6 +516,20 @@ impl State { self.bots.push(b); n += 1; } + // Group members are bot ids from the other machine: follow them to the copies that + // just landed, or a group answers with bots that have nothing to do with it. + if !reid.is_empty() { + for g in self.bots.iter_mut() { + if g.members.is_empty() { + continue; + } + for m in g.members.iter_mut() { + if let Some((_, new)) = reid.iter().find(|(o, _)| *o == *m) { + *m = *new; + } + } + } + } // Keys, skills and connectors this computer doesn't have yet. for (k, v) in &other.providers { let mine = self.providers.entry(k.clone()).or_default(); diff --git a/src/portal.rs b/src/portal.rs index 47c5506..56a7feb 100644 --- a/src/portal.rs +++ b/src/portal.rs @@ -37,10 +37,16 @@ pub struct Screen { struct Live { tx: Sender, screens: Vec, + /// Which started session this is, so its own thread (and only it) can clear LIVE when it ends. + id: u64, } static LIVE: Mutex> = Mutex::new(None); static LAST_ERROR: Mutex = Mutex::new(String::new()); +/// One start at a time: two concurrent ensure() calls used to open two portal sessions, and +/// the first to end then cleared the second's LIVE entry (the window went blind). +static ENSURE: Mutex<()> = Mutex::new(()); +static NEXT_SESSION: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); fn token_file() -> std::path::PathBuf { crate::model::State::path().parent().map(|d| d.join("wayland-remote-token.txt")).unwrap_or_default() @@ -63,9 +69,11 @@ pub fn last_error() -> String { /// Start the session if it isn't running (the first time, the desktop asks the person). fn ensure() -> Option> { + let _start = ENSURE.lock().unwrap_or_else(|e| e.into_inner()); if let Some(l) = LIVE.lock().unwrap().as_ref() { return Some(l.screens.clone()); } + let id = NEXT_SESSION.fetch_add(1, std::sync::atomic::Ordering::Relaxed); let (ready_tx, ready_rx) = channel::, String>>(); let (tx, rx) = channel::(); std::thread::spawn(move || { @@ -147,12 +155,17 @@ fn ensure() -> Option> { } } let _ = session.close().await; - *LIVE.lock().unwrap() = None; + // Only the session that is still the current one retires the slot (an older + // session ending must not wipe the newer one that replaced it). + let mut live = LIVE.lock().unwrap(); + if live.as_ref().map(|l| l.id) == Some(id) { + *live = None; + } }); }); match ready_rx.recv_timeout(Duration::from_secs(120)) { Ok(Ok(screens)) => { - *LIVE.lock().unwrap() = Some(Live { tx, screens: screens.clone() }); + *LIVE.lock().unwrap() = Some(Live { tx, screens: screens.clone(), id }); Some(screens) } Ok(Err(e)) => { diff --git a/src/providers.rs b/src/providers.rs index 43e3fa1..594951f 100644 --- a/src/providers.rs +++ b/src/providers.rs @@ -188,7 +188,9 @@ pub enum StreamEvent { /// The next model step starts: add a fresh reply bubble. NextStep(String), /// The oldest `n` messages were folded into this summary. - Compacted(usize, Message), + /// Older messages were folded into this summary; `cut` is the count in the model's view and + /// `boundary` the timestamp of the first kept message (the window's own copy may be longer). + Compacted(usize, Message, i64), /// A message the person sent mid-turn was handed to the model (add it to the chat). Steered(Message), /// Do something in the app (routines, bots, settings) and answer on the sender. @@ -275,6 +277,10 @@ fn agent() -> ureq::Agent { .http_status_as_error(false) .timeout_connect(Some(Duration::from_millis(CONNECT_TIMEOUT_MS))) .timeout_recv_response(Some(Duration::from_secs(FIRST_RESPONSE_TIMEOUT_SECS))) + // A stream that goes silent (a NAT that dropped the mapping, a server that died without + // a FIN) must end: without a read timeout the lines() loop blocks forever, and every + // such turn leaks a thread and its socket. 5 minutes of total silence is a dead stream. + .timeout_per_read(Some(Duration::from_secs(300))) .build(); config.into() } @@ -549,6 +555,19 @@ pub fn set_eyes(v: Vec) { *EYES.lock().unwrap() = v; } +/// The helper list, the description cache and the failure list are process-wide: tests that +/// touch them must run one at a time, from a clean slate, or they undo each other (the +/// describe cache hit in one test made another's helper never get called). +#[cfg(test)] +pub fn vision_test_guard() -> std::sync::MutexGuard<'static, ()> { + static L: std::sync::Mutex<()> = std::sync::Mutex::new(()); + let g = L.lock().unwrap_or_else(|e| e.into_inner()); + notes().lock().unwrap().clear(); + failed_eyes().lock().unwrap().clear(); + set_eyes(vec![]); + g +} + /// Models likely to see pictures, by provider. pub fn seeing_models(kind: ProviderKind) -> &'static [&'static str] { match kind { @@ -828,7 +847,20 @@ fn stream_openai_compatible(req: &ChatRequest, cancel: &AtomicBool, send: &dyn F } if let Some(calls) = delta.get("tool_calls").and_then(Value::as_array) { for c in calls { - let i = c["index"].as_u64().unwrap_or(pending.len() as u64) as usize; + let i = match c["index"].as_u64() { + Some(i) => i as usize, + // Some OpenAI-compatible proxies drop the index: a part carrying an + // id or a name starts a new call; an arguments-only part continues the + // last one (starting a fresh slot used to lose the call's tail). + None => { + let starts = c["id"].is_string() || c["function"]["name"].is_string(); + if starts || pending.is_empty() { + pending.len() + } else { + pending.len() - 1 + } + } + }; while pending.len() <= i { pending.push(Default::default()); } @@ -1365,6 +1397,7 @@ mod tests { #[test] fn a_model_that_cannot_see_gets_no_pictures() { + let _g = vision_test_guard(); assert!(image_refused("錯誤 400:At most 0 image(s) may be provided in one prompt. (parameter=image)")); assert!(image_refused("This model does not support image input")); assert!(image_refused("unknown variant `image_url`, expected `text`")); @@ -1455,6 +1488,7 @@ mod tests { #[test] fn a_model_that_sees_describes_pictures_for_one_that_cannot() { + let _g = vision_test_guard(); let (blind_url, blind_rx) = fake_model(true, "好的,我看到描述了"); let (eyes_url, eyes_rx) = fake_model(false, "一張報表截圖,上面寫著「十月營收 42 萬」。"); set_eyes(vec![ diff --git a/src/web.rs b/src/web.rs index e0e3520..e3ecdbc 100644 --- a/src/web.rs +++ b/src/web.rs @@ -178,7 +178,9 @@ pub fn normalize_url(u: &str) -> Result { return Err("請給網址".into()); } let full = if u.starts_with("http://") || u.starts_with("https://") { u.to_string() } else { format!("https://{u}") }; - if full.contains(char::is_whitespace) || !full[8..].contains('.') && !full.contains("localhost") { + // Check the host part after the scheme (a fixed [8..] cut one byte short for http://). + let rest = full.split_once("://").map(|(_, r)| r).unwrap_or(&full); + if full.contains(char::is_whitespace) || !rest.contains('.') && !rest.starts_with("localhost") { return Err(format!("「{u}」看起來不是網址")); } Ok(full)