fix(projects): convert shelf documents outside the SQLite write lock (#6305)

* fix(projects): convert shelf documents outside the SQLite write lock

Lazy conversion ran pymupdf/markitdown inside the BEGIN IMMEDIATE
transaction that serializes the derived/converted.md publish against
trash and purge. On SQLite that lock is database-wide, so every unrelated
writer (run status, thread metadata, the scheduler) waited for the whole
conversion and failed with "database is locked" past busy_timeout (30s).

Conversion now writes into .staging/ outside any transaction. The
document row lock is taken only to revalidate the live row and atomically
rename the output into place (publish_under_live_lock, renamed from
convert_under_live_lock); a row trashed or purged meanwhile discards the
staged output, so nothing is published after purge. A read-only probe
skips converting rows that are already gone, a failed conversion is
still revalidated so a purged row reports content_missing, and the
conversion is drained on cancellation so its staging file is always
removed.

* docs(changelog): reference #6305 in the shelf conversion lock entry

* fix(projects): share one conversion across concurrent first reads

A first-read conversion holds a bounded file-IO pool thread for its
whole duration. With the conversion moved out of the row lock,
concurrent first reads of one document each converted on their own
thread, so a burst of reads could starve unrelated offloaded file work.

Route the conversion through a single-flight keyed by (user, document,
project): the first reader leads and the rest wait on its result
without holding a pool thread. The map holds concurrent.futures.Future
values under a threading.Lock because the read tool also runs on the
isolated subagent loop. Followers wait through asyncio.shield, so a
cancelled follower cannot cancel the shared future, and a leader that
is cancelled or raises hands the work to a follower instead of failing
it.

CHANGELOG.md

@@ -466,6 +466,17 @@ This release closes that milestone with **301 merged pull requests**.
 
 ### Fixed
 
+- **projects:** Reading a shelf document for the first time no longer blocks
+  every other database write on SQLite while the document converts. Lazy
+  conversion ran pymupdf/markitdown inside the `BEGIN IMMEDIATE` transaction
+  that serializes the publish against trash and purge, and on SQLite that lock
+  is database-wide, so run status, thread metadata and scheduler writes waited
+  for the whole conversion and failed with `database is locked` after 30
+  seconds. Conversion now writes into `.staging/` outside any transaction; the
+  lock is held only to revalidate the row and atomically rename the output
+  into place, so a document trashed or purged meanwhile still publishes
+  nothing. Concurrent first reads of one document share a single conversion
+  rather than each holding a file-IO worker. ([#6305])
 - **gateway:** Per-run reads now return the rows of IM-channel owners.
   `start_run` stamps run rows and run events with the raw trusted owner id (for
   example `feishu:owner-777`), but several run-scoped routes filtered by the
@@ -7734,3 +7745,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6255]: https://github.com/bytedance/deer-flow/pull/6255
 [#6263]: https://github.com/bytedance/deer-flow/pull/6263
 [#6282]: https://github.com/bytedance/deer-flow/pull/6282
+[#6305]: https://github.com/bytedance/deer-flow/pull/6305

CHANGELOG_zh.md

@@ -423,6 +423,13 @@
 
 ### 修复
 
+- **项目：** 在 SQLite 上首次读取书架文档时，文档转换期间不再阻塞所有其他数据库
+  写入。此前懒转换在 `BEGIN IMMEDIATE` 事务内运行 pymupdf/markitdown，以便与移入
+  回收站和彻底删除串行化发布；而 SQLite 的这把锁作用于整个数据库，运行状态、线程
+  元数据与调度器的写入都要等待整个转换完成，超过 30 秒即报 `database is locked`。
+  现在转换在任何事务之外写入 `.staging/`，只在重新校验文档行并原子重命名输出时
+  持锁，期间被移入回收站或彻底删除的文档仍不会发布任何内容。同一文档的并发首次读取
+  共享一次转换，不再各自占用文件 IO 工作线程。([#6305])
 - **网关：** 单次运行的读取现在能返回 IM 渠道所有者的数据。`start_run` 用原始
   受信所有者 ID（例如 `feishu:owner-777`）标记运行行与运行事件，但多个运行级路由
   按内部调用方规范化后的 ID 过滤，因此在 SQL 存储上，只要所有者 ID 含有
@@ -6455,3 +6462,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6255]: https://github.com/bytedance/deer-flow/pull/6255
 [#6263]: https://github.com/bytedance/deer-flow/pull/6263
 [#6282]: https://github.com/bytedance/deer-flow/pull/6282
+[#6305]: https://github.com/bytedance/deer-flow/pull/6305

backend/docs/ARCHITECTURE.md

@@ -102,7 +102,7 @@ The web conversation delete flow first deletes Gateway-managed thread state thro
 
 **Role authority.** Project data rides a user-role request message, never the system prompt: framework rules stay in SystemMessages; user-authored project configuration is data.
 
-**Shelf storage (spec §6.1-6.3).** `project_documents` rows carry a server-generated content address (`stored_relpath` relative to `users/{user_id}/projects/`, embedding the sha256 and the row's own document ID) so rows never share files and re-upload after trash lands in a fresh namespace. Layout: `{sha256[:2]}/{sha256}/{document_id}/original/{name}` plus an optional `derived/converted.md` companion written via temp file and atomic rename on first read. Inserts are atomic with the active-project row lock: stage → hash → dedup among active `(project_id, sha256)` → place bytes → `INSERT` (file-before-row); a dedup hit returns the existing row (first name wins). Document delete and project delete are pure row transitions to a trash tier (`trashed_at`, `trash_origin` snapshot) — no filesystem work; trashed rows are invisible to the index, the tools, and the listing APIs.
+**Shelf storage (spec §6.1-6.3).** `project_documents` rows carry a server-generated content address (`stored_relpath` relative to `users/{user_id}/projects/`, embedding the sha256 and the row's own document ID) so rows never share files and re-upload after trash lands in a fresh namespace. Layout: `{sha256[:2]}/{sha256}/{document_id}/original/{name}` plus an optional `derived/converted.md` companion created on first read: conversion writes into `.staging/` outside any database transaction, then the document-row lock is taken only to revalidate the live row and atomically rename the output into place (a row trashed or purged meanwhile discards it); concurrent first reads of one document share a single conversion. Inserts are atomic with the active-project row lock: stage → hash → dedup among active `(project_id, sha256)` → place bytes → `INSERT` (file-before-row); a dedup hit returns the existing row (first name wins). Document delete and project delete are pure row transitions to a trash tier (`trashed_at`, `trash_origin` snapshot) — no filesystem work; trashed rows are invisible to the index, the tools, and the listing APIs.
 
 **Bounded live reads (spec §7.3).** `list_project_documents` / `read_project_document` (registered only for runs carrying the pinned key) take `project_id` from the pin and read live shelf rows; binaries and conversion-disabled convertibles are declined with an error naming attach-to-thread, and documents trashed mid-run fail with a "no longer on the shelf" error rather than serving stale content.
 

backend/packages/harness/deerflow/persistence/AGENTS.md

@@ -5,3 +5,5 @@ Postgres bootstrap owns its session-scoped advisory lock until `pg_advisory_unlo
 When `database.postgres_schema` is configured, both async ORM connections and the synchronous SQLAlchemy connections used by DB-backed custom agents and managed subagents must use the same `search_path`; preserve this invariant when adding another persistence entry point.
 
 Alembic stamp/upgrade workers started inside `bootstrap_schema()` remain owned by the bootstrap critical section until the worker finishes. Drain those `asyncio.to_thread()` calls across host cancellation before releasing the in-process SQLite bootstrap lock or PostgreSQL advisory lock; otherwise another bootstrap can overlap a still-running migration worker.
+
+On SQLite, `BEGIN IMMEDIATE` takes the database-wide write lock, so every unrelated writer (run status, thread metadata, the scheduler) waits for the transaction and fails with `database is locked` after `busy_timeout` (30s). Do slow work such as document conversion before opening a locked transaction; lock only to revalidate and publish, as `ProjectDocumentRepository.publish_under_live_lock` does. `tests/test_project_document_tools.py::TestConversionSerialization` pins this.

backend/packages/harness/deerflow/persistence/projects/sql.py

@@ -477,33 +477,37 @@ class ProjectDocumentRepository:
             await session.commit()
             return row, staged
 
-    async def convert_under_live_lock[T](
+    async def publish_under_live_lock[T](
         self,
         document_id: str,
         *,
         project_id: str,
-        convert: Callable[[dict[str, Any]], Awaitable[T]],
+        publish: Callable[[dict[str, Any]], Awaitable[T]],
         user_id: str | None | _AutoSentinel = AUTO,
     ) -> tuple[dict[str, Any], T] | None:
-        """Run *convert* while holding the owned LIVE document row's lock (§6.3).
+        """Run *publish* while holding the owned LIVE document row's lock (§6.3).
 
         Lazy conversion publishes ``derived/converted.md`` serialized against
         trash and purge: this takes the document row lock inside one
         transaction (``BEGIN IMMEDIATE`` on SQLite, ``SELECT … FOR UPDATE``
         elsewhere), revalidates ownership, shelf membership and active state
-        (``trashed_at IS NULL``) AFTER locking, then invokes *convert* — the
-        offloaded convert-to-temp + atomic ``os.replace`` publish — and
-        commits. A concurrent trash/purge either commits first (the locked
-        re-read finds the row gone/trashed and this returns ``None`` without
-        invoking *convert*, so nothing is published) or blocks until the
-        publish commits — a derived file is never published after purge
-        removed the original (§6.3 lock order, §13). Only the document lock
-        is taken, so no lock-ordering hazard exists against trash's
-        project→document order. ``None`` means missing/foreign/trashed; the
-        caller maps it to its not-on-the-shelf / ``content_missing``
-        semantics.
+        (``trashed_at IS NULL``) AFTER locking, then invokes *publish* — the
+        atomic ``os.replace`` of an already-converted file — and commits. A
+        concurrent trash/purge either commits first (the locked re-read finds
+        the row gone/trashed and this returns ``None`` without invoking
+        *publish*) or blocks until the publish commits — a derived file is
+        never published after purge removed the original (§6.3 lock order,
+        §13). Only the document lock is taken, so no lock-ordering hazard
+        exists against trash's project→document order. ``None`` means
+        missing/foreign/trashed; the caller maps it to its not-on-the-shelf /
+        ``content_missing`` semantics.
+
+        *publish* must stay short: on SQLite ``BEGIN IMMEDIATE`` takes the
+        database-wide write lock, so every unrelated writer (run status,
+        thread metadata, the scheduler) waits for it. Convert BEFORE calling
+        this, never inside *publish*.
         """
-        resolved_user_id = resolve_user_id(user_id, method_name="ProjectDocumentRepository.convert_under_live_lock")
+        resolved_user_id = resolve_user_id(user_id, method_name="ProjectDocumentRepository.publish_under_live_lock")
         async with self._sf() as session:
             await self._begin_immediate_if_sqlite(session)
             locked = (await session.execute(select(ProjectDocumentRow).where(ProjectDocumentRow.id == document_id).with_for_update())).scalar_one_or_none()
@@ -511,9 +515,9 @@ class ProjectDocumentRepository:
                 await session.rollback()
                 return None
             row = self._row_to_dict(locked)
-            # Drained: a cancelled conversion finishes its worker (and any
-            # publish) before the document lock unwinds (§6.3).
-            result = await await_drained(convert(row))
+            # Drained: a cancelled publish finishes its worker before the
+            # document lock unwinds (§6.3).
+            result = await await_drained(publish(row))
             await session.commit()
             return row, result
 

backend/packages/harness/deerflow/projects/documents.py

@@ -13,8 +13,9 @@ Files are immutable and hash-qualified: ``stored_relpath`` (relative to
 ``users/{user_id}/projects/``) embeds the content hash and the row's own
 document ID, so rows never share bytes and a re-upload after trash lands in a
 fresh namespace (§6.2). Conversion is lazy: a convertible original is turned
-into ``derived/converted.md`` on first read, written through a temporary file
-and an atomic rename, and only when ``uploads.auto_convert_documents`` is on
+into ``derived/converted.md`` on first read, converted into ``.staging/``
+outside any database transaction and then atomically renamed into place under
+the document row lock, and only when ``uploads.auto_convert_documents`` is on
 (§6.4, §7.3).
 """
 
@@ -22,13 +23,15 @@ from __future__ import annotations
 
 import asyncio
 import codecs
+import concurrent.futures
 import hashlib
 import logging
 import os
 import shutil
+import threading
 import uuid
 from collections import OrderedDict
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Awaitable, Callable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -36,7 +39,7 @@ from typing import TYPE_CHECKING, Any
 from deerflow.config.paths import Paths
 from deerflow.uploads.manager import is_reserved_upload_filename
 from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown
-from deerflow.utils.file_io import run_file_io
+from deerflow.utils.file_io import await_drained, run_file_io
 from deerflow.utils.text_detection import is_text_file_by_content
 
 if TYPE_CHECKING:
@@ -136,6 +139,11 @@ class StagedDocument:
     size_bytes: int
 
 
+def _new_staging_path(paths: Paths, *, user_id: str, project_id: str, prefix: str = "") -> Path:
+    """A fresh ``.staging/{prefix}{uuid}`` path for in-flight bytes; the sweep collects leftovers after 24 hours (§8.3)."""
+    return paths.project_documents_dir(user_id, project_id) / ".staging" / f"{prefix}{uuid.uuid4().hex}"
+
+
 def _open_staging(staging_dir: Path, staging_path: Path):
     """Worker-thread: create the staging directory and open the staging file."""
     staging_dir.mkdir(parents=True, exist_ok=True)
@@ -160,9 +168,8 @@ async def stage_document_bytes(paths: Paths, *, user_id: str, project_id: str, c
     (the route maps it to 413, mirroring the uploads router) and leaves no
     staging file behind.
     """
-    staging_dir = paths.project_documents_dir(user_id, project_id) / ".staging"
-    staging_path = staging_dir / uuid.uuid4().hex
-    handle = await run_file_io(_open_staging, staging_dir, staging_path)
+    staging_path = _new_staging_path(paths, user_id=user_id, project_id=project_id)
+    handle = await run_file_io(_open_staging, staging_path.parent, staging_path)
     digest = hashlib.sha256()
     size = 0
 
@@ -313,44 +320,96 @@ def _convert_in_thread(original: Path, output_path: Path) -> Path | None:
     return asyncio.run(convert_file_to_markdown(original, output_path=output_path))
 
 
-def _convert_and_publish(original: Path, derived: Path) -> bool:
-    """Worker-thread body: convert to a temp file, then atomically publish.
+#: First-read conversions in flight, keyed by ``(user_id, document_id,
+#: project_id)``. A conversion holds a bounded file-IO pool thread for its
+#: whole duration, so concurrent first reads of one document share it instead
+#: of each pinning a thread. The tool also runs on the isolated subagent loop,
+#: so this is a lock-guarded map of ``concurrent.futures.Future`` rather than
+#: a per-loop asyncio primitive. A ``None`` result means the leader was
+#: cancelled or raised: its followers retry and one of them leads.
+_inflight_conversions: dict[tuple[str, str, str], concurrent.futures.Future[tuple[Path | None, str | None] | None]] = {}
+_inflight_conversions_lock = threading.Lock()
 
-    The temp file lives beside the final path (same filesystem) so the
-    rename is atomic: readers only ever see a complete ``converted.md`` or
-    none at all. A conversion failure publishes nothing and removes the temp.
+
+async def _single_flight_conversion(key: tuple[str, str, str], convert: Callable[[], Awaitable[tuple[Path | None, str | None]]]) -> tuple[Path | None, str | None]:
+    """Run *convert* once per *key* across concurrent callers on any event loop."""
+    while True:
+        with _inflight_conversions_lock:
+            shared = _inflight_conversions.get(key)
+            leading = shared is None
+            if leading:
+                shared = _inflight_conversions[key] = concurrent.futures.Future()
+        if not leading:
+            # Shielded: a cancelled follower must not cancel the shared future
+            # the leader still has to resolve.
+            outcome = await asyncio.shield(asyncio.wrap_future(shared))
+            if outcome is not None:
+                return outcome
+            continue
+        outcome = None
+        try:
+            outcome = await convert()
+            return outcome
+        finally:
+            # Unregister before resolving, so a follower that retries after a
+            # ``None`` outcome cannot find this finished future again.
+            with _inflight_conversions_lock:
+                del _inflight_conversions[key]
+            shared.set_result(outcome)
+
+
+def _convert_to_staging(original: Path, staging_path: Path) -> bool:
+    """Worker-thread body: convert *original* into its ``.staging/`` file.
+
+    Runs outside any database transaction — a conversion can take seconds,
+    and SQLite's write lock is database-wide. The output stays outside the
+    document namespace until published, so a purge racing the conversion
+    never finds, or leaves behind, a partial file there. A conversion failure
+    returns ``False``; the caller removes the staging file either way.
     """
-    derived.parent.mkdir(parents=True, exist_ok=True)
-    temp_path = derived.parent / f".{derived.name}.{uuid.uuid4().hex}.tmp"
+    staging_path.parent.mkdir(parents=True, exist_ok=True)
     try:
-        produced = _convert_in_thread(original, temp_path)
-        if produced is None:
-            return False
-        os.replace(produced, derived)
-        return True
+        return _convert_in_thread(original, staging_path) is not None
     except Exception:
         logger.warning("Failed to convert shelf document %s", original, exc_info=True)
-        temp_path.unlink(missing_ok=True)
         return False
-    finally:
-        temp_path.unlink(missing_ok=True)
+
+
+def _publish_converted(staging_path: Path, derived: Path) -> bool:
+    """Worker-thread body: atomically publish a staged conversion as ``derived/converted.md``.
+
+    ``.staging/`` and the namespace share the projects root, so the rename is
+    atomic: readers only ever see a complete ``converted.md`` or none at all.
+    """
+    try:
+        derived.parent.mkdir(parents=True, exist_ok=True)
+        os.replace(staging_path, derived)
+        return True
+    except Exception:
+        logger.warning("Failed to publish converted shelf document %s", derived, exc_info=True)
+        return False
 
 
 async def ensure_converted_markdown(repo: ProjectDocumentRepository, paths: Paths, *, user_id: str, row: dict, auto_convert: bool) -> tuple[Path | None, str | None]:
     """Return ``(converted_path, None)`` or ``(None, reason)``, converting on first read.
 
-    Conversion runs — and ``derived/converted.md`` is published — while the
-    repository holds the document row lock, with ownership, shelf membership
-    and active state revalidated AFTER locking (§6.3): a trash/purge that
-    committed first makes this decline as ``content_missing`` without
-    converting or publishing anything, and one arriving during conversion
-    blocks on the row lock and then proceeds. The temp-file + atomic
-    ``os.replace`` publish stays inside the lock, so no partial converted
-    text is ever exposed. The published companion is valid for the lifetime
-    of the row's immutable original and never needs invalidation (§6.2,
-    §10.9). ``reason`` is one of ``"conversion_disabled"`` (auto-convert
-    off, §7.3), ``"content_missing"`` (row no longer live), or ``"binary"``
-    (not convertible, or the conversion itself failed).
+    The conversion runs into ``.staging/`` outside any database transaction;
+    only the publish holds the document row lock, with ownership, shelf
+    membership and active state revalidated AFTER locking (§6.3). A
+    trash/purge that committed before this read declines as
+    ``content_missing`` without converting; one that commits during the
+    conversion makes the locked revalidation decline, so the staged output is
+    discarded and nothing is published; one arriving during the publish
+    blocks on the row lock and then proceeds. The atomic ``os.replace``
+    publish exposes no partial converted text. Concurrent first reads of one
+    document share a single conversion (``_single_flight_conversion``); one
+    from another process may still convert again and publish identical
+    bytes. The published companion is
+    valid for the lifetime of the row's immutable original and never needs
+    invalidation (§6.2, §10.9). ``reason`` is one of
+    ``"conversion_disabled"`` (auto-convert off, §7.3), ``"content_missing"``
+    (row no longer live), or ``"binary"`` (not convertible, or the conversion
+    itself failed).
     """
     if not auto_convert:
         return None, "conversion_disabled"
@@ -359,18 +418,37 @@ async def ensure_converted_markdown(repo: ProjectDocumentRepository, paths: Path
         return None, "binary"
     if await run_file_io(derived.is_file):
         return derived, None
+    key = (user_id, row["id"], row["project_id"])
+    return await _single_flight_conversion(key, lambda: _convert_and_publish(repo, paths, user_id=user_id, row=row, original=original, derived=derived))
 
-    async def _convert(_locked_row: dict) -> bool:
-        # Re-check under the lock: a conversion that blocked on this row's
-        # lock may already have been published by the lock's previous holder.
-        if await run_file_io(derived.is_file):
-            return True
-        return await run_file_io(_convert_and_publish, original, derived)
 
-    locked = await repo.convert_under_live_lock(row["id"], project_id=row["project_id"], convert=_convert, user_id=user_id)
+async def _convert_and_publish(repo: ProjectDocumentRepository, paths: Paths, *, user_id: str, row: dict, original: Path, derived: Path) -> tuple[Path | None, str | None]:
+    """Convert into ``.staging/`` outside any transaction, then publish under the row lock (§6.3)."""
+    # Read-only probe: skips converting a row that is already gone. It decides
+    # nothing — the locked revalidation before the publish does (§15.5).
+    live = await repo.get(row["id"], user_id=user_id)
+    if live is None or live["project_id"] != row["project_id"]:
+        return None, "content_missing"
+
+    staging_path = _new_staging_path(paths, user_id=user_id, project_id=row["project_id"], prefix="convert-")
+    try:
+        # Drained: a cancelled read waits for the worker, so the cleanup below
+        # runs after its last write.
+        converted = await await_drained(run_file_io(_convert_to_staging, original, staging_path))
+
+        async def _publish(_locked_row: dict) -> bool:
+            if not converted:
+                return False
+            return await run_file_io(_publish_converted, staging_path, derived)
+
+        # Revalidate even after a failed conversion: a row purged meanwhile is
+        # ``content_missing``, not ``binary``.
+        locked = await repo.publish_under_live_lock(row["id"], project_id=row["project_id"], publish=_publish, user_id=user_id)
+    finally:
+        await run_file_io(_remove_staging, staging_path)
     if locked is None:
-        # Purged/trashed (or foreign) between the caller's read and the
-        # locked revalidation: decline without publishing anything (§6.3).
+        # Purged/trashed (or foreign) before the locked revalidation: the
+        # staged output was discarded, nothing published (§6.3).
         return None, "content_missing"
     _locked_row, published = locked
     return (derived, None) if published else (None, "binary")
@@ -558,7 +636,7 @@ async def stage_document_copy_for_attach(
     missing, foreign, trashed, or not on this project's shelf (the route
     maps it to 404); :class:`ShelfContentMissingError` maps to 409.
     """
-    staging_path = paths.project_documents_dir(user_id, project_id) / ".staging" / f"attach-{uuid.uuid4().hex}"
+    staging_path = _new_staging_path(paths, user_id=user_id, project_id=project_id, prefix="attach-")
 
     async def _copy(row: dict) -> Path:
         await run_file_io(_copy_original_under_lock, paths, staging_path, user_id=user_id, row=row)

backend/tests/blocking_io/test_project_documents.py

@@ -2,8 +2,8 @@
 
 Guards the Phase-2 Slice-B filesystem surfaces (spec §13): upload staging
 and the atomic rename into the document namespace, dedup-hit staging
-cleanup, the conversion publish path (temp file + atomic rename of
-``derived/converted.md``), and the ``read_project_document`` tool's sampled
+cleanup, the conversion publish path (``.staging/`` conversion + atomic
+rename into ``derived/converted.md``), and the ``read_project_document`` tool's sampled
 text detection and content reads.
 
 A real ``init_engine`` cannot run under the strict Blockbuster gate
@@ -167,9 +167,10 @@ async def test_tool_read_dispatches_off_the_loop(tmp_path, monkeypatch) -> None:
 
 async def test_conversion_publish_path_dispatches_off_the_loop(tmp_path, monkeypatch) -> None:
     """First read of a convertible document validates the original
-    (``_content_intact``) and runs the whole convert + temp write + atomic
-    rename (``_convert_and_publish``, inside the locked conversion callback)
-    through the offload."""
+    (``_content_intact``), converts into ``.staging/`` (``_convert_to_staging``,
+    outside any transaction), publishes with an atomic rename
+    (``_publish_converted``, inside the locked publish callback) and removes
+    the staging file (``_remove_staging``) — all through the offload."""
 
     async def fake_convert(file_path, output_path=None):
         output_path.write_text("# converted markdown", encoding="utf-8")
@@ -188,7 +189,9 @@ async def test_conversion_publish_path_dispatches_off_the_loop(tmp_path, monkeyp
         assert payload["content"] == "# converted markdown"
         assert "_content_intact" in calls
         assert "resolve_document_paths" in calls
-        assert "_convert_and_publish" in calls
+        assert "_convert_to_staging" in calls
+        assert "_publish_converted" in calls
+        assert "_remove_staging" in calls
         assert "_read_text_window" in calls
         assert "_cached_char_count" in calls
     finally:

backend/tests/test_project_document_tools.py

@@ -41,7 +41,7 @@ async def env(tmp_path, monkeypatch):
     projects = ProjectRepository(sf)
     docs = ProjectDocumentRepository(sf)
     project = await projects.create(name="P", user_id=_USER)
-    yield SimpleNamespace(paths=paths_mod.get_paths(), projects=projects, docs=docs, project=project)
+    yield SimpleNamespace(paths=paths_mod.get_paths(), sf=sf, projects=projects, docs=docs, project=project)
     await close_engine()
 
 
@@ -508,53 +508,166 @@ class TestReadOriginalIntegrity:
 
 
 class TestConversionSerialization:
-    """§6.3/§13: conversion holds the document row lock through publication,
-    revalidating active state after locking — a trash/purge either commits
-    first (conversion declines, nothing published) or blocks until the
-    publish commits and then proceeds."""
+    """§6.3/§13: conversion runs outside any database transaction and stages
+    its output under ``.staging/``; only the publish — revalidate the live row,
+    then atomically rename into ``derived/`` — holds the document row lock. A
+    trash/purge that commits before the publish makes conversion decline with
+    nothing published; one arriving during the publish blocks until it
+    commits. On SQLite that lock is database-wide, so a conversion holding it
+    would stall every unrelated writer."""
 
     @staticmethod
     def _no_temp_left(env, row: dict) -> bool:
-        namespace = env.paths.project_document_path(_USER, row["stored_relpath"])
-        return not namespace.exists() or list(namespace.rglob("*.tmp")) == []
+        staging = env.paths.project_documents_dir(_USER, row["project_id"]) / ".staging"
+        return not staging.exists() or list(staging.iterdir()) == []
 
-    async def test_conversion_first_purge_blocks_then_removes_everything(self, env, monkeypatch):
+    @staticmethod
+    def _gated_convert(started: threading.Event, finish: threading.Event):
+        async def fake_convert(file_path, output_path=None):
+            # Stall after the output exists, so every cleanup path has a file to remove.
+            output_path.write_text("# converted markdown", encoding="utf-8")
+            started.set()
+            assert finish.wait(10)
+            return output_path
+
+        return fake_convert
+
+    async def test_conversion_does_not_block_unrelated_database_writes(self, env, monkeypatch):
+        from deerflow.persistence.thread_meta.sql import ThreadMetaRepository
+        from deerflow.projects import documents as documents_mod
+
+        started = threading.Event()
+        finish = threading.Event()
+        monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._gated_convert(started, finish))
+        monkeypatch.setattr("deerflow.projects.tools._resolve_auto_convert", lambda: True)
+        row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes")
+
+        # Driven through the agent tool, the path a first read actually takes.
+        read = asyncio.create_task(_read_project_document_impl(_runtime(project_id=env.project["id"]), document_id=row["id"], offset=0, limit=100))
+        try:
+            assert await asyncio.to_thread(started.wait, 10)
+            # Another user's thread-metadata write lands while the conversion is still running.
+            await asyncio.wait_for(ThreadMetaRepository(env.sf).create("t-unrelated", user_id="u2"), 5)
+        finally:
+            finish.set()
+        assert json.loads(await read)["content"] == "# converted markdown"
+        assert converted_markdown_path(env.paths, user_id=_USER, row=row).is_file()
+        assert self._no_temp_left(env, row)
+
+    async def test_restored_document_converts_from_its_new_project_into_its_original_namespace(self, env, monkeypatch):
+        from deerflow.projects import documents as documents_mod
+
+        async def fake_convert(file_path, output_path=None):
+            output_path.write_text("# converted markdown", encoding="utf-8")
+            return output_path
+
+        monkeypatch.setattr(documents_mod, "convert_file_to_markdown", fake_convert)
+        monkeypatch.setattr("deerflow.projects.tools._resolve_auto_convert", lambda: True)
+        row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes")
+        target = await env.projects.create(name="T", user_id=_USER)
+        assert await env.docs.trash(row["id"], user_id=_USER) is True
+        outcome, restored = await env.docs.restore(row["id"], target_project_id=target["id"], user_id=_USER)
+        assert outcome == "restored"
+        # Restore re-points the row without moving bytes: the namespace stays
+        # under the source project while staging happens under the target.
+        assert restored["project_id"] == target["id"]
+        assert restored["stored_relpath"].startswith(f"{env.project['id']}/")
+
+        result = json.loads(await _read_project_document_impl(_runtime(project_id=target["id"]), document_id=row["id"], offset=0, limit=100))
+        assert result["content"] == "# converted markdown"
+        assert converted_markdown_path(env.paths, user_id=_USER, row=restored).is_file()
+        assert self._no_temp_left(env, restored)
+
+    async def test_purge_during_conversion_proceeds_and_conversion_publishes_nothing(self, env, monkeypatch):
+        from deerflow.projects import documents as documents_mod
+
+        started = threading.Event()
+        finish = threading.Event()
+        monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._gated_convert(started, finish))
+        row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes")
+
+        conversion = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True))
+        try:
+            assert await asyncio.to_thread(started.wait, 10)
+            assert await asyncio.wait_for(env.docs.trash(row["id"], user_id=_USER), 5) is True
+            assert await asyncio.wait_for(env.docs.purge(row["id"], remove_files=make_purge_file_remover(env.paths, user_id=_USER), user_id=_USER), 5) is True
+        finally:
+            finish.set()
+        served, reason = await conversion
+        # The locked revalidation finds the row gone, so the staged output is
+        # discarded instead of published after the purge.
+        assert (served, reason) == (None, "content_missing")
+        assert await env.docs.get(row["id"], include_trashed=True, user_id=_USER) is None
+        # The output never lived in the namespace, so purge removed all of it.
+        assert not env.paths.project_document_path(_USER, row["stored_relpath"]).exists()
+        assert self._no_temp_left(env, row)
+
+    async def test_trash_during_conversion_proceeds_and_conversion_publishes_nothing(self, env, monkeypatch):
+        from deerflow.projects import documents as documents_mod
+
+        started = threading.Event()
+        finish = threading.Event()
+        monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._gated_convert(started, finish))
+        row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes")
+        derived = converted_markdown_path(env.paths, user_id=_USER, row=row)
+
+        conversion = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True))
+        try:
+            assert await asyncio.to_thread(started.wait, 10)
+            assert await asyncio.wait_for(env.docs.trash(row["id"], user_id=_USER), 5) is True
+        finally:
+            finish.set()
+        served, reason = await conversion
+        assert (served, reason) == (None, "content_missing")
+        assert not derived.exists()
+        assert self._no_temp_left(env, row)
+
+    async def test_failed_conversion_of_a_row_purged_meanwhile_reports_content_missing(self, env, monkeypatch):
         from deerflow.projects import documents as documents_mod
 
         started = threading.Event()
         finish = threading.Event()
 
-        async def fake_convert(file_path, output_path=None):
+        async def failing_convert(file_path, output_path=None):
             started.set()
             assert finish.wait(10)
-            output_path.write_text("# converted markdown", encoding="utf-8")
-            return output_path
+            return None
 
-        monkeypatch.setattr(documents_mod, "convert_file_to_markdown", fake_convert)
+        monkeypatch.setattr(documents_mod, "convert_file_to_markdown", failing_convert)
+        row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes")
+
+        conversion = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True))
+        try:
+            assert await asyncio.to_thread(started.wait, 10)
+            assert await asyncio.wait_for(env.docs.trash(row["id"], user_id=_USER), 5) is True
+            assert await asyncio.wait_for(env.docs.purge(row["id"], remove_files=make_purge_file_remover(env.paths, user_id=_USER), user_id=_USER), 5) is True
+        finally:
+            finish.set()
+        # The failure is revalidated under the lock: the row is gone, so the
+        # reason is content_missing rather than binary.
+        assert await conversion == (None, "content_missing")
+        assert self._no_temp_left(env, row)
+
+    async def test_cancelled_conversion_drains_and_leaves_no_staged_output(self, env, monkeypatch):
+        from deerflow.projects import documents as documents_mod
+
+        started = threading.Event()
+        finish = threading.Event()
+        monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._gated_convert(started, finish))
         row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes")
-        original = original_file_path(env.paths, user_id=_USER, row=row)
         derived = converted_markdown_path(env.paths, user_id=_USER, row=row)
 
         conversion = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True))
         assert await asyncio.to_thread(started.wait, 10)
-
-        async def trash_and_purge() -> bool:
-            assert await env.docs.trash(row["id"], user_id=_USER) is True
-            return await env.docs.purge(row["id"], remove_files=make_purge_file_remover(env.paths, user_id=_USER), user_id=_USER)
-
-        purge_task = asyncio.create_task(trash_and_purge())
+        conversion.cancel()
         await asyncio.sleep(0.2)
-        # Trash+purge is blocked on the row lock the conversion holds.
-        assert not purge_task.done()
+        # The worker cannot be interrupted: the cancellation waits for it, so
+        # the staged output is removed after the worker's last write.
+        assert not conversion.done()
         finish.set()
-        (served, reason), purged = await asyncio.gather(conversion, purge_task)
-        assert reason is None and served == derived
-        assert purged is True
-        # Purge then removed original, derived, and the row; no partial
-        # conversion temp file remains.
-        assert not original.exists()
+        with pytest.raises(asyncio.CancelledError):
+            await conversion
         assert not derived.exists()
-        assert await env.docs.get(row["id"], include_trashed=True, user_id=_USER) is None
         assert self._no_temp_left(env, row)
 
     async def test_purge_committed_first_conversion_publishes_nothing(self, env, monkeypatch):
@@ -601,3 +714,102 @@ class TestConversionSerialization:
         assert called is False
         assert not derived.exists()
         assert self._no_temp_left(env, row)
+
+
+class TestConversionSingleFlight:
+    """Concurrent first reads of one document share a single conversion.
+
+    Conversion occupies a bounded file-IO pool thread for its whole duration,
+    so duplicates would starve unrelated offloaded file work. Followers wait
+    on the leader without holding a pool thread — from any event loop, since
+    the tool also runs on the isolated subagent loop — and a cancelled leader
+    hands the work to a follower instead of failing it."""
+
+    @staticmethod
+    def _counting_gated_convert(started: threading.Event, finish: threading.Event, calls: list[str]):
+        async def fake_convert(file_path, output_path=None):
+            calls.append(str(file_path))
+            started.set()
+            assert finish.wait(10)
+            output_path.write_text("# converted markdown", encoding="utf-8")
+            return output_path
+
+        return fake_convert
+
+    async def test_concurrent_first_reads_share_one_conversion_across_event_loops(self, env, monkeypatch):
+        from deerflow.projects import documents as documents_mod
+
+        started = threading.Event()
+        finish = threading.Event()
+        calls: list[str] = []
+        monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._counting_gated_convert(started, finish, calls))
+        row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes")
+        derived = converted_markdown_path(env.paths, user_id=_USER, row=row)
+
+        def read() -> asyncio.Task:
+            return asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True))
+
+        leader = read()
+        try:
+            assert await asyncio.to_thread(started.wait, 10)
+            followers = [read(), read()]
+            # A follower on another event loop, like the isolated subagent loop.
+            other_loop = asyncio.create_task(asyncio.to_thread(asyncio.run, ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True)))
+            await asyncio.sleep(0.2)
+            assert len(calls) == 1
+        finally:
+            finish.set()
+        results = await asyncio.gather(leader, *followers, other_loop)
+        assert results == [(derived, None)] * 4
+        assert len(calls) == 1
+        assert documents_mod._inflight_conversions == {}
+
+    async def test_cancelled_follower_leaves_the_shared_conversion_running(self, env, monkeypatch):
+        from deerflow.projects import documents as documents_mod
+
+        started = threading.Event()
+        finish = threading.Event()
+        calls: list[str] = []
+        monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._counting_gated_convert(started, finish, calls))
+        row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes")
+        derived = converted_markdown_path(env.paths, user_id=_USER, row=row)
+
+        leader = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True))
+        try:
+            assert await asyncio.to_thread(started.wait, 10)
+            follower = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True))
+            await asyncio.sleep(0.2)
+            follower.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await follower
+        finally:
+            finish.set()
+        assert await leader == (derived, None)
+        assert len(calls) == 1
+        assert documents_mod._inflight_conversions == {}
+
+    async def test_follower_takes_over_when_the_leader_is_cancelled(self, env, monkeypatch):
+        from deerflow.projects import documents as documents_mod
+
+        started = threading.Event()
+        finish = threading.Event()
+        calls: list[str] = []
+        monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._counting_gated_convert(started, finish, calls))
+        row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes")
+        derived = converted_markdown_path(env.paths, user_id=_USER, row=row)
+
+        leader = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True))
+        try:
+            assert await asyncio.to_thread(started.wait, 10)
+            follower = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True))
+            await asyncio.sleep(0.2)
+            leader.cancel()
+        finally:
+            finish.set()
+        with pytest.raises(asyncio.CancelledError):
+            await leader
+        # The cancelled leader published nothing; the follower converted again
+        # instead of inheriting the cancellation.
+        assert await follower == (derived, None)
+        assert len(calls) == 2
+        assert documents_mod._inflight_conversions == {}

backend/tests/test_project_documents_repo.py

@@ -611,18 +611,18 @@ class TestPurge:
                 assert await docs.get(row["id"], user_id="u1") is not None
 
 
-class TestConvertUnderLiveLock:
-    async def test_runs_convert_under_the_lock_and_returns_the_row(self, repos):
+class TestPublishUnderLiveLock:
+    async def test_runs_publish_under_the_lock_and_returns_the_row(self, repos):
         projects, docs = repos
         project = await _project(projects)
         row = await _insert(docs, project["id"], sha="cf" * 32, doc_id="cv-ok")
         seen: list[str] = []
 
-        async def convert(r: dict) -> str:
+        async def publish(r: dict) -> str:
             seen.append(r["id"])
             return "published"
 
-        result = await docs.convert_under_live_lock(row["id"], project_id=project["id"], convert=convert, user_id="u1")
+        result = await docs.publish_under_live_lock(row["id"], project_id=project["id"], publish=publish, user_id="u1")
         assert result is not None
         locked_row, value = result
         assert locked_row["id"] == row["id"]
@@ -637,21 +637,21 @@ class TestConvertUnderLiveLock:
         trashed_project, trashed = await _trashed(docs, projects, sha="d1" * 32, doc_id="cv-trashed")
         calls: list[str] = []
 
-        async def convert(r: dict) -> None:
+        async def publish(r: dict) -> None:
             calls.append(r["id"])
 
-        assert await docs.convert_under_live_lock("missing", project_id=project["id"], convert=convert, user_id="u1") is None
+        assert await docs.publish_under_live_lock("missing", project_id=project["id"], publish=publish, user_id="u1") is None
         # Foreign owner, sibling project, and trashed rows all decline without
-        # ever invoking the conversion callback.
-        assert await docs.convert_under_live_lock(row["id"], project_id=project["id"], convert=convert, user_id="u2") is None
-        assert await docs.convert_under_live_lock(row["id"], project_id=other["id"], convert=convert, user_id="u1") is None
-        assert await docs.convert_under_live_lock(trashed["id"], project_id=trashed_project["id"], convert=convert, user_id="u1") is None
+        # ever invoking the publish callback.
+        assert await docs.publish_under_live_lock(row["id"], project_id=project["id"], publish=publish, user_id="u2") is None
+        assert await docs.publish_under_live_lock(row["id"], project_id=other["id"], publish=publish, user_id="u1") is None
+        assert await docs.publish_under_live_lock(trashed["id"], project_id=trashed_project["id"], publish=publish, user_id="u1") is None
         assert calls == []
 
-    async def test_concurrent_conversion_vs_purge_purge_commits_first(self, repos):
+    async def test_concurrent_publish_vs_purge_purge_commits_first(self, repos):
         """§13: purge holds its document lock across unlink and commit; a
-        conversion that blocked on the row lock then finds the row gone and
-        converts/publishes nothing."""
+        publish that blocked on the row lock then finds the row gone and
+        publishes nothing."""
         projects, docs = repos
         project = await _project(projects)
         row = await _insert(docs, project["id"], sha="e0" * 32, doc_id="cv-purge-first")
@@ -664,26 +664,26 @@ class TestConvertUnderLiveLock:
             entered.set()
             await asyncio.wait_for(release.wait(), 5)
 
-        async def convert(r: dict) -> str:
+        async def publish(r: dict) -> str:
             calls.append(r["id"])
             return "published"
 
-        async def run_conversion() -> tuple[dict, str] | None:
+        async def run_publish() -> tuple[dict, str] | None:
             await entered.wait()
-            task = asyncio.create_task(docs.convert_under_live_lock(row["id"], project_id=project["id"], convert=convert, user_id="u1"))
+            task = asyncio.create_task(docs.publish_under_live_lock(row["id"], project_id=project["id"], publish=publish, user_id="u1"))
             release.set()
             return await task
 
         purged, result = await asyncio.gather(
             docs.purge(row["id"], remove_files=remove_files, user_id="u1"),
-            run_conversion(),
+            run_publish(),
         )
         assert purged is True
         assert result is None
         assert calls == []
 
-    async def test_concurrent_conversion_vs_trash_conversion_commits_first(self, repos):
-        """§13: conversion holds the document lock through the publish; a
+    async def test_concurrent_publish_vs_trash_publish_commits_first(self, repos):
+        """§13: the publish holds the document lock until it commits; a
         trash arriving meanwhile blocks and then proceeds."""
         projects, docs = repos
         project = await _project(projects)
@@ -691,7 +691,7 @@ class TestConvertUnderLiveLock:
         entered = asyncio.Event()
         release = asyncio.Event()
 
-        async def convert(r: dict) -> str:
+        async def publish(r: dict) -> str:
             entered.set()
             await asyncio.wait_for(release.wait(), 5)
             return "published"
@@ -703,7 +703,7 @@ class TestConvertUnderLiveLock:
             return await task
 
         result, trashed = await asyncio.gather(
-            docs.convert_under_live_lock(row["id"], project_id=project["id"], convert=convert, user_id="u1"),
+            docs.publish_under_live_lock(row["id"], project_id=project["id"], publish=publish, user_id="u1"),
             run_trash(),
         )
         assert result is not None and result[1] == "published"
@@ -765,38 +765,38 @@ class TestLockedOffloadDrain:
         still = await docs.get(row["id"], include_trashed=True, user_id="u1")
         assert still is not None and still["trashed_at"] is not None
 
-    async def test_cancelled_conversion_drains_the_worker_before_unlocking(self, repos):
+    async def test_cancelled_publish_drains_the_worker_before_unlocking(self, repos):
         import threading
 
         projects, docs = repos
         project = await _project(projects)
-        row = await _insert(docs, project["id"], sha="e7" * 32, doc_id="convert-cancel")
+        row = await _insert(docs, project["id"], sha="e7" * 32, doc_id="publish-cancel")
 
         entered = threading.Event()
         finish = threading.Event()
         state = {"published": False}
 
-        def slow_convert() -> None:
+        def slow_publish() -> None:
             entered.set()
             assert finish.wait(10)
             state["published"] = True
 
-        async def convert(r: dict) -> bool:
-            await asyncio.to_thread(slow_convert)
+        async def publish(r: dict) -> bool:
+            await asyncio.to_thread(slow_publish)
             return True
 
-        convert_task = asyncio.create_task(docs.convert_under_live_lock(row["id"], project_id=project["id"], convert=convert, user_id="u1"))
+        publish_task = asyncio.create_task(docs.publish_under_live_lock(row["id"], project_id=project["id"], publish=publish, user_id="u1"))
         await asyncio.to_thread(entered.wait, 10)
-        convert_task.cancel()
-        # The document lock stays held while the conversion worker drains: a
+        publish_task.cancel()
+        # The document lock stays held while the publish worker drains: a
         # trash blocks on it, so the publish never lands after the lock is
         # gone (the round-2 serialization survives cancellation).
         trash_task = asyncio.create_task(docs.trash(row["id"], user_id="u1"))
         await asyncio.sleep(0.3)
-        assert not convert_task.done()
+        assert not publish_task.done()
         assert not trash_task.done()
         finish.set()
         with pytest.raises(asyncio.CancelledError):
-            await convert_task
+            await publish_task
         assert state["published"]
         assert await trash_task is True