fix(task-deeplink): 共享任务对话的 md/产物按对话创建者解析路径,修复切换账号后找不到文件

xdfx/rwfx/3qfx 任务深链对话按 metadata.taskId 全局共享、不区分用户,但生成的
md/artifacts 与上传文件物理存在 users/{user_id}/threads/{thread_id}/ 下,user_id
之前取当前登录账号。导致 A 账号生成的文件,B 账号打开同一共享对话时后端去
users/B/ 找 → 404「找不到」。

新增 deerflow/runtime/thread_paths.py(resolve_path_user_id /
aresolve_path_user_id):带 taskId 的共享对话统一解析到对话创建者
(threads_meta.user_id),个人对话仍走当前登录(行为不变)。在所有线程文件路径关口替换:
写入源 ThreadDataMiddleware、present_files、uploads(manager+middleware+router)、
产物读写(path_utils → artifacts/skills router)、嵌入式 client.get_artifact。
正向结果进程内缓存;无 DB/内存态/无行均安全回退到当前登录。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

offline-backend-20260512/backend/CLAUDE.md

@@ -459,6 +459,7 @@ Operators upgrading an existing DB run the whole file (`USE <db>;` then execute)
 - Physical: `backend/.deer-flow/users/{user_id}/threads/{thread_id}/user-data/...`, `deer-flow/skills/`
 - Translation: `replace_virtual_path()` / `replace_virtual_paths_in_command()`
 - Detection: `is_local_sandbox()` checks `sandbox_id == "local"`
+- **`{user_id}` for path resolution = `resolve_path_user_id(thread_id)`, NOT the current login.** Task-deeplink conversations (rwfx/3qfx/xdfx, `threads_meta.metadata.taskId`) are shared across all users (see the `taskId` escape branches in `thread_meta/sql.py`), but generated files/uploads live under a *per-user* dir. Resolving by current login meant account B opening a shared task conversation 404'd on the md/artifacts account A generated ("切换账号后生成的文件找不到"). `deerflow/runtime/thread_paths.py` (`resolve_path_user_id` sync / `aresolve_path_user_id` async) resolves a task-shared thread to its **creator** (`threads_meta.user_id`), so every account reads/writes the same physical directory; personal threads resolve to the current login unchanged (no behavior change). Wired at every thread-path seam: write source `ThreadDataMiddleware.before_agent` (drives all sandbox writes via `thread_data` paths), `present_file_tool`, `uploads/manager.get_uploads_dir` + `uploads_middleware` + uploads router, artifact serving (`path_utils.{a,}resolve_thread_virtual_path` → artifacts/skills routers), embedded `client.get_artifact`. Positive (taskId→creator) results are cached process-wide (immutable mapping); any failure / memory-mode store / missing row falls back to the current login. **Caveat**: files a *non-creator* wrote into a shared thread before this fix stay orphaned under that writer's dir (no data loss); all new writes consolidate under the creator. Tests: `tests/test_thread_paths_owner.py`.
 
 **Sandbox Tools** (in `packages/harness/deerflow/sandbox/tools.py`):
 - `bash` - Execute commands with path translation and error handling

offline-backend-20260512/backend/app/gateway/path_utils.py

@@ -5,16 +5,21 @@ from pathlib import Path
 from fastapi import HTTPException
 
 from deerflow.config.paths import get_paths
-from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.runtime.thread_paths import aresolve_path_user_id, resolve_path_user_id
 
 
-def resolve_thread_virtual_path(thread_id: str, virtual_path: str) -> Path:
+def resolve_thread_virtual_path(thread_id: str, virtual_path: str, *, user_id: str | None = None) -> Path:
     """Resolve a virtual path to the actual filesystem path under thread user-data.
 
     Args:
         thread_id: The thread ID.
         virtual_path: The virtual path as seen inside the sandbox
                       (e.g., /mnt/user-data/outputs/file.txt).
+        user_id: Pre-resolved canonical user bucket. When ``None`` it is
+                 resolved synchronously via :func:`resolve_path_user_id` (the
+                 thread creator for task-shared threads, else the current
+                 login). Async callers should pass the result of
+                 :func:`aresolve_path_user_id` to avoid a blocking DB read.
 
     Returns:
         The resolved filesystem path.
@@ -22,8 +27,21 @@ def resolve_thread_virtual_path(thread_id: str, virtual_path: str) -> Path:
     Raises:
         HTTPException: If the path is invalid or outside allowed directories.
     """
+    if user_id is None:
+        user_id = resolve_path_user_id(thread_id)
     try:
-        return get_paths().resolve_virtual_path(thread_id, virtual_path, user_id=get_effective_user_id())
+        return get_paths().resolve_virtual_path(thread_id, virtual_path, user_id=user_id)
     except ValueError as e:
         status = 403 if "traversal" in str(e) else 400
         raise HTTPException(status_code=status, detail=str(e))
+
+
+async def aresolve_thread_virtual_path(thread_id: str, virtual_path: str) -> Path:
+    """Async variant of :func:`resolve_thread_virtual_path` for FastAPI handlers.
+
+    Resolves the canonical user bucket without blocking the event loop, so a
+    task-shared thread's artifacts resolve to the thread creator's directory
+    regardless of which account is currently logged in.
+    """
+    user_id = await aresolve_path_user_id(thread_id)
+    return resolve_thread_virtual_path(thread_id, virtual_path, user_id=user_id)

offline-backend-20260512/backend/app/gateway/routers/artifacts.py

@@ -10,9 +10,9 @@ from fastapi.responses import FileResponse, PlainTextResponse, Response
 from pydantic import BaseModel
 
 from app.gateway.authz import require_permission
-from app.gateway.path_utils import resolve_thread_virtual_path
+from app.gateway.path_utils import aresolve_thread_virtual_path
 from deerflow.config.paths import get_paths
-from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.runtime.thread_paths import aresolve_path_user_id
 
 logger = logging.getLogger(__name__)
 
@@ -155,7 +155,7 @@ async def list_artifacts(thread_id: str, request: Request) -> dict:
     返回 ``{ "thread_id", "files": [{ path, name, size_bytes, mime_type, modified_at }] }``，
     目录不存在时 ``files`` 为空数组（非 404）。
     """
-    outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=get_effective_user_id())
+    outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=await aresolve_path_user_id(thread_id))
     return {"thread_id": thread_id, "files": collect_outputs_listing(outputs_dir)}
 
 
@@ -206,7 +206,7 @@ async def get_artifact(thread_id: str, path: str, request: Request, download: bo
         skill_file_path = path[: marker_pos + len(".skill")]  # e.g., "mnt/user-data/outputs/my-skill.skill"
         internal_path = path[marker_pos + len(skill_marker) :]  # e.g., "SKILL.md"
 
-        actual_skill_path = resolve_thread_virtual_path(thread_id, skill_file_path)
+        actual_skill_path = await aresolve_thread_virtual_path(thread_id, skill_file_path)
 
         if not actual_skill_path.exists():
             raise HTTPException(status_code=404, detail=f"Skill file not found: {skill_file_path}")
@@ -236,7 +236,7 @@ async def get_artifact(thread_id: str, path: str, request: Request, download: bo
         except UnicodeDecodeError:
             return Response(content=content, media_type=mime_type or "application/octet-stream", headers=cache_headers)
 
-    actual_path = resolve_thread_virtual_path(thread_id, path)
+    actual_path = await aresolve_thread_virtual_path(thread_id, path)
 
     logger.info(f"Resolving artifact path: thread_id={thread_id}, requested_path={path}, actual_path={actual_path}")
 
@@ -285,7 +285,7 @@ async def update_artifact(
     if ".skill/" in path:
         raise HTTPException(status_code=400, detail="Skill archive contents are read-only.")
 
-    actual_path = resolve_thread_virtual_path(thread_id, path)
+    actual_path = await aresolve_thread_virtual_path(thread_id, path)
     logger.info(f"Updating artifact path: thread_id={thread_id}, requested_path={path}, actual_path={actual_path}")
 
     if not actual_path.exists():

offline-backend-20260512/backend/app/gateway/routers/skills.py

@@ -17,7 +17,7 @@ from pydantic import BaseModel, Field
 from sqlalchemy import select
 
 from app.gateway.deps import get_agent_store, get_config, get_skill_store, get_tag_store
-from app.gateway.path_utils import resolve_thread_virtual_path
+from app.gateway.path_utils import aresolve_thread_virtual_path
 from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async
 from deerflow.config.app_config import AppConfig
 from deerflow.config.extensions_config import ExtensionsConfig, SkillDisplayConfig, SkillStateConfig, get_extensions_config, reload_extensions_config
@@ -1058,7 +1058,7 @@ async def install_skill(
         target_category = body.target_category
         if target_category == SkillCategory.PUBLIC and not is_admin:
             raise HTTPException(status_code=403, detail="Only admin users can install skills to the public directory")
-        skill_file_path = resolve_thread_virtual_path(body.thread_id, body.path)
+        skill_file_path = await aresolve_thread_virtual_path(body.thread_id, body.path)
         result = await get_or_new_skill_storage(app_config=config).ainstall_skill_from_archive(skill_file_path, target_category=target_category)
         try:
             from deerflow.skills.usage import mark_uploaded

offline-backend-20260512/backend/app/gateway/routers/uploads.py

@@ -11,7 +11,7 @@ from app.gateway.authz import require_permission
 from app.gateway.deps import get_config
 from deerflow.config.app_config import AppConfig
 from deerflow.config.paths import get_paths
-from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.runtime.thread_paths import aresolve_path_user_id
 from deerflow.sandbox.sandbox_provider import SandboxProvider, get_sandbox_provider
 from deerflow.uploads.manager import (
     PathTraversalError,
@@ -186,7 +186,7 @@ async def upload_files(
         uploads_dir = ensure_uploads_dir(thread_id)
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
-    sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=get_effective_user_id())
+    sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=await aresolve_path_user_id(thread_id))
     uploaded_files = []
     written_paths = []
     sandbox_sync_targets = []
@@ -308,7 +308,7 @@ async def list_uploaded_files(thread_id: str, request: Request) -> dict:
     enrich_file_listing(result, thread_id)
 
     # Gateway additionally includes the sandbox-relative path.
-    sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=get_effective_user_id())
+    sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=await aresolve_path_user_id(thread_id))
     for f in result["files"]:
         f["path"] = str(sandbox_uploads / f["filename"])
 

offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/thread_data_middleware.py

@@ -10,7 +10,7 @@ from langgraph.runtime import Runtime
 
 from deerflow.agents.thread_state import ThreadDataState
 from deerflow.config.paths import Paths, get_paths
-from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.runtime.thread_paths import resolve_path_user_id
 
 logger = logging.getLogger(__name__)
 
@@ -89,7 +89,11 @@ class ThreadDataMiddleware(AgentMiddleware[ThreadDataMiddlewareState]):
         if thread_id is None:
             raise ValueError("Thread ID is required in runtime context or config.configurable")
 
-        user_id = get_effective_user_id()
+        # Task-shared threads (rwfx/3qfx/xdfx, metadata.taskId) resolve to the
+        # thread creator so every account that shares the conversation writes to
+        # the same physical user-data directory. Personal threads resolve to the
+        # current login unchanged.
+        user_id = resolve_path_user_id(thread_id)
 
         if self._lazy_init:
             # Lazy initialization: only compute paths, don't create directories

offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py

@@ -1,10 +1,10 @@
 """Middleware to inject uploaded files information into agent context."""
 
-import logging
-import shutil
-from pathlib import Path
-from typing import NotRequired, override
-from urllib.parse import unquote
+import logging
+import shutil
+from pathlib import Path
+from typing import NotRequired, override
+from urllib.parse import unquote
 
 from langchain.agents import AgentState
 from langchain.agents.middleware import AgentMiddleware
@@ -12,10 +12,10 @@ from langchain_core.messages import HumanMessage
 from langgraph.runtime import Runtime
 
 from deerflow.config.paths import Paths, get_paths
-from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.runtime.thread_paths import resolve_path_user_id
 from deerflow.utils.file_conversion import extract_outline
 
-logger = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 
 _OUTLINE_PREVIEW_LINES = 5
@@ -24,8 +24,8 @@ _OUTLINE_PREVIEW_LINES = 5
 # configured vision model), never by reading their bytes. Listing them with the
 # generic read_file/grep guidance made the agent try to parse image binaries
 # with bash/python, which fails and wastes turns.
-_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp"}
-_ARTIFACT_URL_MARKER = "/artifacts/"
+_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp"}
+_ARTIFACT_URL_MARKER = "/artifacts/"
 
 
 def _extract_outline_for_file(file_path: Path) -> tuple[list[dict], list[str]]:
@@ -72,7 +72,7 @@ class UploadsMiddlewareState(AgentState):
     uploaded_files: NotRequired[list[dict] | None]
 
 
-class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
+class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
     """Middleware to inject uploaded files information into the agent context.
 
     Reads file metadata from the current message's additional_kwargs.files
@@ -125,7 +125,7 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
             lines.append("  Use `grep` to search for keywords (e.g. `grep(pattern='keyword', path='/mnt/user-data/uploads/')`).")
         lines.append("")
 
-    def _create_files_message(self, new_files: list[dict], historical_files: list[dict]) -> str:
+    def _create_files_message(self, new_files: list[dict], historical_files: list[dict]) -> str:
         """Create a formatted message listing uploaded files.
 
         Args:
@@ -163,61 +163,61 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
         lines.append("- Only fall back to web search if the file content is clearly insufficient to answer the question.")
         lines.append("</uploaded_files>")
 
-        return "\n".join(lines)
-
-    def _import_file_from_artifact_url(
-        self,
-        file_info: dict,
-        uploads_dir: Path | None,
-    ) -> bool:
-        """Copy a previously uploaded file into the current thread when possible.
-
-        iframe-style callers may pass file metadata that points at an upload
-        from another DeerFlow thread. The current run thread still needs a real
-        file under its own ``/mnt/user-data/uploads`` directory, otherwise this
-        middleware will filter the entry out during existence checks.
-        """
-        filename = file_info.get("filename") or ""
-        artifact_url = str(file_info.get("artifact_url") or "").strip()
-        if not filename or uploads_dir is None or not artifact_url:
-            return False
-
-        if not artifact_url.startswith("/api/threads/") or _ARTIFACT_URL_MARKER not in artifact_url:
-            return False
-
-        prefix, encoded_virtual_path = artifact_url.split(_ARTIFACT_URL_MARKER, 1)
-        source_thread_id = prefix.removeprefix("/api/threads/").strip("/")
-        if not source_thread_id or not encoded_virtual_path:
-            return False
-
-        virtual_path = "/" + unquote(encoded_virtual_path.lstrip("/"))
-        if Path(virtual_path).name != filename:
-            return False
-
-        try:
-            source_path = self._paths.resolve_virtual_path(
-                source_thread_id,
-                virtual_path,
-                user_id=get_effective_user_id(),
-            )
-        except ValueError:
-            logger.debug("Failed to resolve uploaded artifact path: %s", artifact_url, exc_info=True)
-            return False
-
-        if not source_path.is_file():
-            return False
-
-        uploads_dir.mkdir(parents=True, exist_ok=True)
-        target_path = uploads_dir / filename
-        shutil.copy2(source_path, target_path)
-
-        source_markdown = source_path.with_suffix(".md")
-        if source_markdown.is_file() and source_markdown != source_path:
-            shutil.copy2(source_markdown, target_path.with_suffix(".md"))
-
-        return True
-
-    def _files_from_kwargs(self, message: HumanMessage, uploads_dir: Path | None = None) -> list[dict] | None:
+        return "\n".join(lines)
+
+    def _import_file_from_artifact_url(
+        self,
+        file_info: dict,
+        uploads_dir: Path | None,
+    ) -> bool:
+        """Copy a previously uploaded file into the current thread when possible.
+
+        iframe-style callers may pass file metadata that points at an upload
+        from another DeerFlow thread. The current run thread still needs a real
+        file under its own ``/mnt/user-data/uploads`` directory, otherwise this
+        middleware will filter the entry out during existence checks.
+        """
+        filename = file_info.get("filename") or ""
+        artifact_url = str(file_info.get("artifact_url") or "").strip()
+        if not filename or uploads_dir is None or not artifact_url:
+            return False
+
+        if not artifact_url.startswith("/api/threads/") or _ARTIFACT_URL_MARKER not in artifact_url:
+            return False
+
+        prefix, encoded_virtual_path = artifact_url.split(_ARTIFACT_URL_MARKER, 1)
+        source_thread_id = prefix.removeprefix("/api/threads/").strip("/")
+        if not source_thread_id or not encoded_virtual_path:
+            return False
+
+        virtual_path = "/" + unquote(encoded_virtual_path.lstrip("/"))
+        if Path(virtual_path).name != filename:
+            return False
+
+        try:
+            source_path = self._paths.resolve_virtual_path(
+                source_thread_id,
+                virtual_path,
+                user_id=resolve_path_user_id(source_thread_id),
+            )
+        except ValueError:
+            logger.debug("Failed to resolve uploaded artifact path: %s", artifact_url, exc_info=True)
+            return False
+
+        if not source_path.is_file():
+            return False
+
+        uploads_dir.mkdir(parents=True, exist_ok=True)
+        target_path = uploads_dir / filename
+        shutil.copy2(source_path, target_path)
+
+        source_markdown = source_path.with_suffix(".md")
+        if source_markdown.is_file() and source_markdown != source_path:
+            shutil.copy2(source_markdown, target_path.with_suffix(".md"))
+
+        return True
+
+    def _files_from_kwargs(self, message: HumanMessage, uploads_dir: Path | None = None) -> list[dict] | None:
         """Extract file info from message additional_kwargs.files.
 
         The frontend sends uploaded file metadata in additional_kwargs.files
@@ -238,17 +238,17 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
 
         files = []
         for f in kwargs_files:
-            if not isinstance(f, dict):
-                continue
-            filename = f.get("filename") or ""
-            if not filename or Path(filename).name != filename:
-                continue
-            if uploads_dir is not None and not (uploads_dir / filename).is_file():
-                if not self._import_file_from_artifact_url(f, uploads_dir):
-                    continue
-            files.append(
-                {
-                    "filename": filename,
+            if not isinstance(f, dict):
+                continue
+            filename = f.get("filename") or ""
+            if not filename or Path(filename).name != filename:
+                continue
+            if uploads_dir is not None and not (uploads_dir / filename).is_file():
+                if not self._import_file_from_artifact_url(f, uploads_dir):
+                    continue
+            files.append(
+                {
+                    "filename": filename,
                     "size": int(f.get("size") or 0),
                     "path": f"/mnt/user-data/uploads/{filename}",
                     "extension": Path(filename).suffix,
@@ -294,7 +294,7 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
                 thread_id = get_config().get("configurable", {}).get("thread_id")
             except RuntimeError:
                 pass  # get_config() raises outside a runnable context (e.g. unit tests)
-        uploads_dir = self._paths.sandbox_uploads_dir(thread_id, user_id=get_effective_user_id()) if thread_id else None
+        uploads_dir = self._paths.sandbox_uploads_dir(thread_id, user_id=resolve_path_user_id(thread_id)) if thread_id else None
 
         # Get newly uploaded files from the current message's additional_kwargs.files
         new_files = self._files_from_kwargs(last_message, uploads_dir) or []

offline-backend-20260512/backend/packages/harness/deerflow/client.py

@@ -1178,7 +1178,9 @@ class DeerFlowClient:
             ValueError: If the path is invalid.
         """
         try:
-            actual = get_paths().resolve_virtual_path(thread_id, path, user_id=get_effective_user_id())
+            from deerflow.runtime.thread_paths import resolve_path_user_id
+
+            actual = get_paths().resolve_virtual_path(thread_id, path, user_id=resolve_path_user_id(thread_id))
         except ValueError as exc:
             if "traversal" in str(exc):
                 from deerflow.uploads.manager import PathTraversalError

offline-backend-20260512/backend/packages/harness/deerflow/runtime/thread_paths.py

@@ -0,0 +1,145 @@
+"""Canonical owner resolution for a thread's filesystem paths.
+
+Task-deeplink conversations (those whose ``threads_meta.metadata.taskId`` is
+set — rwfx / 3qfx / xdfx) are **shared across all users**: any logged-in
+account may open the same conversation (enforced by the ``taskId`` escape
+branches in :mod:`deerflow.persistence.thread_meta.sql`). But the agent's
+generated files (md / artifacts) and uploads live under a *per-user* directory::
+
+    {base_dir}/users/{user_id}/threads/{thread_id}/user-data/...
+
+If that ``{user_id}`` were the **current login**, account B opening a shared
+task conversation would resolve to an empty ``users/B/...`` directory and 404
+on every file account A generated — which is exactly the "切换账号后生成的 md
+文件找不到" bug.
+
+To keep the *file* layer consistent with the *conversation-sharing* layer, all
+path resolution for a task-shared thread must use one stable user bucket: the
+thread's **creator** (``threads_meta.user_id``). That value is fixed at thread
+creation and never changes, so every account resolves to the same physical
+directory.
+
+Personal (non-task) threads are unaffected: their creator is always the only /
+current user, so :func:`resolve_path_user_id` returns exactly what
+``get_effective_user_id`` would have. On any failure (no DB, missing row,
+memory-mode store) it falls back to the current login, so behaviour degrades
+safely to the pre-fix semantics.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import threading
+
+from deerflow.runtime.user_context import get_effective_user_id
+
+logger = logging.getLogger(__name__)
+
+# thread_id -> creator user_id, populated only for task-shared threads. The
+# (taskId, creator) pair is immutable for a thread, so positive results are
+# cached forever; the fallback (current login) is never cached because a row may
+# appear after the first lookup.
+_owner_cache: dict[str, str] = {}
+_owner_cache_lock = threading.Lock()
+_OWNER_CACHE_MAX = 4096
+
+
+def _cache_get(thread_id: str) -> str | None:
+    with _owner_cache_lock:
+        return _owner_cache.get(thread_id)
+
+
+def _cache_put(thread_id: str, owner: str) -> None:
+    with _owner_cache_lock:
+        # Crude bound: clear wholesale rather than track LRU order. The mapping
+        # is tiny and immutable, so a rare full clear just re-warms lazily.
+        if len(_owner_cache) >= _OWNER_CACHE_MAX:
+            _owner_cache.clear()
+        _owner_cache[thread_id] = owner
+
+
+async def _fetch_task_owner(thread_id: str) -> str | None:
+    """Return the creator user_id iff *thread_id* is a task-shared thread.
+
+    Returns ``None`` for personal threads, untracked threads, the memory-mode
+    store, or any DB error — callers then fall back to the current login.
+    """
+    from deerflow.persistence.engine import get_session_factory
+    from deerflow.persistence.thread_meta import ThreadMetaRepository
+
+    session_factory = get_session_factory()
+    if session_factory is None:
+        return None
+    store = ThreadMetaRepository(session_factory)
+    # user_id=None bypasses the owner filter so we can read any thread's row.
+    row = await store.get(thread_id, user_id=None)
+    if not row:
+        return None
+    metadata = row.get("metadata") or {}
+    owner = row.get("user_id")
+    if metadata.get("taskId") and owner:
+        return str(owner)
+    return None
+
+
+def _run_coro_blocking(coro):
+    """Run an async coroutine to completion from sync code.
+
+    Mirrors ``lead_agent.prompt._run_coro_blocking``: when already inside a
+    running event loop, hand the coroutine to a fresh loop in a worker thread;
+    otherwise run it directly.
+    """
+    try:
+        asyncio.get_running_loop()
+        in_loop = True
+    except RuntimeError:
+        in_loop = False
+
+    if in_loop:
+        import concurrent.futures
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+            return pool.submit(asyncio.run, coro).result()
+    return asyncio.run(coro)
+
+
+def resolve_path_user_id(thread_id: str | None) -> str:
+    """Canonical user bucket for *thread_id*'s files (sync).
+
+    For a task-shared thread this is the thread creator, so all accounts that
+    share the conversation read/write the same physical directory. For every
+    other case it is the current login (``get_effective_user_id``).
+    """
+    if not thread_id:
+        return get_effective_user_id()
+    cached = _cache_get(thread_id)
+    if cached is not None:
+        return cached
+    try:
+        owner = _run_coro_blocking(_fetch_task_owner(thread_id))
+    except Exception:
+        logger.debug("Failed to resolve task-shared owner for thread %s", thread_id, exc_info=True)
+        owner = None
+    if owner:
+        _cache_put(thread_id, owner)
+        return owner
+    return get_effective_user_id()
+
+
+async def aresolve_path_user_id(thread_id: str | None) -> str:
+    """Async variant of :func:`resolve_path_user_id` for FastAPI handlers."""
+    if not thread_id:
+        return get_effective_user_id()
+    cached = _cache_get(thread_id)
+    if cached is not None:
+        return cached
+    try:
+        owner = await _fetch_task_owner(thread_id)
+    except Exception:
+        logger.debug("Failed to resolve task-shared owner for thread %s", thread_id, exc_info=True)
+        owner = None
+    if owner:
+        _cache_put(thread_id, owner)
+        return owner
+    return get_effective_user_id()

offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/present_file_tool.py

@@ -10,7 +10,7 @@ from langgraph.typing import ContextT
 
 from deerflow.agents.thread_state import ThreadState
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths
-from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.runtime.thread_paths import resolve_path_user_id
 
 logger = logging.getLogger(__name__)
 
@@ -106,7 +106,7 @@ def _normalize_presented_filepath(
 
     if stripped == virtual_prefix or stripped.startswith(virtual_prefix + "/"):
         try:
-            actual_path = get_paths().resolve_virtual_path(thread_id, filepath, user_id=get_effective_user_id())
+            actual_path = get_paths().resolve_virtual_path(thread_id, filepath, user_id=resolve_path_user_id(thread_id))
         except TypeError:
             actual_path = get_paths().resolve_virtual_path(thread_id, filepath)
     else:

offline-backend-20260512/backend/packages/harness/deerflow/uploads/manager.py

@@ -12,7 +12,7 @@ from pathlib import Path
 from urllib.parse import quote
 
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths
-from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.runtime.thread_paths import resolve_path_user_id
 
 
 class PathTraversalError(ValueError):
@@ -40,7 +40,7 @@ def validate_thread_id(thread_id: str) -> None:
 def get_uploads_dir(thread_id: str) -> Path:
     """Return the uploads directory path for a thread (no side effects)."""
     validate_thread_id(thread_id)
-    return get_paths().sandbox_uploads_dir(thread_id, user_id=get_effective_user_id())
+    return get_paths().sandbox_uploads_dir(thread_id, user_id=resolve_path_user_id(thread_id))
 
 
 def ensure_uploads_dir(thread_id: str) -> Path: