feat(persistence): PR6 T5.11 — ORM workspace_id nullable=False

Flip the 4 business ORM models (ThreadMetaRow, RunRow, FeedbackRow,
RunEventRow) to ``workspace_id: Mapped[str]`` with ``nullable=False``.
PR5's alembic 0003 already enforces NOT NULL at the DB layer; this
aligns the ORM-driven ``create_all()`` path (dev / tests) with the same
invariant so a new install ends up at the post-0003 schema without
running alembic.

Test fallout absorbed:

- `tests/conftest.py` autouse seed now produces a fully consistent
  pair: user row (default_workspace_id = test-workspace-autouse) plus
  the workspace itself. The PR5 backfill script's "users without
  default_workspace_id" query no longer picks the fixture up. Insert
  order is user → workspace → UPDATE user, walking around the chicken-
  and-egg FK between `workspaces.owner_id` and `users.default_workspace_id`.
- `tests/test_backfill_workspace_id.py` adds a file-scoped autouse
  fixture that temporarily flips `column.nullable = True` for the four
  business tables (production correctness comes from alembic 0003;
  the script's own job is exactly to fill rows between 0002 and 0003
  so its tests need that transient state to be representable). Its
  `_init_engine` also deletes the autouse seed rows to match the
  "fresh DB" model the tests assume.
- `test_thread_meta_workspace_filter::test_create_workspace_none_bypasses`
  renamed to `test_create_workspace_none_rejected_by_orm` and asserts
  the new IntegrityError on explicit None — write paths can no longer
  bypass workspace scope.
- 5 `test_workspace_context` tests + the auth-middleware reset test
  get `@pytest.mark.no_auto_workspace` so they keep testing the
  unset-contextvar path.
- `test_workspace_repo::test_list_by_user_bypass_returns_all` switches
  to membership assertions instead of strict equality since the
  autouse fixture surfaces under `user_id=None`.

3214 passed, 30 skipped; the remaining 17 are the documented
pre-existing caplog ordering flakes (all pass in isolation).

backend/tests/conftest.py

@@ -66,7 +66,7 @@ sys.modules["deerflow.subagents.executor"] = _executor_mock
 def _register_test_seed_listener() -> None:
     """Attach an after_create hook that seeds the autouse user + workspace."""
     try:
-        from sqlalchemy import event
+        from sqlalchemy import event, update
         from sqlalchemy.dialects.postgresql import insert as pg_insert
         from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 
@@ -86,6 +86,13 @@ def _register_test_seed_listener() -> None:
         dialect = connection.dialect.name
         now = datetime.now(UTC)
 
+        # Seed both rows with a consistent ``default_workspace_id`` so the
+        # PR5 backfill script (which scans ``users.default_workspace_id IS
+        # NULL``) does not pick up the test fixtures as candidates.
+        # Insert user first with NULL default_workspace_id (chicken-and-egg
+        # with workspaces.owner_id FK), then workspace, then UPDATE the user
+        # row to point at the workspace so the PR5 backfill script does not
+        # pick up the autouse user as a candidate.
         user_values = {
             "id": "test-user-autouse",
             "email": "test-user-autouse@local",
@@ -119,6 +126,7 @@ def _register_test_seed_listener() -> None:
 
         connection.execute(user_stmt)
         connection.execute(ws_stmt)
+        connection.execute(update(UserRow.__table__).where(UserRow.__table__.c.id == "test-user-autouse").where(UserRow.__table__.c.default_workspace_id.is_(None)).values(default_workspace_id="test-workspace-autouse"))
 
     event.listen(Base.metadata, "after_create", _seed)
 

backend/tests/test_auth_middleware_workspace.py

@@ -101,6 +101,7 @@ def test_public_path_skips_workspace_check() -> None:
     assert res.status_code == 200
 
 
+@pytest.mark.no_auto_workspace
 def test_workspace_contextvar_resets_between_requests() -> None:
     """After dispatch returns the contextvar must be clear (no leak across requests).
 

backend/tests/test_backfill_workspace_id.py

@@ -40,12 +40,51 @@ def anyio_backend() -> str:
     return "asyncio"
 
 
+_BUSINESS_ROWS = (ThreadMetaRow, RunRow, FeedbackRow, RunEventRow)
+
+
+@pytest.fixture(autouse=True)
+def _relax_workspace_id_nullable():
+    """Simulate alembic 0002 (pre-backfill) state during these tests.
+
+    PR6 T5.11 flipped ``workspace_id`` to ``nullable=False`` on the four
+    business ORM models — production correctness comes from alembic 0003.
+    The backfill script's job is precisely to fill the rows that were
+    inserted between 0002 (column added, nullable) and 0003 (NOT NULL),
+    so tests for it must be able to insert NULL rows. We mutate
+    ``column.nullable`` for the four tables before ``create_all`` runs,
+    then restore on teardown so other tests see the production shape.
+    """
+    saved: list[tuple] = []
+    for model in _BUSINESS_ROWS:
+        col = model.__table__.c.workspace_id
+        saved.append((col, col.nullable))
+        col.nullable = True
+    try:
+        yield
+    finally:
+        for col, original in saved:
+            col.nullable = original
+
+
 async def _init_engine(tmp_path):
+    from sqlalchemy import delete
+
     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()
+    sf = get_session_factory()
+    # The PR6 conftest seeds an autouse user + workspace so business-row FKs
+    # resolve in the wider test suite. Backfill tests model "fresh DB needs
+    # backfill" semantics, so wipe those rows here. Order: clear the FK
+    # pointer first, then the rows.
+    async with sf() as session:
+        await session.execute(delete(WorkspaceMembershipRow))
+        await session.execute(delete(WorkspaceRow).where(WorkspaceRow.id == "test-workspace-autouse"))
+        await session.execute(delete(UserRow).where(UserRow.id == "test-user-autouse"))
+        await session.commit()
+    return sf
 
 
 async def _close():

backend/tests/test_thread_meta_workspace_filter.py

@@ -96,11 +96,18 @@ class TestCreateWorkspace:
             await _cleanup()
 
     @pytest.mark.anyio
-    async def test_create_workspace_none_bypasses(self, tmp_path):
-        """Explicit None creates an orphan row (migration / CLI path)."""
+    async def test_create_workspace_none_rejected_by_orm(self, tmp_path):
+        """After T5.11 (ORM nullable=False) explicit None creates fail at the DB layer.
+
+        Pre-PR6 the migration scripts relied on `workspace_id=None` to insert
+        orphan rows; that use is now restricted to **read** paths (filter
+        bypass). Writes must always carry a workspace.
+        """
+        import sqlalchemy
+
         repo = await _make_repo(tmp_path)
-        record = await repo.create("t1", workspace_id=None)
-        assert record["workspace_id"] is None
+        with pytest.raises((sqlalchemy.exc.IntegrityError, sqlalchemy.exc.DBAPIError)):
+            await repo.create("t1", workspace_id=None)
         await _cleanup()
 
 

backend/tests/test_workspace_context.py

@@ -28,11 +28,13 @@ from deerflow.runtime.workspace_context import (
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.no_auto_workspace
 def test_default_is_none():
     """Before any set, contextvar returns None."""
     assert get_current_workspace() is None
 
 
+@pytest.mark.no_auto_workspace
 def test_set_and_reset_roundtrip():
     """set_current_workspace returns a token that reset restores."""
     workspace = SimpleNamespace(id="ws-1", role="owner")
@@ -44,6 +46,7 @@ def test_set_and_reset_roundtrip():
     assert get_current_workspace() is None
 
 
+@pytest.mark.no_auto_workspace
 def test_require_current_workspace_raises_when_unset():
     """require_current_workspace raises RuntimeError if contextvar is unset."""
     assert get_current_workspace() is None
@@ -93,6 +96,7 @@ def test_default_workspace_id_is_default():
     assert DEFAULT_WORKSPACE_ID == "default"
 
 
+@pytest.mark.no_auto_workspace
 def test_effective_workspace_id_returns_default_when_no_workspace():
     """No workspace in context -> fallback to DEFAULT_WORKSPACE_ID."""
     assert get_effective_workspace_id() == "default"
@@ -132,6 +136,7 @@ def test_resolve_auto_reads_from_contextvar():
         reset_current_workspace(token)
 
 
+@pytest.mark.no_auto_workspace
 def test_resolve_auto_raises_when_unset():
     assert get_current_workspace() is None
     with pytest.raises(RuntimeError, match="workspace_id=AUTO but no workspace"):

backend/tests/test_workspace_repo.py

@@ -284,7 +284,13 @@ async def test_list_by_user_bypass_returns_all(tmp_path):
         await repo.create(name="B", slug="all-b", owner_id="u-B")
 
         workspaces = await repo.list_by_user(user_id=None)
+        # PR6 conftest auto-seeds an "autouse-test" workspace via the
+        # ``Base.metadata.after_create`` hook so business-row FKs resolve.
+        # ``user_id=None`` bypasses the membership filter, so it surfaces
+        # alongside the two rows the test inserted — that is the intended
+        # "no filter" behaviour. Assert the inserted ones are present.
         slugs = sorted(w["slug"] for w in workspaces)
-        assert slugs == ["all-a", "all-b"]
+        assert "all-a" in slugs
+        assert "all-b" in slugs
     finally:
         await _cleanup()