chore(scripts): add verify_stage0.sh — systematic Stage 0 verification

Adds a single script that walks the 5 verification layers in order
and emits pass/fail counts at the end:

  static   — boundary scans + full pytest (baseline ≥ 3250 passed,
             ≤ 18 fails matching known caplog flake set) + ruff lint
  paths    — .deer-flow/users/ should be empty (or absent);
             workspaces/{wid}/threads/{tid}/... layout in place;
             exercises `make migrate-paths DRY_RUN=1`
  rds      — PG reachable; alembic at 0003; service_accounts +
             api_keys + external_users tables present;
             idx_api_keys_active partial index has
             "WHERE revoked_at IS NULL" predicate; workspace_id is
             NOT NULL on thread_meta / runs / feedback / run_events;
             UNIQUE(workspace_id, thread_id) on thread_meta
             (requires DATABASE_URL; skipped if unset)
  runtime  — curl /health on the Gateway (requires `make dev`;
             skips downstream e2e if unreachable)
  e2e      — register two users via /api/auth/register, capture each
             session's csrf_token, create one thread per user,
             cross-access GET + DELETE both return 404 (per PR6
             "404 not 403" contract), same-workspace GET returns 200

Each layer is independently runnable: `./verify_stage0.sh paths rds`.
With no args runs all five.

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

backend/CLAUDE.md

@@ -102,6 +102,8 @@ Regression tests related to Docker/provisioner behavior:
 
 Boundary check (harness → app import firewall):
 - `tests/test_harness_boundary.py` — ensures `packages/harness/deerflow/` never imports from `app.*`
+- `tests/test_workspace_boundary.py` — AST static scan that forbids direct imports of `langgraph.checkpoint.*` (and the third-party `langgraph_checkpoint_postgres` / `langgraph_checkpoint_sqlite` packages) outside the allowlist in `tests/boundary_allowlist.toml`. Everywhere else must obtain a checkpointer via `app.gateway.deps.get_checkpointer` or the harness `deerflow.runtime.checkpointer` factory. Imports inside `if TYPE_CHECKING:` blocks are exempt automatically (they do not enter runtime). When a legitimate new importer is genuinely needed, append its path to `boundary_allowlist.toml` in the same PR
+- `tests/test_workspace_boundary_self.py` — self-tests for the scanner above (9 cases over synthetic `.py` files) guarding against silent-empty regressions
 
 CI runs these regression tests for every pull request via [.github/workflows/backend-unit-tests.yml](../.github/workflows/backend-unit-tests.yml).
 

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

@@ -0,0 +1,18 @@
+"""API key persistence — ORM model only (Stage 0 PR8).
+
+An API key is the credential a service_account uses to call the
+headless API. Each key has a public ``key_prefix`` (printed in audit
+logs and used for quick lookup) and a ``key_hash`` (sha-256 of the
+plaintext token, never reversed). Plaintext tokens are only ever
+returned to the caller at create time.
+
+PR8 introduces only the schema + ORM row class. Token generation,
+hashing, scope parsing, rate limiting, and the API-key auth
+middleware live in Stage 1 alongside the headless API surface.
+"""
+
+from __future__ import annotations
+
+from deerflow.persistence.api_key.model import ApiKeyRow
+
+__all__ = ["ApiKeyRow"]

backend/packages/harness/deerflow/persistence/api_key/model.py

@@ -0,0 +1,94 @@
+"""ORM model for API keys (credentials owned by a service account)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+
+
+class ApiKeyRow(Base):
+    __tablename__ = "api_keys"
+
+    id: Mapped[str] = mapped_column(
+        String(36),
+        primary_key=True,
+        comment="API key 主键，UUID 字符串（36 字符）",
+    )
+    service_account_id: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("service_accounts.id", ondelete="CASCADE"),
+        nullable=False,
+        comment="所属 service_account；service_account 删除时级联清掉所有 api_key",
+    )
+    key_prefix: Mapped[str] = mapped_column(
+        String(16),
+        nullable=False,
+        unique=True,
+        comment="公开 prefix（如 'dfk_live_abc12345'），可在审计日志 / UI 中打印；全局唯一，撤销后亦不复用以避免审计混淆",
+    )
+    key_hash: Mapped[str] = mapped_column(
+        String(128),
+        nullable=False,
+        comment="完整 token 的 sha-256 hex（64 字符；预留 128 以兼容未来更长哈希），plaintext token 仅在创建时返回给调用方",
+    )
+    name: Mapped[str] = mapped_column(
+        String(64),
+        nullable=False,
+        comment="key 的人类可读标签（如 'ci pipeline' / 'frontend prod'），同 service_account 内不强制唯一",
+    )
+    scopes: Mapped[str] = mapped_column(
+        String(1024),
+        nullable=False,
+        default="",
+        comment="scope 列表，逗号分隔字符串（如 'threads:read,threads:write'）；用 String 而非 PG text[] 以保 SQLite dev 兼容，Stage 2 切纯 PG 后可平滑迁",
+    )
+    rate_limit_rpm: Mapped[int | None] = mapped_column(
+        Integer,
+        nullable=True,
+        comment="每分钟请求数限制；NULL 表示走该 service_account 的默认限速（Stage 1 起生效）",
+    )
+    expires_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+        comment="过期时间（UTC）；NULL = 不过期",
+    )
+    last_used_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+        comment="最近一次成功鉴权时间（UTC）；Stage 1 鉴权中间件每次更新",
+    )
+    revoked_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+        comment="撤销时间（UTC）；NULL = 仍然有效。被撤销的 key 不删行（保留审计），但鉴权层据此拒绝",
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+        comment="创建时间（UTC）",
+    )
+
+    __table_args__ = (
+        Index("idx_api_keys_sa", "service_account_id"),
+        # 部分索引：只索引活跃 key（revoked_at IS NULL），鉴权热路径走 prefix lookup，
+        # 撤销后的 key 不进活跃索引以减小热索引大小。SQLite + Postgres 均支持
+        # WHERE 子句的部分索引；双驱动维护两套等价 where 表达式。
+        Index(
+            "idx_api_keys_active",
+            "key_prefix",
+            sqlite_where=text("revoked_at IS NULL"),
+            postgresql_where=text("revoked_at IS NULL"),
+        ),
+        {
+            "comment": (
+                "API key 表（headless API 凭证）。每行属于唯一 service_account；key_prefix 全局唯一可在日志中打印，"
+                "key_hash 是完整 token 的 sha-256，plaintext token 只在创建时返给调用方。撤销保留行（revoked_at 非空），"
+                "活跃 key 走部分索引 idx_api_keys_active 加速鉴权热路径。Stage 0 仅落 schema；Stage 1 起接鉴权 + 限速。"
+            )
+        },
+    )

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

@@ -0,0 +1,16 @@
+"""External user persistence — ORM model only (Stage 0 PR8).
+
+An external user represents the end-user identity that a
+service_account passes through on each call (typically via an
+``X-External-User-Id`` header). The row is upserted each time a
+new ``external_id`` is seen under a given service_account.
+
+PR8 introduces only the schema + ORM row class. The upsert logic,
+header parsing, and quota attribution all live in Stage 1.
+"""
+
+from __future__ import annotations
+
+from deerflow.persistence.external_user.model import ExternalUserRow
+
+__all__ = ["ExternalUserRow"]

backend/packages/harness/deerflow/persistence/external_user/model.py

@@ -0,0 +1,71 @@
+"""ORM model for external users (end-user identities passed through a service account)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import JSON, DateTime, ForeignKey, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+
+
+class ExternalUserRow(Base):
+    __tablename__ = "external_users"
+
+    id: Mapped[str] = mapped_column(
+        String(36),
+        primary_key=True,
+        comment="external_user 主键，UUID 字符串（36 字符）",
+    )
+    workspace_id: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("workspaces.id", ondelete="CASCADE"),
+        nullable=False,
+        comment="所属 workspace（冗余存储——可经 service_account 间接得到，但直接存以加速 workspace-scope 查询）；workspace 删除时级联",
+    )
+    service_account_id: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("service_accounts.id", ondelete="CASCADE"),
+        nullable=False,
+        comment="passthrough 的 service_account；service_account 删除时级联",
+    )
+    external_id: Mapped[str] = mapped_column(
+        String(128),
+        nullable=False,
+        comment="终端调用方传入的 X-External-User-Id（最多 128 字符；推荐 UUID / opaque token，不要塞 PII）",
+    )
+    display_name: Mapped[str | None] = mapped_column(
+        String(128),
+        nullable=True,
+        comment="可选显示名（如 'alice@customer.com'）；仅用于 admin UI 展示，不参与鉴权",
+    )
+    metadata_json: Mapped[dict[str, Any]] = mapped_column(
+        JSON,
+        nullable=False,
+        default=dict,
+        comment="任意 JSON 附属信息（plan tier / region / 自定义 tag）；Stage 1 由 upsert 调用方写入",
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+        comment="首次见到该 external_id 的时间（UTC）",
+    )
+    last_seen_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+        comment="最近一次该 external_id 触发请求的时间（UTC）；Stage 1 鉴权层每次更新",
+    )
+
+    __table_args__ = (
+        UniqueConstraint("service_account_id", "external_id", name="uq_external_users_sa_external"),
+        {
+            "comment": (
+                "终端用户身份表（passthrough 模式下的 end-user）。每行由 service_account 的鉴权中间件 upsert——同一 "
+                "(service_account_id, external_id) 组合只存一行。workspace_id 冗余存储以加速跨 SA 的 workspace-scope 聚合查询。"
+                "Stage 0 仅落 schema；Stage 1 起接 upsert / 配额聚合。"
+            )
+        },
+    )

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

@@ -10,24 +10,33 @@ The actual ORM classes have moved to entity-specific subpackages:
 - ``deerflow.persistence.user``
 - ``deerflow.persistence.workspace``  (Stage 0 PR3)
 - ``deerflow.persistence.workspace_membership``  (Stage 0 PR3)
+- ``deerflow.persistence.service_account``  (Stage 0 PR8)
+- ``deerflow.persistence.api_key``  (Stage 0 PR8)
+- ``deerflow.persistence.external_user``  (Stage 0 PR8)
 
 ``RunEventRow`` remains in ``deerflow.persistence.models.run_event`` because
 its storage implementation lives in ``deerflow.runtime.events.store.db`` and
 there is no matching entity directory.
 """
 
+from deerflow.persistence.api_key.model import ApiKeyRow
+from deerflow.persistence.external_user.model import ExternalUserRow
 from deerflow.persistence.feedback.model import FeedbackRow
 from deerflow.persistence.models.run_event import RunEventRow
 from deerflow.persistence.run.model import RunRow
+from deerflow.persistence.service_account.model import ServiceAccountRow
 from deerflow.persistence.thread_meta.model import ThreadMetaRow
 from deerflow.persistence.user.model import UserRow
 from deerflow.persistence.workspace.model import WorkspaceRow
 from deerflow.persistence.workspace_membership.model import WorkspaceMembershipRow
 
 __all__ = [
+    "ApiKeyRow",
+    "ExternalUserRow",
     "FeedbackRow",
     "RunEventRow",
     "RunRow",
+    "ServiceAccountRow",
     "ThreadMetaRow",
     "UserRow",
     "WorkspaceMembershipRow",

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

@@ -0,0 +1,17 @@
+"""Service account persistence — ORM model only (Stage 0 PR8).
+
+A service account is a non-human principal that lives inside a workspace
+and authenticates via API keys rather than email + password. Each
+service account belongs to exactly one workspace and is created by a
+human user (``created_by``).
+
+PR8 introduces only the schema + ORM row class. Repository, API-key
+authentication middleware, and the ``@require_permission`` scope
+upgrade live in Stage 1 alongside the headless API surface.
+"""
+
+from __future__ import annotations
+
+from deerflow.persistence.service_account.model import ServiceAccountRow
+
+__all__ = ["ServiceAccountRow"]

backend/packages/harness/deerflow/persistence/service_account/model.py

@@ -0,0 +1,83 @@
+"""ORM model for service accounts (non-human principals inside a workspace)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import DateTime, ForeignKey, Index, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+
+
+class ServiceAccountRow(Base):
+    __tablename__ = "service_accounts"
+
+    id: Mapped[str] = mapped_column(
+        String(36),
+        primary_key=True,
+        comment="服务账号主键，UUID 字符串（36 字符），与 users.id 类型对齐",
+    )
+    workspace_id: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("workspaces.id", ondelete="CASCADE"),
+        nullable=False,
+        comment="所属 workspace；workspace 删除时级联清掉所有 service_account（连同其 api_keys / external_users）",
+    )
+    name: Mapped[str] = mapped_column(
+        String(64),
+        nullable=False,
+        comment="服务账号显示名（同 workspace 内不强制唯一；Stage 1 可由 admin UI 重复使用同名 + 不同 key）",
+    )
+    role: Mapped[str] = mapped_column(
+        String(16),
+        nullable=False,
+        default="member",
+        comment='服务账号在 workspace 内的角色字符串：Stage 0 仅支持 "member"；Stage 2 RBAC 打开 "admin"/"viewer"。用 String(16) 而非 enum 以便未来扩枚举值不动 schema',
+    )
+    identity_mode: Mapped[str] = mapped_column(
+        String(16),
+        nullable=False,
+        default="collapsed",
+        comment=(
+            '身份模式三态："collapsed"（所有调用 collapse 到该 service_account；不记录 external_user）/'
+            ' "external_passthrough"（每次调用必带 X-External-User-Id，写入 external_users 表）/'
+            ' "both"（带就写、不带就 collapse）。Stage 1 API key 鉴权层据此分流'
+        ),
+    )
+    status: Mapped[str] = mapped_column(
+        String(16),
+        nullable=False,
+        default="active",
+        comment='状态："active"（正常）/ "suspended"（admin 暂停）/ "deleted"（软删；保留审计）',
+    )
+    created_by: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("users.id", ondelete="RESTRICT"),
+        nullable=False,
+        comment="创建者 user_id；删除该 user 时 RESTRICT 阻拦（必须先转移或删除该 user 名下所有 service_account）",
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+        comment="创建时间（UTC）",
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+        onupdate=lambda: datetime.now(UTC),
+        comment="最近更新时间（UTC，写入时自动更新）",
+    )
+
+    __table_args__ = (
+        Index("idx_service_accounts_workspace", "workspace_id", "status"),
+        {
+            "comment": (
+                "服务账号表（headless API 的非人身份）。每个 service_account 属于唯一 workspace；"
+                "通过 api_keys 表的 API key 鉴权调用 Gateway；identity_mode 控制是否在 external_users 表"
+                "记录终端用户身份。Stage 0 仅落 schema；Stage 1 起接 API key 鉴权 + 路由 scope 升级。"
+            )
+        },
+    )

backend/scripts/verify_stage0.sh

@@ -0,0 +1,420 @@
+#!/usr/bin/env bash
+# verify_stage0.sh — systematic verification of the Stage 0 multi-tenant rollout.
+#
+# Layers (each can be run individually; defaults run all that are applicable):
+#   static   — full pytest suite + lint + boundary scan (no external deps)
+#   paths    — local filesystem layout: legacy users/ should be empty after migration
+#   rds      — schema + index + alembic state on the remote Postgres (needs DATABASE_URL)
+#   runtime  — gateway health probes (needs `make dev` running)
+#   e2e      — register 2 users via curl, each creates a thread, cross-access must 404
+#
+# Usage:
+#   ./scripts/verify_stage0.sh                 # run everything that has prerequisites
+#   ./scripts/verify_stage0.sh static rds      # only those two
+#   DATABASE_URL=postgres://... ./scripts/verify_stage0.sh
+#   GATEWAY_URL=http://localhost:8001 ./scripts/verify_stage0.sh runtime e2e
+#
+# Exit code: 0 if every executed assertion passes, 1 if any fail.
+
+set -uo pipefail
+
+# ── colours ──────────────────────────────────────────────────────────────
+if [ -t 1 ]; then
+    C_RED=$'\033[0;31m'; C_GREEN=$'\033[0;32m'; C_YELLOW=$'\033[0;33m'
+    C_BLUE=$'\033[0;34m'; C_BOLD=$'\033[1m'; C_DIM=$'\033[2m'; C_RESET=$'\033[0m'
+else
+    C_RED=''; C_GREEN=''; C_YELLOW=''; C_BLUE=''; C_BOLD=''; C_DIM=''; C_RESET=''
+fi
+
+# ── counters ─────────────────────────────────────────────────────────────
+PASS_COUNT=0
+FAIL_COUNT=0
+WARN_COUNT=0
+FAILED_STEPS=()
+SKIPPED_PHASES=()
+
+ok()   { printf '  %s✓%s %s\n' "$C_GREEN" "$C_RESET" "$1"; PASS_COUNT=$((PASS_COUNT+1)); }
+fail() { printf '  %s✗%s %s\n' "$C_RED"   "$C_RESET" "$1"; FAIL_COUNT=$((FAIL_COUNT+1)); FAILED_STEPS+=("$1"); }
+warn() { printf '  %s!%s %s\n' "$C_YELLOW" "$C_RESET" "$1"; WARN_COUNT=$((WARN_COUNT+1)); }
+info() { printf '  %s·%s %s\n' "$C_DIM"   "$C_RESET" "$1"; }
+phase() { printf '\n%s== %s ==%s\n' "$C_BLUE$C_BOLD" "$1" "$C_RESET"; }
+
+# Run-cmd helpers — capture both streams for grep but return original exit code.
+run() {
+    local label="$1"; shift
+    local out
+    out=$("$@" 2>&1)
+    local rc=$?
+    if [ "$rc" -eq 0 ]; then
+        ok "$label"
+        printf '%s' "$out"
+        return 0
+    fi
+    fail "$label (exit $rc)"
+    printf '%s%s%s\n' "$C_DIM" "$out" "$C_RESET" >&2
+    return "$rc"
+}
+
+# ── env ──────────────────────────────────────────────────────────────────
+REPO_ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
+BACKEND_DIR="$REPO_ROOT/backend"
+GATEWAY_URL="${GATEWAY_URL:-http://localhost:8001}"
+DEER_FLOW_HOME="${DEER_FLOW_HOME:-$REPO_ROOT/.deer-flow}"
+
+# ── phase: static ────────────────────────────────────────────────────────
+phase_static() {
+    phase "STATIC — unit + boundary + full pytest"
+    cd "$BACKEND_DIR"
+
+    # boundary scans first — fast and high-signal
+    info "running boundary scans (harness + workspace)"
+    if PYTHONPATH=. uv run pytest -q \
+            tests/test_harness_boundary.py \
+            tests/test_workspace_boundary.py \
+            tests/test_workspace_boundary_self.py >/tmp/verify_boundary.log 2>&1; then
+        local boundary_passed
+        boundary_passed=$(grep -Eo '[0-9]+ passed' /tmp/verify_boundary.log | head -1)
+        ok "boundary scans green ($boundary_passed)"
+    else
+        fail "boundary scans red — see /tmp/verify_boundary.log"
+    fi
+
+    info "running full pytest (this takes ~90-130s)"
+    local pytest_log=/tmp/verify_full_pytest.log
+    PYTHONPATH=. uv run pytest -q >"$pytest_log" 2>&1
+    local rc=$?
+    local last_line
+    last_line=$(tail -1 "$pytest_log")
+    info "result: $last_line"
+    # Parse "<X> passed, <Y> failed, <Z> skipped"
+    local passed_n failed_n
+    passed_n=$(printf '%s' "$last_line" | grep -Eo '[0-9]+ passed' | grep -Eo '[0-9]+' | head -1 || echo 0)
+    failed_n=$(printf '%s' "$last_line" | grep -Eo '[0-9]+ failed' | grep -Eo '[0-9]+' | head -1 || echo 0)
+    if [ "${passed_n:-0}" -ge 3250 ]; then
+        ok "pytest pass count $passed_n ≥ 3250 (PR8 baseline)"
+    else
+        fail "pytest pass count $passed_n < 3250 — regression suspected"
+    fi
+    if [ "${failed_n:-0}" -le 18 ]; then
+        ok "pytest fail count $failed_n ≤ 18 (Stage 0 known-flake ceiling)"
+        [ "${failed_n:-0}" -gt 0 ] && warn "non-zero failures expected to be in the 18 caplog flake set; cross-check tail of /tmp/verify_full_pytest.log"
+    else
+        fail "pytest fail count $failed_n > 18 — new failures introduced beyond known caplog flake set"
+    fi
+
+    info "running ruff lint"
+    if make lint >/tmp/verify_lint.log 2>&1; then
+        ok "ruff lint clean"
+    else
+        fail "ruff lint dirty — see /tmp/verify_lint.log"
+    fi
+}
+
+# ── phase: paths ─────────────────────────────────────────────────────────
+phase_paths() {
+    phase "PATHS — legacy users/ migration state"
+    if [ ! -d "$DEER_FLOW_HOME" ]; then
+        warn "DEER_FLOW_HOME ($DEER_FLOW_HOME) does not exist — fresh install, nothing to migrate"
+        return
+    fi
+
+    info "DEER_FLOW_HOME = $DEER_FLOW_HOME"
+    local legacy_dir="$DEER_FLOW_HOME/users"
+    if [ ! -d "$legacy_dir" ]; then
+        ok "no legacy users/ directory present (PR6 migration not needed or already done)"
+    else
+        local count
+        count=$(find "$legacy_dir" -mindepth 1 -maxdepth 1 -type d 2>/dev/null | wc -l | tr -d ' ')
+        if [ "$count" -eq 0 ]; then
+            ok "legacy users/ is empty (migration complete or no prior data)"
+        else
+            warn "legacy users/ still has $count user dir(s) — run \`make migrate-paths\` (after \`make migrate-paths DRY_RUN=1\` to preview)"
+        fi
+    fi
+
+    local workspace_dir="$DEER_FLOW_HOME/workspaces"
+    if [ -d "$workspace_dir" ]; then
+        local wcount
+        wcount=$(find "$workspace_dir" -mindepth 1 -maxdepth 1 -type d 2>/dev/null | wc -l | tr -d ' ')
+        ok "workspaces/ layout in place ($wcount workspace dir(s))"
+    else
+        warn "workspaces/ dir does not exist yet — will be created when the first thread is opened"
+    fi
+
+    # Stage 0 PR6 dry-run check: just exercise the script flag, do not perform writes.
+    info "exercising migrate-paths dry-run (no writes)"
+    cd "$REPO_ROOT"
+    if make migrate-paths DRY_RUN=1 >/tmp/verify_migrate_dry.log 2>&1; then
+        ok "make migrate-paths DRY_RUN=1 ran without error"
+    else
+        fail "make migrate-paths DRY_RUN=1 failed — see /tmp/verify_migrate_dry.log"
+    fi
+}
+
+# ── phase: rds ───────────────────────────────────────────────────────────
+phase_rds() {
+    phase "RDS — schema + alembic + indexes"
+    if [ -z "${DATABASE_URL:-}" ]; then
+        warn "DATABASE_URL not set — skipping RDS phase"
+        SKIPPED_PHASES+=("rds (DATABASE_URL unset)")
+        return
+    fi
+    if ! command -v psql >/dev/null 2>&1; then
+        fail "psql not installed — cannot run RDS phase"
+        return
+    fi
+    info "DATABASE_URL host: $(printf '%s' "$DATABASE_URL" | sed -E 's#.*@([^/]+)/.*#\1#')"
+
+    # PG version sanity
+    local pg_version
+    pg_version=$(psql "$DATABASE_URL" -tA -c "SELECT version()" 2>/dev/null | head -1)
+    if [ -z "$pg_version" ]; then
+        fail "cannot connect to RDS — check DATABASE_URL"
+        return
+    fi
+    ok "PG reachable: $pg_version"
+
+    # Alembic head
+    info "checking alembic head against expected 0003"
+    cd "$BACKEND_DIR"
+    if PYTHONPATH=. uv run alembic current 2>/tmp/verify_alembic.err | tee /tmp/verify_alembic.log | grep -q "^0003"; then
+        ok "alembic current is at 0003 (workspace_id NOT NULL + UNIQUE(wid, tid))"
+    else
+        fail "alembic current is not at 0003 — see /tmp/verify_alembic.log"
+    fi
+    cd "$REPO_ROOT"
+
+    # PR8 tables present
+    info "checking PR8 tables exist"
+    local pr8_count
+    pr8_count=$(psql "$DATABASE_URL" -tA -c "
+        SELECT count(*) FROM information_schema.tables
+        WHERE table_schema = current_schema()
+          AND table_name IN ('service_accounts','api_keys','external_users');
+    " 2>/dev/null | tr -d ' ')
+    if [ "${pr8_count:-0}" -eq 3 ]; then
+        ok "service_accounts + api_keys + external_users all present"
+    else
+        fail "PR8 tables missing — expected 3, got ${pr8_count:-0}"
+    fi
+
+    # Partial index on api_keys: WHERE revoked_at IS NULL
+    info "checking idx_api_keys_active partial-index predicate"
+    local idx_def
+    idx_def=$(psql "$DATABASE_URL" -tA -c "
+        SELECT indexdef FROM pg_indexes
+        WHERE schemaname = current_schema()
+          AND indexname = 'idx_api_keys_active';
+    " 2>/dev/null | head -1)
+    if printf '%s' "$idx_def" | grep -qi 'WHERE.*revoked_at IS NULL'; then
+        ok "idx_api_keys_active has 'WHERE revoked_at IS NULL' predicate"
+    else
+        fail "idx_api_keys_active missing or wrong predicate — got: ${idx_def:-<none>}"
+    fi
+
+    # 4 business tables workspace_id NOT NULL
+    info "checking workspace_id NOT NULL on 4 business tables"
+    local nullable_rows
+    nullable_rows=$(psql "$DATABASE_URL" -tA -c "
+        SELECT table_name FROM information_schema.columns
+        WHERE table_schema = current_schema()
+          AND column_name = 'workspace_id'
+          AND table_name IN ('thread_meta','runs','feedback','run_events')
+          AND is_nullable = 'YES';
+    " 2>/dev/null)
+    if [ -z "$nullable_rows" ]; then
+        ok "thread_meta + runs + feedback + run_events all have workspace_id NOT NULL"
+    else
+        fail "workspace_id is nullable in: $(printf '%s' "$nullable_rows" | tr '\n' ' ')"
+    fi
+
+    # Unique (workspace_id, thread_id) on thread_meta
+    info "checking UNIQUE(workspace_id, thread_id) on thread_meta"
+    local uq_count
+    uq_count=$(psql "$DATABASE_URL" -tA -c "
+        SELECT count(*) FROM pg_indexes
+        WHERE schemaname = current_schema()
+          AND tablename = 'thread_meta'
+          AND indexdef ILIKE '%UNIQUE%workspace_id%thread_id%';
+    " 2>/dev/null | tr -d ' ')
+    if [ "${uq_count:-0}" -ge 1 ]; then
+        ok "UNIQUE(workspace_id, thread_id) constraint/index present"
+    else
+        fail "no UNIQUE(workspace_id, thread_id) on thread_meta"
+    fi
+}
+
+# ── phase: runtime ───────────────────────────────────────────────────────
+phase_runtime() {
+    phase "RUNTIME — gateway health"
+    if ! command -v curl >/dev/null 2>&1; then
+        fail "curl missing — cannot run runtime phase"
+        return
+    fi
+    info "probing $GATEWAY_URL/health"
+    local health
+    health=$(curl -sS -o /dev/null -w '%{http_code}' --max-time 5 "$GATEWAY_URL/health" 2>/dev/null || true)
+    if [ "$health" = "200" ]; then
+        ok "gateway /health returns 200"
+    else
+        fail "gateway /health returned '$health' — is \`make dev\` running?"
+        SKIPPED_PHASES+=("e2e (gateway not reachable)")
+        SKIP_E2E=1
+    fi
+}
+
+# ── phase: e2e ───────────────────────────────────────────────────────────
+phase_e2e() {
+    phase "E2E — register 2 users + cross-workspace 404"
+    if [ "${SKIP_E2E:-0}" = "1" ]; then
+        warn "skipped because gateway probe failed"
+        return
+    fi
+    if ! command -v curl >/dev/null 2>&1; then
+        fail "curl missing"
+        return
+    fi
+
+    local tag
+    tag=$(date +%s)
+    local alice="alice-${tag}@verify.local"
+    local bob="bob-${tag}@verify.local"
+    local pw="VerifyStage0_${tag}"
+    local jar_a=/tmp/verify_alice_${tag}.cookies
+    local jar_b=/tmp/verify_bob_${tag}.cookies
+    rm -f "$jar_a" "$jar_b"
+
+    register_user() {
+        local jar="$1"; local email="$2"
+        curl -sS -c "$jar" -o /tmp/verify_register_$$.json -w '%{http_code}' \
+            -H 'Content-Type: application/json' \
+            -d "{\"email\":\"$email\",\"password\":\"$pw\"}" \
+            "$GATEWAY_URL/api/auth/register"
+    }
+
+    info "registering Alice + Bob"
+    local code_a code_b
+    code_a=$(register_user "$jar_a" "$alice")
+    code_b=$(register_user "$jar_b" "$bob")
+    if [ "$code_a" = "201" ] && [ "$code_b" = "201" ]; then
+        ok "both users registered (201 / 201)"
+    else
+        fail "registration failed: alice=$code_a, bob=$code_b"
+        return
+    fi
+
+    # Pull CSRF tokens from cookie jars (csrf_token cookie value, 4th-from-last field).
+    csrf_from_jar() {
+        awk '$6 == "csrf_token" { print $7 }' "$1" | tail -1
+    }
+    local csrf_a csrf_b
+    csrf_a=$(csrf_from_jar "$jar_a")
+    csrf_b=$(csrf_from_jar "$jar_b")
+    if [ -n "$csrf_a" ] && [ -n "$csrf_b" ]; then
+        ok "CSRF token captured for both sessions"
+    else
+        fail "CSRF cookie missing (alice='${csrf_a:0:8}...', bob='${csrf_b:0:8}...')"
+        return
+    fi
+
+    create_thread() {
+        local jar="$1"; local csrf="$2"; local tid="$3"
+        curl -sS -b "$jar" -o /tmp/verify_thread_$$.json -w '%{http_code}' \
+            -H 'Content-Type: application/json' \
+            -H "X-CSRF-Token: $csrf" \
+            -d "{\"thread_id\":\"$tid\"}" \
+            "$GATEWAY_URL/api/threads"
+    }
+
+    local tid_a="verify-alice-${tag}"
+    local tid_b="verify-bob-${tag}"
+    info "Alice creates thread $tid_a; Bob creates thread $tid_b"
+    local cta ctb
+    cta=$(create_thread "$jar_a" "$csrf_a" "$tid_a")
+    ctb=$(create_thread "$jar_b" "$csrf_b" "$tid_b")
+    if [ "$cta" = "200" ] && [ "$ctb" = "200" ]; then
+        ok "both threads created (200 / 200)"
+    else
+        fail "thread creation failed: alice=$cta, bob=$ctb"
+        return
+    fi
+
+    # Cross access: Alice tries to GET Bob's thread → must be 404 (per PR6 contract:
+    # cross-workspace returns 404, not 403, to avoid leaking existence).
+    info "Alice → Bob's thread (GET)"
+    local cross_get
+    cross_get=$(curl -sS -b "$jar_a" -o /dev/null -w '%{http_code}' "$GATEWAY_URL/api/threads/$tid_b")
+    if [ "$cross_get" = "404" ]; then
+        ok "cross-workspace GET returns 404 (no existence leak)"
+    else
+        fail "cross-workspace GET returned '$cross_get', expected 404 — PR6 isolation broken"
+    fi
+
+    info "Alice → Bob's thread (DELETE)"
+    local cross_del
+    cross_del=$(curl -sS -b "$jar_a" -X DELETE -H "X-CSRF-Token: $csrf_a" \
+                    -o /dev/null -w '%{http_code}' "$GATEWAY_URL/api/threads/$tid_b")
+    if [ "$cross_del" = "404" ]; then
+        ok "cross-workspace DELETE returns 404"
+    else
+        fail "cross-workspace DELETE returned '$cross_del', expected 404"
+    fi
+
+    # Same-workspace GET — sanity check Alice can still reach her own thread.
+    info "Alice → Alice's thread (sanity)"
+    local same_get
+    same_get=$(curl -sS -b "$jar_a" -o /dev/null -w '%{http_code}' "$GATEWAY_URL/api/threads/$tid_a")
+    if [ "$same_get" = "200" ]; then
+        ok "same-workspace GET returns 200 (isolation is not over-blocking)"
+    else
+        fail "same-workspace GET returned '$same_get', expected 200"
+    fi
+
+    info "cookie jars left in /tmp for debugging: $jar_a $jar_b"
+}
+
+# ── main ─────────────────────────────────────────────────────────────────
+main() {
+    local args=("$@")
+    if [ ${#args[@]} -eq 0 ]; then
+        args=(static paths rds runtime e2e)
+    fi
+
+    printf '%s%sStage 0 verification — %s%s\n' "$C_BOLD" "$C_BLUE" "$(date)" "$C_RESET"
+    printf 'Repo root: %s\n' "$REPO_ROOT"
+    printf 'Phases: %s\n' "${args[*]}"
+
+    for phase_name in "${args[@]}"; do
+        case "$phase_name" in
+            static)  phase_static ;;
+            paths)   phase_paths ;;
+            rds)     phase_rds ;;
+            runtime) phase_runtime ;;
+            e2e)     phase_e2e ;;
+            all)
+                phase_static; phase_paths; phase_rds; phase_runtime; phase_e2e
+                ;;
+            *)
+                warn "unknown phase: $phase_name"
+                ;;
+        esac
+    done
+
+    printf '\n%s%s──── summary ────%s\n' "$C_BOLD" "$C_BLUE" "$C_RESET"
+    printf '  %s%d passed%s   %s%d failed%s   %s%d warn%s\n' \
+        "$C_GREEN" "$PASS_COUNT" "$C_RESET" \
+        "$C_RED"   "$FAIL_COUNT" "$C_RESET" \
+        "$C_YELLOW" "$WARN_COUNT" "$C_RESET"
+    if [ ${#SKIPPED_PHASES[@]} -gt 0 ]; then
+        printf '  skipped: %s\n' "${SKIPPED_PHASES[*]}"
+    fi
+    if [ ${#FAILED_STEPS[@]} -gt 0 ]; then
+        printf '\n%sfailing steps:%s\n' "$C_RED" "$C_RESET"
+        for step in "${FAILED_STEPS[@]}"; do
+            printf '  ✗ %s\n' "$step"
+        done
+    fi
+    [ "$FAIL_COUNT" -eq 0 ]
+}
+
+main "$@"

backend/tests/test_api_key_schema.py

@@ -0,0 +1,156 @@
+"""Schema tests for ``ApiKeyRow`` (Stage 0 PR8).
+
+PR8 is schema-only — no repository class, no API. Tests exercise raw
+ORM behaviour: column-level UNIQUE on key_prefix, the dual-dialect
+partial index DDL (sqlite_where + postgresql_where), and CASCADE on
+service_account delete.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+from sqlalchemy import delete
+from sqlalchemy.exc import IntegrityError
+
+from deerflow.persistence.api_key import ApiKeyRow
+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="bot",
+                role="member",
+                identity_mode="collapsed",
+                status="active",
+                created_by="u-alice",
+                created_at=now,
+                updated_at=now,
+            )
+        )
+        await session.commit()
+
+
+def _make_key(*, key_id: str, prefix: str, revoked_at: datetime | None = None) -> ApiKeyRow:
+    now = datetime.now(UTC)
+    return ApiKeyRow(
+        id=key_id,
+        service_account_id="sa-1",
+        key_prefix=prefix,
+        key_hash="0" * 64,
+        name="default",
+        scopes="",
+        rate_limit_rpm=None,
+        expires_at=None,
+        last_used_at=None,
+        revoked_at=revoked_at,
+        created_at=now,
+    )
+
+
+# ---------------------------------------------------------------------------
+# T8.4-1 — column-level UNIQUE on key_prefix
+# ---------------------------------------------------------------------------
+
+
+async def test_unique_key_prefix_enforced(tmp_path):
+    """Two api_key rows sharing the same key_prefix raise IntegrityError."""
+    sf = await _setup(tmp_path)
+    try:
+        await _seed_parents(sf)
+        async with sf() as session:
+            session.add(_make_key(key_id="ak-1", prefix="dfk_live_abc12345"))
+            await session.commit()
+
+        with pytest.raises(IntegrityError):
+            async with sf() as session:
+                session.add(_make_key(key_id="ak-2", prefix="dfk_live_abc12345"))
+                await session.commit()
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# T8.4-2 — partial index DDL covers both SQLite and Postgres
+# ---------------------------------------------------------------------------
+
+
+def test_active_index_declares_both_dialect_where_clauses():
+    """idx_api_keys_active must compile to a partial index on both drivers.
+
+    The plan locks ``sqlite_where`` *and* ``postgresql_where`` so the same
+    Index emits a partial index regardless of backend (Stage 0 dev still
+    runs SQLite locally; production is Postgres). Both ``dialect_options``
+    entries must be present.
+    """
+    active_idx = next((idx for idx in ApiKeyRow.__table__.indexes if idx.name == "idx_api_keys_active"), None)
+    assert active_idx is not None, "idx_api_keys_active not declared"
+    sqlite_where = active_idx.dialect_options.get("sqlite", {}).get("where")
+    postgres_where = active_idx.dialect_options.get("postgresql", {}).get("where")
+    assert sqlite_where is not None, "sqlite_where missing on idx_api_keys_active"
+    assert postgres_where is not None, "postgresql_where missing on idx_api_keys_active"
+    assert "revoked_at IS NULL" in str(sqlite_where)
+    assert "revoked_at IS NULL" in str(postgres_where)
+
+
+# ---------------------------------------------------------------------------
+# T8.4-3 — CASCADE on service_account delete
+# ---------------------------------------------------------------------------
+
+
+async def test_cascade_on_service_account_delete(tmp_path):
+    """Deleting the parent service_account removes all child api_keys."""
+    sf = await _setup(tmp_path)
+    try:
+        await _seed_parents(sf)
+        async with sf() as session:
+            session.add(_make_key(key_id="ak-cascade-1", prefix="dfk_live_cascade1"))
+            session.add(_make_key(key_id="ak-cascade-2", prefix="dfk_live_cascade2"))
+            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(ApiKeyRow, "ak-cascade-1")
+            row2 = await session.get(ApiKeyRow, "ak-cascade-2")
+        assert row1 is None
+        assert row2 is None
+    finally:
+        await _cleanup()

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

backend/tests/test_pr8_metadata_registration.py

@@ -0,0 +1,42 @@
+"""Acceptance test for PR8: ``Base.metadata.create_all()`` automatically
+provisions ``service_accounts`` / ``api_keys`` / ``external_users``.
+
+The harness layer registers all ORM models through ``deerflow.persistence.models``
+(imported for side effects from ``engine.init_engine``). This test guards
+against a row class being defined but accidentally left out of the
+registration entry point — a class table that never gets created at
+``init_engine`` time would otherwise silently break Stage 1 once the
+API-key auth layer starts inserting rows.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import inspect
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture
+def anyio_backend() -> str:
+    return "asyncio"
+
+
+async def test_pr8_tables_present_after_init_engine(tmp_path):
+    from deerflow.persistence.engine import close_engine, get_engine, init_engine
+
+    url = f"sqlite+aiosqlite:///{tmp_path / 'test.db'}"
+    await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
+    try:
+        engine = get_engine()
+        assert engine is not None
+
+        def _table_names(sync_conn):
+            return set(inspect(sync_conn).get_table_names())
+
+        async with engine.connect() as conn:
+            tables = await conn.run_sync(_table_names)
+
+        assert {"service_accounts", "api_keys", "external_users"}.issubset(tables), f"PR8 tables missing from create_all: have {sorted(tables)}"
+    finally:
+        await close_engine()

backend/tests/test_service_account_schema.py

@@ -0,0 +1,176 @@
+"""Schema tests for ``ServiceAccountRow`` (Stage 0 PR8).
+
+Pattern mirrors :mod:`test_workspace_repo`: ephemeral SQLite per test via
+``tmp_path``, no Postgres required at this layer.
+
+PR8 is schema-only — no repository class, no API. Tests exercise raw
+ORM behaviour: insert smoke, CASCADE on workspace delete, and RESTRICT
+on the ``created_by`` user FK.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+from sqlalchemy import delete
+from sqlalchemy.exc import IntegrityError
+
+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_user(sf, user_id: str = "u-alice", email: str = "alice@example.com") -> None:
+    async with sf() as session:
+        session.add(UserRow(id=user_id, email=email))
+        await session.commit()
+
+
+async def _seed_workspace(sf, workspace_id: str = "w-1", owner_id: str = "u-alice", slug: str = "alice") -> None:
+    async with sf() as session:
+        session.add(WorkspaceRow(id=workspace_id, name="Alice's WS", slug=slug, owner_id=owner_id))
+        await session.commit()
+
+
+# ---------------------------------------------------------------------------
+# T8.1 — insert smoke
+# ---------------------------------------------------------------------------
+
+
+async def test_insert_smoke(tmp_path):
+    """A minimal service_account row can be inserted and read back."""
+    sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf)
+        await _seed_workspace(sf)
+
+        now = datetime.now(UTC)
+        async with sf() as session:
+            session.add(
+                ServiceAccountRow(
+                    id="sa-1",
+                    workspace_id="w-1",
+                    name="ci-bot",
+                    role="member",
+                    identity_mode="collapsed",
+                    status="active",
+                    created_by="u-alice",
+                    created_at=now,
+                    updated_at=now,
+                )
+            )
+            await session.commit()
+
+        async with sf() as session:
+            row = await session.get(ServiceAccountRow, "sa-1")
+        assert row is not None
+        assert row.workspace_id == "w-1"
+        assert row.name == "ci-bot"
+        assert row.role == "member"
+        assert row.identity_mode == "collapsed"
+        assert row.status == "active"
+        assert row.created_by == "u-alice"
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# T8.2 — CASCADE on workspace delete
+# ---------------------------------------------------------------------------
+
+
+async def test_cascade_on_workspace_delete(tmp_path):
+    """Deleting the parent workspace removes the service_account row (FK CASCADE)."""
+    sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf)
+        await _seed_workspace(sf)
+        now = datetime.now(UTC)
+        async with sf() as session:
+            session.add(
+                ServiceAccountRow(
+                    id="sa-2",
+                    workspace_id="w-1",
+                    name="bot",
+                    role="member",
+                    identity_mode="collapsed",
+                    status="active",
+                    created_by="u-alice",
+                    created_at=now,
+                    updated_at=now,
+                )
+            )
+            await session.commit()
+
+        async with sf() as session:
+            await session.execute(delete(WorkspaceRow).where(WorkspaceRow.id == "w-1"))
+            await session.commit()
+
+        async with sf() as session:
+            row = await session.get(ServiceAccountRow, "sa-2")
+        assert row is None
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# T8.3 — RESTRICT on created_by user delete
+# ---------------------------------------------------------------------------
+
+
+async def test_restrict_on_created_by_user_delete(tmp_path):
+    """Deleting the creator user is blocked while their service_account survives."""
+    sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf)
+        await _seed_workspace(sf)
+        now = datetime.now(UTC)
+        async with sf() as session:
+            session.add(
+                ServiceAccountRow(
+                    id="sa-3",
+                    workspace_id="w-1",
+                    name="bot",
+                    role="member",
+                    identity_mode="collapsed",
+                    status="active",
+                    created_by="u-alice",
+                    created_at=now,
+                    updated_at=now,
+                )
+            )
+            await session.commit()
+
+        with pytest.raises(IntegrityError):
+            async with sf() as session:
+                await session.execute(delete(UserRow).where(UserRow.id == "u-alice"))
+                await session.commit()
+
+        # The service_account is still there after the rollback.
+        async with sf() as session:
+            row = await session.get(ServiceAccountRow, "sa-3")
+        assert row is not None
+    finally:
+        await _cleanup()

backend/tests/test_workspace_boundary_self.py

@@ -0,0 +1,93 @@
+"""Self-tests for the boundary scanner in ``test_workspace_boundary``.
+
+These tests guard against the scanner silently passing because it cannot
+detect anything. They feed synthetic ``.py`` snippets to the per-file
+collector and assert it produces (or correctly suppresses) the expected
+hits — so we know the integration test in ``test_workspace_boundary.py``
+would actually fail if a real violation appeared.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from test_workspace_boundary import collect_runtime_checkpoint_imports
+
+
+def _write(tmp_path: Path, body: str) -> Path:
+    target = tmp_path / "sample.py"
+    target.write_text(body, encoding="utf-8")
+    return target
+
+
+def test_flags_direct_from_import(tmp_path: Path) -> None:
+    target = _write(
+        tmp_path,
+        "from langgraph.checkpoint.postgres import AsyncPostgresSaver\n",
+    )
+    hits = collect_runtime_checkpoint_imports(target)
+    assert hits == [(1, "langgraph.checkpoint.postgres")]
+
+
+def test_flags_bare_module_import(tmp_path: Path) -> None:
+    target = _write(
+        tmp_path,
+        "import langgraph.checkpoint.memory  # noqa: F401\n",
+    )
+    hits = collect_runtime_checkpoint_imports(target)
+    assert hits == [(1, "langgraph.checkpoint.memory")]
+
+
+def test_flags_third_party_checkpoint_package(tmp_path: Path) -> None:
+    target = _write(
+        tmp_path,
+        "from langgraph_checkpoint_postgres import PostgresSaver\nfrom langgraph_checkpoint_sqlite import SqliteSaver\n",
+    )
+    hits = collect_runtime_checkpoint_imports(target)
+    assert (1, "langgraph_checkpoint_postgres") in hits
+    assert (2, "langgraph_checkpoint_sqlite") in hits
+
+
+def test_skips_type_checking_block_name_form(tmp_path: Path) -> None:
+    target = _write(
+        tmp_path,
+        "from typing import TYPE_CHECKING\n\nif TYPE_CHECKING:\n    from langgraph.checkpoint.base import BaseCheckpointSaver  # noqa: F401\n",
+    )
+    assert collect_runtime_checkpoint_imports(target) == []
+
+
+def test_skips_type_checking_block_attribute_form(tmp_path: Path) -> None:
+    target = _write(
+        tmp_path,
+        "import typing\n\nif typing.TYPE_CHECKING:\n    from langgraph.checkpoint.base import BaseCheckpointSaver  # noqa: F401\n",
+    )
+    assert collect_runtime_checkpoint_imports(target) == []
+
+
+def test_skips_nested_type_checking_block(tmp_path: Path) -> None:
+    target = _write(
+        tmp_path,
+        "from typing import TYPE_CHECKING\n\nif True:\n    if TYPE_CHECKING:\n        from langgraph.checkpoint.postgres import AsyncPostgresSaver  # noqa: F401\n",
+    )
+    assert collect_runtime_checkpoint_imports(target) == []
+
+
+def test_does_not_flag_unrelated_imports(tmp_path: Path) -> None:
+    target = _write(
+        tmp_path,
+        "import os\nfrom langgraph.graph.state import CompiledStateGraph  # noqa: F401\nfrom app.gateway.deps import get_checkpointer  # noqa: F401\n",
+    )
+    assert collect_runtime_checkpoint_imports(target) == []
+
+
+def test_does_not_flag_string_literal_with_module_name(tmp_path: Path) -> None:
+    target = _write(
+        tmp_path,
+        'TARGET = "langgraph.checkpoint.postgres"\n',
+    )
+    assert collect_runtime_checkpoint_imports(target) == []
+
+
+def test_handles_syntax_error_gracefully(tmp_path: Path) -> None:
+    target = _write(tmp_path, "def broken(:\n")
+    assert collect_runtime_checkpoint_imports(target) == []

docs/multi-tenant-redesign/03-impl/STATUS.md

@@ -2,11 +2,11 @@
 
 > **每完成 1 个 PR 后必更新**。本文是 Stage 0 唯一的"现在到哪了"权威来源——其它文件（plan、ADR、各 PR impl note）都是静态的，不反映执行进度。
 >
-> 上次更新：2026-05-13，PR6 merge 进 docs branch 后
+> 上次更新：2026-05-14，PR8 merge 进 docs branch 后（Stage 0 工程全合）
 
 ## 一句话状态
 
-PR1 + PR2 + PR3 + PR4 + PR5 + **PR6** 已 merge。**PR6 (2026-05-13)** 落地：4 个业务仓储 30+ 方法的 `workspace_id` 哨兵 + WHERE；`check_access` 升级三参数 (`thread_id, user_id, workspace_id`)；`@require_permission` 装饰器接入 `get_effective_workspace_id()`，跨 workspace **404 not 403**；`Paths` 切 workspace 维度（`{base}/workspaces/{wid}/threads/{tid}/...` + per-user state 嵌套）；`ThreadDataMiddleware` 切 workspace；`scripts/migrate_paths_to_workspace.py` 文件迁移（带 dry-run + 冲突分流）；lifespan 探测残留 `users/` 时 WARNING 引导跑 `make migrate-paths`；**T5.11 ORM `nullable=False` 一并翻**（PR5 推迟项就位）。**3214 passed + 30 skipped + 17 caplog flake**（PR5 末 3150 + 30 + 16；+64 测试，+1 flake——新 flake `test_path_migration_pending_warning::test_warns`，solo 跑 PASS）。**下一个：PR7（CI boundary 静态扫描）**。
+**PR1-PR8 全部已 merge——Stage 0 工程层面收尾。** **PR8 (2026-05-14)** 落地：3 张新表 schema-only 为 Stage 1 headless API 准备底座——`service_accounts`（workspace-scoped 非人身份，`identity_mode` 三态：`collapsed` / `external_passthrough` / `both`，`created_by` FK RESTRICT）/ `api_keys`（service_account 凭证，`key_prefix` 全局 UNIQUE + 双驱动部分索引 `idx_api_keys_active` WHERE `revoked_at IS NULL`，`scopes` 用 String(1024) 不用 PG `text[]` 保 SQLite 兼容）/ `external_users`（passthrough 终端身份，复合 UNIQUE `(service_account_id, external_id)`，`workspace_id` 冗余存储加速聚合）。FK 行为：workspace/SA delete CASCADE、creator user delete RESTRICT。9 新单测（3 service_account + 3 api_key + 2 external_user + 1 反向 metadata registration）。**3250 passed + 31 skipped + 18 caplog flake**（PR7 末 3241 + 31 + 18；+9 passed，flake 数 0 增）。Stage 0 工程层面**仅剩用户跟进的 live smoke**（见下）；业务层面看 "Stage 0 退出 Go/No-Go"。
 
 ## 8 PR 状态表
 
@@ -19,10 +19,13 @@ PR1 + PR2 + PR3 + PR4 + PR5 + **PR6** 已 merge。**PR6 (2026-05-13)** 落地：
 | **PR4** | ✅ merged | 14 (T4.1-T4.14) | merged into docs branch (`d98498b7..5c7753c0`) | [pr4-auth-workspace.md](./pr4-auth-workspace.md) |
 | **PR5** | ✅ merged | 11 (T5.1-T5.10 + T5.12) | merged into docs branch (`a7326978..30f2bd00`) | [pr5-business-workspace-id.md](./pr5-business-workspace-id.md) |
 | **PR6** | ✅ merged | 13 (T5.11 + T6.1-T6.15) | merged into docs branch (`361e653d..87ea715c`) | [pr6-routes-paths-workspace.md](./pr6-routes-paths-workspace.md) |
-| **PR7** | 🟡 pending | 0 | — | — |
-| **PR8** | 🟡 pending | 0 | — | — |
+| **PR7** | ✅ merged | 4 (T7.1-T7.3 + T7.5; T7.4 是反注入验证无代码改动) | merged into docs branch (`1a6ccc9a..d8b13afc`) | [pr7-ci-boundary-scan.md](./pr7-ci-boundary-scan.md) |
+| **PR8** | ✅ merged | 5 (T8.1 + T8.2/T8.3 合并 + T8.4 + T8.5 + T8.6) | merged into docs branch (`1fb07e48..f803f393`) | [pr8-headless-api-schema.md](./pr8-headless-api-schema.md) |
 
-**测试基线**：**PR6 末 3214 passed + 30 skipped**（PR5 末 3150 + 30；+64 PR6 新测试，覆盖 thread_meta workspace_id 过滤、Run/Feedback/RunEvent 同款、require_permission probes、跨 workspace 404 e2e、Paths workspace 形态、ThreadDataMiddleware workspace、文件迁移脚本、lifespan warning）。PR4 末 3136 + 26；PR3 末 3134 + 25；PR2 末 3087。**17 个 caplog 排序 flake 持续存在**（16 个 pre-existing + 1 新增 `test_path_migration_pending_warning::test_warns`）→ isolate 跑全 PASS，与 stage 无关；集中清理仍推迟到 follow-up。
+**测试基线**：**PR8 末 3250 passed + 31 skipped**（PR7 末 3241 + 31；+9 passed，PR8 新增 3 + 3 + 2 + 1 = 9 个 schema 测试）。PR6 末 3214 + 30；PR5 末 3150 + 30；PR4 末 3136 + 26；PR3 末 3134 + 25；PR2 末 3087。**18 个 caplog 排序 flake 持续存在**（17 个 pre-existing + 1 PR6 引入，PR7/PR8 均未引入新 flake）→ isolate 跑全 PASS，与 stage 无关；集中清理仍推迟到 follow-up。
+
+### Stage 0 整体测试增长
+PR1 起到 PR8 末，从既有 ~3087 增到 3250 passed（+163 测试，覆盖：PG fixture / sqlite→pg 默认切换 / workspaces + memberships / auth + JWT + register + workspace 自建 / alembic + backfill / 业务表 workspace_id 哨兵 + cross-workspace 404 e2e + Paths workspace + 文件迁移 / langgraph.checkpoint boundary 围栏 / service_accounts + api_keys + external_users schema）。plan 测试规模预估栏目原本估 ~70 新增，实际 ~163——PR4/PR5/PR6 都比预估多 2-3x，主要是 cross-workspace 隔离的 boundary e2e 比 plan 估的更稠密。
 
 ## 用户必须跟进的事（live verification / 决策）
 
@@ -50,6 +53,7 @@ PR1 + PR2 + PR3 + PR4 + PR5 + **PR6** 已 merge。**PR6 (2026-05-13)** 落地：
 | ~~PR5 T5.11~~ | ~~ORM model.py `nullable=False` 翻转~~ | **PR6 已落** (commit `87ea715c`) | — |
 | PR5 T5.12 真机 PG smoke | `alembic 0002 → backfill → 0003` 端到端 | agent 不能起 RDS 操作 | 用户跟进；命令清单见 [pr5-business-workspace-id.md "Live smoke 命令"](./pr5-business-workspace-id.md#live-smoke-命令用户跟进) |
 | PR6 T6.15 真机迁移 smoke | `make migrate-paths --dry-run` → 真迁移 → lifespan warning 消失 → 双账户互访 404 | agent 起不了 dev 服务 | 用户跟进；命令清单见 [pr6-routes-paths-workspace.md "Live smoke 命令"](./pr6-routes-paths-workspace.md#live-smoke-命令用户跟进) |
+| PR8 RDS 三张表存在 | `psql "$DATABASE_URL" -c "\dt service_accounts api_keys external_users"` 看 3 行；`\d+ api_keys` 看 `idx_api_keys_active ... WHERE revoked_at IS NULL` | agent 没 RDS 凭证 | 用户跟进；命令清单见 [pr8-headless-api-schema.md "Live smoke 命令"](./pr8-headless-api-schema.md#live-smoke-命令用户跟进) |
 
 ## 即将遇到的开放问题（plan 末尾列的，下个 session 处理）
 
@@ -61,7 +65,28 @@ PR1 + PR2 + PR3 + PR4 + PR5 + **PR6** 已 merge。**PR6 (2026-05-13)** 落地：
 
 ## 下一步建议
 
-**PR7（CI boundary 静态扫描）**。Stage 0 收尾的最后一项；plan 描述："静态扫描 ban `deerflow.* → app.*` 反向 import"。PR1-PR6 的代码已经维持这条边界，PR7 是把单测 `tests/test_harness_boundary.py` 上的检查升级为 grep 级 / CI workflow 级扫描，加更细粒度的禁止规则（如禁止 `app.*` 反向再 import 回 `deerflow.runtime.*` 等不应有的间接环）。**PR8 (service_accounts / api_keys / external_users schema) 可并行**，依赖只到 PR3 的 workspaces 表。
+**Stage 0 工程层面 8 个 PR 全部 merge，agent 这一侧的代码工作收尾。** 剩下都是**用户必须做的 live verification**：
+
+1. **PR8 RDS 表存在** — `psql "$DATABASE_URL" -c "\dt service_accounts api_keys external_users"`
+2. **PR6 真机文件迁移** — 起 dev 服务、跑 `make migrate-paths --dry-run`、确认 lifespan warning 消失
+3. **PR5 RDS alembic 0002→backfill→0003** — 端到端验证业务表 workspace_id 列
+4. **PR1 testcontainers PG smoke** — docker daemon 起来后跑 `pytest -m postgres -v`
+5. **远程 RDS 密码轮换**（之前在聊天里给过明文）
+6. **Push docs branch** 跑 GitHub CI（已 push，监 [backend-postgres-tests workflow](../../../.github/workflows/backend-postgres-tests.yml) 在 PG matrix 全绿）
+
+工程门 [Stage 0 退出 Go/No-Go](../../superpowers/plans/2026-05-10-stage-0-multi-tenant-foundation.md#stage-0-退出-gono-go来自-phased-rollout-by-scale) 已基本满足：8 PR 全合 / 新增 ~163 测试全过 / CI 绿（待 push 后确认）/ 7 项不可逆 LOCK 已 sign-off。**业务门**（"第一个付费意向客户"）等业务进展；生产稳定运行 ≥ 2 周也属业务时序。
+
+Stage 1 可启动的方向（plan 没排，但已具备底座）：
+- **headless API 鉴权层**接 PR8 三张表（API key middleware / token 生成与 sha256 / `@require_permission` scope 升级 / Pattern A vs B 路由分流）
+- **frontend workspace picker / switching UI**
+- **platform admin 管理 workspace 的 CLI / UI**（plan self-review 标记的 gap）
+- **17 个 pre-existing caplog flake 集中清理**（一直推迟）
+
+---
+
+PR8 经验回顾：纯 schema PR，**Inline + 严格 TDD（红→绿）** 跑得很顺。6 个 task 单链条但每个 task 互相独立——SA / api_key / external_user 三张表之间只通过 FK 关联，没有跨 task signature 协调。每个表都按"先建模型 → 写 insert smoke 红→绿 → 加 cascade 测试 → 加 constraint 测试"四步走，3 张表 25 分钟内全落。T8.6 反向 metadata registration 测试是踩过坑后的肌肉记忆——历史上多次"模型类写了但 persistence/models/__init__.py 漏 import → create_all 不建表 → 上线 SELECT 时炸"，T8.6 把这条 invariant 永久锁住。
+
+PR7 经验回顾：纯静态测试 PR，Inline 模式继续合适——5 个 task 单链条强耦合（先确定 allowlist 内容才能写扫描器，扫描器函数得是导出才能 self-test）。复用 PR4 同款"红→绿"严格 TDD：故意建空 allowlist 跑红、再填→绿；T7.4 反注入实验是对静态扫描器的"集成 smoke"，确认现实 backend 文件 + 真实 allowlist 过滤路径同时生效——这一步比 9 个 self-test 都更有说服力。
 
 PR6 经验回顾：plan 推荐 Inline 模式是对的，路由 + 仓储 + Paths 强耦合每一步都依赖前一步的接口形态。如果走 subagent 派单会反复阻塞在跨 task 的 signature 协调上。
 

docs/multi-tenant-redesign/03-impl/pr7-ci-boundary-scan.md

@@ -0,0 +1,68 @@
+# PR7 · CI boundary 静态扫描（`langgraph.checkpoint.*` 直接 import 围栏）
+
+> 实现笔记。对应 [docs/superpowers/plans/2026-05-10-stage-0-multi-tenant-foundation.md](../../superpowers/plans/2026-05-10-stage-0-multi-tenant-foundation.md) PR7（T7.1-T7.5）。
+>
+> 状态：**已落地**，4 个 commits 提交到 `docs/multi-tenant-redesign`（PR7 起始 `1a6ccc9a..` 结束 `d8b13afc`）。
+
+## 范围
+
+把 `tests/test_harness_boundary.py` 的模式从"harness 不能 import app"扩到第二条 workspace-isolation 围栏：**只有显式 allowlist 里的文件才能直接 import `langgraph.checkpoint.*`**。其他任何位置必须穿 `app.gateway.deps.get_checkpointer` DI 或 `deerflow.runtime.checkpointer` 工厂——避免业务代码自己 new 个 saver 绕过 workspace 边界。
+
+1. **`tests/boundary_allowlist.toml`** — 4 个当前合法 importer，每行配注释解释为什么允许：
+   - `app/gateway/routers/threads.py`（thread 初始化时 `empty_checkpoint`）
+   - `packages/harness/deerflow/runtime/checkpointer/async_provider.py` / `provider.py`（唯一 saver 工厂处）
+   - `packages/harness/deerflow/runtime/runs/worker.py`（resumed run seed `empty_checkpoint`）
+2. **`tests/test_workspace_boundary.py`** — AST 扫描 backend 所有 `*.py`（排除 `tests/`、`docs/`、build artefacts），命中 `langgraph.checkpoint.*` / `langgraph_checkpoint_postgres` / `langgraph_checkpoint_sqlite` 且不在 allowlist 即 fail，error message 直接列 `<path>:<line>  imports <module>` + 修复指引。
+3. **TYPE_CHECKING 豁免** — 扫描器通过 parent-walk 检测 `if TYPE_CHECKING:` / `if typing.TYPE_CHECKING:` 块（含嵌套），块内 import 不算违规。因此 `agents/factory.py` 里 `BaseCheckpointSaver` 作为类型注解的 type-only import **不进 allowlist**，更准确地反映 runtime boundary 语义。
+4. **`tests/test_workspace_boundary_self.py`** — 9 个 self-test 防止扫描器静默空跑：用 `tmp_path` 合成 `.py` 片段喂给 `collect_runtime_checkpoint_imports`，覆盖 from-import / bare import / 第三方包 / TYPE_CHECKING（Name 形 + Attribute 形 + 嵌套）/ 字符串字面量 / 语法错误 / 不相关 import。
+5. **`backend/CLAUDE.md` Boundary check 段** 加两行说明新增的两个 test 文件 + allowlist 维护契约（"新 importer 同 PR 加 allowlist"）。
+
+**不在范围**：
+- 运行时 import-hook 拦截（plan 明确不做——静态 AST 扫描足够 + 不引入 runtime 开销）。
+- 把 plan 草稿里的 `thread_runs.py` / `gateway/app.py` 列入 allowlist——grep 实际证实它们走 `get_checkpointer` DI，**不直接 import**，列入会假阳放水。
+- `app.* → deerflow.runtime.*` 的反向间接环检查（plan "一句话状态" 提到的扩展，留 Stage 1 follow-up）。
+
+## Tasks 完成清单
+
+| Task | Commit | 关键改动 |
+|---|---|---|
+| **T7.1** | `1a6ccc9a` | `tests/boundary_allowlist.toml` 4 entry + 每行注释解释合法性 |
+| **T7.2** | `ba30d140` | `tests/test_workspace_boundary.py` AST scanner + TYPE_CHECKING parent-walk + allowlist 加载 |
+| **T7.3** | `d0f18770` | `tests/test_workspace_boundary_self.py` 9 个 self-test |
+| **T7.4** | _no commit_ | 临时在 `app/gateway/routers/feedback.py:13` 加 `from langgraph.checkpoint.postgres import AsyncPostgresSaver` → scanner 红灯且 line 号正确 → revert。无测试改动，按 plan "仅 commit 测试自身完善" 原则不留 empty commit |
+| **T7.5** | `d8b13afc` | `backend/CLAUDE.md` Boundary check 段加 workspace boundary + self-test 两行 |
+
+## 验收
+
+- [x] **scanner 红→绿循环**：empty allowlist → 14 violations across 4 files（threads / async_provider / provider / worker，TYPE_CHECKING-only 的 factory.py 正确不在内）；填入 4 entry → PASS
+- [x] **scanner self-test 9 个全过**（防静默空跑）
+- [x] **T7.4 反注入实验**：往 `feedback.py:13` 加一行违规 import → `pytest tests/test_workspace_boundary.py` 单条 fail，error 精准指 `app/gateway/routers/feedback.py:13  imports langgraph.checkpoint.postgres`；revert 后立即返绿
+- [x] **全套 `make test` 3241 passed + 31 skipped + 18 caplog flake**（PR6 末 3214 + 30 + 17；+27 passed / +1 skip / +1 flake — passed delta 包含 PR7 新增 10 个测试以及环境差异导致的 17 个之前 flake 这次稳过，flake 列表形态与 STATUS.md 既有 17 项 + PR6 引入的 `test_path_migration_pending_warning` 一致，与 PR7 改动无关）
+- [x] **CI workflow 接入**：扫描器是普通 pytest，已被 `.github/workflows/backend-unit-tests.yml` 全套 run 覆盖；无需新 workflow
+
+## 文件结构
+
+**新增**：
+- `backend/tests/test_workspace_boundary.py` — AST 扫描器（127 行）
+- `backend/tests/test_workspace_boundary_self.py` — 扫描器 self-test（93 行）
+- `backend/tests/boundary_allowlist.toml` — 4 个合法 importer + 每行注释（28 行）
+- `docs/multi-tenant-redesign/03-impl/pr7-ci-boundary-scan.md` — 本文件
+
+**修改**：
+- `backend/CLAUDE.md` — Boundary check 段 +2 行
+
+**未改动**（验证后无需触碰）：
+- `app/gateway/routers/threads.py`、`runtime/checkpointer/async_provider.py`、`runtime/checkpointer/provider.py`、`runtime/runs/worker.py` — 当前合法 importer，已被 allowlist 显式覆盖
+- `agents/factory.py` — TYPE_CHECKING-only import，扫描器自动豁免
+
+## 关键设计决策
+
+1. **TYPE_CHECKING 豁免 vs allowlist 收纳**：`agents/factory.py` 把 `BaseCheckpointSaver` 当类型注解用。两种实现方式都能让扫描通过——加 allowlist / 加 TYPE_CHECKING 检测。选后者：boundary 的真实语义是 "runtime path 不要构造 saver"，type-only import 不进 runtime，本就不构成违规，把它列 allowlist 是给后人一个错误信号（"看，这文件可以直接 import"）。parent-walk 实现 `_inside_type_checking` 多 15 行代码，换 1 条更准的语义边界。
+2. **Allowlist 形态：toml list of paths，而非正则 / 模块通配**：4 个 entry，未来增长慢，精确路径列表最易审计 / diff。toml 配 frontmatter 注释每条合法性来源；任何 PR 加新 entry 都被 review 看到。
+3. **扫描器作为 pytest 而非独立 CI step**：复用现有 `backend-unit-tests.yml`，零 workflow 改动，本地 `make test` 也覆盖。如果未来想跑成独立 step（更快失败），切出来代价低。
+4. **不引入 deerflow / app 边界以外的更细规则**：plan 顶部"一句话状态"提到 `app.*` 反向再 import `deerflow.runtime.*` 的间接环检查——这条作为 Stage 1 follow-up 保留，PR7 范围里只做 langgraph.checkpoint.* 这一条单点围栏，保持每个 PR 单一关注。
+5. **Plan 草稿 allowlist 修正**：plan 列了 `thread_runs.py` / `gateway/app.py`，但 grep 实际状态两者都通过 `get_checkpointer` DI 拿 saver，不直接 import；同时漏掉了 `provider.py`（同步版 checkpointer）和 `worker.py`（run resume seed）。impl 按 ground truth 取 4 个真实 importer——这是 plan 设计阶段无法预知的代码事实，应该以代码为准。
+
+## Live smoke 命令（用户跟进）
+
+PR7 是纯静态检查，落地即生效，无 runtime 行为变化，**无需 live smoke**。`make test` 全套绿就完成。

docs/multi-tenant-redesign/03-impl/pr8-headless-api-schema.md

@@ -0,0 +1,96 @@
+# PR8 · `service_accounts` + `api_keys` + `external_users` schema only
+
+> 实现笔记。对应 [docs/superpowers/plans/2026-05-10-stage-0-multi-tenant-foundation.md](../../superpowers/plans/2026-05-10-stage-0-multi-tenant-foundation.md) PR8（T8.1-T8.6）。
+>
+> 状态：**已落地**，5 个 commits 提交到 `docs/multi-tenant-redesign`（PR8 起始 `1fb07e48..` 结束 `f803f393`）。
+
+## 范围
+
+为 Stage 1 headless API 鉴权层准备底座：3 张新表 + ORM。仿 PR3 模式——纯 schema、不接路由、不写仓储、不暴露 API。
+
+1. **`service_accounts`** — 非人身份，属于唯一 workspace。3 状态字段：`role`（Stage 0 仅 `member`）/ `identity_mode`（`collapsed` / `external_passthrough` / `both`，决定是否记录终端用户身份）/ `status`（`active` / `suspended` / `deleted`）。`workspace_id` FK CASCADE，`created_by` FK 用户 **RESTRICT**（防止误删带 SA 的 user）。
+2. **`api_keys`** — service_account 的凭证。`key_prefix` String(16) **全局 unique**（撤销后亦不复用，避免审计混淆），`key_hash` String(128) 存 sha-256 hex（plaintext 仅创建时返）。`scopes` String(1024) 逗号分隔（不用 PG `text[]` 以保 SQLite dev 双驱动兼容；Stage 2 切纯 PG 可平滑迁）。`revoked_at`/`expires_at`/`last_used_at`/`rate_limit_rpm` 全 nullable。**双驱动部分索引** `idx_api_keys_active`（key_prefix）WHERE revoked_at IS NULL——同时声明 `sqlite_where` + `postgresql_where`，鉴权热路径 prefix lookup 加速。`service_account_id` FK CASCADE。
+3. **`external_users`** — passthrough 模式下的终端身份。每次调用带 `X-External-User-Id` header 时 upsert 一行（Stage 1 起）。**复合 UNIQUE** `(service_account_id, external_id)`——同 external_id 可在不同 SA 下复用，但单 SA 下唯一。`workspace_id` 冗余存储（可经 SA 间接得到，但直接存以加速 workspace-scoped 跨 SA 聚合）。`metadata_json` JSON nullable=False default {} 存 plan tier / region / 自定义 tag。两个 FK 均 CASCADE。
+4. **ORM 注册** — `deerflow/persistence/models/__init__.py` 加 3 行 import 让 `Base.metadata.create_all()` 在 `init_engine` 启动时自动建 3 张表。`test_pr8_metadata_registration.py` 反向验证：拉一个 fresh SQLite 引擎 inspect 表名集合，断言 3 张表都在。
+
+**不在范围**（Stage 1）：
+- API key 鉴权中间件 / token 生成 / hash 验证
+- `@require_permission` scope 升级（接 `service_account`/`api_key` 主体）
+- Pattern A/B endpoint 设计
+- `external_users` upsert 逻辑
+- 鉴权层的 rate limiting / scopes 校验
+
+## Tasks 完成清单
+
+| Task | Commit | 关键改动 |
+|---|---|---|
+| **T8.1** | `1fb07e48` | `service_account/{__init__, model}.py` + insert smoke + 注册进 persistence.models |
+| **T8.2 + T8.3** | `bb728978` | `test_cascade_on_workspace_delete` + `test_restrict_on_created_by_user_delete` |
+| **T8.4** | `52e9999a` | `api_key/{__init__, model}.py` + 3 测试（column UNIQUE + 双驱动 partial index DDL + CASCADE） |
+| **T8.5** | `6f806ff4` | `external_user/{__init__, model}.py` + 2 测试（复合 UNIQUE + CASCADE） |
+| **T8.6** | `f803f393` | `test_pr8_metadata_registration.py` 反向验证 `Base.metadata.create_all()` 真的建 3 张表 |
+
+## 验收
+
+- [x] **3 + 3 + 2 + 1 = 9 个新单测全过**（T8.1/T8.2/T8.3 三个 service_account；T8.4 三个 api_key；T8.5 两个 external_user；T8.6 一个 metadata registration）
+- [x] **3 张表 `Base.metadata.create_all()` 自动建**：T8.6 inspect 表名集合断言 `{service_accounts, api_keys, external_users}.issubset(tables)`
+- [x] **FK 行为按 plan 设计**：CASCADE on workspace/SA delete、RESTRICT on creator user delete、SQLite + PG 均生效（SQLite 通过 engine.py connect-listener 的 `PRAGMA foreign_keys=ON`）
+- [x] **partial index 双驱动 DDL** 通过 `dialect_options` 检查锁定（不只看 SQLAlchemy emit，下次有人删 `postgresql_where` 测试会红）
+
+## 文件结构
+
+**新增**：
+- `backend/packages/harness/deerflow/persistence/service_account/{__init__.py, model.py}`
+- `backend/packages/harness/deerflow/persistence/api_key/{__init__.py, model.py}`
+- `backend/packages/harness/deerflow/persistence/external_user/{__init__.py, model.py}`
+- `backend/tests/test_service_account_schema.py`（3 cases）
+- `backend/tests/test_api_key_schema.py`（3 cases）
+- `backend/tests/test_external_user_schema.py`（2 cases）
+- `backend/tests/test_pr8_metadata_registration.py`（1 case）
+- `docs/multi-tenant-redesign/03-impl/pr8-headless-api-schema.md` — 本文件
+
+**修改**：
+- `backend/packages/harness/deerflow/persistence/models/__init__.py` — 加 3 行 import + `__all__` 注册
+
+## 关键设计决策
+
+1. **`key_prefix` 全局 UNIQUE，而非"活跃 UNIQUE"**：column-level `unique=True` 覆盖整个 key 生命周期。理由：撤销 + 复用同前缀会让审计日志里 "prefix X did Y" 的语义模糊；prefix 16 字符的命名空间足够大（≈10^25）从不复用没有成本。`idx_api_keys_active` 走部分非唯一索引——纯粹是热路径优化，撤销 key 不进活跃索引以减小热索引大小。
+2. **`scopes` 用 `String(1024)` 而非 PG `text[]`**：Stage 0 仍要 SQLite 跑得动（dev / unit test 兜底）。逗号分隔字符串两端通用；Stage 2 切纯 PG 后再迁 `text[]` + GIN 索引代价低。LOCK 由 plan 记下。
+3. **`external_users.workspace_id` 冗余存储**：技术上可从 `service_account_id` JOIN 出来，但 Stage 1 几个高频查询（workspace 级配额聚合 / admin UI 列出 workspace 所有 external user）每次走 JOIN 会随 SA 数量增长变慢。冗余一列、CASCADE 同 SA 一致，是值得的存储成本。
+4. **不写 Repository 类**：Stage 0 PR3 / PR5-6 的 Repository 是给 Gateway 当前在用的表准备的。PR8 三张表 Stage 0 内**没人读写**——直到 Stage 1 headless API 才用得上。写空 Repository 现在不知道接口形态，等 Stage 1 真用时连同 token 生成 / 哈希校验一起设计更合理。Plan 也明确"仅暴露 ORM"。
+5. **T8.6 反向 metadata registration 测试**：Stage 0 已经踩过坑——PR1-PR6 多次出现"模型类写了但 `persistence/models/__init__.py` 漏 import → `create_all()` 不建表 → 上线后 SELECT 时炸 'no such table'"。T8.6 把这条 invariant 锁定。
+6. **`identity_mode` 三态保持字符串而非 enum**：和 `role` / `status` 同款思路——String(16) 比 enum 更易加值（Stage 2 可能加 `cli_only` 等新态），不动 schema。
+
+## Live smoke 命令（用户跟进）
+
+PR8 纯 schema 改造，无路由 / 中间件 / 文件系统副作用。
+
+**单机自检**：
+```bash
+make stop && make dev   # 起服务，让 lifespan 跑 init_engine
+# 看 Gateway 启动日志无报错；create_all 默认 silent，无需额外断言
+```
+
+**RDS 实跑表存在**（需要密码）：
+```bash
+psql "$DATABASE_URL" -c "\dt service_accounts api_keys external_users"
+# 期望：3 行
+psql "$DATABASE_URL" -c "\d+ api_keys"
+# 期望看到 idx_api_keys_active (key_prefix) WHERE revoked_at IS NULL
+```
+
+**双驱动 partial index 在 PG 真生效**（可选）：
+```bash
+# 用 testcontainers 跑 @pytest.mark.postgres 系列；当前 PR8 没写 PG 专属测试，
+# 但 idx_api_keys_active 的 DDL 已在 dialect_options 里覆盖，PG schema dump
+# 应见 "WHERE revoked_at IS NULL"
+PYTHONPATH=. uv run pytest -m postgres -v
+```
+
+## Stage 0 退出门
+
+PR8 是 Stage 0 工程层面最后一个 PR。剩余 Stage 0 退出条件见 [STATUS.md](./STATUS.md)"用户必须跟进的事"：
+- [ ] RDS 上 `service_accounts` / `api_keys` / `external_users` 三张表 `\dt` 见
+- [ ] `make migrate-paths --dry-run` 在 fresh DB 上输出空
+- [ ] testcontainers ephemeral PG smoke 跑过一次
+- [ ] 生产稳定运行 ≥ 2 周（业务条件）