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>
2026-06-19 14:35:22 +08:00
1 parent 13ef1942f2
commit 23e6685977
@@ -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
@@ -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)
@@ -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():
@@ -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
@@ -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"])

@@ -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
@@ -12,7 +12,7 @@ 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__)
@@ -198,7 +198,7 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
            source_path = self._paths.resolve_virtual_path(
                source_thread_id,
                virtual_path,
-                user_id=get_effective_user_id(),
+                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)
@@ -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 []
@@ -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
@@ -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()
@@ -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:
@@ -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: