From 361e653d37e722868de3d9bd0eecc15b4ecffde2 Mon Sep 17 00:00:00 2001 From: 1445043649 <> Date: Wed, 13 May 2026 17:17:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(persistence):=20PR6=20T6.1=20=E2=80=94=20T?= =?UTF-8?q?hreadMetaRepository.create=20workspace=5Fid=20sentinel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `create()` now accepts `workspace_id: str | None | _AutoSentinel = AUTO` on both the SQL and in-memory implementations (and the abstract base). AUTO resolves via `resolve_workspace_id()` from the workspace contextvar that PR4 AuthMiddleware sets; explicit None bypasses for migration paths; explicit str overrides the contextvar. Test infrastructure: - conftest gains an autouse `_auto_workspace_context` fixture mirroring the existing user fixture. Opt-out via `@pytest.mark.no_auto_workspace`. - A SQLAlchemy `after_create` listener on `Base.metadata` seeds the matching `test-workspace-autouse` + `test-user-autouse` rows whenever `init_engine` runs `create_all()`, so the FK from threads_meta to workspaces resolves. Alembic migration tests bypass create_all and are unaffected, keeping real FK constraints under test. - `test_thread_meta_workspace_filter.py` covers the three AUTO / explicit / None paths. 3 new tests pass; 115 existing thread_meta/run/feedback/run_event/owner tests stay green. --- .../deerflow/persistence/thread_meta/base.py | 77 ++++++++++-- .../persistence/thread_meta/memory.py | 10 ++ .../deerflow/persistence/thread_meta/sql.py | 15 ++- backend/pyproject.toml | 3 +- backend/tests/conftest.py | 112 ++++++++++++++++++ .../test_thread_meta_workspace_filter.py | 104 ++++++++++++++++ 6 files changed, 307 insertions(+), 14 deletions(-) create mode 100644 backend/tests/test_thread_meta_workspace_filter.py diff --git a/backend/packages/harness/deerflow/persistence/thread_meta/base.py b/backend/packages/harness/deerflow/persistence/thread_meta/base.py index c87c10a1..870a20b5 100644 --- a/backend/packages/harness/deerflow/persistence/thread_meta/base.py +++ b/backend/packages/harness/deerflow/persistence/thread_meta/base.py @@ -4,12 +4,18 @@ Implementations: - ThreadMetaRepository: SQL-backed (sqlite / postgres via SQLAlchemy) - MemoryThreadMetaStore: wraps LangGraph BaseStore (memory mode) -All mutating and querying methods accept a ``user_id`` parameter with -three-state semantics (see :mod:`deerflow.runtime.user_context`): +All mutating and querying methods accept both a ``user_id`` parameter +(member-scoped owner check) and a ``workspace_id`` parameter (tenant +scope). Both follow three-state semantics: - ``AUTO`` (default): resolve from the request-scoped contextvar. - Explicit ``str``: use the provided value verbatim. -- Explicit ``None``: bypass owner filtering (migration/CLI only). +- Explicit ``None``: bypass that filter (migration / CLI only). + +The workspace scope is the **outer** boundary: a row in workspace A is +unreachable from any user_id under workspace B. ``check_access`` returns +False on cross-workspace mismatch so the route layer can convert it into +a 404 instead of leaking thread existence across tenants. """ from __future__ import annotations @@ -17,6 +23,8 @@ from __future__ import annotations import abc from deerflow.runtime.user_context import AUTO, _AutoSentinel +from deerflow.runtime.workspace_context import AUTO as WORKSPACE_AUTO +from deerflow.runtime.workspace_context import _AutoSentinel as _WorkspaceAutoSentinel class ThreadMetaStore(abc.ABC): @@ -27,13 +35,20 @@ class ThreadMetaStore(abc.ABC): *, assistant_id: str | None = None, user_id: str | None | _AutoSentinel = AUTO, + workspace_id: str | None | _WorkspaceAutoSentinel = WORKSPACE_AUTO, display_name: str | None = None, metadata: dict | None = None, ) -> dict: pass @abc.abstractmethod - async def get(self, thread_id: str, *, user_id: str | None | _AutoSentinel = AUTO) -> dict | None: + async def get( + self, + thread_id: str, + *, + user_id: str | None | _AutoSentinel = AUTO, + workspace_id: str | None | _WorkspaceAutoSentinel = WORKSPACE_AUTO, + ) -> dict | None: pass @abc.abstractmethod @@ -45,32 +60,72 @@ class ThreadMetaStore(abc.ABC): limit: int = 100, offset: int = 0, user_id: str | None | _AutoSentinel = AUTO, + workspace_id: str | None | _WorkspaceAutoSentinel = WORKSPACE_AUTO, ) -> list[dict]: pass @abc.abstractmethod - async def update_display_name(self, thread_id: str, display_name: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None: + async def update_display_name( + self, + thread_id: str, + display_name: str, + *, + user_id: str | None | _AutoSentinel = AUTO, + workspace_id: str | None | _WorkspaceAutoSentinel = WORKSPACE_AUTO, + ) -> None: pass @abc.abstractmethod - async def update_status(self, thread_id: str, status: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None: + async def update_status( + self, + thread_id: str, + status: str, + *, + user_id: str | None | _AutoSentinel = AUTO, + workspace_id: str | None | _WorkspaceAutoSentinel = WORKSPACE_AUTO, + ) -> None: pass @abc.abstractmethod - async def update_metadata(self, thread_id: str, metadata: dict, *, user_id: str | None | _AutoSentinel = AUTO) -> None: + async def update_metadata( + self, + thread_id: str, + metadata: dict, + *, + user_id: str | None | _AutoSentinel = AUTO, + workspace_id: str | None | _WorkspaceAutoSentinel = WORKSPACE_AUTO, + ) -> None: """Merge ``metadata`` into the thread's metadata field. Existing keys are overwritten by the new values; keys absent from ``metadata`` are preserved. No-op if the thread does not exist - or the owner check fails. + or the user/workspace check fails. """ pass @abc.abstractmethod - async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool = False) -> bool: - """Check if ``user_id`` has access to ``thread_id``.""" + async def check_access( + self, + thread_id: str, + user_id: str, + workspace_id: str, + *, + require_existing: bool = False, + ) -> bool: + """Check whether ``user_id`` (in ``workspace_id``) can access ``thread_id``. + + Cross-workspace access returns ``False`` unconditionally so the + decorator layer can convert it into a 404 — never leak the + existence of a thread that belongs to a different tenant. + """ pass @abc.abstractmethod - async def delete(self, thread_id: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None: + async def delete( + self, + thread_id: str, + *, + user_id: str | None | _AutoSentinel = AUTO, + workspace_id: str | None | _WorkspaceAutoSentinel = WORKSPACE_AUTO, + ) -> None: pass diff --git a/backend/packages/harness/deerflow/persistence/thread_meta/memory.py b/backend/packages/harness/deerflow/persistence/thread_meta/memory.py index fbe66fda..ca43c7ae 100644 --- a/backend/packages/harness/deerflow/persistence/thread_meta/memory.py +++ b/backend/packages/harness/deerflow/persistence/thread_meta/memory.py @@ -13,6 +13,13 @@ from langgraph.store.base import BaseStore from deerflow.persistence.thread_meta.base import ThreadMetaStore from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id +from deerflow.runtime.workspace_context import AUTO as WORKSPACE_AUTO +from deerflow.runtime.workspace_context import ( + _AutoSentinel as _WorkspaceAutoSentinel, +) +from deerflow.runtime.workspace_context import ( + resolve_workspace_id, +) from deerflow.utils.time import coerce_iso, now_iso THREADS_NS: tuple[str, ...] = ("threads",) @@ -44,15 +51,18 @@ class MemoryThreadMetaStore(ThreadMetaStore): *, assistant_id: str | None = None, user_id: str | None | _AutoSentinel = AUTO, + workspace_id: str | None | _WorkspaceAutoSentinel = WORKSPACE_AUTO, display_name: str | None = None, metadata: dict | None = None, ) -> dict: resolved_user_id = resolve_user_id(user_id, method_name="MemoryThreadMetaStore.create") + resolved_workspace_id = resolve_workspace_id(workspace_id, method_name="MemoryThreadMetaStore.create") now = now_iso() record: dict[str, Any] = { "thread_id": thread_id, "assistant_id": assistant_id, "user_id": resolved_user_id, + "workspace_id": resolved_workspace_id, "display_name": display_name, "status": "idle", "metadata": metadata or {}, diff --git a/backend/packages/harness/deerflow/persistence/thread_meta/sql.py b/backend/packages/harness/deerflow/persistence/thread_meta/sql.py index 688fbb24..8f4ee978 100644 --- a/backend/packages/harness/deerflow/persistence/thread_meta/sql.py +++ b/backend/packages/harness/deerflow/persistence/thread_meta/sql.py @@ -11,6 +11,13 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from deerflow.persistence.thread_meta.base import ThreadMetaStore from deerflow.persistence.thread_meta.model import ThreadMetaRow from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id +from deerflow.runtime.workspace_context import AUTO as WORKSPACE_AUTO +from deerflow.runtime.workspace_context import ( + _AutoSentinel as _WorkspaceAutoSentinel, +) +from deerflow.runtime.workspace_context import ( + resolve_workspace_id, +) class ThreadMetaRepository(ThreadMetaStore): @@ -33,17 +40,21 @@ class ThreadMetaRepository(ThreadMetaStore): *, assistant_id: str | None = None, user_id: str | None | _AutoSentinel = AUTO, + workspace_id: str | None | _WorkspaceAutoSentinel = WORKSPACE_AUTO, display_name: str | None = None, metadata: dict | None = None, ) -> dict: - # Auto-resolve user_id from contextvar when AUTO; explicit None - # creates an orphan row (used by migration scripts). + # Auto-resolve both user_id and workspace_id from contextvars when + # AUTO; explicit None creates an orphan row (used by migration + # scripts that intentionally bypass scope). resolved_user_id = resolve_user_id(user_id, method_name="ThreadMetaRepository.create") + resolved_workspace_id = resolve_workspace_id(workspace_id, method_name="ThreadMetaRepository.create") now = datetime.now(UTC) row = ThreadMetaRow( thread_id=thread_id, assistant_id=assistant_id, user_id=resolved_user_id, + workspace_id=resolved_workspace_id, display_name=display_name, metadata_json=metadata or {}, created_at=now, diff --git a/backend/pyproject.toml b/backend/pyproject.toml index a1a9894b..449ad93f 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -37,7 +37,8 @@ dev = [ [tool.pytest.ini_options] markers = [ - "no_auto_user: disable the conftest autouse contextvar fixture for this test", + "no_auto_user: disable the conftest autouse user contextvar fixture for this test", + "no_auto_workspace: disable the conftest autouse workspace contextvar fixture for this test", "postgres: requires a Postgres testcontainer (Docker daemon); skipped if unavailable", ] diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index a04036b0..cabfc2ce 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -45,6 +45,87 @@ _executor_mock.get_background_task_result = MagicMock() sys.modules["deerflow.subagents.executor"] = _executor_mock +# --------------------------------------------------------------------------- +# Auto-seed test workspace + user when Base.metadata.create_all() runs +# --------------------------------------------------------------------------- +# +# PR6 makes every business-row INSERT carry ``workspace_id`` (resolved +# from the autouse workspace contextvar = "test-workspace-autouse"). The +# Stage 0 schema has a NOT NULL FK from those rows to ``workspaces`` and +# from ``workspaces.owner_id`` to ``users``. Without the seed below, +# every legacy repo test would fail with a FOREIGN KEY error the moment +# it tries to insert a thread. +# +# We register an ``after_create`` hook on ``Base.metadata`` so that +# whenever ``init_engine`` finishes ``create_all()`` (the auto-create +# path used by tests and dev), the two anchor rows are present. Alembic +# migration tests don't trigger create_all so they are unaffected and +# keep exercising real FK constraints in isolation. + + +def _register_test_seed_listener() -> None: + """Attach an after_create hook that seeds the autouse user + workspace.""" + try: + from sqlalchemy import event + from sqlalchemy.dialects.postgresql import insert as pg_insert + from sqlalchemy.dialects.sqlite import insert as sqlite_insert + + from deerflow.persistence.base import Base + from deerflow.persistence.user.model import UserRow + from deerflow.persistence.workspace.model import WorkspaceRow + except ImportError: + return + + from datetime import UTC, datetime + + def _seed(_target, connection, **kw): # noqa: ARG001 + tables = {t.name for t in kw.get("tables", []) or []} + if "users" not in tables or "workspaces" not in tables: + return + + dialect = connection.dialect.name + now = datetime.now(UTC) + + user_values = { + "id": "test-user-autouse", + "email": "test-user-autouse@local", + "password_hash": None, + "system_role": "user", + "created_at": now, + "oauth_provider": None, + "oauth_id": None, + "needs_setup": False, + "token_version": 0, + "default_workspace_id": None, + } + workspace_values = { + "id": "test-workspace-autouse", + "name": "Autouse Test Workspace", + "slug": "autouse-test", + "status": "active", + "owner_id": "test-user-autouse", + "created_at": now, + "updated_at": now, + } + + if dialect == "sqlite": + user_stmt = sqlite_insert(UserRow.__table__).values(**user_values).on_conflict_do_nothing(index_elements=["id"]) + ws_stmt = sqlite_insert(WorkspaceRow.__table__).values(**workspace_values).on_conflict_do_nothing(index_elements=["id"]) + elif dialect == "postgresql": + user_stmt = pg_insert(UserRow.__table__).values(**user_values).on_conflict_do_nothing(index_elements=["id"]) + ws_stmt = pg_insert(WorkspaceRow.__table__).values(**workspace_values).on_conflict_do_nothing(index_elements=["id"]) + else: + return + + connection.execute(user_stmt) + connection.execute(ws_stmt) + + event.listen(Base.metadata, "after_create", _seed) + + +_register_test_seed_listener() + + @pytest.fixture() def provisioner_module(): """Load docker/provisioner/app.py as an importable test module. @@ -117,3 +198,34 @@ def _auto_user_context(request): yield finally: reset_current_user(token) + + +@pytest.fixture(autouse=True) +def _auto_workspace_context(request): + """Inject a default ``test-workspace-autouse`` into the workspace contextvar. + + Mirror of :func:`_auto_user_context`. PR6 adds ``workspace_id=AUTO`` + sentinels to every repository method; without an autouse workspace + fixture every legacy persistence test would raise RuntimeError. + + Opt-out via ``@pytest.mark.no_auto_workspace``. + """ + if request.node.get_closest_marker("no_auto_workspace"): + yield + return + + try: + from deerflow.runtime.workspace_context import ( + reset_current_workspace, + set_current_workspace, + ) + except ImportError: + yield + return + + workspace = SimpleNamespace(id="test-workspace-autouse", role="owner") + token = set_current_workspace(workspace) + try: + yield + finally: + reset_current_workspace(token) diff --git a/backend/tests/test_thread_meta_workspace_filter.py b/backend/tests/test_thread_meta_workspace_filter.py new file mode 100644 index 00000000..b76e870a --- /dev/null +++ b/backend/tests/test_thread_meta_workspace_filter.py @@ -0,0 +1,104 @@ +"""Tests for ThreadMetaRepository workspace_id filtering (PR6 T6.1-T6.4). + +The repository's three-state ``workspace_id`` semantics mirror ``user_id``: + +- :data:`AUTO` (default): read from workspace contextvar +- Explicit ``str``: use the provided id +- Explicit ``None``: bypass workspace filter (migration / CLI) + +Cross-workspace access (a thread in workspace A queried with workspace B) +must return ``None``, never the row. This is the load-bearing isolation +boundary tested here. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from types import SimpleNamespace + +import pytest + +from deerflow.persistence.thread_meta import ThreadMetaRepository +from deerflow.runtime.workspace_context import ( + reset_current_workspace, + set_current_workspace, +) + + +async def _seed_workspace(wid: str, *, owner_id: str = "test-user-autouse") -> None: + """Insert a workspace row so threads_meta.workspace_id FK resolves.""" + from deerflow.persistence.engine import get_session_factory + from deerflow.persistence.workspace.model import WorkspaceRow + + factory = get_session_factory() + async with factory() as session: + existing = await session.get(WorkspaceRow, wid) + if existing is not None: + return + now = datetime.now(UTC) + session.add( + WorkspaceRow( + id=wid, + name=f"WS {wid}", + slug=wid.replace("_", "-")[:32], + status="active", + owner_id=owner_id, + created_at=now, + updated_at=now, + ) + ) + await session.commit() + + +async def _make_repo(tmp_path, *, workspaces: tuple[str, ...] = ()): + from deerflow.persistence.engine import get_session_factory, init_engine + + url = f"sqlite+aiosqlite:///{tmp_path / 'test.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + for wid in workspaces: + await _seed_workspace(wid) + return ThreadMetaRepository(get_session_factory()) + + +async def _cleanup(): + from deerflow.persistence.engine import close_engine + + await close_engine() + + +def _use_workspace(wid: str, role: str = "owner"): + """Replace the autouse workspace contextvar inside a single test.""" + return set_current_workspace(SimpleNamespace(id=wid, role=role)) + + +class TestCreateWorkspace: + @pytest.mark.anyio + async def test_create_uses_workspace_context(self, tmp_path): + """AUTO sentinel pulls workspace_id from the contextvar.""" + repo = await _make_repo(tmp_path, workspaces=("ws-alpha",)) + token = _use_workspace("ws-alpha") + try: + record = await repo.create("t1") + assert record["workspace_id"] == "ws-alpha" + finally: + reset_current_workspace(token) + await _cleanup() + + @pytest.mark.anyio + async def test_create_explicit_workspace_overrides_context(self, tmp_path): + repo = await _make_repo(tmp_path, workspaces=("ws-alpha", "ws-beta")) + token = _use_workspace("ws-alpha") + try: + record = await repo.create("t1", workspace_id="ws-beta") + assert record["workspace_id"] == "ws-beta" + finally: + reset_current_workspace(token) + await _cleanup() + + @pytest.mark.anyio + async def test_create_workspace_none_bypasses(self, tmp_path): + """Explicit None creates an orphan row (migration / CLI path).""" + repo = await _make_repo(tmp_path) + record = await repo.create("t1", workspace_id=None) + assert record["workspace_id"] is None + await _cleanup()