feat(persistence): WorkspaceMembershipRepository + 8 unit tests

新建 backend/packages/harness/deerflow/persistence/workspace_membership/sql.py，方法：
  - add(workspace_id, user_id, role, invited_by=None) 校验 role ∈
    {owner, admin, member}（Stage 0 仅写 owner，schema 已支持其余）
  - remove(workspace_id, user_id) → bool（rowcount > 0）
  - list_by_user(user_id) 按 joined_at desc
  - list_by_workspace(workspace_id) 按 joined_at asc
  - get_role(workspace_id, user_id) → str | None
  - change_role(workspace_id, user_id, new_role) → bool

MembershipValidationError 自定义异常（role 非法）。

8 test 覆盖：
  - add → get_role round-trip
  - remove 命中/未命中返回 True/False
  - 同 workspace 第 2 个 owner 触发 IntegrityError（partial unique）
  - 同 workspace 多个 admin 不触发（Stage 2 forward compat）
  - CASCADE: 删 user 自动清理 memberships
  - list_by_user 按 joined_at desc 排序（多 workspace）
  - role 校验拒绝 'viewer'
  - change_role 命中改值 + 未命中返 False

注：owner 转让需要事务内两行原子 swap（先把现 owner 改 admin，再把新 owner
插 owner），放 PR4+ auth router 内做带权限检查的版本；本仓储不暴露
transfer_ownership 方法以保持单一职责。

Stage 0 PR3 T3.7。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/packages/harness/deerflow/persistence/workspace_membership/__init__.py

@@ -12,5 +12,13 @@ unique index. Owner transfer is a two-row transactional swap.
 from __future__ import annotations
 
 from deerflow.persistence.workspace_membership.model import WorkspaceMembershipRow
+from deerflow.persistence.workspace_membership.sql import (
+    MembershipValidationError,
+    WorkspaceMembershipRepository,
+)
 
-__all__ = ["WorkspaceMembershipRow"]
+__all__ = [
+    "MembershipValidationError",
+    "WorkspaceMembershipRepository",
+    "WorkspaceMembershipRow",
+]

backend/packages/harness/deerflow/persistence/workspace_membership/sql.py

@@ -0,0 +1,149 @@
+"""SQLAlchemy-backed workspace membership repository.
+
+Manages the ``workspace_memberships`` join table. Stage 0 only writes
+``role='owner'`` (single-user workspaces); the repository accepts the
+full Stage 2 RBAC role enum so the schema is forward-compatible.
+
+Owner transfer is intentionally NOT modelled here as a single method —
+it requires a two-row transactional swap with careful retry semantics,
+and belongs in the auth router (PR4+) where it can be wrapped in a
+permission check.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import delete, select, update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.workspace_membership.model import WorkspaceMembershipRow
+
+# 允许的 role 字面值。Stage 0 实际仅写 owner；admin/member 留给 Stage 2 RBAC。
+_VALID_ROLES = frozenset({"owner", "admin", "member"})
+
+
+class MembershipValidationError(ValueError):
+    """Raised when role is not in the allowed enum."""
+
+
+def _validate_role(role: str) -> None:
+    if role not in _VALID_ROLES:
+        raise MembershipValidationError(f"role {role!r} is not in allowed set {_VALID_ROLES!r}")
+
+
+class WorkspaceMembershipRepository:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    @staticmethod
+    def _row_to_dict(row: WorkspaceMembershipRow) -> dict[str, Any]:
+        return {
+            "workspace_id": row.workspace_id,
+            "user_id": row.user_id,
+            "role": row.role,
+            "invited_by": row.invited_by,
+            "joined_at": row.joined_at.isoformat() if row.joined_at else None,
+        }
+
+    async def add(
+        self,
+        *,
+        workspace_id: str,
+        user_id: str,
+        role: str,
+        invited_by: str | None = None,
+    ) -> dict[str, Any]:
+        """Insert a new membership row.
+
+        Raises:
+          - :class:`MembershipValidationError` for unknown role values
+          - :class:`sqlalchemy.exc.IntegrityError` for:
+            - duplicate (workspace_id, user_id) — composite PK collision
+            - second ``owner`` in the same workspace — partial unique index
+            - invalid workspace_id / user_id / invited_by FK
+        """
+        _validate_role(role)
+        row = WorkspaceMembershipRow(
+            workspace_id=workspace_id,
+            user_id=user_id,
+            role=role,
+            invited_by=invited_by,
+            joined_at=datetime.now(UTC),
+        )
+        async with self._sf() as session:
+            session.add(row)
+            await session.commit()
+            await session.refresh(row)
+            return self._row_to_dict(row)
+
+    async def remove(self, *, workspace_id: str, user_id: str) -> bool:
+        """Delete a membership row; returns True if a row was deleted.
+
+        Owner removal is allowed at the repository layer — auth router
+        (PR4) layers the "cannot remove last owner" rule on top.
+        """
+        async with self._sf() as session:
+            result = await session.execute(
+                delete(WorkspaceMembershipRow).where(
+                    WorkspaceMembershipRow.workspace_id == workspace_id,
+                    WorkspaceMembershipRow.user_id == user_id,
+                )
+            )
+            await session.commit()
+            return (result.rowcount or 0) > 0
+
+    async def list_by_user(self, *, user_id: str) -> list[dict[str, Any]]:
+        """Return all memberships for ``user_id``, ordered by joined_at desc.
+
+        No contextvar resolution here — caller is responsible for passing
+        the correct user_id. Used by ``/auth/me`` to list workspaces a
+        user belongs to.
+        """
+        async with self._sf() as session:
+            result = await session.execute(select(WorkspaceMembershipRow).where(WorkspaceMembershipRow.user_id == user_id).order_by(WorkspaceMembershipRow.joined_at.desc()))
+            return [self._row_to_dict(r) for r in result.scalars()]
+
+    async def list_by_workspace(self, *, workspace_id: str) -> list[dict[str, Any]]:
+        """Return all members of a workspace, ordered by joined_at asc."""
+        async with self._sf() as session:
+            result = await session.execute(select(WorkspaceMembershipRow).where(WorkspaceMembershipRow.workspace_id == workspace_id).order_by(WorkspaceMembershipRow.joined_at.asc()))
+            return [self._row_to_dict(r) for r in result.scalars()]
+
+    async def get_role(self, *, workspace_id: str, user_id: str) -> str | None:
+        """Return the role string, or None if user is not a member."""
+        async with self._sf() as session:
+            result = await session.execute(
+                select(WorkspaceMembershipRow.role).where(
+                    WorkspaceMembershipRow.workspace_id == workspace_id,
+                    WorkspaceMembershipRow.user_id == user_id,
+                )
+            )
+            return result.scalar_one_or_none()
+
+    async def change_role(
+        self,
+        *,
+        workspace_id: str,
+        user_id: str,
+        new_role: str,
+    ) -> bool:
+        """Update a member's role; returns True iff a row was updated.
+
+        Validates ``new_role`` against the allowed enum. Owner-transfer
+        flow needs to swap two rows atomically — do that with a manual
+        transaction in the caller; this method is for non-owner changes.
+        """
+        _validate_role(new_role)
+        async with self._sf() as session:
+            result = await session.execute(
+                update(WorkspaceMembershipRow)
+                .where(
+                    WorkspaceMembershipRow.workspace_id == workspace_id,
+                    WorkspaceMembershipRow.user_id == user_id,
+                )
+                .values(role=new_role)
+            )
+            await session.commit()
+            return (result.rowcount or 0) > 0

backend/tests/test_workspace_membership_repo.py

@@ -0,0 +1,216 @@
+"""Tests for WorkspaceMembershipRepository (Stage 0 PR3 T3.7)."""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+from deerflow.persistence.user.model import UserRow
+from deerflow.persistence.workspace import WorkspaceRepository
+from deerflow.persistence.workspace_membership import (
+    MembershipValidationError,
+    WorkspaceMembershipRepository,
+)
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture
+def anyio_backend() -> str:
+    return "asyncio"
+
+
+async def _setup(tmp_path):
+    """Create both repos against a fresh SQLite DB."""
+    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))
+    sf = get_session_factory()
+    return (
+        WorkspaceRepository(sf),
+        WorkspaceMembershipRepository(sf),
+        sf,
+    )
+
+
+async def _cleanup():
+    from deerflow.persistence.engine import close_engine
+
+    await close_engine()
+
+
+async def _seed_user(sf, user_id: str, email: str) -> None:
+    async with sf() as session:
+        session.add(UserRow(id=user_id, email=email))
+        await session.commit()
+
+
+# ---------------------------------------------------------------------------
+# add / remove smoke
+# ---------------------------------------------------------------------------
+
+
+async def test_add_then_get_role(tmp_path):
+    ws_repo, m_repo, sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf, "u-1", "u1@example.com")
+        ws = await ws_repo.create(name="W1", slug="w-1", owner_id="u-1")
+
+        added = await m_repo.add(workspace_id=ws["id"], user_id="u-1", role="owner")
+        assert added["role"] == "owner"
+
+        role = await m_repo.get_role(workspace_id=ws["id"], user_id="u-1")
+        assert role == "owner"
+    finally:
+        await _cleanup()
+
+
+async def test_remove_returns_true_when_deleted_false_when_missing(tmp_path):
+    ws_repo, m_repo, sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf, "u-1", "u1@example.com")
+        ws = await ws_repo.create(name="W1", slug="rm-w", owner_id="u-1")
+        await m_repo.add(workspace_id=ws["id"], user_id="u-1", role="owner")
+
+        assert await m_repo.remove(workspace_id=ws["id"], user_id="u-1") is True
+        # second remove of same row -> nothing to delete
+        assert await m_repo.remove(workspace_id=ws["id"], user_id="u-1") is False
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# partial unique: exactly one owner per workspace
+# ---------------------------------------------------------------------------
+
+
+async def test_cannot_add_second_owner(tmp_path):
+    """Inserting a second role='owner' in the same workspace must raise IntegrityError."""
+    ws_repo, m_repo, sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf, "u-1", "u1@example.com")
+        await _seed_user(sf, "u-2", "u2@example.com")
+        ws = await ws_repo.create(name="W", slug="one-owner", owner_id="u-1")
+
+        await m_repo.add(workspace_id=ws["id"], user_id="u-1", role="owner")
+        with pytest.raises(IntegrityError):
+            await m_repo.add(workspace_id=ws["id"], user_id="u-2", role="owner")
+    finally:
+        await _cleanup()
+
+
+async def test_admin_and_member_dont_trigger_partial_unique(tmp_path):
+    """Multiple admin/member rows in one workspace are fine (Stage 2 forward compat)."""
+    ws_repo, m_repo, sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf, "u-1", "u1@example.com")
+        await _seed_user(sf, "u-2", "u2@example.com")
+        await _seed_user(sf, "u-3", "u3@example.com")
+        ws = await ws_repo.create(name="W", slug="multi-admin", owner_id="u-1")
+
+        await m_repo.add(workspace_id=ws["id"], user_id="u-1", role="owner")
+        await m_repo.add(workspace_id=ws["id"], user_id="u-2", role="admin")
+        # Second admin OK
+        await m_repo.add(workspace_id=ws["id"], user_id="u-3", role="admin")
+
+        members = await m_repo.list_by_workspace(workspace_id=ws["id"])
+        assert len(members) == 3
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# CASCADE: deleting a user wipes their memberships
+# ---------------------------------------------------------------------------
+
+
+async def test_cascade_delete_user_removes_memberships(tmp_path):
+    """FK ON DELETE CASCADE on user_id."""
+    from sqlalchemy import delete
+
+    ws_repo, m_repo, sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf, "u-keep", "keep@example.com")
+        await _seed_user(sf, "u-purge", "purge@example.com")
+        ws = await ws_repo.create(name="W", slug="cascade", owner_id="u-keep")
+        await m_repo.add(workspace_id=ws["id"], user_id="u-keep", role="owner")
+        await m_repo.add(workspace_id=ws["id"], user_id="u-purge", role="admin")
+
+        async with sf() as session:
+            await session.execute(delete(UserRow).where(UserRow.id == "u-purge"))
+            await session.commit()
+
+        members = await m_repo.list_by_workspace(workspace_id=ws["id"])
+        member_ids = [m["user_id"] for m in members]
+        assert "u-purge" not in member_ids
+        assert "u-keep" in member_ids
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# list_by_user ordering (most recent joined_at first)
+# ---------------------------------------------------------------------------
+
+
+async def test_list_by_user_orders_recent_first(tmp_path):
+    import asyncio
+
+    ws_repo, m_repo, sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf, "u-1", "u1@example.com")
+        await _seed_user(sf, "u-2", "u2@example.com")
+
+        # u-1 owns workspace A
+        ws_a = await ws_repo.create(name="A", slug="ord-a", owner_id="u-1")
+        await m_repo.add(workspace_id=ws_a["id"], user_id="u-1", role="owner")
+        await asyncio.sleep(0.01)  # ensure distinct joined_at
+
+        # u-1 later joins workspace C as a member (u-2 owns it)
+        ws_c = await ws_repo.create(name="C", slug="ord-c", owner_id="u-2")
+        await m_repo.add(workspace_id=ws_c["id"], user_id="u-1", role="member")
+
+        memberships = await m_repo.list_by_user(user_id="u-1")
+        slugs_in_order = [(m["workspace_id"], m["role"]) for m in memberships]
+        # 'ws_c member' joined AFTER 'ws_a owner' → ws_c first
+        assert slugs_in_order[0] == (ws_c["id"], "member")
+        assert slugs_in_order[1] == (ws_a["id"], "owner")
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# role validation + change_role
+# ---------------------------------------------------------------------------
+
+
+async def test_add_rejects_unknown_role(tmp_path):
+    ws_repo, m_repo, sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf, "u-1", "u1@example.com")
+        ws = await ws_repo.create(name="W", slug="invrole", owner_id="u-1")
+        with pytest.raises(MembershipValidationError, match="allowed set"):
+            await m_repo.add(workspace_id=ws["id"], user_id="u-1", role="viewer")
+    finally:
+        await _cleanup()
+
+
+async def test_change_role_admin_to_member(tmp_path):
+    ws_repo, m_repo, sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf, "u-1", "u1@example.com")
+        await _seed_user(sf, "u-2", "u2@example.com")
+        ws = await ws_repo.create(name="W", slug="chg", owner_id="u-1")
+        await m_repo.add(workspace_id=ws["id"], user_id="u-1", role="owner")
+        await m_repo.add(workspace_id=ws["id"], user_id="u-2", role="admin")
+
+        ok = await m_repo.change_role(workspace_id=ws["id"], user_id="u-2", new_role="member")
+        assert ok is True
+        assert await m_repo.get_role(workspace_id=ws["id"], user_id="u-2") == "member"
+
+        # change_role on a non-member returns False
+        miss = await m_repo.change_role(workspace_id=ws["id"], user_id="u-nonexistent", new_role="member")
+        assert miss is False
+    finally:
+        await _cleanup()