feat(persistence): backfill Step 3 — orphan rows -> legacy_workspace

_ensure_legacy_workspace creates the nil-UUID anchor (slug=legacy)
owned by the platform admin (or oldest user as fallback). Raises a
clear error if the DB has no users at all so we never silently create
an orphaned workspace. Step 3 UPDATEs each table's remaining
workspace_id IS NULL rows to LEGACY_WORKSPACE_ID. Orchestrator wires
ensure-then-loop between Step 2 and Step 3. 3 new tests: orphan
fan-out, no-users error, end-to-end orchestrator with mixed owned +
orphan rows.

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

backend/scripts/backfill_workspace_id.py

@@ -158,6 +158,57 @@ async def _step2_update_table_from_users(
     return int(rowcount)
 
 
+async def _ensure_legacy_workspace(
+    session_factory: async_sessionmaker[AsyncSession],
+    *,
+    dry_run: bool,
+) -> bool:
+    """Create the ``legacy_workspace`` anchor row idempotently.
+
+    The anchor is needed before Step 3 can point orphan rows at it. We
+    pick the platform admin (``system_role='admin'``) as owner; if no
+    admin exists yet we fall back to the oldest user. If the database
+    has no users at all we refuse to continue — running this script on
+    an unbootstrapped DB would create a workspace with no owner and the
+    FK to ``users`` would fail anyway.
+
+    Returns True if the workspace was just created (or would be, under
+    ``dry_run``). False if it already existed.
+    """
+    from deerflow.persistence.workspace.model import WorkspaceRow
+
+    async with session_factory() as session:
+        existing = await session.get(WorkspaceRow, LEGACY_WORKSPACE_ID)
+    if existing is not None:
+        return False
+
+    async with session_factory() as session:
+        admin_id = (await session.execute(select(UserRow.id).where(UserRow.system_role == "admin").order_by(UserRow.created_at).limit(1))).scalar_one_or_none()
+        if admin_id is None:
+            admin_id = (await session.execute(select(UserRow.id).order_by(UserRow.created_at).limit(1))).scalar_one_or_none()
+
+    if admin_id is None:
+        raise RuntimeError(
+            "Cannot create legacy_workspace: no users exist. Bootstrap an admin via /auth/initialize before running backfill.",
+        )
+
+    if dry_run:
+        logger.info("WOULD create legacy_workspace (id=%s) owned by user=%s", LEGACY_WORKSPACE_ID, admin_id)
+        return True
+
+    ws_repo = WorkspaceRepository(session_factory)
+    await ws_repo.create(
+        workspace_id=LEGACY_WORKSPACE_ID,
+        name=LEGACY_WORKSPACE_NAME,
+        slug=LEGACY_WORKSPACE_SLUG,
+        owner_id=admin_id,
+    )
+    m_repo = WorkspaceMembershipRepository(session_factory)
+    await m_repo.add(workspace_id=LEGACY_WORKSPACE_ID, user_id=admin_id, role="owner")
+    logger.info("Created legacy_workspace (id=%s) owned by user=%s", LEGACY_WORKSPACE_ID, admin_id)
+    return True
+
+
 async def _step3_assign_legacy_workspace(
     session_factory: async_sessionmaker[AsyncSession],
     table: str,
@@ -166,12 +217,28 @@ async def _step3_assign_legacy_workspace(
 ) -> int:
     """Assign LEGACY_WORKSPACE_ID to *table* rows still missing workspace_id.
 
-    Filled in by T5.7 (also responsible for ensuring the legacy workspace row exists).
+    Callers should ensure :func:`_ensure_legacy_workspace` has run first;
+    the orchestrator does this between Step 2 and Step 3. Orphan rows are
+    rows whose ``user_id`` was already NULL (or pointed at a deleted user)
+    so Step 2's correlated subquery left them untouched.
     """
-    _ = session_factory
+    model = _TABLE_MODELS[table]
-    _ = table
+    workspace_col = model.workspace_id
-    _ = dry_run
+
-    return 0
+    if dry_run:
+        count_stmt = select(func.count()).select_from(model).where(workspace_col.is_(None))
+        async with session_factory() as session:
+            count = (await session.execute(count_stmt)).scalar_one() or 0
+        logger.info("WOULD assign %d orphan row(s) in %s to legacy_workspace", count, table)
+        return int(count)
+
+    stmt = update(model).where(workspace_col.is_(None)).values(workspace_id=LEGACY_WORKSPACE_ID)
+    async with session_factory() as session:
+        result = await session.execute(stmt)
+        await session.commit()
+        rowcount = result.rowcount or 0
+    logger.info("Assigned %d orphan row(s) in %s to legacy_workspace", rowcount, table)
+    return int(rowcount)
 
 
 async def backfill(
@@ -189,6 +256,7 @@ async def backfill(
     report["users_workspaces_created"] = await _step1_create_workspaces_for_users(session_factory, dry_run=dry_run)
     for table in _BUSINESS_TABLES:
         report[f"{table}_from_users"] = await _step2_update_table_from_users(session_factory, table, dry_run=dry_run)
+    report["legacy_workspace_created"] = await _ensure_legacy_workspace(session_factory, dry_run=dry_run)
     for table in _BUSINESS_TABLES:
         report[f"{table}_legacy"] = await _step3_assign_legacy_workspace(session_factory, table, dry_run=dry_run)
 

backend/tests/test_backfill_workspace_id.py

@@ -24,8 +24,12 @@ from deerflow.persistence.user.model import UserRow
 from deerflow.persistence.workspace.model import WorkspaceRow
 from deerflow.persistence.workspace_membership.model import WorkspaceMembershipRow
 from scripts.backfill_workspace_id import (
+    LEGACY_WORKSPACE_ID,
+    _ensure_legacy_workspace,
     _step1_create_workspaces_for_users,
     _step2_update_table_from_users,
+    _step3_assign_legacy_workspace,
+    backfill,
 )
 
 pytestmark = pytest.mark.anyio
@@ -50,10 +54,10 @@ async def _close():
     await close_engine()
 
 
-async def _seed_user(sf, *, email: str, default_workspace_id: str | None = None) -> str:
+async def _seed_user(sf, *, email: str, default_workspace_id: str | None = None, system_role: str = "user") -> str:
     user_id = str(uuid.uuid4())
     async with sf() as session:
-        session.add(UserRow(id=user_id, email=email, default_workspace_id=default_workspace_id))
+        session.add(UserRow(id=user_id, email=email, default_workspace_id=default_workspace_id, system_role=system_role))
         await session.commit()
     return user_id
 
@@ -196,3 +200,85 @@ async def test_step1_skips_blacklisted_base_slug(tmp_path):
         assert ws.slug == "admin-2"
     finally:
         await _close()
+
+
+# ---------------------------------------------------------------------------
+# Step 3: orphan rows -> legacy_workspace
+# ---------------------------------------------------------------------------
+
+
+async def test_backfill_orphan_rows_go_to_legacy_workspace(tmp_path):
+    """Rows with user_id=NULL get assigned the legacy_workspace UUID after Step 3."""
+    sf = await _init_engine(tmp_path)
+    try:
+        # Seed a platform admin so the legacy workspace has an owner.
+        await _seed_user(sf, email="admin@example.com", system_role="admin")
+        # Orphan business rows (user_id=NULL): legacy data from before auth.
+        await _seed_business_row(sf, ThreadMetaRow, thread_id="t-orphan", user_id=None)
+        await _seed_business_row(sf, RunRow, run_id="r-orphan", thread_id="t-orphan", user_id=None)
+        await _seed_business_row(sf, FeedbackRow, feedback_id="f-orphan", thread_id="t-orphan", run_id="r-orphan", user_id=None, rating=1)
+        await _seed_business_row(sf, RunEventRow, thread_id="t-orphan", run_id="r-orphan", user_id=None, event_type="legacy", category="lifecycle", seq=1)
+
+        # Ensure the anchor + reassign per table.
+        created = await _ensure_legacy_workspace(sf, dry_run=False)
+        assert created is True
+        for table in ("threads_meta", "runs", "feedback", "run_events"):
+            count = await _step3_assign_legacy_workspace(sf, table, dry_run=False)
+            assert count == 1, table
+
+        # Re-running the anchor helper is a no-op.
+        assert await _ensure_legacy_workspace(sf, dry_run=False) is False
+
+        async with sf() as session:
+            tm = (await session.execute(select(ThreadMetaRow))).scalar_one()
+            run = (await session.execute(select(RunRow))).scalar_one()
+            fb = (await session.execute(select(FeedbackRow))).scalar_one()
+            ev = (await session.execute(select(RunEventRow))).scalar_one()
+            legacy = (await session.execute(select(WorkspaceRow).where(WorkspaceRow.id == LEGACY_WORKSPACE_ID))).scalar_one()
+            legacy_mem = (await session.execute(select(WorkspaceMembershipRow).where(WorkspaceMembershipRow.workspace_id == LEGACY_WORKSPACE_ID))).scalar_one()
+
+        assert tm.workspace_id == LEGACY_WORKSPACE_ID
+        assert run.workspace_id == LEGACY_WORKSPACE_ID
+        assert fb.workspace_id == LEGACY_WORKSPACE_ID
+        assert ev.workspace_id == LEGACY_WORKSPACE_ID
+        assert legacy.slug == "legacy"
+        assert legacy_mem.role == "owner"
+    finally:
+        await _close()
+
+
+async def test_ensure_legacy_workspace_refuses_when_no_users(tmp_path):
+    """ensure_legacy_workspace raises a clear error if the DB has no users."""
+    sf = await _init_engine(tmp_path)
+    try:
+        with pytest.raises(RuntimeError, match="no users exist"):
+            await _ensure_legacy_workspace(sf, dry_run=False)
+    finally:
+        await _close()
+
+
+async def test_full_backfill_orchestrator(tmp_path):
+    """End-to-end: backfill() runs all three steps and reports per-step counts."""
+    sf = await _init_engine(tmp_path)
+    try:
+        await _seed_user(sf, email="admin@example.com", system_role="admin")
+        user_id = await _seed_user(sf, email="gina@example.com")
+        await _seed_business_row(sf, ThreadMetaRow, thread_id="t-owned", user_id=user_id)
+        await _seed_business_row(sf, ThreadMetaRow, thread_id="t-orphan", user_id=None)
+
+        report = await backfill(sf, dry_run=False)
+        assert report["dry_run"] is False
+        # Two users were missing a default workspace (admin too — we
+        # didn't pre-populate admin's default_workspace_id).
+        assert report["users_workspaces_created"] == 2
+        assert report["threads_meta_from_users"] == 1
+        assert report["legacy_workspace_created"] is True
+        assert report["threads_meta_legacy"] == 1
+
+        async with sf() as session:
+            rows = {r.thread_id: r.workspace_id for r in (await session.execute(select(ThreadMetaRow))).scalars().all()}
+        assert rows["t-orphan"] == LEGACY_WORKSPACE_ID
+        assert rows["t-owned"] != LEGACY_WORKSPACE_ID
+        assert rows["t-owned"] is not None
+    finally:
+        await _close()