Merge pull request #3832 from c020627/fix-linkai-bounded-download

fix(linkai): cap the file download in the LinkAI sender

agent/memory/conversation_store.py

@@ -1690,14 +1690,10 @@ class ConversationStore:
         the scheduler threads the push into (``_remember_delivered_output``), so
         this joins back to that session to recover the full body.
 
-        Correlation is by ``(session_id, agent_id)`` from the run, then the
-        scheduler-injected assistant message whose ``created_at`` is closest at
-        or after the run's ``started_at`` (the pair this run wrote). It is
-        best-effort: the session copy is itself capped (~2000 chars) and old
-        scheduler pairs are pruned, so ``full_output`` may be ``None`` — the
-        caller falls back to the preview. Messages are scoped to the *run's*
-        ``agent_id``, not this store's bound id, since a run may belong to any
-        Agent in the global ledger.
+        Correlation uses the scheduler run id persisted on the delivered pair,
+        scoped to its Agent and receiver session. If that pair was pruned or
+        predates run-id association, leave ``full_output`` unset so the caller
+        uses this run's durable preview instead of another task's message.
         """
         run = self.get_run(run_id)
         if run is None:
@@ -1718,45 +1714,39 @@ class ConversationStore:
             return detail
 
         run_agent_id = run.get("agent_id") or ""
-        started_at = run.get("started_at") or 0
+        try:
+            from agent.registry import get_agent_registry
+            if run_agent_id == get_agent_registry().default_agent_id:
+                run_agent_id = ""
+        except Exception:
+            pass
         with self._lock:
             conn = self._connect()
             try:
                 rows = conn.execute(
                     """
-                    SELECT seq, role, content, created_at
+                    SELECT role, content
                     FROM messages
-                    WHERE agent_id = ? AND session_id = ?
+                    WHERE agent_id = ? AND session_id = ? AND run_id = ?
                     ORDER BY seq ASC
                     """,
-                    (run_agent_id, session_id),
+                    (run_agent_id, session_id, run_id),
                 ).fetchall()
             except sqlite3.OperationalError:
                 rows = []
             finally:
                 conn.close()
 
-        # Assistant messages that answer a [SCHEDULED] user turn are the delivered
-        # bodies. Pick the one written closest at/after this run started; fall
-        # back to the newest scheduler body if none lands after the timestamp
-        # (clock skew, or the pair persisted a hair before the run row).
-        candidates: List[tuple] = []  # (created_at, text)
-        for i, (seq, role, raw, created_at) in enumerate(rows):
-            if role != "user":
+        candidates = []
+        for index, (role, raw) in enumerate(rows):
+            if role != "user" or not _message_is_scheduled_marker(raw):
                 continue
-            if not _message_is_scheduled_marker(raw):
-                continue
-            if i + 1 < len(rows) and rows[i + 1][1] == "assistant":
-                text = _message_plain_text(rows[i + 1][2])
+            if index + 1 < len(rows) and rows[index + 1][0] == "assistant":
+                text = _message_plain_text(rows[index + 1][1])
                 if text:
-                    candidates.append((rows[i + 1][3] or 0, text))
-
-        if candidates:
-            after = [c for c in candidates if c[0] >= started_at]
-            chosen = min(after, key=lambda c: c[0]) if after else max(
-                candidates, key=lambda c: c[0]
-            )
-            detail["full_output"] = chosen[1]
+                    candidates.append(text)
+        if len(candidates) == 1:
+            detail["full_output"] = candidates[0]
         return detail
 
     def load_history_page(

agent/tools/scheduler/integration.py

@@ -593,6 +593,7 @@ def _remember_delivered_output(
     channel_type: str,
     content: str,
     agent_id: str = None,
+    run_id: str = "",
 ) -> None:
     """Best-effort persistence of the message the scheduler sent to a user.
 
@@ -637,6 +638,18 @@ def _remember_delivered_output(
             }
             if hasattr(agent_bridge, "agent_registry"):
                 kwargs["agent_id"] = agent_id
+            if run_id:
+                import inspect
+
+                try:
+                    params = inspect.signature(remember).parameters.values()
+                    if any(
+                        p.name == "run_id" or p.kind == inspect.Parameter.VAR_KEYWORD
+                        for p in params
+                    ):
+                        kwargs["run_id"] = run_id
+                except (TypeError, ValueError):
+                    pass
             remember(session_id, str(content), **kwargs)
     except Exception as e:
         logger.warning(
@@ -732,7 +745,7 @@ def _run_scheduled_task(
         return False
 
     run = _record_scheduler_run(task, agent_id, trigger=trigger)
-    sink: Dict[str, str] = {}
+    sink: Dict[str, str] = {"run_id": run[1]} if run is not None else {}
     status = "done"
     error = ""
     try:
@@ -888,7 +901,8 @@ def _execute_agent_task(
             if output_sink is not None:
                 output_sink["preview"] = str(reply.content)
             _remember_delivered_output(
-                agent_bridge, task, channel_type, reply.content, agent_id
+                agent_bridge, task, channel_type, reply.content, agent_id,
+                run_id=(output_sink or {}).get("run_id", ""),
             )
             logger.info(f"[Scheduler] Task {task['id']} executed successfully, result sent to {receiver}")
             return True
@@ -980,7 +994,8 @@ def _execute_send_message(
         if output_sink is not None:
             output_sink["preview"] = str(content)
         _remember_delivered_output(
-            agent_bridge, task, channel_type, content, agent_id
+            agent_bridge, task, channel_type, content, agent_id,
+            run_id=(output_sink or {}).get("run_id", ""),
         )
         logger.info(f"[Scheduler] Task {task['id']} executed: sent message to {receiver}")
         return True
@@ -1060,7 +1075,8 @@ def _execute_tool_call(
         if output_sink is not None:
             output_sink["preview"] = str(content)
         _remember_delivered_output(
-            agent_bridge, task, channel_type, content, agent_id
+            agent_bridge, task, channel_type, content, agent_id,
+            run_id=(output_sink or {}).get("run_id", ""),
         )
         logger.info(f"[Scheduler] Task {task['id']} executed: sent tool result to {receiver}")
         return True
@@ -1164,7 +1180,8 @@ def _execute_skill_call(
         if output_sink is not None:
             output_sink["preview"] = str(summary)
         _remember_delivered_output(
-            agent_bridge, task, channel_type, summary, agent_id
+            agent_bridge, task, channel_type, summary, agent_id,
+            run_id=(output_sink or {}).get("run_id", ""),
         )
         logger.info(f"[Scheduler] Task {task['id']} executed: skill result sent to {receiver}")
         return True

agent/tools/utils/file_state.py

@@ -2,7 +2,9 @@
 Cross-tool record of when the agent last read each file.
 
 The tool manager builds a fresh tool instance per call, so this state has to
-live at module level to be shared between read, edit and write.
+live at module level to be shared between read, edit and write. Observations
+are scoped to the current conversation: another conversation reading the same
+file must not replace the snapshot this agent actually saw.
 
 It powers a staleness warning: if a file changed on disk after the agent read
 it, the agent is probably about to overwrite someone else's change (the user
@@ -17,12 +19,12 @@ import threading
 from collections import OrderedDict
 from typing import Optional
 
-# Plenty for one conversation; bounded so a long-running process can't grow
+# Bound all conversations together so a long-running process cannot grow
 # without limit.
 _MAX_TRACKED = 512
 
 _lock = threading.Lock()
-_read_mtimes = OrderedDict()  # realpath -> mtime at the time we read it
+_read_mtimes = OrderedDict()  # conversation + realpath -> observed mtime
 
 
 def _key(path: str) -> Optional[str]:
@@ -32,6 +34,15 @@ def _key(path: str) -> Optional[str]:
         return None
 
 
+def _observation_key(path: str) -> tuple:
+    from common.runtime_identity import current_identity
+
+    identity = current_identity()
+    # Keep observations across turns of the same conversation. A fresh run id
+    # must not discard a file view still present in that conversation's context.
+    return identity.agent_id, identity.user_id, identity.session_id, path
+
+
 def _current_mtime(path: str) -> Optional[float]:
     try:
         return os.path.getmtime(path)
@@ -47,9 +58,10 @@ def note_read(path: str) -> None:
     mtime = _current_mtime(key)
     if mtime is None:
         return
+    observation = _observation_key(key)
     with _lock:
-        _read_mtimes[key] = mtime
-        _read_mtimes.move_to_end(key)
+        _read_mtimes[observation] = mtime
+        _read_mtimes.move_to_end(observation)
         while len(_read_mtimes) > _MAX_TRACKED:
             _read_mtimes.popitem(last=False)
 
@@ -70,7 +82,7 @@ def staleness_warning(path: str) -> Optional[str]:
     if not key:
         return None
     with _lock:
-        seen = _read_mtimes.get(key)
+        seen = _read_mtimes.get(_observation_key(key))
     if seen is None:
         return None
     current = _current_mtime(key)

bridge/agent_bridge.py

@@ -2212,6 +2212,7 @@ class AgentBridge:
         channel_type: str = "",
         task_description: str = "",
         agent_id: str = None,
+        run_id: str = "",
     ) -> None:
         """Add the visible output of a scheduled task to the receiver's session.
 
@@ -2227,6 +2228,9 @@ class AgentBridge:
                 Maximum scheduler-injected user/assistant pairs retained per
                 session. Older injections are pruned automatically.
 
+        ``run_id`` associates the stored pair with its scheduler execution;
+        callers without an id remain supported without guessing association.
+
         Content is truncated to 4000 chars to prevent a single high-volume task
         from bloating one entry, while staying long enough that the history
         detail view (which recovers this copy) shows the full message for the
@@ -2251,9 +2255,14 @@ class AgentBridge:
             {"role": "assistant", "content": [{"type": "text", "text": content}]},
         ]
 
+        # Association is storage metadata, not a field in model messages.
+        persisted_messages = (
+            [dict(message, run_id=run_id) for message in messages] if run_id else messages
+        )
+
         # Persist first so the new pair gets a stable seq, then prune old
         # scheduler pairs in DB, then sync the in-memory agent.messages buffer.
-        self._persist_messages(session_id, messages, channel_type, agent_id)
+        self._persist_messages(session_id, persisted_messages, channel_type, agent_id)
 
         keep_last_n = max(int(conf().get("scheduler_inject_max_per_session", 3) or 0), 0)
         try:

channel/chat_channel.py

@@ -242,19 +242,24 @@ class ChatChannel(Channel):
                     any_to_wav(file_path, wav_path)
                 except Exception as e:  # 转换失败，直接使用mp3，对于某些api，mp3也可以识别
                     logger.warning("[chat_channel]any to wav error, use raw path. " + str(e))
-                    wav_path = file_path
+                    # Keep wav_path pointing at the file the conversion may
+                    # have created: pydub opens the destination wb+ before it
+                    # runs ffmpeg, so an encoder that exits nonzero leaves a
+                    # file there, and reassigning the name here orphaned it in
+                    # the tmp dir for good.
+                    stt_path = file_path
+                else:
+                    stt_path = wav_path
                 # 语音识别
-                reply = self._safe_voice_to_text(wav_path)
+                reply = self._safe_voice_to_text(stt_path)
                 if reply is None:
                     reply = Reply(ReplyType.ERROR, "语音识别失败，请重试或发送文字消息")
-                # 删除临时文件
-                try:
-                    os.remove(file_path)
-                    if wav_path != file_path:
-                        os.remove(wav_path)
-                except Exception:
-                    pass
-                    # logger.warning("[chat_channel]delete temp file error: " + str(e))
+                # 删除临时文件：逐个删除，一个失败不会连带跳过另一个
+                for path in {file_path, wav_path}:
+                    try:
+                        os.remove(path)
+                    except Exception:
+                        pass
 
                 if reply.type == ReplyType.TEXT:
                     new_context = self._compose_context(ContextType.TEXT, reply.content, **context.kwargs)

channel/dingtalk/dingtalk_channel.py

@@ -25,7 +25,7 @@ from bridge.reply import Reply, ReplyType
 from channel.chat_channel import ChatChannel
 from channel.chat_message import safe_filename
 from common import state_dir
-from common.media_download import MAX_FILE_BYTES, MediaTooLargeError, download_to_file
+from common.media_download import MAX_FILE_BYTES, MediaTooLargeError, download_to_file, remove_download
 from channel.dingtalk.dingtalk_message import DingTalkMessage
 from channel.dingtalk.dingtalk_stream_card import (
     DingTalkCardStreamer,
@@ -449,6 +449,7 @@ class DingTalkChanel(ChatChannel, dingtalk_stream.ChatbotHandler):
             file_path = file_path[7:]
         
         # 如果是 HTTP URL，先下载（带字节上限，防止超大响应耗尽内存/磁盘）
+        temp_file = None
         if file_path.startswith("http://") or file_path.startswith("https://"):
             try:
                 import uuid
@@ -458,7 +459,10 @@ class DingTalkChanel(ChatChannel, dingtalk_stream.ChatbotHandler):
                     safe_filename(unquote(os.path.basename(urlparse(file_path).path)))
                     or f"media_{uuid.uuid4()}"
                 )
-                temp_file = os.path.join(str(state_dir.tmp_dir()), file_name)
+                # A generated unique name, not the URL's: the tmp dir also holds
+                # inbound attachments, so two downloads of the same URL would
+                # overwrite each other under a name taken from the URL.
+                temp_file = state_dir.tmp_file("dingtalk_media", os.path.splitext(file_name)[1])
                 try:
                     download_to_file(
                         file_path, temp_file, MAX_FILE_BYTES,
@@ -502,6 +506,11 @@ class DingTalkChanel(ChatChannel, dingtalk_stream.ChatbotHandler):
         except Exception as e:
             logger.error(f"[DingTalk] Error uploading media: {e}")
             return None
+        finally:
+            # A remote download belongs to this call to clean up, whatever the
+            # upload did with it -- the way the weixin / wecom_bot senders do.
+            if temp_file:
+                remove_download(temp_file)
     
     def send_image_with_media_id(self, access_token: str, media_id: str, incoming_message, is_group: bool) -> bool:
         """
@@ -1026,31 +1035,43 @@ class DingTalkChanel(ChatChannel, dingtalk_stream.ChatbotHandler):
 
             amr_path = voice_path
             duration_ms = 0
-            if not voice_path.lower().endswith((".amr", ".ogg")):
-                try:
-                    from voice.audio_convert import any_to_amr
-                    amr_path = os.path.splitext(voice_path)[0] + ".amr"
-                    duration_ms = int(any_to_amr(voice_path, amr_path) or 0)
-                except Exception as e:
-                    logger.error(f"[DingTalk] Failed to convert voice to amr: {e}")
-                    self.reply_text("抱歉，语音转码失败", incoming_message)
+            # The amr this send creates is ours to clean up; the TTS source
+            # beside it belongs to voice synthesis and is left in place, the
+            # way the weixin / wecom_bot senders leave a local source alone.
+            temp_files = []
+            try:
+                if not voice_path.lower().endswith((".amr", ".ogg")):
+                    try:
+                        from voice.audio_convert import any_to_amr
+                        amr_path = os.path.splitext(voice_path)[0] + ".amr"
+                        # Register before converting: pydub opens the target
+                        # file before it spawns ffmpeg, so a conversion that
+                        # dies partway leaves a half-written amr behind too.
+                        temp_files.append(amr_path)
+                        duration_ms = int(any_to_amr(voice_path, amr_path) or 0)
+                    except Exception as e:
+                        logger.error(f"[DingTalk] Failed to convert voice to amr: {e}")
+                        self.reply_text("抱歉，语音转码失败", incoming_message)
+                        return
+
+                media_id = self.upload_media(amr_path, media_type="voice")
+                if not media_id:
+                    logger.error("[DingTalk] Failed to upload voice media")
+                    self.reply_text("抱歉，语音上传失败", incoming_message)
                     return
 
-            media_id = self.upload_media(amr_path, media_type="voice")
-            if not media_id:
-                logger.error("[DingTalk] Failed to upload voice media")
-                self.reply_text("抱歉，语音上传失败", incoming_message)
-                return
-
-            msg_param = {
-                "mediaId": media_id,
-                "duration": str(duration_ms or 1000),
-            }
-            success = self._send_file_message(
-                access_token, incoming_message, "sampleAudio", msg_param, isgroup
-            )
-            if not success:
-                self.reply_text("抱歉，语音发送失败", incoming_message)
+                msg_param = {
+                    "mediaId": media_id,
+                    "duration": str(duration_ms or 1000),
+                }
+                success = self._send_file_message(
+                    access_token, incoming_message, "sampleAudio", msg_param, isgroup
+                )
+                if not success:
+                    self.reply_text("抱歉，语音发送失败", incoming_message)
+            finally:
+                for path in temp_files:
+                    remove_download(path)
             return
 
         # 处理文本消息

channel/web/core/_common.py

@@ -833,8 +833,8 @@ def _live_channel_manager():
 def _serve_allowed_roots() -> list:
     """Roots that /api/file and /preview may read from (symlinks resolved).
 
-    Includes the configured serve root, the Agent workspace, and any project
-    directory a session has opened. Project dirs may live outside the serve
+    Includes the configured serve root, all configured Agent workspaces
+    (including disabled Agents), and any project directory a session has opened. Project dirs may live outside the serve
     root (e.g. ``/tmp/foo``), so previewing files in an opened project would
     otherwise be denied.
     """
@@ -843,6 +843,14 @@ def _serve_allowed_roots() -> list:
         os.path.realpath(os.path.expanduser(serve_root)),
         os.path.realpath(_get_workspace_root()),
     ]
+    # Artifact URLs are global, and the artifact index includes disabled
+    # Agents too. Their explicitly configured workspaces remain readable.
+    try:
+        from agent.registry import get_agent_registry
+        for profile in get_agent_registry().list(include_disabled=True):
+            roots.append(os.path.realpath(profile.workspace))
+    except Exception:
+        pass
     try:
         from agent.workspace import project_store
         for rec in project_store.list_recents():

channel/web/static/js/chat/message-actions.js

@@ -121,6 +121,26 @@ function copyToClipboard(text) {
     });
 }
 
+// Keep visible history in sync with the store before editing, regenerating,
+// or deleting a persisted turn. Callers keep their existing deletion options.
+async function deletePersistedMessage(userSeq, options = {}) {
+    try {
+        const resp = await fetch('/api/messages/delete', {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ session_id: sessionId, user_seq: parseInt(userSeq), ...options })
+        });
+        if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
+        const data = await resp.json();
+        if (data.status !== 'success') throw new Error(data.message || 'Delete failed');
+        return true;
+    } catch (err) {
+        console.error('Failed to delete messages:', err);
+        alert(t('records_delete_failed'));
+        return false;
+    }
+}
+
 // Edit user message: extract content, remove this and subsequent messages, fill input
 async function editUserMessage(msgEl) {
     if (isCurrentSessionConversationActive()) return;
@@ -130,24 +150,7 @@ async function editUserMessage(msgEl) {
     // Delete this message and ALL subsequent messages from database (cascade)
     // Must await to ensure delete completes before user sends a new message
     const userSeq = msgEl.dataset.seq;
-    if (userSeq) {
-        try {
-            const resp = await fetch('/api/messages/delete', {
-                method: 'POST',
-                headers: { 'Content-Type': 'application/json' },
-                body: JSON.stringify({ 
-                    session_id: sessionId, 
-                    user_seq: parseInt(userSeq),
-                    delete_user: true,
-                    cascade: true
-                })
-            });
-            const data = await resp.json();
-            if (data.status === 'success') console.log(`Deleted ${data.deleted} old messages`);
-        } catch (err) {
-            console.error('Failed to delete old messages:', err);
-        }
-    }
+    if (userSeq && !await deletePersistedMessage(userSeq, { delete_user: true, cascade: true })) return;
 
     // Remove this message bubble and every later bubble that belongs to
     // this or a subsequent turn. We mirror the backend cascade contract:

channel/web/static/js/chat/send.js

@@ -24,23 +24,7 @@ async function regenerateResponse(botMsgEl) {
     // (because /message will create a fresh user message + new bot reply)
     // Must await to ensure delete completes before /message is sent
     const userSeq = prevEl.dataset.seq;
-    if (userSeq) {
-        try {
-            const resp = await fetch('/api/messages/delete', {
-                method: 'POST',
-                headers: { 'Content-Type': 'application/json' },
-                body: JSON.stringify({ 
-                    session_id: sessionId, 
-                    user_seq: parseInt(userSeq),
-                    delete_user: true
-                })
-            });
-            const data = await resp.json();
-            if (data.status === 'success') console.log(`Deleted ${data.deleted} old messages`);
-        } catch (err) {
-            console.error('Failed to delete old messages:', err);
-        }
-    }
+    if (userSeq && !await deletePersistedMessage(userSeq, { delete_user: true })) return;
 
     // Remove both the old user message and bot message from DOM
     if (prevEl.parentNode) prevEl.parentNode.removeChild(prevEl);

channel/web/static/js/chat/state.js

@@ -604,7 +604,7 @@ messagesDiv.addEventListener('click', (e) => {
         const userMsgEl = deleteBtn.closest('.user-message-group');
         if (!userMsgEl) return;
 
-        showConfirmModal(t('delete_message_title'), t('delete_message_confirm'), () => {
+        showConfirmModal(t('delete_message_title'), t('delete_message_confirm'), async () => {
             // Find the next bot reply for this turn (skip non-message nodes).
             let botReplyEl = null;
             let sibling = userMsgEl.nextElementSibling;
@@ -615,21 +615,13 @@ messagesDiv.addEventListener('click', (e) => {
                 }
                 sibling = sibling.nextElementSibling;
             }
+            const userSeq = userMsgEl.dataset.seq;
+            if (userSeq && !await deletePersistedMessage(userSeq)) return;
+
             userMsgEl.remove();
             if (botReplyEl) botReplyEl.remove();
-
-            const userSeq = userMsgEl.dataset.seq;
-            if (userSeq) {
-                fetch('/api/messages/delete', {
-                    method: 'POST',
-                    headers: { 'Content-Type': 'application/json' },
-                    body: JSON.stringify({ session_id: sessionId, user_seq: parseInt(userSeq) })
-                }).then(r => r.json()).then(data => {
-                    if (data.status === 'success') console.log(`Deleted ${data.deleted} messages`);
-                    // Drop the removed question from the navigation rail.
-                    if (typeof refreshTimeline === 'function') refreshTimeline();
-                }).catch(err => console.error('Failed to delete:', err));
-            }
+            // Drop the removed question from the navigation rail.
+            if (userSeq && typeof refreshTimeline === 'function') refreshTimeline();
         });
         return;
     }

cli/update_service.py

@@ -403,7 +403,13 @@ def _worker_alive(pid: int) -> bool:
 def _update_is_stale(status: Dict[str, Any]) -> bool:
     """A running/restarting status whose worker is gone (killed, OOM, reboot)
     would otherwise block every later update."""
-    worker_pid = int(status.get("worker_pid") or 0)
+    try:
+        worker_pid = int(status.get("worker_pid") or 0)
+    except (TypeError, ValueError):
+        # A corrupt pid tells us nothing about the worker, so fall through to the
+        # timestamp check instead of raising. Reporting the run stale here would
+        # unblock an update that may still be running.
+        worker_pid = 0
     if worker_pid and not _worker_alive(worker_pid):
         return True
     started = status.get("started_at") or status.get("updated_at")

common/cloud_client.py

@@ -22,7 +22,7 @@ from bridge.reply import Reply, ReplyType
 from common.atomic_write import write_json_atomic
 from common.log import logger
 from linkai import LinkAIClient, PushMsg
-from config import conf, pconf, plugin_config, available_setting, write_plugin_config, get_root, get_weixin_credentials_path, drag_sensitive
+from config import conf, pconf, plugin_config, available_setting, write_plugin_config, get_data_root, get_weixin_credentials_path, drag_sensitive
 from plugins import PluginManager
 from contextlib import contextmanager
 from contextvars import ContextVar
@@ -1743,7 +1743,7 @@ class CloudClient(LinkAIClient):
         Save configuration to config.json file.
         """
         try:
-            config_path = os.path.join(get_root(), "config.json")
+            config_path = os.path.join(get_data_root(), "config.json")
             if not os.path.exists(config_path):
                 logger.warning(f"[CloudClient] config.json not found at {config_path}, skip saving")
                 return

common/utils.py

@@ -25,10 +25,10 @@ def fsize(file):
 def compress_imgfile(file, max_size):
     if fsize(file) <= max_size:
         return file
-    from PIL import Image
+    from PIL import Image, ImageOps
     file.seek(0)
     img = Image.open(file)
-    rgb_image = img.convert("RGB")
+    rgb_image = ImageOps.exif_transpose(img).convert("RGB")
     quality = 95
     min_quality = 10
     while True:

docs/channels/web.mdx

@@ -28,7 +28,7 @@ The Web Console is CowAgent's default channel. It runs automatically once starte
 | `external_api_enabled` | Turn on the [Chat API](/extensions/chat-api) | `false` |
 | `external_api_token` | API key of the [Chat API](/extensions/chat-api), kept while the API is off | `""` |
 | `web_session_expire_days` | Login session validity in days | `30` |
-| `web_file_serve_root` | Root directory the web console can directly read/send files from. Defaults to the user home dir and agent workspace only; set to `/` to allow the whole filesystem | `"~"` |
+| `web_file_serve_root` | Additional root directory the web console can directly read/send files from (defaults to the user home directory). Configured Agent workspaces, including disabled Agents, and opened project directories are also allowed; set to `/` to allow the whole filesystem | `"~"` |
 | `enable_thinking` | Whether to enable deep thinking mode | `false` |
 
 Once a password is configured, you must enter it to log in when accessing the console. The login session is kept for 30 days by default, so restarting the service during that period does not require re-login. The password can also be changed online from the "Configuration" page in the console.

models/claudeapi/claude_api_bot.py

@@ -960,6 +960,22 @@ class ClaudeAPIBot(Bot, OpenAIImage):
                                             }]
                                         }
                             
+                            elif event_type == "error":
+                                # Anthropic can fail a stream after the 200
+                                # response has started (e.g. overloaded_error).
+                                # Surface it so the agent's retry/fallback logic
+                                # sees a real error instead of a truncated reply.
+                                err = event.get("error") or {}
+                                err_type = err.get("type", "")
+                                err_msg = err.get("message") or err_type or "Unknown stream error"
+                                logger.error(f"[Claude] Stream error event: {err_type}: {err_msg}")
+                                yield {
+                                    "error": True,
+                                    "message": f"{err_type}: {err_msg}" if err_type else err_msg,
+                                    "status_code": 529 if err_type == "overloaded_error" else 500,
+                                }
+                                return
+
                             elif event_type == "message_stop":
                                 # Final event - log completion
                                 logger.debug(f"[Claude] Stream completed with stop_reason: {stop_reason}")

models/dashscope/dashscope_bot.py

@@ -661,6 +661,7 @@ class DashscopeBot(Bot):
                 # Check if this is a tool result message (user role with tool_result blocks)
                 if role == "user" and any(block.get("type") == "tool_result" for block in content):
                     # Convert each tool_result block to a separate tool message
+                    text_parts = []
                     for block in content:
                         if block.get("type") == "tool_result":
                             dashscope_messages.append({
@@ -668,6 +669,12 @@ class DashscopeBot(Bot):
                                 "content": block.get("content", ""),
                                 "tool_call_id": block.get("tool_use_id")  # DashScope uses 'tool_call_id'
                             })
+                        elif block.get("type") == "text":
+                            text_parts.append(block.get("text", ""))
+                    # Text next to tool results (e.g. a steering message sent
+                    # while tools ran) follows them as a user message.
+                    if text_parts:
+                        dashscope_messages.append({"role": "user", "content": "\n".join(text_parts)})
                 
                 # Check if this is an assistant message with tool_use blocks
                 elif role == "assistant":

models/gemini/google_gemini_bot.py

@@ -22,7 +22,6 @@ from bridge.reply import Reply, ReplyType
 from common.log import logger
 from config import conf
 from models.chatgpt.chat_gpt_session import ChatGPTSession
-from models.baidu.baidu_wenxin_session import BaiduWenxinSession
 
 
 # OpenAI对话模型API (可用)
@@ -434,6 +433,11 @@ class GoogleGeminiBot(Bot):
                                     tool_result_data = tool_content
                             except Exception:
                                 tool_result_data = {"result": tool_content}
+                            # functionResponse.response must be a JSON object;
+                            # tool text that parses to a list or scalar (e.g. an
+                            # MCP tool returning a JSON array) is wrapped.
+                            if not isinstance(tool_result_data, dict):
+                                tool_result_data = {"result": tool_result_data}
                             
                             # Find the tool name from previous messages
                             tool_name = None

models/linkai/link_ai_bot.py

@@ -16,6 +16,7 @@ from common.log import logger
 from config import conf, pconf
 import threading
 from common import memory, state_dir, utils
+from common.media_download import MAX_FILE_BYTES, download_to_file
 import base64
 import os
 from urllib.parse import urlparse
@@ -515,10 +516,11 @@ def _download_file(url: str):
         # the process CWD, which the packaged desktop app does not control and may
         # not be able to write, and state_dir owns the layout anyway.
         file_path = state_dir.tmp_dir() / file_name
-        response = requests.get(url, timeout=(5, 60))
-        # An expired signed link answers with an HTML error page, not the file.
-        response.raise_for_status()
-        file_path.write_bytes(response.content)
+        # Stream it under the shared cap, the way the LinkAI voice download
+        # already does: the vendor decides the size, and an unbounded body
+        # would be buffered whole before anything could object. A refused
+        # download leaves no partial file behind.
+        download_to_file(url, str(file_path), MAX_FILE_BYTES, timeout=(5, 60))
         return str(file_path)
     except Exception as e:
         logger.warn(e)

models/zhipuai/zhipuai_bot.py

@@ -520,6 +520,7 @@ class ZHIPUAIBot(Bot, ZhipuAIImage):
                 # Check if this is a tool result message (user role with tool_result blocks)
                 if role == "user" and any(block.get("type") == "tool_result" for block in content):
                     # Convert each tool_result block to a separate tool message
+                    text_parts = []
                     for block in content:
                         if block.get("type") == "tool_result":
                             zhipu_messages.append({
@@ -527,6 +528,12 @@ class ZHIPUAIBot(Bot, ZhipuAIImage):
                                 "tool_call_id": block.get("tool_use_id"),
                                 "content": block.get("content", "")
                             })
+                        elif block.get("type") == "text":
+                            text_parts.append(block.get("text", ""))
+                    # Text next to tool results (e.g. a steering message sent
+                    # while tools ran) follows them as a user message.
+                    if text_parts:
+                        zhipu_messages.append({"role": "user", "content": "\n".join(text_parts)})
                 
                 # Check if this is an assistant message with tool_use blocks
                 elif role == "assistant":

tests/test_chat_channel_voice_tmp_cleanup.py

@@ -0,0 +1,123 @@
+# encoding:utf-8
+"""The inbound voice branch must not leave recordings behind in the tmp dir.
+
+_generate_reply converts the inbound audio to wav next to it, then deletes the
+temporary files. Two things leaked:
+
+* On a conversion failure the code reassigned ``wav_path = file_path`` before
+  the cleanup ran, so the output the failing encoder had already created was no
+  longer named anywhere and stayed in the tmp dir for good. pydub opens the
+  destination ``wb+`` before it runs ffmpeg, so an encoder that exits nonzero
+  leaves an empty file at that path.
+* The two removals shared one ``try``, so a failure on the first one skipped
+  the second. That one strands the converted wav, which does hold the user's
+  own recording.
+"""
+import os
+
+import pytest
+
+from bridge.context import Context, ContextType
+from bridge.reply import Reply, ReplyType
+from channel import chat_channel as mod
+
+
+class _PassThroughPlugins:
+    def emit_event(self, e_context):
+        return e_context
+
+
+class _FakeMsg:
+    def prepare(self):
+        pass
+
+
+@pytest.fixture
+def channel(monkeypatch):
+    monkeypatch.setattr(mod, "PluginManager", _PassThroughPlugins)
+    # Skip __init__: it starts the consumer thread, which this test does not need.
+    ch = mod.ChatChannel.__new__(mod.ChatChannel)
+    ch.NOT_SUPPORT_REPLYTYPE = []
+    ch.stt_paths = []
+
+    def build_voice_to_text(_self, path):
+        ch.stt_paths.append(path)
+        return Reply(ReplyType.TEXT, "transcript")
+
+    monkeypatch.setattr(mod.Channel, "build_voice_to_text", build_voice_to_text)
+    # The transcript would otherwise be composed into a follow-up text turn.
+    monkeypatch.setattr(mod.ChatChannel, "_compose_context", lambda self, *a, **k: None)
+    return ch
+
+
+def _voice_context(tmp_path, name="wx_123.silk"):
+    src = tmp_path / name
+    src.write_bytes(b"silk-audio")
+    return Context(ContextType.VOICE, str(src), kwargs={"msg": _FakeMsg()})
+
+
+def test_a_failed_conversion_leaves_no_orphaned_output(channel, monkeypatch, tmp_path):
+    def failing_any_to_wav(any_path, wav_path):
+        # pydub opens the destination wb+ before it runs ffmpeg, so an encoder
+        # that exits nonzero leaves the (empty) file behind. Stubbed here
+        # because pydub is an optional dependency and is absent in CI.
+        with open(wav_path, "wb"):
+            pass
+        raise RuntimeError("ffmpeg: Encoder not found")
+
+    monkeypatch.setattr(mod, "any_to_wav", failing_any_to_wav)
+
+    context = _voice_context(tmp_path)
+    channel._generate_reply(context)
+
+    # The fallback still transcribes the raw audio, as before.
+    assert channel.stt_paths == [context.content]
+    assert os.listdir(tmp_path) == []
+
+
+def test_a_failed_source_removal_still_removes_the_converted_wav(channel, monkeypatch, tmp_path):
+    context = _voice_context(tmp_path)
+    source = context.content
+
+    def converting_any_to_wav(any_path, wav_path):
+        with open(wav_path, "wb") as f:
+            f.write(b"RIFF")
+
+    monkeypatch.setattr(mod, "any_to_wav", converting_any_to_wav)
+
+    real_remove = os.remove
+
+    def remove(path, *args, **kwargs):
+        if path == source:
+            raise PermissionError(path)
+        return real_remove(path, *args, **kwargs)
+
+    monkeypatch.setattr(mod.os, "remove", remove)
+
+    channel._generate_reply(context)
+
+    # The source could not be removed, but that must not strand the wav too.
+    assert os.listdir(tmp_path) == [os.path.basename(source)]
+
+
+def test_an_inbound_wav_is_removed_exactly_once(channel, monkeypatch, tmp_path):
+    """file_path and wav_path are the same name here; cleanup must not double-remove."""
+    context = _voice_context(tmp_path, name="wx_123.wav")
+
+    def same_file(any_path, wav_path):
+        raise OSError("are the same file")
+
+    monkeypatch.setattr(mod, "any_to_wav", same_file)
+    removed = []
+    real_remove = os.remove
+
+    def remove(path, *args, **kwargs):
+        removed.append(path)
+        return real_remove(path, *args, **kwargs)
+
+    monkeypatch.setattr(mod.os, "remove", remove)
+
+    channel._generate_reply(context)
+
+    assert removed == [context.content]
+    assert os.listdir(tmp_path) == []

tests/test_claude_stream_error_event.py

@@ -0,0 +1,71 @@
+# encoding:utf-8
+"""Anthropic can fail a stream with an ``error`` SSE event after HTTP 200.
+
+The bot must surface it as an error chunk; ignoring it ends the stream with
+no tool call and a truncated (or empty) reply that the agent treats as final.
+"""
+import json
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+
+class _FakeStreamResponse:
+    status_code = 200
+    text = ""
+
+    def __init__(self, events):
+        self._lines = [f"data: {json.dumps(e)}".encode() for e in events]
+
+    def iter_lines(self):
+        return iter(self._lines)
+
+    def close(self):
+        pass
+
+
+def _stream(monkeypatch, events):
+    from config import conf
+    from models.claudeapi import claude_api_bot
+    from models.claudeapi.claude_api_bot import ClaudeAPIBot
+
+    monkeypatch.setitem(conf(), "claude_api_key", "test")
+    monkeypatch.setitem(conf(), "proxy", "")
+    monkeypatch.setattr(claude_api_bot.requests, "post",
+                        lambda *a, **kw: _FakeStreamResponse(events))
+    bot = ClaudeAPIBot.__new__(ClaudeAPIBot)
+    return list(bot._handle_stream_response({"model": "claude-sonnet-4-6", "messages": []}))
+
+
+def test_mid_stream_overloaded_error_is_surfaced(monkeypatch):
+    chunks = _stream(monkeypatch, [
+        {"type": "message_start", "message": {"usage": {"input_tokens": 5}}},
+        {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
+        {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Let me"}},
+        {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}},
+    ])
+
+    errors = [c for c in chunks if c.get("error")]
+    assert len(errors) == 1
+    assert errors[0]["status_code"] == 529
+    # The agent retries/falls back on errors whose text mentions "overloaded".
+    assert "overloaded" in errors[0]["message"].lower()
+    assert chunks[-1] is errors[0]
+
+
+def test_error_event_during_tool_use_is_not_dropped(monkeypatch):
+    chunks = _stream(monkeypatch, [
+        {"type": "message_start", "message": {"usage": {"input_tokens": 5}}},
+        {"type": "content_block_start", "index": 0,
+         "content_block": {"type": "tool_use", "id": "toolu_1", "name": "read", "input": {}}},
+        {"type": "content_block_delta", "index": 0,
+         "delta": {"type": "input_json_delta", "partial_json": "{\"path\": "}},
+        {"type": "error", "error": {"type": "api_error", "message": "Internal server error"}},
+    ])
+
+    assert not [c for c in chunks if (c.get("choices") or [{}])[0].get("delta", {}).get("tool_calls")]
+    errors = [c for c in chunks if c.get("error")]
+    assert len(errors) == 1
+    assert errors[0]["status_code"] == 500
+    assert "Internal server error" in errors[0]["message"]

tests/test_cloud_config_linkai_guard.py

@@ -24,7 +24,7 @@ def _push(root, remote_config, plugin_config):
     client.client_id = "test-client"
     client._peer_transport = None
     client.channel_mgr = None
-    with patch.object(cloud_client, "get_root", return_value=str(root)), \
+    with patch.object(cloud_client, "get_data_root", return_value=str(root)), \
             patch.object(cloud_client, "conf", return_value={}), \
             patch.object(cloud_client, "pconf", return_value=plugin_config):
         client.on_config(remote_config)

tests/test_compress_imgfile_orientation.py

@@ -0,0 +1,74 @@
+"""Image compression must preserve the displayed orientation of real EXIF images."""
+
+import io
+
+import pytest
+
+from common.utils import compress_imgfile
+
+Image = pytest.importorskip("PIL.Image")
+
+COLORS = ((240, 20, 20), (20, 220, 20), (20, 20, 240), (230, 220, 20))
+# Corners TL, TR, BL, BR after the EXIF display transformation, hand-derived.
+DISPLAY_CORNERS = {
+    1: (0, 1, 2, 3),
+    2: (1, 0, 3, 2),
+    3: (3, 2, 1, 0),
+    4: (2, 3, 0, 1),
+    5: (0, 2, 1, 3),
+    6: (2, 0, 3, 1),
+    7: (3, 1, 2, 0),
+    8: (1, 3, 0, 2),
+}
+
+
+def oriented_jpeg(orientation):
+    image = Image.new("RGB", (120, 80))
+    for index, color in enumerate(COLORS):
+        left, top = (index % 2) * 60, (index // 2) * 40
+        image.paste(color, (left, top, left + 60, top + 40))
+    exif = Image.Exif()
+    exif[274] = orientation
+    output = io.BytesIO()
+    image.save(output, "JPEG", quality=95, exif=exif)
+    output.seek(0)
+    return output
+
+
+def corners(image):
+    width, height = image.size
+    points = (
+        (width // 4, height // 4),
+        (3 * width // 4, height // 4),
+        (width // 4, 3 * height // 4),
+        (3 * width // 4, 3 * height // 4),
+    )
+    return tuple(
+        min(
+            range(4),
+            key=lambda i: sum(
+                (a - b) ** 2 for a, b in zip(image.getpixel(point), COLORS[i])
+            ),
+        )
+        for point in points
+    )
+
+
+@pytest.mark.parametrize("orientation", range(1, 9))
+def test_compression_preserves_exif_display_geometry_and_corners(orientation):
+    source = oriented_jpeg(orientation)
+    # Force real JPEG re-encoding through the public byte-budget helper.
+    result = compress_imgfile(source, len(source.getvalue()) - 1)
+    result.seek(0)
+    with Image.open(result) as decoded:
+        expected_size = (80, 120) if orientation >= 5 else (120, 80)
+        assert decoded.size == expected_size
+        assert corners(decoded) == DISPLAY_CORNERS[orientation]
+        assert decoded.getexif().get(274, 1) == 1
+
+
+def test_under_budget_image_is_returned_unchanged():
+    source = oriented_jpeg(6)
+    original = source.getvalue()
+    assert compress_imgfile(source, len(original)) is source
+    assert source.getvalue() == original

tests/test_conversation_runs.py

@@ -152,8 +152,16 @@ def test_get_run_detail_recovers_full_output_from_session():
         store.append_messages(
             session_id,
             [
-                {"role": "user", "content": [{"type": "text", "text": "[SCHEDULED] run it"}]},
-                {"role": "assistant", "content": [{"type": "text", "text": long_body}]},
+                {
+                    "role": "user",
+                    "content": [{"type": "text", "text": "[SCHEDULED] run it"}],
+                    "run_id": "r-detail",
+                },
+                {
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": long_body}],
+                    "run_id": "r-detail",
+                },
             ],
         )
 

tests/test_dashscope_steering_with_tool_results.py

@@ -0,0 +1,41 @@
+"""A steer that lands while tools run shares the user message with the
+tool_result blocks (AgentStreamExecutor._append_steering). The DashScope converter
+must keep that text instead of forwarding only the tool results."""
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+
+def _steered_tool_result_message():
+    from agent.protocol.agent_stream import AgentStreamExecutor
+
+    executor = AgentStreamExecutor.__new__(AgentStreamExecutor)
+    executor._emit_event = lambda *a, **kw: None
+    blocks = [{"type": "tool_result", "tool_use_id": "call_1", "content": "file listing"}]
+    executor._append_steering(
+        ["Stop and summarise what you found so far"],
+        pending_tool_calls=[{"id": "call_2", "name": "bash", "arguments": {}}],
+        content_blocks=blocks,
+    )
+    return {"role": "user", "content": blocks}
+
+
+def test_steering_text_next_to_tool_results_reaches_qwen():
+    from models.dashscope.dashscope_bot import DashscopeBot
+
+    bot = DashscopeBot.__new__(DashscopeBot)
+    history = [
+        {"role": "user", "content": "look around"},
+        {"role": "assistant", "content": [
+            {"type": "tool_use", "id": "call_1", "name": "ls", "input": {}},
+            {"type": "tool_use", "id": "call_2", "name": "bash", "input": {}},
+        ]},
+        _steered_tool_result_message(),
+    ]
+
+    converted = bot._convert_messages_to_dashscope_format(history)
+
+    assert [m["role"] for m in converted] == ["user", "assistant", "tool", "tool", "user"]
+    assert [m["tool_call_id"] for m in converted[2:4]] == ["call_1", "call_2"]
+    assert "Stop and summarise what you found so far" in converted[-1]["content"]

tests/test_dingtalk_bounded_media_download.py

@@ -2,9 +2,14 @@
 """DingTalk upload_media() must cap remote downloads (cf. #3300, Feishu).
 
 It used a bare requests.get + f.write, so an oversized URL filled memory/disk.
-Pin it to the shared download_to_file helper.
+Pin it to the shared download_to_file helper -- and, a download being this
+call's to make, to cleaning it up again on every way out, the way the
+weixin / wecom_bot senders do. The voice send converts the TTS output to amr
+beside it; that amr is transient too, including the half-written one a failed
+conversion leaves behind.
 """
-from tests.test_dingtalk_streaming_cards import _bare_channel
+from bridge.reply import Reply, ReplyType
+from tests.test_dingtalk_streaming_cards import _bare_channel, _context
 
 
 class FakeResp:
@@ -80,12 +85,28 @@ def test_http_url_over_size_limit_rejected(monkeypatch, tmp_path):
 
 
 def test_http_url_query_string_stays_out_of_the_file_name(monkeypatch, tmp_path):
+    from channel.dingtalk import dingtalk_channel as mod
+
     get = FakeResp(content=b"<media data>")
     post = FakeResp(json_data={"errcode": 0, "media_id": "mid-3"})
     _stub(monkeypatch, get, post, tmp_path)
+    real_download = mod.download_to_file
+    seen = {}
+
+    def recording_download(url, path, *a, **k):
+        seen["path"] = path
+        return real_download(url, path, *a, **k)
+
+    monkeypatch.setattr(mod, "download_to_file", recording_download)
     url = "https://cdn.example/dir/%E6%8A%A5%E5%91%8A.pdf?token=secret&x=a/b"
     assert _channel().upload_media(url, "file") == "mid-3"
-    assert [p.name for p in tmp_path.iterdir()] == ["报告.pdf"]
+    # The download target is a generated unique name, so nothing from the URL
+    # -- query string included -- ever becomes part of a path.
+    assert "?" not in seen["path"]
+    assert "token" not in seen["path"]
+    assert "secret" not in seen["path"]
+    # A download belongs to this call: nothing is left in the tmp dir.
+    assert [p.name for p in tmp_path.iterdir()] == []
 
 
 def test_local_file_url_still_uploaded(monkeypatch, tmp_path):
@@ -109,3 +130,63 @@ def test_http_url_download_is_guarded(monkeypatch, tmp_path):
     monkeypatch.setattr(mod, "download_to_file", download_to_file)
     assert _channel().upload_media("https://cdn.example/a.pdf", "file") is None
     assert seen.get("guarded") is True
+
+
+def test_http_url_download_is_cleaned_up_when_the_upload_fails(monkeypatch, tmp_path):
+    get = FakeResp(content=b"<media data>")
+    post = FakeResp(json_data={"errcode": 40001, "errmsg": "invalid token"})
+    _stub(monkeypatch, get, post, tmp_path)
+    assert _channel().upload_media("https://cdn.example/ok.mp4", "video") is None
+    assert [p.name for p in tmp_path.iterdir()] == []
+
+
+def test_voice_amr_is_cleaned_up_after_the_send(monkeypatch, tmp_path):
+    from voice import audio_convert
+
+    voice = tmp_path / "tts.mp3"
+    voice.write_bytes(b"<mp3 data>")
+    amr = tmp_path / "tts.amr"
+
+    def fake_convert(src, dst):
+        # pydub opens the target, then spawns ffmpeg into it.
+        with open(dst, "wb") as f:
+            f.write(b"#!AMR\n")
+        return 1.5
+
+    monkeypatch.setattr(audio_convert, "any_to_amr", fake_convert)
+    _stub(monkeypatch, FakeResp(), FakeResp(json_data={"errcode": 0, "media_id": "mid-v"}), tmp_path)
+    ch = _channel()
+    ch._send_file_message = lambda *a, **k: True
+    ch.reply_text = lambda *a, **k: None
+    ch.send(Reply(ReplyType.VOICE, str(voice)), _context())
+
+    assert not amr.exists()
+    # The TTS source belongs to voice synthesis, not to this send.
+    assert [p.name for p in tmp_path.iterdir()] == ["tts.mp3"]
+
+
+def test_voice_amr_is_cleaned_up_when_the_conversion_fails(monkeypatch, tmp_path):
+    from voice import audio_convert
+
+    voice = tmp_path / "tts.mp3"
+    voice.write_bytes(b"<mp3 data>")
+    amr = tmp_path / "tts.amr"
+
+    def failing_convert(src, dst):
+        # pydub opens the target before it spawns ffmpeg: a conversion that
+        # dies partway still leaves a half-written amr behind.
+        with open(dst, "wb") as f:
+            f.write(b"#!AMR\nhalf")
+        raise RuntimeError("ffmpeg died")
+
+    monkeypatch.setattr(audio_convert, "any_to_amr", failing_convert)
+    _stub(monkeypatch, FakeResp(), FakeResp(), tmp_path)
+    ch = _channel()
+    ch._send_file_message = lambda *a, **k: True
+    texts = []
+    ch.reply_text = lambda content, incoming: texts.append(content)
+    ch.send(Reply(ReplyType.VOICE, str(voice)), _context())
+
+    assert not amr.exists()
+    assert texts == ["抱歉，语音转码失败"]
+    assert [p.name for p in tmp_path.iterdir()] == ["tts.mp3"]

tests/test_file_observation_scope.py

@@ -0,0 +1,110 @@
+"""One conversation's fresh read must not replace another's stale file view."""
+
+import os
+
+import pytest
+
+from agent.tools.edit.edit import Edit
+from agent.tools.read.read import Read
+from agent.tools.utils import file_state
+from agent.tools.write.write import Write
+from common.runtime_identity import identity_scope
+
+
+@pytest.fixture(autouse=True)
+def isolated_observations():
+    file_state.reset()
+    yield
+    file_state.reset()
+
+
+def observed_file(tmp_path):
+    path = tmp_path / "report.txt"
+    path.write_text("first line\nold second line\n", encoding="utf-8")
+    os.utime(path, (1700000000, 1700000000))
+    with identity_scope(agent_id="writer", session_id="chat-A"):
+        assert (
+            Read({"cwd": str(tmp_path)}).execute({"path": str(path)}).status
+            == "success"
+        )
+    path.write_text("first line\nupdated outside\n", encoding="utf-8")
+    os.utime(path, (1700000002, 1700000002))
+    return path
+
+
+@pytest.mark.parametrize(
+    "other",
+    [
+        {"agent_id": "writer", "session_id": "chat-B"},
+        {"agent_id": "teammate", "session_id": "chat-A"},
+    ],
+)
+@pytest.mark.parametrize("tool", ["edit", "write"])
+@pytest.mark.parametrize("other_action", ["read", "write"])
+def test_other_observation_does_not_hide_original_conversations_warning(
+    tmp_path, other, tool, other_action
+):
+    path = observed_file(tmp_path)
+    with identity_scope(**other):
+        if other_action == "read":
+            result = Read({"cwd": str(tmp_path)}).execute({"path": str(path)})
+            assert "updated outside" in result.result["content"]
+        else:
+            result = Write({"cwd": str(tmp_path)}).execute(
+                {"path": str(path), "content": "first line\nwritten by other actor\n"}
+            )
+            assert result.status == "success"
+    with identity_scope(agent_id="writer", session_id="chat-A"):
+        if tool == "edit":
+            result = Edit({"cwd": str(tmp_path)}).execute(
+                {
+                    "path": str(path),
+                    "oldText": "first line",
+                    "newText": "revised first line",
+                }
+            )
+        else:
+            result = Write({"cwd": str(tmp_path)}).execute(
+                {"path": str(path), "content": "revised first line\nold second line\n"}
+            )
+        assert result.status == "success", result.result
+        assert "modified after you last read it" in result.result.get("warning", "")
+        assert file_state.staleness_warning(str(path)) is None
+
+
+def test_unread_conversation_has_no_borrowed_expectation(tmp_path):
+    path = observed_file(tmp_path)
+    with identity_scope(agent_id="writer", session_id="chat-B"):
+        result = Write({"cwd": str(tmp_path)}).execute(
+            {"path": str(path), "content": "new document\n"}
+        )
+    assert result.status == "success"
+    assert "warning" not in result.result
+
+
+def test_same_conversation_keeps_observation_across_turns(tmp_path):
+    path = tmp_path / "report.txt"
+    path.write_text("initial\n", encoding="utf-8")
+    os.utime(path, (1700000000, 1700000000))
+    with identity_scope(agent_id="writer", session_id="chat-A", run_id="first-turn"):
+        Read({"cwd": str(tmp_path)}).execute({"path": str(path)})
+    path.write_text("outside edit\n", encoding="utf-8")
+    os.utime(path, (1700000002, 1700000002))
+    with identity_scope(agent_id="writer", session_id="chat-A", run_id="next-turn"):
+        assert "modified after you last read it" in file_state.staleness_warning(
+            str(path)
+        )
+
+
+def test_same_conversation_reread_refreshes_its_own_view(tmp_path):
+    path = observed_file(tmp_path)
+    with identity_scope(agent_id="writer", session_id="chat-A"):
+        assert (
+            Read({"cwd": str(tmp_path)}).execute({"path": str(path)}).status
+            == "success"
+        )
+        result = Write({"cwd": str(tmp_path)}).execute(
+            {"path": str(path), "content": "new content based on fresh read\n"}
+        )
+    assert result.status == "success"
+    assert "warning" not in result.result

tests/test_gemini_function_response_object.py

@@ -0,0 +1,66 @@
+"""Gemini's functionResponse.response is a JSON object (Struct).
+
+Tool results arrive as text; MCP tools in particular return whatever the
+server printed, which is often a JSON array or a bare number. Those must be
+wrapped in an object, or Gemini rejects the whole request.
+"""
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+
+class _Resp:
+    status_code = 200
+    text = ""
+
+    def json(self):
+        return {"candidates": [{"content": {"parts": [{"text": "ok"}]}, "finishReason": "STOP"}]}
+
+
+def _function_responses(monkeypatch, tool_content):
+    from config import conf
+    from models.gemini import google_gemini_bot
+    from models.gemini.google_gemini_bot import GoogleGeminiBot
+
+    monkeypatch.setitem(conf(), "gemini_api_key", "test")
+    captured = {}
+
+    def fake_post(url, headers=None, json=None, **kwargs):
+        captured["payload"] = json
+        return _Resp()
+
+    monkeypatch.setattr(google_gemini_bot.requests, "post", fake_post)
+    bot = GoogleGeminiBot.__new__(GoogleGeminiBot)
+    messages = [
+        {"role": "user", "content": "list my issues"},
+        {"role": "assistant", "content": [
+            {"type": "tool_use", "id": "call_1", "name": "list_issues", "input": {}},
+        ]},
+        {"role": "user", "content": [
+            {"type": "tool_result", "tool_use_id": "call_1", "content": tool_content},
+        ]},
+    ]
+    bot.call_with_tools(messages, tools=[], stream=False, model="gemini-2.5-flash")
+    return [
+        part["functionResponse"]
+        for content in captured["payload"]["contents"]
+        for part in content["parts"]
+        if "functionResponse" in part
+    ]
+
+
+def test_json_array_tool_result_is_wrapped_in_an_object(monkeypatch):
+    (fr,) = _function_responses(monkeypatch, '[{"number": 1}, {"number": 2}]')
+    assert fr["name"] == "list_issues"
+    assert fr["response"] == {"result": [{"number": 1}, {"number": 2}]}
+
+
+def test_scalar_tool_result_is_wrapped_in_an_object(monkeypatch):
+    (fr,) = _function_responses(monkeypatch, "42")
+    assert fr["response"] == {"result": 42}
+
+
+def test_object_tool_result_is_passed_through(monkeypatch):
+    (fr,) = _function_responses(monkeypatch, '{"output": "done"}')
+    assert fr["response"] == {"output": "done"}

tests/test_linkai_bounded_download.py

@@ -0,0 +1,89 @@
+# encoding:utf-8
+"""LinkAI's file sender must cap its download.
+
+``_download_file`` did a bare requests.get and wrote resp.content, so the
+whole body was buffered before anything could object -- ``media_download``'s
+docstring spells out why a socket timeout alone never fires on a server that
+keeps trickling bytes. The LinkAI voice download already routes through the
+shared helper; this file download was the one that did not.
+"""
+
+import pytest
+
+
+class _Stream:
+    """A minimal streamed response: chunks, headers, close."""
+
+    def __init__(self, chunks, headers=None):
+        self._chunks = list(chunks)
+        self.headers = headers or {}
+        # The pre-fix code path read the whole body instead of streaming.
+        self.content = b"".join(self._chunks)
+
+    def raise_for_status(self):
+        return None
+
+    def iter_content(self, chunk_size=1):
+        for chunk in self._chunks:
+            yield chunk
+
+    def close(self):
+        return None
+
+
+def _oversized(chunks=8, size=512):
+    return [b"x" * size] * chunks
+
+
+def _cap(monkeypatch, module, size=1024):
+    """Shrink the shared cap so an oversized body is a few chunks, not 100 MB.
+
+    ``raising=False`` keeps this working on the unfixed code, where the
+    module does not import the constant yet: there the real 100 MB cap
+    applies and the small body sails through -- which is the bug.
+    """
+    monkeypatch.setattr(module, "MAX_FILE_BYTES", size, raising=False)
+
+
+def test_linkai_file_download_refuses_an_oversized_body(tmp_path, monkeypatch):
+    from models.linkai import link_ai_bot
+
+    _cap(monkeypatch, link_ai_bot)
+    monkeypatch.setattr(link_ai_bot.state_dir, "tmp_dir", lambda: tmp_path)
+    monkeypatch.setattr(link_ai_bot.requests, "get",
+                        lambda *a, **k: _Stream(_oversized()))
+
+    assert link_ai_bot._download_file("https://files.example.com/report.pdf") is None
+    # A refused download leaves no partial file behind.
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_linkai_file_download_still_writes_ordinary_bodies(tmp_path, monkeypatch):
+    from models.linkai import link_ai_bot
+
+    monkeypatch.setattr(link_ai_bot.state_dir, "tmp_dir", lambda: tmp_path)
+    monkeypatch.setattr(link_ai_bot.requests, "get",
+                        lambda *a, **k: _Stream([b"%PDF-1.7"]))
+
+    result = link_ai_bot._download_file("https://files.example.com/report.pdf")
+
+    assert result == str(tmp_path / "report.pdf")
+    assert (tmp_path / "report.pdf").read_bytes() == b"%PDF-1.7"
+
+
+@pytest.mark.parametrize("status_code", [403, 404])
+def test_linkai_file_download_refuses_an_error_page(tmp_path, monkeypatch, status_code):
+    """An expired signed link answers with an HTML page, not the file."""
+    from models.linkai import link_ai_bot
+
+    class _Error(_Stream):
+        def raise_for_status(self):
+            raise link_ai_bot.requests.exceptions.HTTPError(
+                f"{status_code} Client Error")
+
+    monkeypatch.setattr(link_ai_bot.state_dir, "tmp_dir", lambda: tmp_path)
+    monkeypatch.setattr(link_ai_bot.requests, "get",
+                        lambda *a, **k: _Error([b"<html>expired</html>"]))
+
+    assert link_ai_bot._download_file("https://files.example.com/report.pdf") is None
+    assert list(tmp_path.iterdir()) == []

tests/test_linkai_download_status.py

@@ -19,6 +19,9 @@ def _download(tmp_path, status_code, content):
     response = requests.Response()
     response.status_code = status_code
     response._content = content
+    # The download is streamed now, so the response has to hand its body out
+    # in chunks; a Response whose content was already read can do that.
+    response._content_consumed = True
     with patch.object(link_ai_bot.state_dir, "tmp_dir", return_value=tmp_path), \
             patch.object(link_ai_bot.requests, "get", return_value=response):
         return link_ai_bot._download_file(URL)

tests/test_linkai_media_urls.py

@@ -53,7 +53,8 @@ def test_downloaded_file_name_excludes_query(monkeypatch, tmp_path):
     from models.linkai import link_ai_bot
 
     # The download checks the status before writing, so the stand-in response
-    # has to carry one the way a real requests.Response does.
+    # has to carry one the way a real requests.Response does. It is streamed
+    # now, so it also has to hand its body out in chunks.
     monkeypatch.setattr(
         link_ai_bot.requests,
         "get",
@@ -61,7 +62,10 @@ def test_downloaded_file_name_excludes_query(monkeypatch, tmp_path):
             "R", (), {
                 "content": b"x",
                 "status_code": 200,
+                "headers": {},
                 "raise_for_status": lambda self: None,
+                "iter_content": lambda self, chunk_size=1: iter([b"x"]),
+                "close": lambda self: None,
             }
         )(),
     )

tests/test_registered_workspace_file_serving.py

@@ -0,0 +1,104 @@
+"""Global artifact routes can serve files in configured Agent workspaces."""
+
+from urllib.parse import urlencode
+
+import pytest
+
+web = pytest.importorskip("web")
+
+from agent.registry import AgentProfile, AgentRegistry, set_agent_registry  # noqa: E402
+from channel.web.api import files  # noqa: E402
+from channel.web.core import _common  # noqa: E402
+from config import conf  # noqa: E402
+
+
+@pytest.fixture
+def registered_files(tmp_path, monkeypatch):
+    default = tmp_path / "default"
+    helper = tmp_path / "helper"
+    disabled = tmp_path / "disabled"
+    for folder in (default, helper, disabled):
+        folder.mkdir()
+    registry = AgentRegistry(
+        [
+            AgentProfile("default", "Default", str(default)),
+            AgentProfile("helper", "Helper", str(helper)),
+            AgentProfile("disabled", "Disabled", str(disabled), enabled=False),
+        ],
+        "default",
+    )
+    set_agent_registry(registry)
+    monkeypatch.setitem(conf(), "web_password", "")
+    monkeypatch.setitem(conf(), "web_file_serve_root", str(default))
+    monkeypatch.setattr(_common, "_PREVIEW_SECRET", b"local-test-preview-secret")
+    app = web.application(
+        ("/api/file", "FileServeHandler", "/preview/(.*)", "PreviewHandler"),
+        vars(files),
+    )
+    return app, {"default": default, "helper": helper, "disabled": disabled}
+
+
+def request_file(app, path, preview):
+    url = (
+        _common._build_preview_url(str(path))
+        if preview
+        else "/api/file?" + urlencode({"path": str(path)})
+    )
+    return app.request(url)
+
+
+@pytest.mark.parametrize("preview", [False, True])
+@pytest.mark.parametrize("owner", ["default", "helper", "disabled"])
+def test_registered_workspace_artifact_bytes_are_readable(
+    registered_files, owner, preview
+):
+    app, folders = registered_files
+    path = folders[owner] / "report.txt"
+    path.write_bytes(b"Native artifact bytes\n")
+    response = request_file(app, path, preview)
+    assert response.status == "200 OK"
+    assert response.data == path.read_bytes()
+    assert response.headers["Content-Type"] == "text/plain"
+
+
+@pytest.mark.parametrize("preview", [False, True])
+@pytest.mark.parametrize("alias", [False, True])
+def test_unregistered_paths_and_workspace_symlink_escapes_stay_denied(
+    registered_files, tmp_path, alias, preview
+):
+    app, folders = registered_files
+    unrelated = tmp_path / "helper-unregistered"
+    unrelated.mkdir()
+    path = unrelated / "private.txt"
+    path.write_bytes(b"outside configured roots")
+    if alias:
+        link = folders["helper"] / "linked.txt"
+        try:
+            link.symlink_to(path)
+        except OSError:
+            pytest.skip("symlinks unavailable")
+        path = link
+    response = request_file(app, path, preview)
+    assert response.status == "404 Not Found"
+    assert b"outside configured roots" not in response.data
+
+
+def test_global_file_route_still_requires_authentication(registered_files, monkeypatch):
+    app, folders = registered_files
+    path = folders["helper"] / "report.txt"
+    path.write_bytes(b"requires login")
+    monkeypatch.setitem(conf(), "web_password", "test-password")
+    response = request_file(app, path, False)
+    assert response.status == "401 Unauthorized"
+    assert b"requires login" not in response.data
+
+
+def test_preview_route_rejects_unsigned_directory_tokens(registered_files):
+    app, folders = registered_files
+    path = folders["helper"] / "report.txt"
+    path.write_bytes(b"requires signature")
+    url = _common._build_preview_url(str(path))
+    token, name = url.removeprefix("/preview/").split("/", 1)
+    response = app.request("/preview/" + token.rsplit(".", 1)[0] + ".invalid/" + name)
+    assert response.status == "404 Not Found"
+    assert b"requires signature" not in response.data

tests/test_root_config_atomic_write.py

@@ -74,7 +74,7 @@ def _a_cloud_client(tmp_path, monkeypatch):
     ``__new__`` skips ``__init__``, which would open the cloud transport.
     """
     config_path, original = _an_existing_config(tmp_path)
-    monkeypatch.setattr(cloud_client, "get_root", lambda: str(tmp_path))
+    monkeypatch.setattr(cloud_client, "get_data_root", lambda: str(tmp_path))
     return cloud_client.CloudClient.__new__(cloud_client.CloudClient), config_path, original
 
 
@@ -117,6 +117,32 @@ def test_a_cloud_save_that_succeeds_replaces_the_file(tmp_path, monkeypatch):
     assert _leftovers(tmp_path) == []
 
 
+def test_a_cloud_save_lands_in_the_data_root(tmp_path, monkeypatch):
+    """The desktop build sets COW_DATA_DIR=~/.cow and the docker image points it
+    at /home/agent/.cow, and ``load_config`` reads the data root -- so a
+    cloud-pushed config must be written where it will be read back. The save
+    used to go to ``get_root()`` (the bundle), which on those deployments is a
+    directory nobody reads: it either holds no config.json at all -- the push
+    is then dropped with a warning -- or a stale one that is never loaded.
+    """
+    data_root = tmp_path / "data_root"
+    data_root.mkdir()
+    data_config = data_root / "config.json"
+    data_config.write_text(json.dumps(ORIGINAL, indent=4, ensure_ascii=False), encoding="utf-8")
+
+    # Drive the real resolver: whatever it answers for COW_DATA_DIR is where a
+    # save has to land, exactly as on a desktop or docker deployment.
+    monkeypatch.setenv("COW_DATA_DIR", str(data_root))
+    client = cloud_client.CloudClient.__new__(cloud_client.CloudClient)
+
+    client._save_config_to_file({"model": "gpt-4o"})
+
+    saved = json.loads(data_config.read_text(encoding="utf-8"))
+    assert saved.get("model") == "gpt-4o"
+    assert saved["web_password"] == ORIGINAL["web_password"]
+    assert _leftovers(data_root) == []
+
+
 # ---------------------------------------------------------------------------
 # The feishu one-click setup writes the credentials it just created.
 # ---------------------------------------------------------------------------

tests/test_scheduler_run_output_identity.py

@@ -0,0 +1,184 @@
+"""A scheduler run detail must recover only the message that run delivered."""
+
+import sqlite3
+
+import pytest
+
+from agent.memory import clear_conversation_store_cache
+from agent.registry import AgentProfile, AgentRegistry, set_agent_registry
+from agent.tools.scheduler import integration
+from bridge.agent_bridge import AgentBridge
+from config import conf
+
+
+class LocalDelivery:
+    """No account or network: capture only the outbound delivery boundary."""
+
+    def __init__(self):
+        self.delivered = []
+
+    def send(self, reply, context):
+        self.delivered.append(reply.content)
+
+
+@pytest.fixture
+def runtime(tmp_path, monkeypatch):
+    import agent.evolution.trigger as evolution
+
+    registry = AgentRegistry(
+        [
+            AgentProfile("default", "Default", str(tmp_path / "default")),
+            AgentProfile("helper", "Helper", str(tmp_path / "helper")),
+        ],
+        "default",
+    )
+    set_agent_registry(registry)
+    clear_conversation_store_cache()
+    monkeypatch.setitem(conf(), "conversation_persistence", True)
+    monkeypatch.setitem(conf(), "scheduler_inject_to_session", True)
+    monkeypatch.setitem(conf(), "scheduler_inject_max_per_session", 3)
+    # Suppress background jobs only; constructor, persistence, dispatch and
+    # history recovery below are the real implementations.
+    monkeypatch.setattr(integration, "init_scheduler", lambda *args: False)
+    monkeypatch.setattr(evolution, "start_evolution_trigger", lambda *args: None)
+    channel = LocalDelivery()
+    monkeypatch.setattr(integration, "_resolve_delivery_channel", lambda *args: channel)
+    monkeypatch.setattr("agent.memory.conversation_store.time.time", lambda: 1700000000)
+    bridge = AgentBridge(object())
+    try:
+        yield bridge, channel
+    finally:
+        clear_conversation_store_cache()
+        set_agent_registry(None)
+
+
+def deliver(bridge, owner, task_id, content):
+    task = {
+        "id": task_id,
+        "name": task_id,
+        "action": {
+            "type": "send_message",
+            "channel_type": "web",
+            "receiver": "chat",
+            "content": content,
+        },
+    }
+    assert integration._run_scheduled_task(task, bridge, owner)
+    store = bridge.get_conversation_store(owner)
+    runs = store.list_runs(task_id=task_id, agent_id=owner)
+    assert len(runs) == 1
+    return store, runs[0]["run_id"]
+
+
+@pytest.mark.parametrize("owner", ["default", "helper"])
+def test_same_second_deliveries_recover_their_own_full_body(runtime, owner):
+    bridge, channel = runtime
+    bodies = ["First report " + "A" * 1100, "Second report " + "B" * 1100]
+    records = [
+        deliver(bridge, owner, f"task-{i}", body) for i, body in enumerate(bodies)
+    ]
+    assert channel.delivered == bodies
+    for (store, run_id), body in zip(records, bodies):
+        detail = store.get_run_detail(run_id)
+        assert detail["full_output"] == body
+        assert len(detail["output_preview"]) == 1001
+        with sqlite3.connect(store._db_path) as connection:
+            assert (
+                connection.execute(
+                    "SELECT COUNT(*) FROM messages WHERE run_id = ?", (run_id,)
+                ).fetchone()[0]
+                == 2
+            )
+
+
+@pytest.mark.parametrize("owner", ["default", "helper"])
+def test_pruned_run_uses_its_preview_instead_of_another_delivery(runtime, owner):
+    bridge, _ = runtime
+    store, first = deliver(bridge, owner, "first", "First delivered report")
+    _, second = deliver(bridge, owner, "second", "Second delivered report")
+    assert store.prune_scheduled_messages("chat", keep_last_n=1) == 2
+    detail = store.get_run_detail(first)
+    assert detail["full_output"] is None
+    assert detail["output_preview"] == "First delivered report"
+    assert store.get_run_detail(second)["full_output"] == "Second delivered report"
+
+
+def test_legacy_unassociated_run_never_borrows_current_message(runtime):
+    bridge, _ = runtime
+    store = bridge.get_conversation_store("helper")
+    store.create_run(
+        "legacy",
+        agent_id="helper",
+        session_id="chat",
+        task_source="scheduler",
+        extras={"output_preview": "Original historical preview"},
+    )
+    store.append_messages(
+        "chat",
+        [
+            {"role": "user", "content": "[SCHEDULED] later task"},
+            {"role": "assistant", "content": "Unrelated newer report"},
+        ],
+    )
+    detail = store.get_run_detail("legacy")
+    assert detail["full_output"] is None
+    assert detail["output_preview"] == "Original historical preview"
+
+
+def test_agent_scopes_and_normal_chat_do_not_replace_delivered_output(runtime):
+    bridge, _ = runtime
+    default_store, default_run = deliver(
+        bridge, "default", "default-task", "Default report"
+    )
+    helper_store, helper_run = deliver(bridge, "helper", "helper-task", "Helper report")
+    helper_store.append_messages(
+        "chat",
+        [
+            {"role": "user", "content": "A normal follow-up"},
+            {"role": "assistant", "content": "Normal chat answer"},
+        ],
+    )
+    assert default_store.get_run_detail(default_run)["full_output"] == "Default report"
+    assert helper_store.get_run_detail(helper_run)["full_output"] == "Helper report"
+
+
+def test_legacy_remember_callback_still_receives_delivery(runtime):
+    bridge, channel = runtime
+
+    class LegacyBridge:
+        def remember_scheduled_output(
+            self, session_id, content, channel_type="", task_description=""
+        ):
+            bridge.remember_scheduled_output(
+                session_id, content, channel_type, task_description
+            )
+
+    task = {
+        "id": "legacy-callback",
+        "name": "Legacy callback",
+        "action": {
+            "type": "send_message",
+            "channel_type": "web",
+            "receiver": "chat",
+            "content": "Legacy delivered text",
+        },
+    }
+    assert integration._run_scheduled_task(task, LegacyBridge(), "default")
+    assert channel.delivered == ["Legacy delivered text"]
+    store = bridge.get_conversation_store("default")
+    assert (
+        store.load_messages("chat")[-1]["content"][0]["text"] == "Legacy delivered text"
+    )
+
+
+@pytest.mark.parametrize("owner", ["default", "helper"])
+def test_run_identity_stays_out_of_model_message_payload(runtime, owner):
+    from agent.protocol import Agent
+
+    bridge, _ = runtime
+    cached = Agent("Local fixture", enable_skills=False, skip_context_files=True)
+    bridge._agent_instances[(owner, "chat")] = cached
+    store, run_id = deliver(bridge, owner, "memory-copy", "Visible scheduled text")
+    assert len(cached.messages) == 2
+    assert all(set(message) == {"role", "content"} for message in cached.messages)
+    assert store.get_run_detail(run_id)["full_output"] == "Visible scheduled text"

tests/test_web_console_update.py

@@ -264,6 +264,30 @@ def test_a_running_status_left_by_a_dead_worker_does_not_block(tmp_path, monkeyp
     ) is True
 
 
+def test_a_corrupt_worker_pid_does_not_crash_the_in_progress_guard(tmp_path, monkeypatch):
+    """A status file carrying a non-numeric worker_pid must not raise.
+
+    read_update_status already tolerates a missing file, unreadable bytes, invalid
+    JSON and a non-dict payload, but it does not validate the fields it hands on.
+    int() on a corrupt pid raised ValueError out of the "already in progress" check.
+    """
+    from cli import update_service
+
+    monkeypatch.setattr(update_service, "_worker_alive", lambda pid: True)
+    fresh = update_service.datetime.now(update_service.timezone.utc).isoformat()
+
+    for bad in ("abc", "", "12abc", [], {}, 1.5e3):
+        status = {"state": "running", "worker_pid": bad, "started_at": fresh}
+        # A corrupt pid is no evidence the worker died, so a fresh run still blocks.
+        assert update_service._update_is_stale(status) is False, bad
+
+    # ... and an old run is still reported stale rather than blocking forever.
+    for bad in ("abc", [], {}):
+        assert update_service._update_is_stale(
+            {"state": "running", "worker_pid": bad, "started_at": "2020-01-01T00:00:00+00:00"}
+        ) is True, bad
+
+
 def test_web_handlers_auth_and_no_github_on_version(monkeypatch):
     if "web" not in sys.modules:
         web_stub = types.ModuleType("web")

tests/test_zhipu_steering_with_tool_results.py

@@ -0,0 +1,47 @@
+"""A steer that lands while tools run shares the user message with the
+tool_result blocks (AgentStreamExecutor._append_steering). The GLM converter
+must keep that text instead of forwarding only the tool results."""
+import os
+import sys
+import types
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+if "zai" not in sys.modules:
+    fake_zai = types.ModuleType("zai")
+    fake_zai.ZhipuAiClient = object
+    sys.modules["zai"] = fake_zai
+
+
+def _steered_tool_result_message():
+    from agent.protocol.agent_stream import AgentStreamExecutor
+
+    executor = AgentStreamExecutor.__new__(AgentStreamExecutor)
+    executor._emit_event = lambda *a, **kw: None
+    blocks = [{"type": "tool_result", "tool_use_id": "call_1", "content": "file listing"}]
+    executor._append_steering(
+        ["Stop and summarise what you found so far"],
+        pending_tool_calls=[{"id": "call_2", "name": "bash", "arguments": {}}],
+        content_blocks=blocks,
+    )
+    return {"role": "user", "content": blocks}
+
+
+def test_steering_text_next_to_tool_results_reaches_glm():
+    from models.zhipuai.zhipuai_bot import ZHIPUAIBot
+
+    bot = ZHIPUAIBot.__new__(ZHIPUAIBot)
+    history = [
+        {"role": "user", "content": "look around"},
+        {"role": "assistant", "content": [
+            {"type": "tool_use", "id": "call_1", "name": "ls", "input": {}},
+            {"type": "tool_use", "id": "call_2", "name": "bash", "input": {}},
+        ]},
+        _steered_tool_result_message(),
+    ]
+
+    converted = bot._convert_messages_to_zhipu_format(history)
+
+    assert [m["role"] for m in converted] == ["user", "assistant", "tool", "tool", "user"]
+    assert [m["tool_call_id"] for m in converted[2:4]] == ["call_1", "call_2"]
+    assert "Stop and summarise what you found so far" in converted[-1]["content"]

tests/web_message_mutation.test.cjs

@@ -0,0 +1,141 @@
+const { test } = require('node:test');
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const vm = require('node:vm');
+const path = require('node:path');
+const ROOT = path.join(__dirname, '..');
+
+// Follow web_scroll.test.cjs: exercise the actual classic-script consumers
+// with controlled browser collaborators and deferred persistence responses.
+function setup({ active = false, persisted = true } = {}) {
+    const events = [], requests = [], alerts = [], nodes = [];
+    let click, confirm, resolveDelete, rejectDelete;
+    const deletion = new Promise((resolve, reject) => { resolveDelete = resolve; rejectDelete = reject; });
+    const box = {
+        removeChild(item) { events.push(`remove:${item.dataset.seq || item.role}`); nodes.splice(nodes.indexOf(item), 1); },
+        addEventListener(name, callback) { if (name === 'click') click = callback; },
+    };
+    function node(role, seq, content) {
+        const item = {
+            role, dataset: { rawContent: content },
+            classList: { contains(name) { return name === `${role}-message-group`; } },
+            get parentNode() { return nodes.includes(item) ? box : null; },
+            get nextElementSibling() { return nodes[nodes.indexOf(item) + 1] || null; },
+            get previousElementSibling() { return nodes[nodes.indexOf(item) - 1] || null; },
+            remove() { if (item.parentNode) box.removeChild(item); },
+        };
+        if (seq != null) item.dataset.seq = String(seq);
+        nodes.push(item);
+        return item;
+    }
+    const first = node('user', persisted ? 1 : null, 'First question');
+    const answer = node('bot', persisted ? 2 : null, 'First answer');
+    node('user', 3, 'Later question'); node('bot', 4, 'Later answer');
+    const input = { value: 'Unsent draft', dispatchEvent() {}, focus() {} };
+    const context = vm.createContext({
+        messagesDiv: box, chatInput: input, sessionId: 'owned-session', currentLang: 'en',
+        isCurrentSessionConversationActive: () => active,
+        t: key => key, alert: message => alerts.push(message),
+        console: { log() {}, warn() {}, error() {} }, Event: class {},
+        fetch(url, options) {
+            requests.push({ url, body: JSON.parse(options.body) });
+            if (url === '/api/messages/delete') {
+                events.push('delete:request');
+                return deletion.then(response => { events.push('delete:response'); return response; });
+            }
+            assert.equal(url, '/message'); events.push('regenerate:send');
+            return Promise.resolve({ json: async () => ({ status: 'success', inline_reply: 'Fresh answer' }) });
+        },
+        showConfirmModal(_title, _message, callback) { confirm = callback; },
+        refreshTimeline() { events.push('timeline:refresh'); }, scrollChatToBottom() {},
+        addUserMessage(content) { events.push('user:add'); node('user', null, content); },
+        addBotMessage() { events.push('bot:add'); }, addLoadingIndicator() { return { remove() {} }; },
+        addressedAgentId: () => '', rememberLiveSpeaker() {}, setLoadingSpeaker() {},
+        loadingContainers: {}, Date, setTimeout,
+    });
+    const source = name => fs.readFileSync(path.join(ROOT, 'channel/web/static/js/chat', name), 'utf8');
+    vm.runInContext(source('message-actions.js'), context);
+    vm.runInContext(source('send.js'), context);
+    const state = source('state.js');
+    vm.runInContext(state.slice(state.indexOf('// Intercept internal navigation links'), state.indexOf('const attachmentPreview')), context);
+    function deleteClick() {
+        const button = { closest(selector) { return selector === '.user-message-group' ? first : null; } };
+        click({ preventDefault() {}, target: { closest(selector) { return selector === '.delete-msg-btn' ? button : null; } } });
+    }
+    return {
+        nodes, requests, alerts, events, input,
+        ids: () => nodes.filter(item => item.dataset.seq).map(item => item.dataset.seq),
+        start(action) {
+            if (action === 'edit') return context.editUserMessage(first);
+            if (action === 'regenerate') return context.regenerateResponse(answer);
+            deleteClick(); return confirm ? confirm() : Promise.resolve();
+        },
+        cancelDelete() { deleteClick(); },
+        respond(body = { status: 'success', deleted: 2 }, ok = true) {
+            resolveDelete({ ok, status: ok ? 200 : 500, json: async () => body });
+        },
+        invalidJson() { resolveDelete({ ok: true, json: async () => { throw new SyntaxError('Invalid JSON'); } }); },
+        networkError() { rejectDelete(new Error('Connection lost')); },
+    };
+}
+const settle = () => new Promise(resolve => setImmediate(resolve));
+
+for (const action of ['edit', 'regenerate', 'delete']) {
+    for (const failure of ['api-error', 'network-error', 'invalid-json', 'http-error']) {
+        test(`${action}: ${failure} preserves history and draft without a new send`, async () => {
+            const s = setup(), operation = s.start(action);
+            await settle();
+            assert.deepEqual(s.ids(), ['1', '2', '3', '4'], 'wait for persistence before removing a turn');
+            assert.equal(s.input.value, 'Unsent draft');
+            assert.equal(s.requests.filter(r => r.url === '/message').length, 0);
+            if (failure === 'network-error') s.networkError();
+            else if (failure === 'invalid-json') s.invalidJson();
+            else if (failure === 'http-error') s.respond({ status: 'success' }, false);
+            else s.respond({ status: 'error', message: 'Store failed' });
+            await operation; await settle();
+            assert.deepEqual(s.ids(), ['1', '2', '3', '4']);
+            assert.equal(s.input.value, 'Unsent draft');
+            assert.deepEqual(s.alerts, ['records_delete_failed']);
+            assert.equal(s.requests.filter(r => r.url === '/message').length, 0);
+            assert.equal(s.events.includes('timeline:refresh'), false);
+        });
+    }
+    test(`${action}: acknowledgment preserves the existing successful mutation`, async () => {
+        const s = setup(), operation = s.start(action);
+        await settle(); assert.deepEqual(s.ids(), ['1', '2', '3', '4']);
+        s.respond(); await operation; await settle();
+        const expected = { session_id: 'owned-session', user_seq: 1 };
+        if (action !== 'delete') expected.delete_user = true;
+        if (action === 'edit') expected.cascade = true;
+        assert.deepEqual(s.requests[0], { url: '/api/messages/delete', body: expected });
+        assert.deepEqual(s.ids(), action === 'edit' ? [] : ['3', '4']);
+        const removed = s.events.findIndex(event => event.startsWith('remove:'));
+        assert.ok(removed > s.events.indexOf('delete:response'));
+        assert.deepEqual(s.alerts, []);
+        assert.equal(s.input.value, action === 'edit' ? 'First question' : 'Unsent draft');
+        const sends = s.requests.filter(r => r.url === '/message');
+        assert.equal(sends.length, action === 'regenerate' ? 1 : 0);
+        if (sends.length) assert.equal(sends[0].body.message, 'First question');
+        if (action === 'delete') assert.ok(s.events.indexOf('timeline:refresh') > removed);
+    });
+}
+
+test('canceling direct Delete does not submit or change the conversation', () => {
+    const s = setup(); s.cancelDelete();
+    assert.deepEqual(s.requests, []); assert.deepEqual(s.ids(), ['1', '2', '3', '4']);
+    assert.equal(s.input.value, 'Unsent draft');
+});
+test('active reply still blocks Edit and direct Delete', async () => {
+    for (const action of ['edit', 'delete']) {
+        const s = setup({ active: true }); await s.start(action);
+        assert.deepEqual(s.requests, []); assert.deepEqual(s.ids(), ['1', '2', '3', '4']);
+    }
+});
+for (const action of ['edit', 'regenerate', 'delete']) {
+    test(`${action}: nonpersisted turns still work without a deletion request`, async () => {
+        const s = setup({ persisted: false }); await s.start(action); await settle();
+        assert.equal(s.requests.filter(r => r.url === '/api/messages/delete').length, 0);
+        assert.deepEqual(s.ids(), action === 'edit' ? [] : ['3', '4']); assert.deepEqual(s.alerts, []);
+        if (action === 'delete') assert.equal(s.events.includes('timeline:refresh'), false);
+    });
+}