deer-flow/backend/tests/test_owner_isolation.py
heart-scalpel 3bc3af2530
fix(runs): close multi-worker ownership gaps in run atomicity (#3948) (#4003)
* feat(runs): cross-process run ownership with lease + reconciliation (#3948)

Implements work items 2 and 3 of the multi-worker P0 plan
(docs/multi_worker.md). Work item 1 (Postgres startup gate, #3960)
already landed; this PR makes run creation race-safe across worker
processes and lets Postgres deployments recover orphaned inflight runs
from crashed workers without mis-marking live runs as orphans.

Work item 2 — cross-process atomic create_or_reject

- Alembic revision 0004_run_ownership adds runs.owner_worker_id,
  runs.lease_expires_at, idx_runs_lease, and a partial unique index
  uq_runs_thread_active (one pending/running run per thread). The
  index is declared on RunRow.__table_args__ with sqlite_where +
  postgresql_where (mirroring uq_channel_connection_active_identity)
  so the empty-DB bootstrap path — which runs Base.metadata.create_all
  + alembic stamp head without executing any revision's upgrade() —
  also lands it on fresh deployments. Migration 0004 additionally
  creates it idempotently for legacy/versioned upgrades.
- RunRepository.create_run_atomic is the new atomic primitive:
  - reject: INSERT directly; the partial unique index catches
    duplicate active runs; the manager surfaces the result as
    ConflictError.
  - interrupt/rollback: SELECT FOR UPDATE the conflicting rows,
    skip rows whose lease is still valid AND owned by another live
    worker (raise ConflictError — the INSERT would have failed on
    the index anyway, and a retry loop cannot make progress),
    cancel the rest in the same transaction, then INSERT the new
    row. Rows owned by this worker are interruptible regardless of
    lease state.
- RunManager.create_or_reject dispatches to the store under the
  existing local lock; same-worker in-memory cancellation runs after
  the store commit succeeds. MemoryRunStore mirrors the same
  semantics for tests and database.backend=memory.

Work item 3 — lease heartbeat + Postgres reconciliation

- RunOwnershipConfig (lease_seconds=30, grace_seconds=10,
  heartbeat_enabled=false by default), registered as startup-only in
  reload_boundary.STARTUP_ONLY_FIELDS because the heartbeat background
  task is created once in langgraph_runtime() and is not rebuilt on
  config.yaml edits.
- When heartbeat_enabled, each worker renewes leases on its own
  active runs with interval = lease_seconds / 3. The loop is bounded
  and stop-event-cancellable so shutdown is prompt.
- reconcile_orphaned_inflight_runs now runs on every backend — the
  sqlite-only gate in app/gateway/deps.py is dropped in the same
  commit so there is no window where Postgres would mis-mark live
  Worker A runs as orphans. Reconciliation errors only runs whose
  lease is NULL (legacy pre-ownership rows) or older than
  grace_seconds. In single-worker mode (heartbeat off, NULL leases)
  all inflight rows reclaim immediately, preserving the pre-ownership
  recovery latency.
- Heartbeat starts AFTER startup reconciliation and stops BEFORE the
  in-flight run drain on shutdown so the two cannot race.

GATEWAY_WORKERS=1 with heartbeat_enabled=false keeps current behavior.

Verified: 170 related tests + full backend suite (minus Docker-gated
live tests) green; ruff check + ruff format clean.

* fix(runs): tighten unique-violation handling and document clock-sync budget

Three follow-up fixes to the cross-process run ownership work in #3948,
surfacing during review.

1. _is_unique_violation: detect by driver-native signal, not message text

   The previous substring heuristic ("unique" + "violat", or "duplicate")
   missed SQLite's actual phrasing "UNIQUE constraint failed: <table>.<index>"
   — SQLite says "failed", not "violates", and never "duplicate". On SQLite
   the detector returned False, the reject path re-raised the raw
   IntegrityError, and clients saw HTTP 500 instead of ConflictError 409.
   The conversion is the load-bearing piece of the "store is source of
   truth" design but was untested — every atomic test used MemoryRunStore,
   which raises ConflictError directly and never reached this branch.

   Now prefers driver-native signals: psycopg pgcode/sqlcode "23505" and
   sqlite3 sqlite_errorcode SQLITE_CONSTRAINT_UNIQUE (reachable through
   SQLAlchemy IntegrityError.orig). Message matching stays as a fallback
   with SQLite's exact "unique constraint failed" phrase added.

2. interrupt/rollback: convert exhausted-retry IntegrityError to ConflictError

   The reject branch converts unique violations to ConflictError. The
   interrupt/rollback retry loop did not — on the 3rd attempt it re-raised
   the raw IntegrityError, leaking HTTP 500 for the same race condition
   that reject surfaces as 409. Symmetric conversion added after the loop;
   callers now see a consistent ConflictError regardless of strategy.

3. Document clock-sync requirement for multi-worker lease reconciliation

   reconcile_orphaned_inflight_runs compares another worker's UTC
   lease_expires_at against this worker's datetime.now(UTC). The only skew
   budget is grace_seconds (default 10s) — worst case, with the owning
   worker's heartbeat just about to fire, a peer whose clock is more than
   ~grace_seconds ahead can mis-reclaim a still-live run as an orphan.

   Documented in RunOwnershipConfig's docstring (with the math) and in
   config.example.yaml (with operational guidance), so operators in
   NTP-poor environments know to raise grace_seconds. Default unchanged:
   10s is reasonable for NTP-synced K8s/cloud, and bumping it would slow
   recovery of genuinely dead workers (lease_seconds + grace_seconds from
   last heartbeat to reclaim).

Tests:
- test_create_run_atomic_reject_propagates_conflict_on_unique_violation:
  end-to-end against a real SQLite-backed RunRepository, pre-inserts an
  active run, asserts reject-strategy create surfaces as ConflictError
  rather than raw IntegrityError.
- test_is_unique_violation_detects_real_sqlite_integrity_error: unit test
  for the detector against a real SQLite-raised IntegrityError; asserts
  driver-level sqlite_errorcode is SQLITE_CONSTRAINT_UNIQUE.
- test_interrupt_exhausted_retries_surface_as_conflict_error: pins the
  symmetric 409 behavior after the retry loop exhausts.

Verified: ruff check + ruff format clean; multi-worker + run_repository
+ owner_isolation + reload_boundary suites green.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): close multi-worker ownership gaps in lease heartbeat and unique-violation detection

Five code-review fixes from docs/multi_worker.md:

1. Drop unused ``claim_inflight_runs`` primitive — no caller anywhere.
   ``create_run_atomic`` does its own inline claim (SELECT FOR UPDATE +
   cancel) inside the INSERT transaction; a separate claim primitive
   would split that into two transactions and open a claim→INSERT race.
   Removes ~40 lines across base.py / memory.py / sql.py plus the
   unused ``now_iso`` parameter, freeing future RunStore implementations
   from providing it.

2. Broaden ``_renew_leases`` filter to renew pending/running runs owned
   by this worker even when ``record.task is None``. The previous
   ``task is not None`` requirement skipped the brief window between
   ``create_run_atomic`` inserting the row and the worker spawning the
   agent task; under event-loop load that window can approach
   ``lease_seconds``, after which peer reconciliation marks the run
   ``error`` (visible) or a peer's ``create_or_reject("interrupt")``
   silently kills the queued run. Filter now:
   ``task is None or not task.done()``.

3. Document the unsynchronised ``record.lease_expires_at = new_expiry``
   write. ``lease_expires_at`` is the only field on an existing record
   this path mutates; ``set_status`` / ``_persist_status`` touch other
   fields, so there is no concurrent writer to race against. Re-acquiring
   ``self._lock`` would serialise unrelated run mutations for no gain.

4. Gate ``_is_unique_violation`` message fallbacks on
   ``isinstance(current, (SAIntegrityError, sqlite3.IntegrityError))``.
   The driver-code path (pgcode/sqlite_errorcode) remains load-bearing;
   substring fallbacks are now belt-and-suspenders only for cases where
   the driver attribute isn't reachable through the cause chain. Without
   the gate, any application exception whose ``str()`` happens to contain
   "duplicate key" / "unique" + "violat" (CHECK constraint, validation
   error) would silently surface as HTTP 409 instead of 500.

5. Route ``update_lease`` through ``_call_store_with_retry`` for
   consistency with every other store call, and wrap
   ``await self._renew_leases()`` in ``_heartbeat_loop`` with
   ``except Exception: logger.warning(...)``. Previously a transient
   error from the snapshot path or an unexpected exception would kill
   the heartbeat task silently — after which no lease is ever renewed
   again and every active run eventually looks orphaned.
   ``except Exception`` lets ``CancelledError`` (BaseException since
   3.8) propagate so shutdown cancellation still works.

Regression tests:
- ``test_heartbeat_renews_pending_run_before_task_is_spawned``
- ``test_is_unique_violation_does_not_misclassify_application_exception``

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): harden multi-worker migration, memory atomicity, and tz-naive lease comparison

Three follow-up fixes to the multi-worker run ownership work:

- migration 0004 dedupe pass: cancel superseded duplicate active rows per
  thread before creating the partial UNIQUE index ``uq_runs_thread_active``
  so dirty DBs (Postgres deployments that had reconciliation skipped by the
  old sqlite-only gate, or any env that ran GATEWAY_WORKERS>1 before this PR)
  do not abort the alembic upgrade and block gateway startup. Keeps the
  newest active row per thread, marks the rest as error with an explanatory
  message.

- MemoryRunStore.create_run_atomic interrupt/rollback path: split the single-
  pass loop into two passes (collect candidates, validate, then mutate) so a
  ConflictError raised on a later candidate does not leave earlier candidates
  half-interrupted. Mirrors the SQL store's transactional rollback semantics;
  the entire test_multi_worker_run_ownership.py suite runs against memory so
  this divergence was giving false confidence.

- RunRepository.create_run_atomic interrupt path: coerce tz-naive
  ``row.lease_expires_at`` to UTC before comparing against the aware
  ``cutoff``. SQLite drops tzinfo on read despite ``DateTime(timezone=True)``
  (this file's own comment acknowledges it), so the Python-side comparison
  raised ``TypeError: can't compare offset-naive and offset-aware datetimes``
  whenever heartbeat was enabled on SQLite and a lease was non-NULL. Defaults
  (heartbeat off -> leases always NULL) masked it, but there was no guard
  against the combination. Follows the existing "naive is UTC" convention
  from ``coerce_iso``.

Each fix ships with a regression test pinning the behavior.

Co-Authored-By: heart-scalpel <heart-scalpel@users.noreply.github.com>
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): enforce heartbeat for multi-worker, fix memory-store datetime comparison, lazy-import ConflictError in store layer

Three fixes from code review:

1. Extend the startup gate (GATEWAY_WORKERS>1) to also require
   run_ownership.heartbeat_enabled=true. Without heartbeat every run has
   a NULL lease, so reconciliation treats all inflight rows as orphans
   and Worker B would kill Worker A's live runs on every rolling update
   or scale-up.

2. Fix MemoryRunStore.list_inflight_with_expired_lease to parse
   created_at as datetime instead of ISO string lexical comparison,
   and handle tz-naive lease values uniformly with the SQL store.

3. Store layer (sql.py, memory.py) now lazy-imports ConflictError
   inside create_run_atomic instead of importing from the higher
   RunManager layer at module level.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): add owner check to update_lease, document create() assumption, restore deleted comment

- update_lease (SQL + memory) now requires owner_worker_id match in WHERE
  clause so the primitive is safe by construction against misuse
- create() docstring notes it bypasses atomic create_run_atomic and
  assumes no active run exists for the thread
- restore explanatory comment in MemoryRunStore.aggregate_tokens_by_thread
  that was dropped in an earlier commit

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): add psycopg3 sqlstate detection and periodic orphan reconciliation

- _is_unique_violation now checks sqlstate attribute (psycopg3 uses this
  instead of pgcode). On Postgres, the only supported multi-worker backend,
  detection was falling through to the message-substring fallback.
- _heartbeat_loop now runs reconcile_orphaned_inflight_runs every 3rd
  cycle (every lease_seconds) to catch orphans whose lease expires between
  pod restarts. Single-worker deployments are unaffected (heartbeat off).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: heart-scalpel <heart-scalpel@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-07-11 16:05:30 +08:00

466 lines
16 KiB
Python

"""Cross-user isolation tests — non-negotiable safety gate.
Mirrors TC-API-17..20 from backend/docs/AUTH_TEST_PLAN.md. A failure
here means users can see each other's data; PR must not merge.
Architecture note
-----------------
These tests bypass the HTTP layer and exercise the storage-layer
owner filter directly by switching the ``user_context`` contextvar
between two users. The safety property under test is:
After a repository write with user_id=A, a subsequent read with
user_id=B must not return the row, and vice versa.
The HTTP layer is covered by test_auth_middleware.py, which proves
that a request cookie reaches the ``set_current_user`` call. Together
the two suites prove the full chain:
cookie → middleware → contextvar → repository → isolation
Every test in this file opts out of the autouse contextvar fixture
(``@pytest.mark.no_auto_user``) so it can set the contextvar to the
specific users it cares about.
"""
from __future__ import annotations
from types import SimpleNamespace
import pytest
from deerflow.runtime.user_context import (
reset_current_user,
set_current_user,
)
USER_A = SimpleNamespace(id="user-a", email="a@test.local")
USER_B = SimpleNamespace(id="user-b", email="b@test.local")
async def _make_engines(tmp_path):
"""Initialize the shared engine against a per-test SQLite DB.
Returns a cleanup coroutine the caller should await at the end.
"""
from deerflow.persistence.engine import close_engine, init_engine
url = f"sqlite+aiosqlite:///{tmp_path / 'isolation.db'}"
await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
return close_engine
def _as_user(user):
"""Context manager-like helper that set/reset the contextvar."""
class _Ctx:
def __enter__(self):
self._token = set_current_user(user)
return user
def __exit__(self, *exc):
reset_current_user(self._token)
return _Ctx()
# ── TC-API-17 — threads_meta isolation ────────────────────────────────────
@pytest.mark.anyio
@pytest.mark.no_auto_user
async def test_thread_meta_cross_user_isolation(tmp_path):
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.thread_meta import ThreadMetaRepository
cleanup = await _make_engines(tmp_path)
try:
repo = ThreadMetaRepository(get_session_factory())
# User A creates a thread.
with _as_user(USER_A):
await repo.create("t-alpha", display_name="A's private thread")
# User B creates a thread.
with _as_user(USER_B):
await repo.create("t-beta", display_name="B's private thread")
# User A must see only A's thread.
with _as_user(USER_A):
a_view = await repo.get("t-alpha")
assert a_view is not None
assert a_view["display_name"] == "A's private thread"
# CRITICAL: User A must NOT see B's thread.
leaked = await repo.get("t-beta")
assert leaked is None, f"User A leaked User B's thread: {leaked}"
# Search should only return A's threads.
results = await repo.search()
assert [r["thread_id"] for r in results] == ["t-alpha"]
# User B must see only B's thread.
with _as_user(USER_B):
b_view = await repo.get("t-beta")
assert b_view is not None
assert b_view["display_name"] == "B's private thread"
leaked = await repo.get("t-alpha")
assert leaked is None, f"User B leaked User A's thread: {leaked}"
results = await repo.search()
assert [r["thread_id"] for r in results] == ["t-beta"]
finally:
await cleanup()
@pytest.mark.anyio
@pytest.mark.no_auto_user
async def test_thread_meta_cross_user_mutation_denied(tmp_path):
"""User B cannot update or delete a thread owned by User A."""
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.thread_meta import ThreadMetaRepository
cleanup = await _make_engines(tmp_path)
try:
repo = ThreadMetaRepository(get_session_factory())
with _as_user(USER_A):
await repo.create("t-alpha", display_name="original")
# User B tries to rename A's thread — must be a no-op.
with _as_user(USER_B):
await repo.update_display_name("t-alpha", "hacked")
# Verify the row is unchanged from A's perspective.
with _as_user(USER_A):
row = await repo.get("t-alpha")
assert row is not None
assert row["display_name"] == "original"
# User B tries to delete A's thread — must be a no-op.
with _as_user(USER_B):
await repo.delete("t-alpha")
# A's thread still exists.
with _as_user(USER_A):
row = await repo.get("t-alpha")
assert row is not None
finally:
await cleanup()
# ── TC-API-18 — runs isolation ────────────────────────────────────────────
@pytest.mark.anyio
@pytest.mark.no_auto_user
async def test_runs_cross_user_isolation(tmp_path):
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.run import RunRepository
cleanup = await _make_engines(tmp_path)
try:
repo = RunRepository(get_session_factory())
with _as_user(USER_A):
await repo.put("run-a1", thread_id="t-alpha", status="success")
await repo.put("run-a2", thread_id="t-alpha", status="pending")
with _as_user(USER_B):
await repo.put("run-b1", thread_id="t-beta")
# User A must see only A's runs.
with _as_user(USER_A):
r = await repo.get("run-a1")
assert r is not None
assert r["run_id"] == "run-a1"
leaked = await repo.get("run-b1")
assert leaked is None, "User A leaked User B's run"
a_runs = await repo.list_by_thread("t-alpha")
assert {r["run_id"] for r in a_runs} == {"run-a1", "run-a2"}
# Listing B's thread from A's perspective: empty
empty = await repo.list_by_thread("t-beta")
assert empty == []
# User B must see only B's runs.
with _as_user(USER_B):
leaked = await repo.get("run-a1")
assert leaked is None, "User B leaked User A's run"
b_runs = await repo.list_by_thread("t-beta")
assert [r["run_id"] for r in b_runs] == ["run-b1"]
finally:
await cleanup()
@pytest.mark.anyio
@pytest.mark.no_auto_user
async def test_runs_cross_user_delete_denied(tmp_path):
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.run import RunRepository
cleanup = await _make_engines(tmp_path)
try:
repo = RunRepository(get_session_factory())
with _as_user(USER_A):
await repo.put("run-a1", thread_id="t-alpha")
# User B tries to delete A's run — no-op.
with _as_user(USER_B):
await repo.delete("run-a1")
# A's run still exists.
with _as_user(USER_A):
row = await repo.get("run-a1")
assert row is not None
finally:
await cleanup()
# ── TC-API-19 — run_events isolation (CRITICAL: content leak) ─────────────
@pytest.mark.anyio
@pytest.mark.no_auto_user
async def test_run_events_cross_user_isolation(tmp_path):
"""run_events holds raw conversation content — most sensitive leak vector."""
from deerflow.persistence.engine import get_session_factory
from deerflow.runtime.events.store.db import DbRunEventStore
cleanup = await _make_engines(tmp_path)
try:
store = DbRunEventStore(get_session_factory())
with _as_user(USER_A):
await store.put(
thread_id="t-alpha",
run_id="run-a1",
event_type="human_message",
category="message",
content="User A private question",
)
await store.put(
thread_id="t-alpha",
run_id="run-a1",
event_type="ai_message",
category="message",
content="User A private answer",
)
with _as_user(USER_B):
await store.put(
thread_id="t-beta",
run_id="run-b1",
event_type="human_message",
category="message",
content="User B private question",
)
# User A must see only A's events — CRITICAL.
with _as_user(USER_A):
msgs = await store.list_messages("t-alpha")
contents = [m["content"] for m in msgs]
assert "User A private question" in contents
assert "User A private answer" in contents
# CRITICAL: User B's content must not appear.
assert "User B private question" not in contents
# Attempt to read B's thread by guessing thread_id.
leaked = await store.list_messages("t-beta")
assert leaked == [], f"User A leaked User B's messages: {leaked}"
leaked_events = await store.list_events("t-beta", "run-b1")
assert leaked_events == [], "User A leaked User B's events"
# count_messages must also be zero for B's thread from A's view.
count = await store.count_messages("t-beta")
assert count == 0
# User B must see only B's events.
with _as_user(USER_B):
msgs = await store.list_messages("t-beta")
contents = [m["content"] for m in msgs]
assert "User B private question" in contents
assert "User A private question" not in contents
assert "User A private answer" not in contents
count = await store.count_messages("t-alpha")
assert count == 0
finally:
await cleanup()
@pytest.mark.anyio
@pytest.mark.no_auto_user
async def test_run_events_cross_user_delete_denied(tmp_path):
"""User B cannot delete User A's event stream."""
from deerflow.persistence.engine import get_session_factory
from deerflow.runtime.events.store.db import DbRunEventStore
cleanup = await _make_engines(tmp_path)
try:
store = DbRunEventStore(get_session_factory())
with _as_user(USER_A):
await store.put(
thread_id="t-alpha",
run_id="run-a1",
event_type="human_message",
category="message",
content="hello",
)
# User B tries to wipe A's thread events.
with _as_user(USER_B):
removed = await store.delete_by_thread("t-alpha")
assert removed == 0, f"User B deleted {removed} of User A's events"
# A's events still exist.
with _as_user(USER_A):
count = await store.count_messages("t-alpha")
assert count == 1
finally:
await cleanup()
# ── TC-API-20 — feedback isolation ────────────────────────────────────────
@pytest.mark.anyio
@pytest.mark.no_auto_user
async def test_feedback_cross_user_isolation(tmp_path):
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.feedback import FeedbackRepository
cleanup = await _make_engines(tmp_path)
try:
repo = FeedbackRepository(get_session_factory())
# User A submits positive feedback.
with _as_user(USER_A):
a_feedback = await repo.create(
run_id="run-a1",
thread_id="t-alpha",
rating=1,
comment="A liked this",
)
# User B submits negative feedback.
with _as_user(USER_B):
b_feedback = await repo.create(
run_id="run-b1",
thread_id="t-beta",
rating=-1,
comment="B disliked this",
)
# User A must see only A's feedback.
with _as_user(USER_A):
retrieved = await repo.get(a_feedback["feedback_id"])
assert retrieved is not None
assert retrieved["comment"] == "A liked this"
# CRITICAL: cannot read B's feedback by id.
leaked = await repo.get(b_feedback["feedback_id"])
assert leaked is None, "User A leaked User B's feedback"
# list_by_run for B's run must be empty.
empty = await repo.list_by_run("t-beta", "run-b1")
assert empty == []
# User B must see only B's feedback.
with _as_user(USER_B):
leaked = await repo.get(a_feedback["feedback_id"])
assert leaked is None, "User B leaked User A's feedback"
b_list = await repo.list_by_run("t-beta", "run-b1")
assert len(b_list) == 1
assert b_list[0]["comment"] == "B disliked this"
finally:
await cleanup()
@pytest.mark.anyio
@pytest.mark.no_auto_user
async def test_feedback_cross_user_delete_denied(tmp_path):
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.feedback import FeedbackRepository
cleanup = await _make_engines(tmp_path)
try:
repo = FeedbackRepository(get_session_factory())
with _as_user(USER_A):
fb = await repo.create(run_id="run-a1", thread_id="t-alpha", rating=1)
# User B tries to delete A's feedback — must return False (no-op).
with _as_user(USER_B):
deleted = await repo.delete(fb["feedback_id"])
assert deleted is False, "User B deleted User A's feedback"
# A's feedback still retrievable.
with _as_user(USER_A):
row = await repo.get(fb["feedback_id"])
assert row is not None
finally:
await cleanup()
# ── Regression: AUTO sentinel without contextvar must raise ───────────────
@pytest.mark.anyio
@pytest.mark.no_auto_user
async def test_repository_without_context_raises(tmp_path):
"""Defense-in-depth: calling repo methods without a user context errors."""
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.thread_meta import ThreadMetaRepository
cleanup = await _make_engines(tmp_path)
try:
repo = ThreadMetaRepository(get_session_factory())
# Contextvar is explicitly unset under @pytest.mark.no_auto_user.
with pytest.raises(RuntimeError, match="no user context is set"):
await repo.get("anything")
finally:
await cleanup()
# ── Escape hatch: explicit user_id=None bypasses filter (for migration) ──
@pytest.mark.anyio
@pytest.mark.no_auto_user
async def test_explicit_none_bypasses_filter(tmp_path):
"""Migration scripts pass user_id=None to see all rows regardless of owner."""
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.thread_meta import ThreadMetaRepository
cleanup = await _make_engines(tmp_path)
try:
repo = ThreadMetaRepository(get_session_factory())
# Seed data as two different users.
with _as_user(USER_A):
await repo.create("t-alpha")
with _as_user(USER_B):
await repo.create("t-beta")
# Migration-style read: no contextvar, explicit None bypass.
all_rows = await repo.search(user_id=None)
thread_ids = {r["thread_id"] for r in all_rows}
assert thread_ids == {"t-alpha", "t-beta"}
# Explicit get with None does not apply the filter either.
row_a = await repo.get("t-alpha", user_id=None)
assert row_a is not None
row_b = await repo.get("t-beta", user_id=None)
assert row_b is not None
finally:
await cleanup()