增强消息运行可靠性

.agents/MEMORY.md

@@ -5,6 +5,7 @@
 ## 用户偏好与长期原则
 
 - 一律使用中文与用户沟通。
+- Git commit message 一律使用中文。
 - Windows Python 脚本 stdout 可能是 GBK：使用 ASCII 状态标签，不用 emoji 或特殊装饰字符。
 - `CHANGELOG.md` 提到海外模型时只用“国际旗舰模型”等泛称；具体型号只放在 `PROGRESS.md`、配置和 git log。
 - 版本、CHANGELOG、PROGRESS 在 push 前统一更新；DESIGN 跟随产生架构或决策变化的 commit。
@@ -61,4 +62,3 @@ Mermaid/Chromium 的历史故障最终有三个根因：
 ## 领域
 
 用户单位是中国建筑材料科学研究总院。代码、库、模板和示例默认服务于水泥/混凝土、玻璃、陶瓷、耐火和新型建材的材料研发、表征分析、实验建模与科研写作；不是建筑施工、BIM 或结构设计语境。
-

CHANGELOG.md

@@ -5,6 +5,10 @@
 > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。
 > 工程口径的完整记录见 `PROGRESS.md` / git log。
 
+## 0.60.2 — 2026-07-27
+
+- 提升发送消息的可靠性：服务在接收消息后即使遇到重启或后台任务调度异常，也不会静默丢失用户刚发送的内容；非法的生图、视频模型选项也不再导致任务卡在运行中。
+
 ## 0.60.1 — 2026-07-27
 
 - 联网搜索现在会以当前北京时间判断时效：未指定时间时不再擅自给搜索词添加旧年份，查询“最新、当前、近期”内容时会采用合适的近期结果范围；用户明确指定的年份或时间窗保持优先。

DESIGN.md

@@ -29,6 +29,8 @@ zcbot/
 │   ├── skills.py           # SkillRegistry(渐进披露,多来源)
 │   ├── task.py             # TaskState
 │   ├── memory.py           # per-user .memory/ 双层记忆
+│   ├── file_store.py       # 原子文件替换 + 跨进程 advisory file lock
+│   ├── kb_lock.py          # .kb/<库> mutation 单写者锁(Web/入库/fs tool 共用)
 │   ├── shortcuts.py        # 快捷指令(入口层确定性展开)
 │   ├── paths.py            # task_dir db form 归一
 │   ├── storage/            # SQLAlchemy 2.x ORM;usage(计费写)/telemetry(失败埋点)/usage_report(聚合读)三分
@@ -94,7 +96,7 @@ Session = 消息列表,ORM 直写 PG `messages`(append-only,jsonb 存 LiteLLM
 用户自建资料(规范/报告/标准/内部文档)的长期查阅层,与 §3.7 记忆同范式:**纯文件 + prompt 注入契约,无向量无 DB**(判据同"真实文件为准":个人库几十~百余文件,agentic search 足够;索引若引入只能是可重建派生缓存)。**两层格局**:zcbot 内建 `.kb/` 私有小库(本节)+ 院检索服务共享大库(document_search,zcbot 只当客户端)——分工标准 = 文件数 × 查询频次,路由靠各自工具/契约描述自然分流,不在 kb 契约里点名 document_search(2026-07-22 收窄:契约只管自己怎么用,少一层耦合)。
 
 - **做成机制而非 skill**(判据:有独立于会话的持久状态需用户管理 → 机制):落盘 `user_root/.kb/<库名>/`(INDEX.md + docs/ 转换后 md + sources/ 原件)。**"已入库"判据 = INDEX.md 有条目**,sources 有而 INDEX 无 = 待入库 → 入库幂等、崩溃可恢复、零 migration。dotfile 命名同 `.memory` 双向防呆,GET /v1/files 天然隐藏。
-- **入库管线**(`core/kb_ingest.py`,上传即触发 + 手动兜底):markitdown Python API 转 md → 扫描件 PDF(文本近零)走方舟文档理解 OCR 兜底(§8.13 同通道)→ deepseek flash 单次 chat 写 标题/摘要/关键词(失败降级文件名+正文开头,不阻塞)→ 追加 INDEX 行。编排照定时执行器:create_task + to_thread + per-(user,库) 内存锁去重;进度存内存供前端轮询,崩了靠 FS 判据续跑。
+- **入库管线**(`core/kb_ingest.py`,上传即触发 + 手动兜底):markitdown Python API 转 md → 扫描件 PDF(文本近零)走方舟文档理解 OCR 兜底(§8.13 同通道)→ deepseek flash 单次 chat 写 标题/摘要/关键词(失败降级文件名+正文开头,不阻塞)→ 追加 INDEX 行。编排照定时执行器:create_task + to_thread;**写并发收口为共享 FS advisory lock**(`.kb/.locks/<hash>.lock`):Web 上传/删除、后台入库与 agent `write/edit` 对同一库共用一把跨进程锁,蓝绿实例间只允许一个写者,锁占用返 409/工具可重试;进程退出由 OS 自动释锁,不靠清理锁文件。文档正文、原件与 INDEX 全走同目录临时文件 + fsync + `os.replace` 原子发布,读者只会看到完整旧版或完整新版。进度详情仍以内存保存细节,但会探测跨进程锁补出 `running`;崩了靠 FS 判据续跑。
 - **agent 侧零新工具**:`kb_block`(照 memory_block)把 INDEX 全文 + 契约(主动查阅无需点名 / INDEX 行格式 / 答题标来源)注 prompt;**零库时注极简冷启动契约**(建库步骤 + INDEX 行格式 + "成篇资料进 KB / 短事实进记忆"分工,~百 token)——原"有库才注入"省 token,但零注入让模型不知道 KB 机制存在,用户说"放进知识库"被就近写进 `.memory/`(2026-07-24 真实事故),与 memory 空契约常驻是同一课:教会第一次,建库落盘后下轮 build_agent 自然切全量。fs 工具在 user_root 内可读写、docker 沙箱整 user_root bind → `.kb` 天然可达。INDEX 行格式是对话内手动入库与后台产出的同一契约。
 - **API 薄壳**(`/v1/kb*` 8 端点):列/建/删库、详情(带入库进度)、上传即入库、手动 ingest、看/删单篇。**不设 HTTP 检索端点**——检索是 agent 的事。前端两栏 modal(kb.js)管上传/删除,查询全走对话。
 - **记账**:`usage_events` kind="kb_ingest"(OCR 那笔走 kind="vision"),无 task 上下文 → 0022 放宽 task_id 可 NULL,溯源靠 units JSONB `{"kb", "source"}`。
@@ -158,7 +160,8 @@ Tasks     POST/GET/PATCH/DELETE /v1/tasks*(POST 可选 auto_title;分页+筛选+
             DELETE=软删,FS 不动)
           GET /v1/folders(working_dir + task 计数)
           GET/POST /v1/tasks/{id}/messages(POST 起 run;单活 run:running/cancelling→409,
-            SELECT FOR UPDATE 锁 task 行防 idx race)
+            先校验请求，再用 SELECT FOR UPDATE 将 user 消息与 running 同事务提交；
+            BG worker 消费已持久化轮次，不重复追加 user，防 202 后崩溃丢输入/idx race)
           GET /v1/tasks/{id}/events(SSE)   POST /v1/tasks/{id}/cancel(协作式,202)
 Auth      POST /v1/auth/login(platform_key)/ login_password / change_password;GET /v1/me
 Files     GET /v1/files?path= / upload / download / delete / rename
@@ -244,7 +247,7 @@ scheduled_jobs(§8.5)   channel_bindings(§8.7,判别列+JSONB)
 | running task 被 rename/delete | 后端校验 + UI 禁按钮 |
 | DB-then-FS 中断孤儿 | rename DB 先行可回滚;delete 后台 GC 扫"FS 有 DB 无" |
 | 同 wd 多 task 并发写同名 | known limitation,频率近 0;软警告 banner;宪法文件已按 short_id 命名隔离 |
-| 并发 POST 撞 messages.idx | 单活 run gate(FOR UPDATE + 409)+ lifespan reaper;multi-worker 再换 lease |
+| 并发 POST 撞 messages.idx / 202 后进程退出丢输入 | 单活 run gate(FOR UPDATE + 409)下原子提交 user 消息与 running；worker 只消费已持久化轮次；lifespan reaper 收敛残留 running，multi-worker 再换 lease |
 | shell/run_python 无沙箱开放外部 = 主机沦陷 | **Stage C 是 hard prereq**;`BLOCKED_PATTERNS` 是 trivial-bypass 装饰品,不再加规则(黑名单 fundamentally broken),防线在 OS 层 |
 | sandbox 出站越权 / 资源滥用 | default-deny + 受控 proxy;硬限制 + 软配额 + idle 回收 |
 

PROGRESS.md

@@ -2,7 +2,7 @@
 
 > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。
 
-最后更新:2026-07-27(联网搜索动态时效约束,bump 0.60.1)
+最后更新:2026-07-27(Web 消息接收事务化,bump 0.60.2)
 
 ---
 
@@ -23,6 +23,7 @@
 
 ### 2026-07
 
+- **07-27 / 0.60.2 / Web 消息接收事务化（消除 202 后丢输入窗口）**:`POST /messages` 先完成媒体 variant 校验，再在 task 行锁保护下把用户消息与 `run_status=running` 同事务提交；Web worker 新增已持久化轮次入口，只消费 Session 末尾 user 而不重复 append，CLI/渠道/定时入口继续走旧入口保持兼容。后台 coroutine 调度失败会把 task 收敛为 error，不再留下假 running；不新增 runs 表、队列或 migration。全量 375 测试通过（17 项按环境跳过），新增 DB 路由测试因未设置 `ZCBOT_TEST_DB_URL` 按安全门控跳过。
 - **07-27 / 0.60.1 / 联网搜索动态时效约束**:真实 task `24d6b609` 在 2026-07-27、用户未指定年份时，`deepseek_v4.flash` 两次自行给 `web_search` query 加 `2025`；工具原样转发且默认 `freshness=noLimit`，根因是 system prompt 虽注入 today 却仅标作宪法文件命名用途，工具 schema 也无年份策略。修为 system prompt 首段按 `Asia/Shanghai` 动态注入当前时间/日期/年份，并给唯一规则：未指定时间则 query 无年份，“最新/当前/近期”按当前日期配合 freshness，明确时间窗则服从用户；`web_search` 的 query/freshness schema 再局部强化。5 项定向测试通过。
 - **07-27 / 0.60.0 / 目录优先的新对话草稿页 + 首条消息自动命名**:登录未选任务与左栏「+ 新对话」统一进入不落 DB 的草稿页，选择/新建 working_dir 后首发才创建 task，避免空任务；原完整表单保留为「自定义」。新增 0023 `tasks.auto_title_pending` 一次性闸与 `auto_title` 可选创建字段，首条消息并行生成短标题（`usage_events.kind=task_title`），只改 task 显示名不动目录，人工 PATCH name 清闸且条件 UPDATE 防在途覆盖；失败保留「新对话」不阻塞主 run。
 - **07-27 / 0.59.6 / 对话内安全重命名 working_dir**:新增 `rename_working_dir` 受控工具，agent run 内只登记目标 leaf 名，正常回复结束并把当前 task 退出 running 后再落地，避免 executor/system prompt/宿主工具仍握旧 cwd；取消或失败不执行。文件面板与对话路径共用 `core/working_dirs.py` DB-aware 原语：锁定并同步更新共享目录的全部 task、活跃邻居与 no-subtask 冲突安全拒绝、DB UPDATE 后再做 FS rename。system prompt 明确当前 task_dir 禁走 shell/run_python 直接改名，定时 run 不挂该工具；无新增 schema/migration。全量 357 测试通过（17 项按环境跳过），DB 路由测试未设 `ZCBOT_TEST_DB_URL` 时按安全门控跳过。

core/__init__.py

@@ -1,3 +1,3 @@
 # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。
 # 改版本只动这一行。
-__version__ = "0.60.1"
+__version__ = "0.60.2"

core/file_store.py

@@ -0,0 +1,135 @@
+"""Durable local-file primitives shared by host and sandbox file mutations.
+
+Writes are staged in the destination directory, fsynced, then published with
+``os.replace`` so readers see either the old complete file or the new complete
+file.  ``interprocess_file_lock`` uses the operating system's advisory lock;
+the lock file may remain on disk, but the lock itself is released automatically
+when a process exits.
+"""
+from __future__ import annotations
+
+import os
+import tempfile
+import time
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Iterator, Optional
+
+
+class FileLockBusy(RuntimeError):
+    """A non-blocking or timed inter-process lock could not be acquired."""
+
+
+def _atomic_replace(path: Path, data: bytes) -> None:
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    old_mode: Optional[int] = None
+    try:
+        old_mode = path.stat().st_mode
+    except FileNotFoundError:
+        pass
+
+    fd, raw_tmp = tempfile.mkstemp(
+        prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent)
+    )
+    tmp = Path(raw_tmp)
+    try:
+        with os.fdopen(fd, "wb") as f:
+            f.write(data)
+            f.flush()
+            os.fsync(f.fileno())
+        if old_mode is not None:
+            os.chmod(tmp, old_mode)
+        os.replace(tmp, path)
+        # Persist the directory entry on POSIX. Windows cannot open directories
+        # this way; os.replace still gives atomic visibility there.
+        if os.name != "nt":
+            try:
+                dir_fd = os.open(path.parent, os.O_RDONLY)
+                try:
+                    os.fsync(dir_fd)
+                finally:
+                    os.close(dir_fd)
+            except OSError:
+                # Some network/virtual filesystems reject directory fsync.
+                # The file itself is already fsynced and atomically visible.
+                pass
+    finally:
+        try:
+            tmp.unlink()
+        except FileNotFoundError:
+            pass
+
+
+def atomic_write_bytes(path: Path, data: bytes) -> None:
+    _atomic_replace(Path(path), data)
+
+
+def atomic_write_text(path: Path, text: str, encoding: str = "utf-8") -> None:
+    _atomic_replace(Path(path), text.encode(encoding))
+
+
+def _try_lock(f) -> bool:
+    f.seek(0)
+    if os.name == "nt":
+        import msvcrt
+
+        try:
+            msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1)
+            return True
+        except OSError:
+            return False
+
+    import fcntl
+
+    try:
+        fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        return True
+    except BlockingIOError:
+        return False
+
+
+def _unlock(f) -> None:
+    f.seek(0)
+    if os.name == "nt":
+        import msvcrt
+
+        msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
+        return
+
+    import fcntl
+
+    fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+
+
+@contextmanager
+def interprocess_file_lock(
+    path: Path,
+    *,
+    timeout_seconds: Optional[float] = 0,
+    poll_seconds: float = 0.05,
+) -> Iterator[None]:
+    """Acquire an advisory exclusive lock.
+
+    ``timeout_seconds=0`` is non-blocking; ``None`` waits indefinitely.
+    """
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "a+b") as f:
+        f.seek(0, os.SEEK_END)
+        if f.tell() == 0:
+            f.write(b"\0")
+            f.flush()
+
+        deadline = (
+            None if timeout_seconds is None
+            else time.monotonic() + max(0.0, timeout_seconds)
+        )
+        while not _try_lock(f):
+            if deadline is not None and time.monotonic() >= deadline:
+                raise FileLockBusy(str(path))
+            time.sleep(poll_seconds)
+        try:
+            yield
+        finally:
+            _unlock(f)

core/kb.py

@@ -21,9 +21,12 @@ from __future__ import annotations
 import re
 import shutil
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Iterable, List, Optional
 from uuid import UUID
 
+from .file_store import atomic_write_bytes, atomic_write_text
+from .kb_lock import kb_mutation_lock
+
 # INDEX 单行格式(全角 ｜ 分隔,摘要/关键词内允许半角标点)。agent 对话内手动入库
 # 与后台 ingest 产出同一格式 —— 契约文本(kb_block)里原样给出。
 INDEX_LINE_FORMAT = "- [标题](docs/<文件名>.md)｜来源 sources/<原件名>｜摘要：<两三句>｜关键词：<逗号分隔>"
@@ -143,11 +146,12 @@ def create_kb(workspace_dir: Path, user_id: UUID, name: str) -> Optional[Path]:
     d = kb_dir(workspace_dir, user_id, name)
     if d is None:
         return None
+    with kb_mutation_lock(workspace_dir, user_id, name):
         (d / "docs").mkdir(parents=True, exist_ok=True)
         (d / "sources").mkdir(parents=True, exist_ok=True)
         idx = d / "INDEX.md"
         if not idx.exists():
-        idx.write_text(f"# {name}\n\n", encoding="utf-8")
+            atomic_write_text(idx, f"# {name}\n\n")
     return d
 
 
@@ -156,6 +160,9 @@ def delete_kb(workspace_dir: Path, user_id: UUID, name: str) -> bool:
     d = kb_dir(workspace_dir, user_id, name)
     if d is None or not d.is_dir():
         return False
+    with kb_mutation_lock(workspace_dir, user_id, name):
+        if not d.is_dir():
+            return False
         shutil.rmtree(d)
     return True
 
@@ -186,11 +193,10 @@ def read_doc(workspace_dir: Path, user_id: UUID, name: str, filename: str) -> Op
         return None
 
 
-def delete_doc(workspace_dir: Path, user_id: UUID, name: str, filename: str) -> bool:
-    """删单篇:docs 文件 + INDEX 对应行 + 对应 source 原件一起删(否则原件会被当
-    待入库重新转一遍 —— 判据使然)。"""
-    d = kb_dir(workspace_dir, user_id, name)
-    if d is None or not is_safe_file_name(filename) or not filename.endswith(".md"):
+def _delete_doc_unlocked(
+    d: Path, filename: str, *, delete_source: bool = True
+) -> bool:
+    if not is_safe_file_name(filename) or not filename.endswith(".md"):
         return False
     doc_rel = f"docs/{filename}"
     entries = _read_index(d)
@@ -200,14 +206,11 @@ def delete_doc(workspace_dir: Path, user_id: UUID, name: str, filename: str) ->
         return False
     if not target.is_file() and hit is None:
         return False
-    if target.is_file():
-        target.unlink()
+    src: Optional[Path] = None
     if hit is not None:
         src_name = hit["source"].removeprefix("sources/")
         if is_safe_file_name(src_name):
             src = d / "sources" / src_name
-            if src.is_file():
-                src.unlink()
         idx = d / "INDEX.md"
         try:
             lines = idx.read_text(encoding="utf-8").splitlines()
@@ -215,28 +218,74 @@ def delete_doc(workspace_dir: Path, user_id: UUID, name: str, filename: str) ->
                 ln for ln in lines
                 if not (_INDEX_LINE_RE.match(ln.strip()) and f"({doc_rel})" in ln)
             ]
-            idx.write_text("\n".join(kept).rstrip() + "\n", encoding="utf-8")
+            atomic_write_text(idx, "\n".join(kept).rstrip() + "\n")
         except (OSError, UnicodeDecodeError):
-            pass
+            return False
+    # Publish the recoverable INDEX state before deleting data:if the process
+    # dies here, an undeleted source is merely pending and can be re-ingested.
+    if target.is_file():
+        target.unlink()
+    if delete_source and src is not None and src.is_file():
+        src.unlink()
     return True
 
 
-def save_source(workspace_dir: Path, user_id: UUID, name: str, filename: str, data: bytes) -> Optional[str]:
-    """上传原件落 sources/(同名覆盖 —— 重传即重新入库的自然语义)。
-    返回落盘文件名;库不存在 / 文件名非法 → None。"""
+def delete_doc(workspace_dir: Path, user_id: UUID, name: str, filename: str) -> bool:
+    """删单篇:docs 文件 + INDEX 对应行 + 对应 source 原件一起删。"""
     d = kb_dir(workspace_dir, user_id, name)
-    if d is None or not d.is_dir() or not is_safe_file_name(filename):
+    if d is None:
+        return False
+    with kb_mutation_lock(workspace_dir, user_id, name):
+        return _delete_doc_unlocked(d, filename)
+
+
+def _save_source_unlocked(d: Path, filename: str, data: bytes) -> Optional[str]:
+    if not d.is_dir() or not is_safe_file_name(filename):
         return None
     src_dir = d / "sources"
     src_dir.mkdir(parents=True, exist_ok=True)
     # 覆盖旧 doc 判据:同名 source 若已在 INDEX,删掉旧条目让它重新排队入库
     doc_rel_hits = [e for e in _read_index(d) if e["source"] == f"sources/{filename}"]
     for e in doc_rel_hits:
-        delete_doc(workspace_dir, user_id, name, e["doc"].removeprefix("docs/"))
-    (src_dir / filename).write_bytes(data)
+        if not _delete_doc_unlocked(
+            d, e["doc"].removeprefix("docs/"), delete_source=False
+        ):
+            return None
+    atomic_write_bytes(src_dir / filename, data)
     return filename
 
 
+def save_sources(
+    workspace_dir: Path,
+    user_id: UUID,
+    name: str,
+    items: Iterable[tuple[str, bytes]],
+) -> Optional[List[str]]:
+    """一把库锁内批量保存原件;任一文件名非法时整批不写。"""
+    d = kb_dir(workspace_dir, user_id, name)
+    batch = list(items)
+    if (
+        d is None
+        or not d.is_dir()
+        or any(not is_safe_file_name(filename) for filename, _data in batch)
+    ):
+        return None
+    with kb_mutation_lock(workspace_dir, user_id, name):
+        saved: List[str] = []
+        for filename, data in batch:
+            result = _save_source_unlocked(d, filename, data)
+            if result is None:
+                return None
+            saved.append(result)
+        return saved
+
+
+def save_source(workspace_dir: Path, user_id: UUID, name: str, filename: str, data: bytes) -> Optional[str]:
+    """上传单个原件;保留给内部调用方的兼容入口。"""
+    saved = save_sources(workspace_dir, user_id, name, [(filename, data)])
+    return saved[0] if saved else None
+
+
 # ── prompt 注入(照 memory_block 范式) ────────────────────────────────
 
 # 零库时的冷启动契约:曾因零库零注入,用户说「放进知识库」被 agent 就近写进

core/kb_ingest.py

@@ -19,7 +19,6 @@ from __future__ import annotations
 import base64
 import hashlib
 import re
-import threading
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
@@ -28,6 +27,8 @@ from uuid import UUID
 from core.ark_client import ArkClient, ArkError
 from core.capabilities import ModelCapabilities
 from core.kb import format_index_line, kb_dir, parse_index, pending_sources
+from core.file_store import atomic_write_text
+from core.kb_lock import KbBusyError, kb_is_locked, kb_mutation_lock
 from core.llm import LLM
 from core.storage.usage import record_chat_usage, record_vision_usage
 
@@ -54,20 +55,18 @@ _SUMMARY_PROMPT = """\
 {content}"""
 
 # ── 进度状态(内存,供 API 轮询) ─────────────────────────────────────
-_guard = threading.Lock()
-_locks: Dict[Tuple[str, str], threading.Lock] = {}
 _status: Dict[Tuple[str, str], Dict[str, Any]] = {}
 
 
-def _lock_for(key: Tuple[str, str]) -> threading.Lock:
-    with _guard:
-        return _locks.setdefault(key, threading.Lock())
-
-
-def ingest_status(user_id: UUID, name: str) -> Dict[str, Any]:
+def ingest_status(
+    user_id: UUID, name: str, workspace_dir: Optional[Path] = None
+) -> Dict[str, Any]:
     """当前/最近一次入库进度(无记录返回 idle);前端轮询 + 库详情附带。"""
     st = _status.get((str(user_id), name))
-    return dict(st) if st else {"running": False, "total": 0, "done": 0, "errors": []}
+    out = dict(st) if st else {"running": False, "total": 0, "done": 0, "errors": []}
+    if workspace_dir is not None and not out.get("running"):
+        out["running"] = kb_is_locked(workspace_dir, user_id, name)
+    return out
 
 
 # ── 转换 ─────────────────────────────────────────────────────────────
@@ -227,7 +226,7 @@ def _append_index(d: Path, line: str) -> None:
     text = idx.read_text(encoding="utf-8") if idx.is_file() else ""
     if text and not text.endswith("\n"):
         text += "\n"
-    idx.write_text(text + line + "\n", encoding="utf-8")
+    atomic_write_text(idx, text + line + "\n")
 
 
 def run_ingest(workspace_dir: Path, user_id: UUID, kb_name: str, models_dir: Path) -> bool:
@@ -240,9 +239,20 @@ def run_ingest(workspace_dir: Path, user_id: UUID, kb_name: str, models_dir: Pat
     if d is None or not d.is_dir():
         return False
     key = (str(user_id), kb_name)
-    lock = _lock_for(key)
-    if not lock.acquire(blocking=False):
+    try:
+        with kb_mutation_lock(workspace_dir, user_id, kb_name):
+            return _run_ingest_locked(d, key, user_id, kb_name, models_dir)
+    except KbBusyError:
         return False
+
+
+def _run_ingest_locked(
+    d: Path,
+    key: Tuple[str, str],
+    user_id: UUID,
+    kb_name: str,
+    models_dir: Path,
+) -> bool:
     try:
         pending = pending_sources(d)
         st: Dict[str, Any] = {
@@ -273,7 +283,7 @@ def run_ingest(workspace_dir: Path, user_id: UUID, kb_name: str, models_dir: Pat
                     raise ValueError(f"抽出文本过少({len(text)} 字符),无法入库")
                 doc_name = _doc_name_for(d, name)
                 (d / "docs").mkdir(parents=True, exist_ok=True)
-                (d / "docs" / doc_name).write_text(text, encoding="utf-8")
+                atomic_write_text(d / "docs" / doc_name, text)
                 if llm is not None and caps is not None:
                     title, summary, keywords = _summarize(
                         llm, caps, text=text, filename=name, user_id=user_id, kb_name=kb_name
@@ -296,4 +306,3 @@ def run_ingest(workspace_dir: Path, user_id: UUID, kb_name: str, models_dir: Pat
             st["running"] = False
             st["current"] = None
             st["finished_at"] = datetime.now().isoformat(timespec="seconds")
-        lock.release()

core/kb_lock.py

@@ -0,0 +1,81 @@
+"""Cross-process mutation lock for one user's knowledge-base library."""
+from __future__ import annotations
+
+import hashlib
+from contextlib import contextmanager, nullcontext
+from pathlib import Path
+from typing import Iterator, Optional
+from uuid import UUID
+
+from .file_store import FileLockBusy, interprocess_file_lock
+
+
+class KbBusyError(RuntimeError):
+    """The target knowledge base is being mutated by another process."""
+
+
+def _user_root(workspace_dir: Path, user_id: UUID) -> Path:
+    return Path(workspace_dir) / "users" / str(user_id)
+
+
+def kb_lock_path(user_root: Path, kb_name: str) -> Path:
+    digest = hashlib.sha256(kb_name.encode("utf-8")).hexdigest()[:32]
+    return Path(user_root) / ".kb" / ".locks" / f"{digest}.lock"
+
+
+@contextmanager
+def kb_mutation_lock(
+    workspace_dir: Path,
+    user_id: UUID,
+    kb_name: str,
+    *,
+    timeout_seconds: Optional[float] = 0,
+) -> Iterator[None]:
+    try:
+        with interprocess_file_lock(
+            kb_lock_path(_user_root(workspace_dir, user_id), kb_name),
+            timeout_seconds=timeout_seconds,
+        ):
+            yield
+    except FileLockBusy as e:
+        raise KbBusyError(kb_name) from e
+
+
+@contextmanager
+def kb_mutation_lock_for_path(
+    target: Path,
+    user_root: Optional[Path],
+    *,
+    timeout_seconds: Optional[float] = 0,
+) -> Iterator[None]:
+    """Lock the containing ``.kb/<name>`` library; no-op outside ``.kb``."""
+    if user_root is None:
+        with nullcontext():
+            yield
+        return
+    try:
+        rel = Path(target).resolve().relative_to(Path(user_root).resolve())
+    except (OSError, ValueError):
+        with nullcontext():
+            yield
+        return
+    if len(rel.parts) < 3 or rel.parts[0] != ".kb" or rel.parts[1].startswith("."):
+        with nullcontext():
+            yield
+        return
+    try:
+        with interprocess_file_lock(
+            kb_lock_path(Path(user_root), rel.parts[1]),
+            timeout_seconds=timeout_seconds,
+        ):
+            yield
+    except FileLockBusy as e:
+        raise KbBusyError(rel.parts[1]) from e
+
+
+def kb_is_locked(workspace_dir: Path, user_id: UUID, kb_name: str) -> bool:
+    try:
+        with kb_mutation_lock(workspace_dir, user_id, kb_name):
+            return False
+    except KbBusyError:
+        return True

core/loop.py

@@ -245,7 +245,18 @@ class AgentLoop:
             })
 
     def run(self, user_message: str) -> str:
+        """运行一个尚未落库的用户轮次（CLI、渠道与调度任务的兼容入口）。"""
+        return self._run(user_message)
+
+    def run_persisted_turn(self) -> str:
+        """运行已由调用方原子持久化的用户轮次（Web POST 入口）。"""
+        if not self.session.messages or self.session.messages[-1].get("role") != "user":
+            raise RuntimeError("persisted turn requires the latest message to be user")
+        return self._run(None)
+
+    def _run(self, user_message: Optional[str]) -> str:
         self._maybe_fold_context()
+        if user_message is not None:
             self.session.append({"role": "user", "content": user_message})
 
         for _ in range(self.max_iterations):

core/session.py

@@ -10,7 +10,6 @@ system prompt **不入库** —— 每次 build_agent 重建拼到 messages[0](
 """
 from __future__ import annotations
 
-import os
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 from uuid import UUID
@@ -19,6 +18,7 @@ from sqlalchemy import delete, func, select
 
 from .storage import session_scope
 from .storage.models import Message, Task
+from .file_store import atomic_write_text
 
 
 def _to_dict(msg: Any) -> Any:
@@ -31,21 +31,6 @@ def _to_dict(msg: Any) -> Any:
     return msg
 
 
-def atomic_write_text(path: Path, text: str, encoding: str = "utf-8") -> None:
-    """原子写: 先写到 path.tmp 再 os.replace 到 path。
-
-    防止写中途异常(磁盘满 / surrogate 编码错 / 进程被杀)留下 0 字节或半文件。
-    skill 产物(*.spec.md / sections/*.md 等)走这里,messages 已改走 PG。
-    """
-    path.parent.mkdir(parents=True, exist_ok=True)
-    tmp = path.with_suffix(path.suffix + ".tmp")
-    with open(tmp, "w", encoding=encoding, newline="\n") as f:
-        f.write(text)
-        f.flush()
-        os.fsync(f.fileno())
-    os.replace(tmp, path)
-
-
 class Session:
     """消息列表 anchored on task_id。
 

deploy/sandbox/Dockerfile

@@ -172,6 +172,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends file \
 # edit/glob/grep 全在容器内执行,物理边界替代代码护栏。tools/ 目录与 host 同步
 # (build 时 COPY,不挂 mount ── 容器内代码不应跟随 host repo 修改重启)。
 COPY tools/ /sandbox/tools/
+COPY core/__init__.py core/file_store.py core/kb_lock.py /sandbox/core/
 COPY core/sandbox/tool_runner.py /sandbox/tool_runner.py
 
 COPY deploy/sandbox/init.sh /init.sh

tests/test_file_store.py

@@ -0,0 +1,175 @@
+"""原子文件写与知识库跨进程写锁回归。"""
+from __future__ import annotations
+
+import tempfile
+import unittest
+from multiprocessing import get_context
+from pathlib import Path
+from unittest.mock import patch
+from uuid import UUID
+
+from core.file_store import (
+    FileLockBusy,
+    atomic_write_text,
+    interprocess_file_lock,
+)
+from core.kb import (
+    create_kb,
+    delete_doc,
+    delete_kb,
+    format_index_line,
+    save_source,
+    save_sources,
+)
+from core.kb_lock import KbBusyError, kb_is_locked, kb_mutation_lock
+from tools.fs import WriteTool
+
+
+_UID = UUID("11111111-2222-3333-4444-555555555555")
+
+
+def _hold_lock_in_child(path: str, ready, release) -> None:
+    with interprocess_file_lock(Path(path)):
+        ready.set()
+        release.wait(10)
+
+
+class AtomicWriteTests(unittest.TestCase):
+    def test_atomic_write_replaces_complete_file_and_cleans_temp(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "a.txt"
+            path.write_text("old", encoding="utf-8")
+            atomic_write_text(path, "新内容")
+            self.assertEqual(path.read_text(encoding="utf-8"), "新内容")
+            self.assertEqual(list(path.parent.glob(".a.txt.*.tmp")), [])
+
+    def test_replace_failure_keeps_old_file_and_cleans_temp(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "a.txt"
+            path.write_text("old", encoding="utf-8")
+            with patch("core.file_store.os.replace", side_effect=OSError("boom")):
+                with self.assertRaises(OSError):
+                    atomic_write_text(path, "new")
+            self.assertEqual(path.read_text(encoding="utf-8"), "old")
+            self.assertEqual(list(path.parent.glob(".a.txt.*.tmp")), [])
+
+
+class FileLockTests(unittest.TestCase):
+    def test_second_handle_cannot_acquire_locked_file(self):
+        with tempfile.TemporaryDirectory() as td:
+            lock = Path(td) / "x.lock"
+            with interprocess_file_lock(lock):
+                with self.assertRaises(FileLockBusy):
+                    with interprocess_file_lock(lock):
+                        pass
+            with interprocess_file_lock(lock):
+                pass
+
+    def test_lock_is_visible_across_processes(self):
+        with tempfile.TemporaryDirectory() as td:
+            lock = Path(td) / "cross-process.lock"
+            ctx = get_context("spawn")
+            ready = ctx.Event()
+            release = ctx.Event()
+            child = ctx.Process(
+                target=_hold_lock_in_child, args=(str(lock), ready, release)
+            )
+            child.start()
+            try:
+                self.assertTrue(ready.wait(5), "child did not acquire lock")
+                with self.assertRaises(FileLockBusy):
+                    with interprocess_file_lock(lock):
+                        pass
+            finally:
+                release.set()
+                child.join(5)
+                if child.is_alive():
+                    child.terminate()
+                    child.join(5)
+            self.assertEqual(child.exitcode, 0)
+
+
+class SandboxPackagingTests(unittest.TestCase):
+    def test_sandbox_copies_file_store_dependencies(self):
+        root = Path(__file__).resolve().parents[1]
+        dockerfile = (root / "deploy" / "sandbox" / "Dockerfile").read_text("utf-8")
+        self.assertIn("core/file_store.py", dockerfile)
+        self.assertIn("core/kb_lock.py", dockerfile)
+
+
+class KnowledgeBaseMutationTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.ws = Path(self.tmp.name)
+        self.user_root = self.ws / "users" / str(_UID)
+        self.kb = create_kb(self.ws, _UID, "标准库")
+        assert self.kb is not None
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_lock_blocks_service_and_agent_mutations(self):
+        idx = self.kb / "INDEX.md"
+        old = idx.read_text(encoding="utf-8")
+        tool = WriteTool(base_dir=self.user_root, user_root=self.user_root)
+
+        with kb_mutation_lock(self.ws, _UID, "标准库"):
+            self.assertTrue(kb_is_locked(self.ws, _UID, "标准库"))
+            with self.assertRaises(KbBusyError):
+                delete_kb(self.ws, _UID, "标准库")
+            result = tool.execute(".kb/标准库/INDEX.md", "bad")
+            self.assertTrue(result.startswith("[Error]"))
+            self.assertEqual(idx.read_text(encoding="utf-8"), old)
+
+        self.assertFalse(kb_is_locked(self.ws, _UID, "标准库"))
+        result = tool.execute(".kb/标准库/INDEX.md", "ok")
+        self.assertTrue(result.startswith("[wrote"))
+        self.assertEqual(idx.read_text(encoding="utf-8"), "ok")
+
+    def test_invalid_batch_is_rejected_before_any_write(self):
+        result = save_sources(
+            self.ws,
+            _UID,
+            "标准库",
+            [("good.txt", b"good"), ("../bad.txt", b"bad")],
+        )
+        self.assertIsNone(result)
+        self.assertFalse((self.kb / "sources" / "good.txt").exists())
+
+    def _seed_indexed_doc(self):
+        source = self.kb / "sources" / "a.txt"
+        doc = self.kb / "docs" / "a.md"
+        source.write_bytes(b"old")
+        doc.write_text("old doc", encoding="utf-8")
+        line = format_index_line(
+            title="A",
+            doc="docs/a.md",
+            source="sources/a.txt",
+            summary="old",
+            keywords="a",
+        )
+        (self.kb / "INDEX.md").write_text(
+            f"# 标准库\n\n{line}\n", encoding="utf-8"
+        )
+        return source, doc
+
+    def test_overwrite_removes_old_index_and_doc_then_publishes_source(self):
+        source, doc = self._seed_indexed_doc()
+        self.assertEqual(
+            save_source(self.ws, _UID, "标准库", "a.txt", b"new"), "a.txt"
+        )
+        self.assertEqual(source.read_bytes(), b"new")
+        self.assertFalse(doc.exists())
+        self.assertNotIn("docs/a.md", (self.kb / "INDEX.md").read_text("utf-8"))
+
+    def test_delete_keeps_data_when_index_publish_fails(self):
+        source, doc = self._seed_indexed_doc()
+        with patch("core.kb.atomic_write_text", side_effect=OSError("disk full")):
+            self.assertFalse(delete_doc(self.ws, _UID, "标准库", "a.md"))
+        self.assertTrue(source.exists())
+        self.assertTrue(doc.exists())
+        self.assertIn("docs/a.md", (self.kb / "INDEX.md").read_text("utf-8"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_loop_persisted_turn.py

@@ -0,0 +1,65 @@
+"""AgentLoop 已持久化用户轮次入口的回归测试。"""
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+from uuid import uuid4
+
+from core.loop import AgentLoop
+
+
+class _Session:
+    def __init__(self, messages=None):
+        self.messages = list(messages or [])
+        self.appended = []
+
+    def append(self, message):
+        self.messages.append(message)
+        self.appended.append(message)
+        return uuid4()
+
+
+def _loop(session: _Session) -> AgentLoop:
+    loop = AgentLoop(
+        llm=MagicMock(),
+        executor=MagicMock(),
+        session=session,
+        capabilities=SimpleNamespace(max_iterations=1),
+        user_id=uuid4(),
+        working_dir=Path("."),
+        cancel_check=lambda: True,
+    )
+    loop._maybe_fold_context = MagicMock()
+    return loop
+
+
+class PersistedTurnTests(unittest.TestCase):
+    def test_persisted_turn_does_not_append_duplicate_user_message(self) -> None:
+        session = _Session([{"role": "user", "content": "已落库"}])
+        result = _loop(session).run_persisted_turn()
+
+        self.assertEqual(result, "[cancelled]")
+        self.assertEqual(session.appended, [])
+        self.assertEqual(session.messages, [{"role": "user", "content": "已落库"}])
+
+    def test_persisted_turn_requires_latest_user_message(self) -> None:
+        session = _Session([{"role": "assistant", "content": "旧回复"}])
+
+        with self.assertRaisesRegex(RuntimeError, "latest message"):
+            _loop(session).run_persisted_turn()
+
+    def test_legacy_run_still_appends_user_message(self) -> None:
+        session = _Session()
+        result = _loop(session).run("新消息")
+
+        self.assertEqual(result, "[cancelled]")
+        self.assertEqual(
+            session.appended,
+            [{"role": "user", "content": "新消息"}],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_rename_working_dir.py

@@ -58,6 +58,40 @@ class RenameWorkingDirToolTests(unittest.TestCase):
 
 
 class DeferredRenameWorkerTests(unittest.TestCase):
+    def test_persisted_web_turn_uses_non_appending_agent_entry(self) -> None:
+        from web import runs
+
+        tid = uuid4()
+        uid = uuid4()
+        agent = SimpleNamespace(
+            run=MagicMock(return_value="wrong entry"),
+            run_persisted_turn=MagicMock(return_value="ok"),
+            deferred_actions=DeferredTaskActions(),
+            sink=None,
+        )
+
+        @contextmanager
+        def fake_scope():
+            yield SimpleNamespace(execute=MagicMock())
+
+        with tempfile.TemporaryDirectory() as tmp:
+            task_dir = Path(tmp)
+            with (
+                patch("core.agent_builder.build_agent", return_value=(
+                    agent, MagicMock(), str(tid), MagicMock(), task_dir,
+                )),
+                patch("core.agent_builder.sync_task_tokens"),
+                patch.object(runs, "session_scope", fake_scope),
+                patch.object(runs, "broker", MagicMock()),
+            ):
+                runs.run_agent_bg(
+                    tid, uid, "已落库",
+                    user_message_persisted=True,
+                )
+
+        agent.run_persisted_turn.assert_called_once_with()
+        agent.run.assert_not_called()
+
     def test_normal_run_renames_after_status_commit_before_done(self) -> None:
         from web import runs
 

tests/test_web_routes_db.py

@@ -10,18 +10,21 @@
   move 被引用→409、递归删被引用→409、软删后放行
 - upload(磁盘配额 gate 路径)+ 根目录列表(system_wd_names DB 查询)
 - clear / cancel 的状态闸;schedules 404 路径;/v1/models 档位过滤
+- messages 参数失败不改状态；202 前用户消息已持久化且 worker 不重复写入
 
 隔离:测试专属 User 行 + 随机 uid 子树,teardown DB 行与 FS 整树删除;
-POST messages / optimize_prompt(会起真 LLM)不在此测。
+optimize_prompt(会起真 LLM)不在此测。
 """
 from __future__ import annotations
 
 import os
 import shutil
 import sys
+import threading
 import unittest
 import uuid
 from pathlib import Path
+from unittest.mock import patch
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
@@ -59,6 +62,8 @@ if _DB_OK:
     from web.auth import AuthConfig, mint_token
 
     _app = create_app()
+    # 不进入 lifespan，手工补消息路由登记后台任务所需的运行态容器。
+    _app.state.inflight = {}
     _client = TestClient(_app)   # 不进 with:不跑 lifespan
     _UID = uuid.uuid4()
     _TOKEN, _ = mint_token(AuthConfig.from_env(), _UID)
@@ -186,6 +191,84 @@ class TasksCrudTests(unittest.TestCase):
         self.assertEqual(_client.get(f"/v1/tasks/{tid}/outline", headers=_AUTH).json()["items"], [])
 
 
+class MessageRunDurabilityTests(unittest.TestCase):
+    def _mk_task(self, name: str) -> str:
+        r = _client.post("/v1/tasks", json={"name": name}, headers=_AUTH)
+        self.assertEqual(r.status_code, 201, r.text)
+        return r.json()["task_id"]
+
+    def test_validation_failure_keeps_task_idle_and_writes_no_message(self):
+        from fastapi import HTTPException
+        from sqlalchemy import select
+
+        tid = self._mk_task("非法媒体参数")
+        with patch(
+            "web.routers.messages.resolve_image_model",
+            side_effect=HTTPException(400, "invalid image model"),
+        ):
+            r = _client.post(
+                f"/v1/tasks/{tid}/messages",
+                json={"content": "不会落库", "image_model": "invalid"},
+                headers=_AUTH,
+            )
+
+        self.assertEqual(r.status_code, 400, r.text)
+        with session_scope() as s:
+            task = s.execute(
+                select(Task.run_status).where(Task.task_id == uuid.UUID(tid))
+            ).scalar_one()
+            messages = s.execute(
+                select(Message).where(Message.task_id == uuid.UUID(tid))
+            ).scalars().all()
+        self.assertEqual(task, "idle")
+        self.assertEqual(messages, [])
+
+    def test_message_is_committed_before_worker_and_not_duplicated(self):
+        from sqlalchemy import select
+
+        tid = self._mk_task("消息持久化")
+        seen = {}
+        worker_done = threading.Event()
+
+        def fake_worker(task_id, user_id, user_message, *args, **kwargs):
+            with session_scope() as s:
+                payloads = s.execute(
+                    select(Message.payload)
+                    .where(Message.task_id == task_id)
+                    .order_by(Message.idx)
+                ).scalars().all()
+            seen["payloads"] = payloads
+            seen["persisted"] = kwargs.get("user_message_persisted")
+            worker_done.set()
+
+        with (
+            patch("web.routers.messages.resolve_image_model", return_value=""),
+            patch("web.routers.messages.resolve_video_model", return_value=""),
+            patch("web.routers.messages.run_agent_bg", side_effect=fake_worker),
+        ):
+            r = _client.post(
+                f"/v1/tasks/{tid}/messages",
+                json={"content": "必须先落库"},
+                headers=_AUTH,
+            )
+            self.assertTrue(worker_done.wait(5), "后台 worker 未启动")
+
+        self.assertEqual(r.status_code, 202, r.text)
+        self.assertTrue(seen["persisted"])
+        self.assertEqual(
+            seen["payloads"],
+            [{"role": "user", "content": "必须先落库"}],
+        )
+        with session_scope() as s:
+            payloads = s.execute(
+                select(Message.payload)
+                .where(Message.task_id == uuid.UUID(tid))
+                .order_by(Message.idx)
+            ).scalars().all()
+        self.assertEqual(payloads, [{"role": "user", "content": "必须先落库"}])
+        _set_run_status(tid, "idle")
+
+
 class FilesDbAwareTests(unittest.TestCase):
     """顶层目录 = task.working_dir 的 DB-aware 分支(§7.4 唯一 mutation 入口)。"""
 

tools/fs.py

@@ -9,6 +9,8 @@ import re
 from pathlib import Path
 
 from .base import Tool
+from core.file_store import atomic_write_text
+from core.kb_lock import KbBusyError, kb_mutation_lock_for_path
 
 
 class ReadTool(Tool):
@@ -64,8 +66,11 @@ class WriteTool(Tool):
 
     def execute(self, path: str, content: str) -> str:
         p = self._resolve(path)
-        p.parent.mkdir(parents=True, exist_ok=True)
-        p.write_text(content, encoding="utf-8")
+        try:
+            with kb_mutation_lock_for_path(p, self.user_root):
+                atomic_write_text(p, content)
+        except KbBusyError:
+            return "[Error] knowledge base is being updated; wait and retry"
         return f"[wrote {len(content)} chars to {self._display(p)}]"
 
 
@@ -90,13 +95,17 @@ class EditTool(Tool):
         disp = self._display(p)
         if not p.exists():
             return f"[Error] file not found: {disp}"
+        try:
+            with kb_mutation_lock_for_path(p, self.user_root):
                 content = p.read_text(encoding="utf-8")
                 count = content.count(old_str)
                 if count == 0:
                     return f"[Error] old_str not found in {disp}"
                 if count > 1:
                     return f"[Error] old_str appears {count} times in {disp}, must be unique — add more context"
-        p.write_text(content.replace(old_str, new_str), encoding="utf-8")
+                atomic_write_text(p, content.replace(old_str, new_str))
+        except KbBusyError:
+            return "[Error] knowledge base is being updated; wait and retry"
         return f"[edited {disp}: 1 replacement]"
 
 

web/routers/kb.py

@@ -43,9 +43,13 @@ def register_kb_routes(app, *, require_user) -> None:
         """建库(幂等)。名字非法 → 400。"""
         from core.agent_builder import resolve_workspace
         from core.kb import create_kb
+        from core.kb_lock import KbBusyError
         name = (body.name or "").strip()
+        try:
             if create_kb(resolve_workspace(None), user_id, name) is None:
                 raise HTTPException(400, f"invalid kb name: {body.name!r}")
+        except KbBusyError:
+            raise HTTPException(409, "该库正在被其他操作修改")
         return {"name": name}
 
     @app.get("/v1/kb/{name}", tags=["kb"])
@@ -54,10 +58,11 @@ def register_kb_routes(app, *, require_user) -> None:
         from core.agent_builder import resolve_workspace
         from core.kb import kb_detail
         from core.kb_ingest import ingest_status
-        detail = kb_detail(resolve_workspace(None), user_id, name)
+        ws = resolve_workspace(None)
+        detail = kb_detail(ws, user_id, name)
         if detail is None:
             raise HTTPException(404, f"kb not found: {name!r}")
-        detail["ingest"] = ingest_status(user_id, name)
+        detail["ingest"] = ingest_status(user_id, name, ws)
         return detail
 
     @app.delete("/v1/kb/{name}", tags=["kb"])
@@ -65,11 +70,12 @@ def register_kb_routes(app, *, require_user) -> None:
         """整库删除(原件 + docs + INDEX)。入库进行中 → 409(避免半截写盘)。"""
         from core.agent_builder import resolve_workspace
         from core.kb import delete_kb
-        from core.kb_ingest import ingest_status
-        if ingest_status(user_id, name).get("running"):
-            raise HTTPException(409, "该库正在入库,等入库结束再删除")
+        from core.kb_lock import KbBusyError
+        try:
             if not delete_kb(resolve_workspace(None), user_id, name):
                 raise HTTPException(404, f"kb not found: {name!r}")
+        except KbBusyError:
+            raise HTTPException(409, "该库正在入库或被其他操作修改")
         return {"deleted": name}
 
     @app.post("/v1/kb/{name}/upload", tags=["kb"])
@@ -84,7 +90,8 @@ def register_kb_routes(app, *, require_user) -> None:
         磁盘配额 gate 与 /v1/files/upload 同款。
         """
         from core.agent_builder import load_config as _load_cfg, resolve_workspace
-        from core.kb import kb_dir, save_source
+        from core.kb import kb_dir, save_sources
+        from core.kb_lock import KbBusyError
         from core.storage.disk_quota import check_disk_quota, parse_bytes
         _quotas_cfg = (_load_cfg().get("quotas") or {})
         _limit = parse_bytes(_quotas_cfg.get("disk_bytes_per_user"))
@@ -97,16 +104,23 @@ def register_kb_routes(app, *, require_user) -> None:
         d = kb_dir(ws, user_id, name)
         if d is None or not d.is_dir():
             raise HTTPException(404, f"kb not found: {name!r}")
-        saved: list[dict] = []
+        uploads: list[tuple[str, bytes]] = []
         for up in files or []:
             raw_name = up.filename or ""
             data = await up.read()
-            ok = save_source(ws, user_id, name, raw_name, data)
-            if ok is None:
-                raise HTTPException(400, f"invalid filename: {raw_name!r}")
-            saved.append({"name": ok, "size": len(data)})
-        if not saved:
+            uploads.append((raw_name, data))
+        if not uploads:
             raise HTTPException(400, "no files uploaded")
+        try:
+            saved_names = save_sources(ws, user_id, name, uploads)
+        except KbBusyError:
+            raise HTTPException(409, "该库正在入库或被其他操作修改")
+        if saved_names is None:
+            raise HTTPException(400, "存在非法文件名")
+        saved = [
+            {"name": saved_name, "size": len(data)}
+            for saved_name, (_raw_name, data) in zip(saved_names, uploads)
+        ]
         _spawn_kb_ingest(user_id, name)
         return {"count": len(saved), "saved": saved}
 
@@ -135,6 +149,10 @@ def register_kb_routes(app, *, require_user) -> None:
         """删单篇(docs 文件 + INDEX 行 + source 原件,防原件被重新入库)。"""
         from core.agent_builder import resolve_workspace
         from core.kb import delete_doc
+        from core.kb_lock import KbBusyError
+        try:
             if not delete_doc(resolve_workspace(None), user_id, name, filename):
                 raise HTTPException(404, f"doc not found: {filename!r}")
+        except KbBusyError:
+            raise HTTPException(409, "该库正在入库或被其他操作修改")
         return {"deleted": filename}

web/routers/messages.py

@@ -207,6 +207,10 @@ def register_message_routes(app, *, require_user) -> None:
         content, _sc_hit = await asyncio.to_thread(_shortcuts.expand, _ws, user_id, content)
         if _sc_hit:
             print(f"[shortcut] {str(user_id)[:8]} '{_sc_hit}' expanded")
+        # 所有可能返回 4xx 的 variant 校验必须发生在事务写入之前。否则 task 已提交
+        # running 后才发现参数非法，会留下一个实际上没有 worker 的假活跃任务。
+        image_variant = resolve_image_model(body.image_model, user_id=user_id)
+        video_variant = resolve_video_model(body.video_model, user_id=user_id)
         with session_scope() as s:
             row = s.execute(
                 select(
@@ -248,22 +252,49 @@ def register_message_routes(app, *, require_user) -> None:
                     fb_profile, fb_model_id = resolve_model_profile(FALLBACK_MODEL_PROFILE)
                     values["model_profile"] = fb_profile
                     values["model"] = fb_model_id
+            # task 行锁串行化同一会话的 idx 分配。用户消息与 running 状态同事务提交：
+            # 只要客户端拿到 202，这一轮输入就已可从 DB 恢复；worker 不再承担首条
+            # user 消息的持久化责任。
+            next_idx = s.execute(
+                select(func.coalesce(func.max(Message.idx), -1) + 1)
+                .where(Message.task_id == tid)
+            ).scalar_one()
+            s.add(Message(
+                task_id=tid,
+                idx=int(next_idx),
+                payload={"role": "user", "content": content},
+            ))
             s.execute(
                 update(Task).where(Task.task_id == tid).values(**values)
             )
             title_profile = values.get("model_profile", cur_profile)
             should_auto_title = bool(row.auto_title_pending)
-        # image_model / video_model 在 POST 时校验,避免 BG 线程里抛在 sink 之外难追;空串透传不查 yaml。
-        # 显式选中非空 variant 时过档位门控(档外 → 403);空串不查(走 yaml 默认)。
-        image_variant = resolve_image_model(body.image_model, user_id=user_id)
-        video_variant = resolve_video_model(body.video_model, user_id=user_id)
         broker.start(tid)  # 清上一轮 done 标记,新订阅者才能看到流式
         # commit 后 lock 释放;BG 线程接管(sink 通过 broker 把 event 桥回 asyncio loop)。
         # 登记到 app.state.inflight:① 关停 drain 时 await 它收尾 ② 持强引用防 task 被 GC
         # 中途回收(asyncio.create_task 不留引用是已知坑)。done 回调自摘除。
-        run_task = asyncio.create_task(asyncio.to_thread(
+        run_coro = asyncio.to_thread(
             run_agent_bg, tid, user_id, content, image_variant, video_variant,
-        ))
+            user_message_persisted=True,
+        )
+        try:
+            run_task = asyncio.create_task(run_coro)
+        except Exception as e:
+            # 极少见的 event-loop 调度失败也不能让 task 永久留在 running。消息已经
+            # 持久化，明确标 error，后续可直接续跑；关闭未调度 coroutine 避免告警。
+            run_coro.close()
+            err = f"background scheduling failed: {type(e).__name__}: {e}"
+            with session_scope() as s:
+                s.execute(
+                    update(Task).where(Task.task_id == tid).values(
+                        run_status="error", run_error=err,
+                    )
+                )
+            broker.close(tid)
+            raise HTTPException(
+                500,
+                "message persisted, but background scheduling failed; retry this task",
+            )
         app.state.inflight[run_task] = tid
         run_task.add_done_callback(lambda t: app.state.inflight.pop(t, None))
         # 快速入口只在首条消息时 pending=true。辅助标题与主 run 并行，不阻塞

web/runs.py

@@ -26,6 +26,7 @@ def run_agent_bg(
     task_id: UUID, user_id: UUID, user_message: str,
     image_variant: str = "", video_variant: str = "",
     scheduled: bool = False,
+    *, user_message_persisted: bool = False,
 ) -> None:
     """工作线程:`build_agent(resume=True)` → 装 WebEventSink + cancel_check → `agent.run` → 写 tasks.run_status。
 
@@ -40,6 +41,8 @@ def run_agent_bg(
 
     image_variant / video_variant:本 run 用哪个 image/video variant 装 tool(空 → yaml 第一个)。
     随消息 POST 传进来,不入 DB —— UI 下拉的选择就跟在这一条消息上生效。
+    user_message_persisted=True 仅供 Web POST：用户消息已和 tasks.running 同事务提交，
+    worker 从 Session 恢复后直接处理最后一条 user，避免重复落库。其余入口保持旧行为。
     """
     from core.agent_builder import build_agent, sync_task_tokens
     cancel_check = lambda tid=task_id: broker.is_cancelled(tid)
@@ -53,7 +56,11 @@ def run_agent_bg(
             scheduled_run=scheduled,
         )
         agent.sink = WebEventSink(broker, task_id)
-        result = agent.run(user_message)
+        result = (
+            agent.run_persisted_turn()
+            if user_message_persisted
+            else agent.run(user_message)
+        )
         sync_task_tokens(task_state)
         # 收尾终态:agent.run 在任一取消路径都 return "[cancelled]"(loop.py)——
         # 用户停止 → 落持久 cancelled(前端 renderPersistedRunTerminal 据此补「已停止」卡,