feat(persistence): PR6 T6.1 — ThreadMetaRepository.create workspace_id sentinel

`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.

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
+All mutating and querying methods accept both a ``user_id`` parameter
-three-state semantics (see :mod:`deerflow.runtime.user_context`):
+(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:
+    async def check_access(
-        """Check if ``user_id`` has access to ``thread_id``."""
+        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

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 {},

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
+        # Auto-resolve both user_id and workspace_id from contextvars when
-        # creates an orphan row (used by migration scripts).
+        # 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,

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",
 ]
 

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)

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()