.agents/MEMORY.md

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

CHANGELOG.md

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

DESIGN.md

@@ -29,8 +29,6 @@ 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(聚合读)三分
@@ -96,7 +94,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;**写并发收口为共享 FS advisory lock**(`.kb/.locks/<hash>.lock`):Web 上传/删除、后台入库与 agent `write/edit` 对同一库共用一把跨进程锁,蓝绿实例间只允许一个写者,锁占用返 409/工具可重试;进程退出由 OS 自动释锁,不靠清理锁文件。文档正文、原件与 INDEX 全走同目录临时文件 + fsync + `os.replace` 原子发布,读者只会看到完整旧版或完整新版。进度详情仍以内存保存细节,但会探测跨进程锁补出 `running`;崩了靠 FS 判据续跑。
+- **入库管线**(`core/kb_ingest.py`,上传即触发 + 手动兜底):markitdown Python API 转 md → 扫描件 PDF(文本近零)走方舟文档理解 OCR 兜底(§8.13 同通道)→ deepseek flash 单次 chat 写 标题/摘要/关键词(失败降级文件名+正文开头,不阻塞)→ 追加 INDEX 行。编排照定时执行器:create_task + to_thread + per-(user,库) 内存锁去重;进度存内存供前端轮询,崩了靠 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"}`。
@@ -160,8 +158,7 @@ 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 将 user 消息与 running 同事务提交；
-            BG worker 消费已持久化轮次，不重复追加 user，防 202 后崩溃丢输入/idx race)
+            SELECT FOR UPDATE 锁 task 行防 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
@@ -247,7 +244,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 / 202 后进程退出丢输入 | 单活 run gate(FOR UPDATE + 409)下原子提交 user 消息与 running；worker 只消费已持久化轮次；lifespan reaper 收敛残留 running，multi-worker 再换 lease |
+| 并发 POST 撞 messages.idx | 单活 run gate(FOR UPDATE + 409)+ lifespan reaper;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(Web 消息接收事务化,bump 0.60.2)
+最后更新:2026-07-27(联网搜索动态时效约束,bump 0.60.1)
 
 ---
 
@@ -23,7 +23,6 @@
 
 ### 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.2"
+__version__ = "0.60.1"

core/file_store.py

@@ -1,135 +0,0 @@
-"""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,12 +21,9 @@ from __future__ import annotations
 import re
 import shutil
 from pathlib import Path
-from typing import Any, Dict, Iterable, List, Optional
+from typing import Any, Dict, 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/<原件名>｜摘要：<两三句>｜关键词：<逗号分隔>"
@@ -146,12 +143,11 @@ 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():
-            atomic_write_text(idx, f"# {name}\n\n")
+    (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")
     return d
 
 
@@ -160,10 +156,7 @@ 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)
+    shutil.rmtree(d)
     return True
 
 
@@ -193,10 +186,11 @@ def read_doc(workspace_dir: Path, user_id: UUID, name: str, filename: str) -> Op
         return None
 
 
-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"):
+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"):
         return False
     doc_rel = f"docs/{filename}"
     entries = _read_index(d)
@@ -206,11 +200,14 @@ def _delete_doc_unlocked(
         return False
     if not target.is_file() and hit is None:
         return False
-    src: Optional[Path] = None
+    if target.is_file():
+        target.unlink()
     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()
@@ -218,74 +215,28 @@ def _delete_doc_unlocked(
                 ln for ln in lines
                 if not (_INDEX_LINE_RE.match(ln.strip()) and f"({doc_rel})" in ln)
             ]
-            atomic_write_text(idx, "\n".join(kept).rstrip() + "\n")
+            idx.write_text("\n".join(kept).rstrip() + "\n", encoding="utf-8")
         except (OSError, UnicodeDecodeError):
-            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()
+            pass
     return True
 
 
-def delete_doc(workspace_dir: Path, user_id: UUID, name: str, filename: str) -> bool:
-    """删单篇:docs 文件 + INDEX 对应行 + 对应 source 原件一起删。"""
+def save_source(workspace_dir: Path, user_id: UUID, name: str, filename: str, data: bytes) -> Optional[str]:
+    """上传原件落 sources/(同名覆盖 —— 重传即重新入库的自然语义)。
+    返回落盘文件名;库不存在 / 文件名非法 → None。"""
     d = kb_dir(workspace_dir, user_id, name)
-    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):
+    if d is None or 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:
-        if not _delete_doc_unlocked(
-            d, e["doc"].removeprefix("docs/"), delete_source=False
-        ):
-            return None
-    atomic_write_bytes(src_dir / filename, data)
+        delete_doc(workspace_dir, user_id, name, e["doc"].removeprefix("docs/"))
+    (src_dir / filename).write_bytes(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,6 +19,7 @@ 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
@@ -27,8 +28,6 @@ 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
 
@@ -55,18 +54,20 @@ _SUMMARY_PROMPT = """\
 {content}"""
 
 # ── 进度状态(内存,供 API 轮询) ─────────────────────────────────────
+_guard = threading.Lock()
+_locks: Dict[Tuple[str, str], threading.Lock] = {}
 _status: Dict[Tuple[str, str], Dict[str, Any]] = {}
 
 
-def ingest_status(
-    user_id: UUID, name: str, workspace_dir: Optional[Path] = None
-) -> 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]:
     """当前/最近一次入库进度(无记录返回 idle);前端轮询 + 库详情附带。"""
     st = _status.get((str(user_id), name))
-    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
+    return dict(st) if st else {"running": False, "total": 0, "done": 0, "errors": []}
 
 
 # ── 转换 ─────────────────────────────────────────────────────────────
@@ -226,7 +227,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"
-    atomic_write_text(idx, text + line + "\n")
+    idx.write_text(text + line + "\n", encoding="utf-8")
 
 
 def run_ingest(workspace_dir: Path, user_id: UUID, kb_name: str, models_dir: Path) -> bool:
@@ -239,20 +240,9 @@ 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)
-    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:
+    lock = _lock_for(key)
+    if not lock.acquire(blocking=False):
         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] = {
@@ -283,7 +273,7 @@ def _run_ingest_locked(
                     raise ValueError(f"抽出文本过少({len(text)} 字符),无法入库")
                 doc_name = _doc_name_for(d, name)
                 (d / "docs").mkdir(parents=True, exist_ok=True)
-                atomic_write_text(d / "docs" / doc_name, text)
+                (d / "docs" / doc_name).write_text(text, encoding="utf-8")
                 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
@@ -306,3 +296,4 @@ def _run_ingest_locked(
             st["running"] = False
             st["current"] = None
             st["finished_at"] = datetime.now().isoformat(timespec="seconds")
+        lock.release()

core/kb_lock.py

@@ -1,81 +0,0 @@
-"""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,19 +245,8 @@ 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})
+        self.session.append({"role": "user", "content": user_message})
 
         for _ in range(self.max_iterations):
             if self._is_cancelled():

core/session.py

@@ -10,6 +10,7 @@ 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
@@ -18,7 +19,6 @@ 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,6 +31,21 @@ 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,7 +172,6 @@ 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

@@ -1,175 +0,0 @@
-"""原子文件写与知识库跨进程写锁回归。"""
-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

@@ -1,65 +0,0 @@
-"""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,40 +58,6 @@ 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,21 +10,18 @@
   move 被引用→409、递归删被引用→409、软删后放行
 - upload(磁盘配额 gate 路径)+ 根目录列表(system_wd_names DB 查询)
 - clear / cancel 的状态闸;schedules 404 路径;/v1/models 档位过滤
-- messages 参数失败不改状态；202 前用户消息已持久化且 worker 不重复写入
 
 隔离:测试专属 User 行 + 随机 uid 子树,teardown DB 行与 FS 整树删除;
-optimize_prompt(会起真 LLM)不在此测。
+POST messages / 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]))
 
@@ -62,8 +59,6 @@ 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)
@@ -191,84 +186,6 @@ 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,8 +9,6 @@ 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):
@@ -66,11 +64,8 @@ class WriteTool(Tool):
 
     def execute(self, path: str, content: str) -> str:
         p = self._resolve(path)
-        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"
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(content, encoding="utf-8")
         return f"[wrote {len(content)} chars to {self._display(p)}]"
 
 
@@ -95,17 +90,13 @@ 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"
-                atomic_write_text(p, content.replace(old_str, new_str))
-        except KbBusyError:
-            return "[Error] knowledge base is being updated; wait and retry"
+        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")
         return f"[edited {disp}: 1 replacement]"
 
 

web/routers/kb.py

@@ -43,13 +43,9 @@ 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, "该库正在被其他操作修改")
+        if create_kb(resolve_workspace(None), user_id, name) is None:
+            raise HTTPException(400, f"invalid kb name: {body.name!r}")
         return {"name": name}
 
     @app.get("/v1/kb/{name}", tags=["kb"])
@@ -58,11 +54,10 @@ 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
-        ws = resolve_workspace(None)
-        detail = kb_detail(ws, user_id, name)
+        detail = kb_detail(resolve_workspace(None), user_id, name)
         if detail is None:
             raise HTTPException(404, f"kb not found: {name!r}")
-        detail["ingest"] = ingest_status(user_id, name, ws)
+        detail["ingest"] = ingest_status(user_id, name)
         return detail
 
     @app.delete("/v1/kb/{name}", tags=["kb"])
@@ -70,12 +65,11 @@ 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_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, "该库正在入库或被其他操作修改")
+        from core.kb_ingest import ingest_status
+        if ingest_status(user_id, name).get("running"):
+            raise HTTPException(409, "该库正在入库,等入库结束再删除")
+        if not delete_kb(resolve_workspace(None), user_id, name):
+            raise HTTPException(404, f"kb not found: {name!r}")
         return {"deleted": name}
 
     @app.post("/v1/kb/{name}/upload", tags=["kb"])
@@ -90,8 +84,7 @@ 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_sources
-        from core.kb_lock import KbBusyError
+        from core.kb import kb_dir, save_source
         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"))
@@ -104,23 +97,16 @@ 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}")
-        uploads: list[tuple[str, bytes]] = []
+        saved: list[dict] = []
         for up in files or []:
             raw_name = up.filename or ""
             data = await up.read()
-            uploads.append((raw_name, data))
-        if not uploads:
+            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:
             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}
 
@@ -149,10 +135,6 @@ 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, "该库正在入库或被其他操作修改")
+        if not delete_doc(resolve_workspace(None), user_id, name, filename):
+            raise HTTPException(404, f"doc not found: {filename!r}")
         return {"deleted": filename}

web/routers/messages.py

@@ -207,10 +207,6 @@ 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(
@@ -252,49 +248,22 @@ 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_coro = asyncio.to_thread(
+        run_task = asyncio.create_task(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,7 +26,6 @@ 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。
 
@@ -41,8 +40,6 @@ 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)
@@ -56,11 +53,7 @@ def run_agent_bg(
             scheduled_run=scheduled,
         )
         agent.sink = WebEventSink(broker, task_id)
-        result = (
-            agent.run_persisted_turn()
-            if user_message_persisted
-            else agent.run(user_message)
-        )
+        result = agent.run(user_message)
         sync_task_tokens(task_state)
         # 收尾终态:agent.run 在任一取消路径都 return "[cancelled]"(loop.py)——
         # 用户停止 → 落持久 cancelled(前端 renderPersistedRunTerminal 据此补「已停止」卡,