From a40df0352106578828da76de7f7c77a4b2b77ac3 Mon Sep 17 00:00:00 2001
From: 1445043649 <>
Date: Tue, 12 May 2026 21:11:28 +0800
Subject: [PATCH] feat(persistence): WorkspaceRepository + 23 unit tests
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

新建 backend/packages/harness/deerflow/persistence/workspace/sql.py，方法：
  - create(name, slug, owner_id, *, workspace_id=None, status='active')
    UUID v4 自动生成；强校验 slug 格式（regex ^[a-z0-9](-?[a-z0-9])*\$ + 3-32
    长度）+ slug 黑名单（25 个保留字）+ status 枚举
  - get(workspace_id, *, user_id=AUTO) JOIN workspace_memberships 做成员校验；
    user_id=None 显式 bypass（迁移/admin）
  - get_by_slug(slug) 不带成员校验（path-based routing 用：先 slug→workspace_id
    再到 route handler 里查成员）
  - list_by_user(*, user_id=AUTO) 列 user 所属所有 workspace
  - update_status / delete platform-admin 操作，不带成员校验

WorkspaceValidationError 自定义异常（slug 格式 / 黑名单 / status）。

23 test 覆盖：
  - CRUD smoke + get_by_slug missing
  - 重复 slug → IntegrityError
  - 8 个 invalid slug pattern（短/长/大写/空格/破折号位置/连续破折号/下划线）
  - 6 个 blacklisted slug
  - status 状态机 + 非法值拒绝
  - delete CASCADE 到 memberships（SQLite FK PRAGMA 已开启）
  - get/list 成员过滤（user-A 看不见 user-B 的 workspace）
  - list user_id=None 显式 bypass

全部在 SQLite ephemeral DB 上跑（< 1s）。partial-unique 双驱动验证留给 T3.8。

Stage 0 PR3 T3.4 + T3.5。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 .../persistence/workspace/__init__.py         |   3 +-
 .../deerflow/persistence/workspace/sql.py     | 230 ++++++++++++++
 backend/tests/test_workspace_repo.py          | 290 ++++++++++++++++++
 3 files changed, 522 insertions(+), 1 deletion(-)
 create mode 100644 backend/packages/harness/deerflow/persistence/workspace/sql.py
 create mode 100644 backend/tests/test_workspace_repo.py

diff --git a/backend/packages/harness/deerflow/persistence/workspace/__init__.py b/backend/packages/harness/deerflow/persistence/workspace/__init__.py
index c60e57cf..9e4b9f6d 100644
--- a/backend/packages/harness/deerflow/persistence/workspace/__init__.py
+++ b/backend/packages/harness/deerflow/persistence/workspace/__init__.py
@@ -12,5 +12,6 @@ to FK back to ``workspaces.id``.
 from __future__ import annotations
 
 from deerflow.persistence.workspace.model import WorkspaceRow
+from deerflow.persistence.workspace.sql import WorkspaceRepository, WorkspaceValidationError
 
-__all__ = ["WorkspaceRow"]
+__all__ = ["WorkspaceRepository", "WorkspaceRow", "WorkspaceValidationError"]
diff --git a/backend/packages/harness/deerflow/persistence/workspace/sql.py b/backend/packages/harness/deerflow/persistence/workspace/sql.py
new file mode 100644
index 00000000..5a213633
--- /dev/null
+++ b/backend/packages/harness/deerflow/persistence/workspace/sql.py
@@ -0,0 +1,230 @@
+"""SQLAlchemy-backed workspace repository.
+
+CRUD + slug lookup for ``workspaces``. Membership-aware methods
+(``get``, ``list_by_user``) JOIN against ``workspace_memberships`` so
+callers cannot read workspaces they don't belong to.
+
+Three-state ``user_id`` parameter (same convention as
+:class:`ThreadMetaRepository`):
+  - ``AUTO`` → read from contextvar; raise if unset
+  - explicit ``str`` → override contextvar (admin / tests)
+  - explicit ``None`` → no filter (migration / CLI only)
+"""
+
+from __future__ import annotations
+
+import re
+import uuid
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.workspace.model import WorkspaceRow
+from deerflow.persistence.workspace_membership.model import WorkspaceMembershipRow
+from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
+
+# slug 字符集 / 长度（与 workspace-schema-design §2.1 锁定一致）。
+_SLUG_PATTERN = re.compile(r"^[a-z0-9](-?[a-z0-9])*$")
+_SLUG_MIN_LEN = 3
+_SLUG_MAX_LEN = 32
+
+# slug 黑名单（应用层校验，不写 DB constraint）。包含 ADR-007 §4 保留 slug
+# + 路径 + Next.js 保留 + 业务保留词。
+_SLUG_BLACKLIST = frozenset(
+    {
+        "admin",
+        "api",
+        "auth",
+        "login",
+        "signup",
+        "accept-invite",
+        "pricing",
+        "docs",
+        "status",
+        "platform",
+        "system",
+        "health",
+        "static",
+        "public",
+        "favicon.ico",
+        "robots.txt",
+        "sitemap.xml",
+        "_next",
+        ".well-known",
+        "settings",
+        "billing",
+        "onboarding",
+        "select-workspace",
+    }
+)
+
+# 允许的 status 集合。
+_VALID_STATUSES = frozenset({"active", "suspended", "deleted"})
+
+
+class WorkspaceValidationError(ValueError):
+    """Raised when workspace input fails application-layer validation
+    (slug format / blacklist / status enum)."""
+
+
+def _validate_slug(slug: str) -> None:
+    """Raise :class:`WorkspaceValidationError` if slug is invalid."""
+    if not isinstance(slug, str):
+        raise WorkspaceValidationError(f"slug must be a string, got {type(slug).__name__}")
+    if not (_SLUG_MIN_LEN <= len(slug) <= _SLUG_MAX_LEN):
+        raise WorkspaceValidationError(f"slug length must be between {_SLUG_MIN_LEN} and {_SLUG_MAX_LEN}, got {len(slug)}")
+    if not _SLUG_PATTERN.fullmatch(slug):
+        raise WorkspaceValidationError(f"slug {slug!r} does not match required pattern ^[a-z0-9](-?[a-z0-9])*$")
+    if slug in _SLUG_BLACKLIST:
+        raise WorkspaceValidationError(f"slug {slug!r} is reserved")
+
+
+def _validate_status(status: str) -> None:
+    if status not in _VALID_STATUSES:
+        raise WorkspaceValidationError(f"status {status!r} is not in allowed set {_VALID_STATUSES!r}")
+
+
+class WorkspaceRepository:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    @staticmethod
+    def _row_to_dict(row: WorkspaceRow) -> dict[str, Any]:
+        return {
+            "id": row.id,
+            "name": row.name,
+            "slug": row.slug,
+            "status": row.status,
+            "owner_id": row.owner_id,
+            "created_at": row.created_at.isoformat() if row.created_at else None,
+            "updated_at": row.updated_at.isoformat() if row.updated_at else None,
+        }
+
+    async def create(
+        self,
+        *,
+        name: str,
+        slug: str,
+        owner_id: str,
+        workspace_id: str | None = None,
+        status: str = "active",
+    ) -> dict[str, Any]:
+        """Create a new workspace.
+
+        ``workspace_id`` is optional — UUID v4 is generated if omitted.
+        Caller is responsible for creating the matching ``owner`` row
+        in ``workspace_memberships`` (this is typically done in the same
+        transaction by the registration flow; we deliberately don't bundle
+        it here to keep the repository single-responsibility).
+
+        Raises :class:`WorkspaceValidationError` for invalid slug / status.
+        Raises :class:`sqlalchemy.exc.IntegrityError` for slug collision
+        or invalid owner_id FK.
+        """
+        _validate_slug(slug)
+        _validate_status(status)
+        wid = workspace_id or str(uuid.uuid4())
+        now = datetime.now(UTC)
+        row = WorkspaceRow(
+            id=wid,
+            name=name,
+            slug=slug,
+            status=status,
+            owner_id=owner_id,
+            created_at=now,
+            updated_at=now,
+        )
+        async with self._sf() as session:
+            session.add(row)
+            await session.commit()
+            await session.refresh(row)
+            return self._row_to_dict(row)
+
+    async def get(
+        self,
+        workspace_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> dict[str, Any] | None:
+        """Return workspace row IFF caller is a member.
+
+        ``user_id=None`` bypasses the membership filter (migration / CLI).
+        Returns ``None`` if the workspace does not exist or the caller is
+        not a member.
+        """
+        resolved_user_id = resolve_user_id(user_id, method_name="WorkspaceRepository.get")
+        async with self._sf() as session:
+            row = await session.get(WorkspaceRow, workspace_id)
+            if row is None:
+                return None
+            if resolved_user_id is None:
+                # Explicit bypass (migration / admin path).
+                return self._row_to_dict(row)
+            # Membership check via separate SELECT (cheap; index covers it).
+            membership = await session.execute(
+                select(WorkspaceMembershipRow).where(
+                    WorkspaceMembershipRow.workspace_id == workspace_id,
+                    WorkspaceMembershipRow.user_id == resolved_user_id,
+                )
+            )
+            if membership.scalar_one_or_none() is None:
+                return None
+            return self._row_to_dict(row)
+
+    async def get_by_slug(self, slug: str) -> dict[str, Any] | None:
+        """Public slug lookup — does NOT check membership.
+
+        Used by path-based routing (``/{slug}/...``) where we need to
+        resolve slug → workspace_id BEFORE we know if caller belongs.
+        Membership check happens downstream in the route handler.
+        """
+        async with self._sf() as session:
+            result = await session.execute(select(WorkspaceRow).where(WorkspaceRow.slug == slug))
+            row = result.scalar_one_or_none()
+            return self._row_to_dict(row) if row else None
+
+    async def list_by_user(
+        self,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> list[dict[str, Any]]:
+        """Return all workspaces caller is a member of, ordered by joined_at desc.
+
+        ``user_id=None`` lists ALL workspaces (migration / admin path).
+        """
+        resolved_user_id = resolve_user_id(user_id, method_name="WorkspaceRepository.list_by_user")
+        async with self._sf() as session:
+            stmt = select(WorkspaceRow).order_by(WorkspaceRow.created_at.desc())
+            if resolved_user_id is not None:
+                stmt = stmt.join(
+                    WorkspaceMembershipRow,
+                    WorkspaceMembershipRow.workspace_id == WorkspaceRow.id,
+                ).where(WorkspaceMembershipRow.user_id == resolved_user_id)
+            result = await session.execute(stmt)
+            return [self._row_to_dict(r) for r in result.scalars()]
+
+    async def update_status(self, workspace_id: str, status: str) -> None:
+        """Platform-admin operation: change workspace status (active/suspended/deleted).
+
+        No membership check — this is for platform-level operations. Audit
+        logging belongs at the route layer.
+        """
+        _validate_status(status)
+        async with self._sf() as session:
+            await session.execute(update(WorkspaceRow).where(WorkspaceRow.id == workspace_id).values(status=status, updated_at=datetime.now(UTC)))
+            await session.commit()
+
+    async def delete(self, workspace_id: str) -> None:
+        """Hard-delete a workspace. CASCADE drops all memberships.
+
+        Intentionally no membership check — caller (platform admin route)
+        must enforce authorization. Stage 0 doesn't expose this to end
+        users; Stage 2+ adds it behind owner_only permission.
+        """
+        async with self._sf() as session:
+            row = await session.get(WorkspaceRow, workspace_id)
+            if row is not None:
+                await session.delete(row)
+                await session.commit()
diff --git a/backend/tests/test_workspace_repo.py b/backend/tests/test_workspace_repo.py
new file mode 100644
index 00000000..6a537e42
--- /dev/null
+++ b/backend/tests/test_workspace_repo.py
@@ -0,0 +1,290 @@
+"""Tests for WorkspaceRepository (Stage 0 PR3).
+
+Pattern mirrors :mod:`test_feedback`: SQLite ephemeral DB per test via
+tmp_path, no real Postgres needed at the unit-test layer. Partial-unique
+double-driver validation lives in :mod:`test_workspace_partial_unique`
+(T3.8, runs against both backends).
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+from deerflow.persistence.user.model import UserRow
+from deerflow.persistence.workspace import WorkspaceRepository, WorkspaceValidationError
+
+
+async def _make_repo(tmp_path):
+    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))
+    return WorkspaceRepository(get_session_factory())
+
+
+async def _cleanup():
+    from deerflow.persistence.engine import close_engine
+
+    await close_engine()
+
+
+async def _seed_user(repo, user_id: str = "u-alice", email: str = "alice@example.com") -> None:
+    """Create a user row so workspace.owner_id FK is satisfied."""
+    async with repo._sf() as session:
+        session.add(UserRow(id=user_id, email=email))
+        await session.commit()
+
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture
+def anyio_backend() -> str:
+    return "asyncio"
+
+
+# ---------------------------------------------------------------------------
+# create / get_by_slug round-trip
+# ---------------------------------------------------------------------------
+
+
+async def test_create_then_lookup_by_slug(tmp_path):
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_user(repo)
+        created = await repo.create(name="Alice's Workspace", slug="alice", owner_id="u-alice")
+        assert created["slug"] == "alice"
+        assert created["status"] == "active"
+        assert created["owner_id"] == "u-alice"
+        assert len(created["id"]) == 36  # UUID v4
+
+        fetched = await repo.get_by_slug("alice")
+        assert fetched is not None
+        assert fetched["id"] == created["id"]
+    finally:
+        await _cleanup()
+
+
+async def test_get_by_slug_returns_none_when_missing(tmp_path):
+    repo = await _make_repo(tmp_path)
+    try:
+        assert await repo.get_by_slug("nonexistent") is None
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# slug uniqueness + format + blacklist
+# ---------------------------------------------------------------------------
+
+
+async def test_create_rejects_duplicate_slug(tmp_path):
+    """Two workspaces with the same slug — second raises IntegrityError."""
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_user(repo)
+        await repo.create(name="A", slug="dup", owner_id="u-alice")
+        with pytest.raises(IntegrityError):
+            await repo.create(name="B", slug="dup", owner_id="u-alice")
+    finally:
+        await _cleanup()
+
+
+@pytest.mark.parametrize(
+    "bad_slug",
+    [
+        "ab",  # too short
+        "x" * 33,  # too long
+        "UPPER",  # uppercase
+        "has space",  # space
+        "-start-with-dash",  # bad start
+        "end-with-dash-",  # bad end
+        "double--dash",  # consecutive dashes
+        "underscore_not_ok",  # underscore
+    ],
+)
+async def test_create_rejects_invalid_slug_pattern(tmp_path, bad_slug):
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_user(repo)
+        with pytest.raises(WorkspaceValidationError, match="(pattern|length)"):
+            await repo.create(name="x", slug=bad_slug, owner_id="u-alice")
+    finally:
+        await _cleanup()
+
+
+@pytest.mark.parametrize("reserved", ["admin", "api", "auth", "settings", "billing", "select-workspace"])
+async def test_create_rejects_reserved_slug(tmp_path, reserved):
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_user(repo)
+        with pytest.raises(WorkspaceValidationError, match="reserved"):
+            await repo.create(name="x", slug=reserved, owner_id="u-alice")
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# status state machine
+# ---------------------------------------------------------------------------
+
+
+async def test_status_state_transitions(tmp_path):
+    """active → suspended → deleted are all accepted."""
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_user(repo)
+        ws = await repo.create(name="x", slug="trans", owner_id="u-alice")
+        assert ws["status"] == "active"
+
+        await repo.update_status(ws["id"], "suspended")
+        async with repo._sf() as session:
+            from deerflow.persistence.workspace.model import WorkspaceRow
+
+            row = await session.get(WorkspaceRow, ws["id"])
+            assert row.status == "suspended"
+
+        await repo.update_status(ws["id"], "deleted")
+        async with repo._sf() as session:
+            from deerflow.persistence.workspace.model import WorkspaceRow
+
+            row = await session.get(WorkspaceRow, ws["id"])
+            assert row.status == "deleted"
+    finally:
+        await _cleanup()
+
+
+async def test_update_status_rejects_unknown_value(tmp_path):
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_user(repo)
+        ws = await repo.create(name="x", slug="rejstat", owner_id="u-alice")
+        with pytest.raises(WorkspaceValidationError, match="allowed set"):
+            await repo.update_status(ws["id"], "weird-state")
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# CASCADE: workspace.delete() drops dependent memberships
+# ---------------------------------------------------------------------------
+
+
+async def test_delete_cascades_to_memberships(tmp_path):
+    """Deleting a workspace removes its membership rows (FK CASCADE)."""
+    from sqlalchemy import select
+
+    from deerflow.persistence.workspace_membership.model import WorkspaceMembershipRow
+
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_user(repo)
+        ws = await repo.create(name="x", slug="casc", owner_id="u-alice")
+        # Insert an owner membership manually (repository pattern is single-
+        # responsibility; registration flow will normally insert both rows
+        # in one transaction).
+        async with repo._sf() as session:
+            session.add(WorkspaceMembershipRow(workspace_id=ws["id"], user_id="u-alice", role="owner"))
+            await session.commit()
+
+        await repo.delete(ws["id"])
+
+        async with repo._sf() as session:
+            remaining = (await session.execute(select(WorkspaceMembershipRow).where(WorkspaceMembershipRow.workspace_id == ws["id"]))).scalars().all()
+        assert remaining == [], "memberships should be CASCADE-deleted with workspace"
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# membership-aware get + list_by_user
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.no_auto_user
+async def test_get_returns_none_for_non_member(tmp_path):
+    """User-A creates a workspace; User-B's `get(wsA)` returns None."""
+    from deerflow.persistence.workspace_membership.model import WorkspaceMembershipRow
+    from deerflow.runtime.user_context import reset_current_user, set_current_user
+
+    repo = await _make_repo(tmp_path)
+    try:
+        # Seed two users
+        async with repo._sf() as session:
+            session.add(UserRow(id="u-A", email="a@example.com"))
+            session.add(UserRow(id="u-B", email="b@example.com"))
+            await session.commit()
+
+        # User A creates workspace + becomes owner
+        ws = await repo.create(name="A's WS", slug="a-ws", owner_id="u-A")
+        async with repo._sf() as session:
+            session.add(WorkspaceMembershipRow(workspace_id=ws["id"], user_id="u-A", role="owner"))
+            await session.commit()
+
+        # User B attempts to read it via contextvar
+        user_b = SimpleNamespace(id="u-B")
+        token = set_current_user(user_b)
+        try:
+            assert await repo.get(ws["id"]) is None
+        finally:
+            reset_current_user(token)
+
+        # User A's own get succeeds
+        user_a = SimpleNamespace(id="u-A")
+        token = set_current_user(user_a)
+        try:
+            row = await repo.get(ws["id"])
+            assert row is not None
+            assert row["slug"] == "a-ws"
+        finally:
+            reset_current_user(token)
+    finally:
+        await _cleanup()
+
+
+@pytest.mark.no_auto_user
+async def test_list_by_user_excludes_other_workspaces(tmp_path):
+    from deerflow.persistence.workspace_membership.model import WorkspaceMembershipRow
+    from deerflow.runtime.user_context import reset_current_user, set_current_user
+
+    repo = await _make_repo(tmp_path)
+    try:
+        async with repo._sf() as session:
+            session.add(UserRow(id="u-A", email="a@example.com"))
+            session.add(UserRow(id="u-B", email="b@example.com"))
+            await session.commit()
+
+        ws_a = await repo.create(name="A", slug="ws-a", owner_id="u-A")
+        ws_b = await repo.create(name="B", slug="ws-b", owner_id="u-B")
+        async with repo._sf() as session:
+            session.add(WorkspaceMembershipRow(workspace_id=ws_a["id"], user_id="u-A", role="owner"))
+            session.add(WorkspaceMembershipRow(workspace_id=ws_b["id"], user_id="u-B", role="owner"))
+            await session.commit()
+
+        token = set_current_user(SimpleNamespace(id="u-A"))
+        try:
+            workspaces = await repo.list_by_user()
+            assert [w["slug"] for w in workspaces] == ["ws-a"]
+        finally:
+            reset_current_user(token)
+    finally:
+        await _cleanup()
+
+
+async def test_list_by_user_bypass_returns_all(tmp_path):
+    """user_id=None opts out of membership filter (migration path)."""
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_user(repo, "u-A", "a@example.com")
+        await _seed_user(repo, "u-B", "b@example.com")
+        await repo.create(name="A", slug="all-a", owner_id="u-A")
+        await repo.create(name="B", slug="all-b", owner_id="u-B")
+
+        workspaces = await repo.list_by_user(user_id=None)
+        slugs = sorted(w["slug"] for w in workspaces)
+        assert slugs == ["all-a", "all-b"]
+    finally:
+        await _cleanup()