feat(persistence): PR8 T8.5 — ExternalUserRow ORM + 2 tests

Adds deerflow/persistence/external_user/{__init__,model}.py with
ExternalUserRow:
  - id: UUID36 PK
  - workspace_id: FK workspaces ON DELETE CASCADE (redundant with SA's
    workspace_id but stored directly to speed workspace-scoped queries
    that span multiple SAs)
  - service_account_id: FK service_accounts ON DELETE CASCADE
  - external_id: String(128) — caller-supplied X-External-User-Id
  - display_name: String(128) nullable (admin UI only, not auth-relevant)
  - metadata_json: JSON nullable=False default {} — plan tier / region /
    custom tags
  - created_at / last_seen_at (UTC)
  - UniqueConstraint (service_account_id, external_id)
    name=uq_external_users_sa_external — the same external_id may be
    reused under a different SA, but is upsert-unique under a single SA

T8.5 tests:
  - test_unique_service_account_id_plus_external_id: second row with
    same (SA, external_id) raises IntegrityError
  - test_cascade_on_service_account_delete: deleting parent SA removes
    all external_users rows under it

Registered in deerflow/persistence/models/__init__.py — all three PR8
tables (service_accounts / api_keys / external_users) are now wired
into Base.metadata.create_all().

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

backend/tests/test_external_user_schema.py

@@ -0,0 +1,134 @@
+"""Schema tests for ``ExternalUserRow`` (Stage 0 PR8).
+
+PR8 is schema-only — no repository class, no API. Tests exercise raw
+ORM behaviour: the composite UNIQUE constraint (service_account_id,
+external_id) and CASCADE on service_account delete.
+
+An external user is the end-user identity passed through by a
+service_account whose ``identity_mode`` is ``external_passthrough`` or
+``both``: every call carries an ``X-External-User-Id`` header which is
+upserted into this table for audit / quota attribution.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+from sqlalchemy import delete
+from sqlalchemy.exc import IntegrityError
+
+from deerflow.persistence.external_user import ExternalUserRow
+from deerflow.persistence.service_account import ServiceAccountRow
+from deerflow.persistence.user.model import UserRow
+from deerflow.persistence.workspace.model import WorkspaceRow
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture
+def anyio_backend() -> str:
+    return "asyncio"
+
+
+async def _setup(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 get_session_factory()
+
+
+async def _cleanup():
+    from deerflow.persistence.engine import close_engine
+
+    await close_engine()
+
+
+async def _seed_parents(sf) -> None:
+    now = datetime.now(UTC)
+    async with sf() as session:
+        session.add(UserRow(id="u-alice", email="alice@example.com"))
+        await session.commit()
+    async with sf() as session:
+        session.add(WorkspaceRow(id="w-1", name="Alice WS", slug="alice", owner_id="u-alice"))
+        await session.commit()
+    async with sf() as session:
+        session.add(
+            ServiceAccountRow(
+                id="sa-1",
+                workspace_id="w-1",
+                name="passthrough-bot",
+                role="member",
+                identity_mode="external_passthrough",
+                status="active",
+                created_by="u-alice",
+                created_at=now,
+                updated_at=now,
+            )
+        )
+        await session.commit()
+
+
+def _make_external_user(*, eu_id: str, external_id: str) -> ExternalUserRow:
+    now = datetime.now(UTC)
+    return ExternalUserRow(
+        id=eu_id,
+        workspace_id="w-1",
+        service_account_id="sa-1",
+        external_id=external_id,
+        display_name=None,
+        metadata_json={},
+        created_at=now,
+        last_seen_at=None,
+    )
+
+
+# ---------------------------------------------------------------------------
+# T8.5-1 — UNIQUE (service_account_id, external_id)
+# ---------------------------------------------------------------------------
+
+
+async def test_unique_service_account_id_plus_external_id(tmp_path):
+    """The same external_id may be inserted twice only under different SAs."""
+    sf = await _setup(tmp_path)
+    try:
+        await _seed_parents(sf)
+        async with sf() as session:
+            session.add(_make_external_user(eu_id="eu-1", external_id="client-42"))
+            await session.commit()
+
+        with pytest.raises(IntegrityError):
+            async with sf() as session:
+                session.add(_make_external_user(eu_id="eu-2", external_id="client-42"))
+                await session.commit()
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# T8.5-2 — CASCADE on service_account delete
+# ---------------------------------------------------------------------------
+
+
+async def test_cascade_on_service_account_delete(tmp_path):
+    """Deleting the parent service_account removes all child external_users rows."""
+    sf = await _setup(tmp_path)
+    try:
+        await _seed_parents(sf)
+        async with sf() as session:
+            session.add(_make_external_user(eu_id="eu-c1", external_id="endpoint-A"))
+            session.add(_make_external_user(eu_id="eu-c2", external_id="endpoint-B"))
+            await session.commit()
+
+        async with sf() as session:
+            await session.execute(delete(ServiceAccountRow).where(ServiceAccountRow.id == "sa-1"))
+            await session.commit()
+
+        async with sf() as session:
+            row1 = await session.get(ExternalUserRow, "eu-c1")
+            row2 = await session.get(ExternalUserRow, "eu-c2")
+        assert row1 is None
+        assert row2 is None
+    finally:
+        await _cleanup()