diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index 4985fc2d..17eb0b9b 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/offline-backend-20260512/backend/CLAUDE.md @@ -459,6 +459,7 @@ Operators upgrading an existing DB run the whole file (`USE ;` 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 diff --git a/offline-backend-20260512/backend/app/gateway/path_utils.py b/offline-backend-20260512/backend/app/gateway/path_utils.py index c67e7a25..403aa930 100644 --- a/offline-backend-20260512/backend/app/gateway/path_utils.py +++ b/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) diff --git a/offline-backend-20260512/backend/app/gateway/routers/artifacts.py b/offline-backend-20260512/backend/app/gateway/routers/artifacts.py index 164b768b..0bf0935e 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/artifacts.py +++ b/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(): diff --git a/offline-backend-20260512/backend/app/gateway/routers/skills.py b/offline-backend-20260512/backend/app/gateway/routers/skills.py index dad9758f..18a3a2c0 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/skills.py +++ b/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 diff --git a/offline-backend-20260512/backend/app/gateway/routers/uploads.py b/offline-backend-20260512/backend/app/gateway/routers/uploads.py index 489e7274..5b2bc581 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/uploads.py +++ b/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"]) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/thread_data_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/thread_data_middleware.py index fba06cf4..9da88c66 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/thread_data_middleware.py +++ b/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 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py index e0be6761..37c937c2 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py +++ b/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("") - 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 [] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/client.py b/offline-backend-20260512/backend/packages/harness/deerflow/client.py index 35374e24..1e76d27a 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/client.py +++ b/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 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/thread_paths.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/thread_paths.py new file mode 100644 index 00000000..fdbae69a --- /dev/null +++ b/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() diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/present_file_tool.py b/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/present_file_tool.py index 8aec0ce5..ace9c057 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/present_file_tool.py +++ b/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: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/uploads/manager.py b/offline-backend-20260512/backend/packages/harness/deerflow/uploads/manager.py index 428c3209..444a2a32 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/uploads/manager.py +++ b/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: