Stream deltas keep tool-call slots; chunked bodies accepted; portal sessions race-safe

src/httpd.rs

@@ -90,24 +90,35 @@ 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<u8> {
+        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'+' => {
-            b'%' if i + 2 < b.len() => {
+                out.push(b' ');
-                if let Ok(v) = u8::from_str_radix(&s[i + 1..i + 3], 16) {
-                    out.push(v);
-                    i += 2;
-                } else {
-                    out.push(b'%');
-                }
-            }
-            c => out.push(c),
-        }
                 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;
+            }
+        }
+    }
     String::from_utf8_lossy(&out).to_string()
 }
 
@@ -143,11 +154,52 @@ pub fn read_request(c: &mut TcpStream) -> Result<Req, u16> {
     let method = fp.next().unwrap_or("").to_string();
     let target = fp.next().unwrap_or("/").to_string();
     let headers: HashMap<String, String> = lines.filter_map(|l| l.split_once(':')).map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string())).collect();
+    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();
+    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<u8> = 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 = 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);
         }
-    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 {
@@ -156,6 +208,7 @@ pub fn read_request(c: &mut TcpStream) -> Result<Req, u16> {
             body.extend_from_slice(&buf[..n]);
         }
         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() })
 }

src/model.rs

@@ -407,10 +407,21 @@ impl State {
     }
 
     pub fn load() -> State {
-        let mut s = std::fs::read_to_string(Self::path())
+        let p = Self::path();
-            .ok()
+        let raw = std::fs::read_to_string(&p);
-            .and_then(|s| serde_json::from_str::<State>(&s).ok())
+        let parsed = raw.as_ref().ok().and_then(|s| serde_json::from_str::<State>(s).ok());
-            .unwrap_or_else(State::first_run);
+        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,11 +437,16 @@ 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() {
+            // 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);
                 }
             }
         }
+    }
 
     pub fn provider(&self, kind: ProviderKind) -> ProviderSettings {
         self.providers.get(kind.key()).cloned().unwrap_or_default()
@@ -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();

src/portal.rs

@@ -37,10 +37,16 @@ pub struct Screen {
 struct Live {
     tx: Sender<Cmd>,
     screens: Vec<Screen>,
+    /// Which started session this is, so its own thread (and only it) can clear LIVE when it ends.
+    id: u64,
 }
 
 static LIVE: Mutex<Option<Live>> = Mutex::new(None);
 static LAST_ERROR: Mutex<String> = 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<Vec<Screen>> {
+    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::<Result<Vec<Screen>, String>>();
     let (tx, rx) = channel::<Cmd>();
     std::thread::spawn(move || {
@@ -147,12 +155,17 @@ fn ensure() -> Option<Vec<Screen>> {
                 }
             }
             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)) => {

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>) {
     *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![

src/web.rs

@@ -178,7 +178,9 @@ pub fn normalize_url(u: &str) -> Result<String, String> {
         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)