From 2d3b546bf9a2966b5664cbaa54240d78cd791722 Mon Sep 17 00:00:00 2001 From: 1445043649 <> Date: Wed, 13 May 2026 17:45:06 +0800 Subject: [PATCH] =?UTF-8?q?feat(agents):=20PR6=20T6.11=20=E2=80=94=20Threa?= =?UTF-8?q?dDataMiddleware=20switches=20to=20workspace=20layout?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ThreadDataMiddleware.before_agent` now reads `get_effective_workspace_id()` and routes the per-thread directory tree through `Paths.sandbox_*_dir(thread_id, workspace_id=...)`, producing `{base_dir}/workspaces/{wid}/threads/{tid}/user-data/...`. The legacy user-id-only layout is no longer written by this middleware; the migration script in T6.12 will lift any pre-existing `users/{uid}/...` trees into the new shape. In no-auth dev mode the contextvar is empty so `get_effective_workspace_id` returns `"default"` and writes land at `workspaces/default/...` — the layout invariant ("threads always live inside a workspace") holds without a real auth setup. `thread_data` now also exposes `user_id` and `workspace_id` so downstream middlewares (sandbox, memory, etc.) can read them without re-resolving the contextvar themselves. 4 new tests cover: contextvar workspace → expected path, `no_auto_workspace` falls back to `default`, eager mode creates the right dirs, and the `get_config` fallback still routes through workspace. Existing 4 thread_data middleware tests stay green. --- .../middlewares/thread_data_middleware.py | 55 +++++------- .../test_thread_data_middleware_workspace.py | 88 +++++++++++++++++++ 2 files changed, 111 insertions(+), 32 deletions(-) create mode 100644 backend/tests/test_thread_data_middleware_workspace.py diff --git a/backend/packages/harness/deerflow/agents/middlewares/thread_data_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/thread_data_middleware.py index 8d93de4f..6ca5c14f 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/thread_data_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/thread_data_middleware.py @@ -11,6 +11,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.workspace_context import get_effective_workspace_id logger = logging.getLogger(__name__) @@ -24,10 +25,15 @@ class ThreadDataMiddlewareState(AgentState): class ThreadDataMiddleware(AgentMiddleware[ThreadDataMiddlewareState]): """Create thread data directories for each thread execution. - Creates the following directory structure: - - {base_dir}/threads/{thread_id}/user-data/workspace - - {base_dir}/threads/{thread_id}/user-data/uploads - - {base_dir}/threads/{thread_id}/user-data/outputs + PR6 routes thread storage through the workspace dimension. When a + workspace contextvar is set (production via AuthMiddleware; tests via + the autouse fixture), directories live at + ``{base_dir}/workspaces/{wid}/threads/{thread_id}/user-data/{workspace,uploads,outputs}``. + In no-auth dev mode ``get_effective_workspace_id()`` returns + ``"default"`` so the layout stays valid; the ``user_id`` falls through + to the same default constant. Either way thread state lives below a + workspace bucket, never directly under ``{base_dir}/threads`` (legacy) + or ``{base_dir}/users`` (PR4 layout). Lifecycle Management: - With lazy_init=True (default): Only compute paths, directories created on-demand @@ -49,34 +55,18 @@ class ThreadDataMiddleware(AgentMiddleware[ThreadDataMiddlewareState]): self._paths = Paths(base_dir) if base_dir else get_paths() self._lazy_init = lazy_init - def _get_thread_paths(self, thread_id: str, user_id: str | None = None) -> dict[str, str]: - """Get the paths for a thread's data directories. - - Args: - thread_id: The thread ID. - user_id: Optional user ID for per-user path isolation. - - Returns: - Dictionary with workspace_path, uploads_path, and outputs_path. - """ + def _get_thread_paths(self, thread_id: str, *, workspace_id: str, user_id: str) -> dict[str, str]: return { - "workspace_path": str(self._paths.sandbox_work_dir(thread_id, user_id=user_id)), - "uploads_path": str(self._paths.sandbox_uploads_dir(thread_id, user_id=user_id)), - "outputs_path": str(self._paths.sandbox_outputs_dir(thread_id, user_id=user_id)), + "workspace_path": str(self._paths.sandbox_work_dir(thread_id, workspace_id=workspace_id)), + "uploads_path": str(self._paths.sandbox_uploads_dir(thread_id, workspace_id=workspace_id)), + "outputs_path": str(self._paths.sandbox_outputs_dir(thread_id, workspace_id=workspace_id)), + "user_id": user_id, + "workspace_id": workspace_id, } - def _create_thread_directories(self, thread_id: str, user_id: str | None = None) -> dict[str, str]: - """Create the thread data directories. - - Args: - thread_id: The thread ID. - user_id: Optional user ID for per-user path isolation. - - Returns: - Dictionary with the created directory paths. - """ - self._paths.ensure_thread_dirs(thread_id, user_id=user_id) - return self._get_thread_paths(thread_id, user_id=user_id) + def _create_thread_directories(self, thread_id: str, *, workspace_id: str, user_id: str) -> dict[str, str]: + self._paths.ensure_thread_dirs(thread_id, workspace_id=workspace_id) + return self._get_thread_paths(thread_id, workspace_id=workspace_id, user_id=user_id) @override def before_agent(self, state: ThreadDataMiddlewareState, runtime: Runtime) -> dict | None: @@ -90,14 +80,15 @@ class ThreadDataMiddleware(AgentMiddleware[ThreadDataMiddlewareState]): raise ValueError("Thread ID is required in runtime context or config.configurable") user_id = get_effective_user_id() + workspace_id = get_effective_workspace_id() if self._lazy_init: # Lazy initialization: only compute paths, don't create directories - paths = self._get_thread_paths(thread_id, user_id=user_id) + paths = self._get_thread_paths(thread_id, workspace_id=workspace_id, user_id=user_id) else: # Eager initialization: create directories immediately - paths = self._create_thread_directories(thread_id, user_id=user_id) - logger.debug("Created thread data directories for thread %s", thread_id) + paths = self._create_thread_directories(thread_id, workspace_id=workspace_id, user_id=user_id) + logger.debug("Created thread data directories for thread %s under workspace %s", thread_id, workspace_id) messages = list(state.get("messages", [])) last_message = messages[-1] if messages else None diff --git a/backend/tests/test_thread_data_middleware_workspace.py b/backend/tests/test_thread_data_middleware_workspace.py new file mode 100644 index 00000000..7ea3698e --- /dev/null +++ b/backend/tests/test_thread_data_middleware_workspace.py @@ -0,0 +1,88 @@ +"""PR6 T6.11 — ThreadDataMiddleware writes under workspace layout.""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from deerflow.agents.middlewares.thread_data_middleware import ThreadDataMiddleware +from deerflow.config.paths import Paths +from deerflow.runtime.workspace_context import ( + reset_current_workspace, + set_current_workspace, +) + + +class _FakeRuntime: + def __init__(self, *, thread_id: str = "t1", run_id: str = "r1"): + self.context = {"thread_id": thread_id, "run_id": run_id} + + +def test_paths_resolve_under_workspace(tmp_path): + paths = Paths(tmp_path) + middleware = ThreadDataMiddleware(base_dir=str(tmp_path), lazy_init=True) + middleware._paths = paths + + token = set_current_workspace(SimpleNamespace(id="ws-alpha", role="owner")) + try: + out = middleware.before_agent({"messages": []}, _FakeRuntime()) + finally: + reset_current_workspace(token) + + expected_root = tmp_path / "workspaces" / "ws-alpha" / "threads" / "t1" / "user-data" + assert out["thread_data"]["workspace_path"] == str(expected_root / "workspace") + assert out["thread_data"]["uploads_path"] == str(expected_root / "uploads") + assert out["thread_data"]["outputs_path"] == str(expected_root / "outputs") + assert out["thread_data"]["workspace_id"] == "ws-alpha" + + +@pytest.mark.no_auto_workspace +def test_falls_back_to_default_workspace(tmp_path): + """Without a workspace contextvar, `get_effective_workspace_id` returns 'default'.""" + paths = Paths(tmp_path) + middleware = ThreadDataMiddleware(base_dir=str(tmp_path), lazy_init=True) + middleware._paths = paths + + out = middleware.before_agent({"messages": []}, _FakeRuntime()) + + expected_root = tmp_path / "workspaces" / "default" / "threads" / "t1" / "user-data" + assert out["thread_data"]["workspace_path"] == str(expected_root / "workspace") + assert out["thread_data"]["workspace_id"] == "default" + + +def test_eager_creates_directories_under_workspace(tmp_path): + paths = Paths(tmp_path) + middleware = ThreadDataMiddleware(base_dir=str(tmp_path), lazy_init=False) + middleware._paths = paths + + token = set_current_workspace(SimpleNamespace(id="ws-beta", role="owner")) + try: + middleware.before_agent({"messages": []}, _FakeRuntime(thread_id="t2")) + finally: + reset_current_workspace(token) + + root = tmp_path / "workspaces" / "ws-beta" / "threads" / "t2" / "user-data" + assert (root / "workspace").is_dir() + assert (root / "uploads").is_dir() + assert (root / "outputs").is_dir() + + +def test_get_config_fallback_still_workspace_scoped(tmp_path): + """Thread_id resolution via LangGraph config still routes through workspace.""" + paths = Paths(tmp_path) + middleware = ThreadDataMiddleware(base_dir=str(tmp_path), lazy_init=True) + middleware._paths = paths + + class _Runtime: + context: dict = {} + + with patch("deerflow.agents.middlewares.thread_data_middleware.get_config", return_value={"configurable": {"thread_id": "t-cfg"}}): + token = set_current_workspace(SimpleNamespace(id="ws-gamma", role="owner")) + try: + out = middleware.before_agent({"messages": []}, _Runtime()) + finally: + reset_current_workspace(token) + + assert "workspaces/ws-gamma/threads/t-cfg/user-data/workspace" in out["thread_data"]["workspace_path"]