mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
* 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>
466 lines
16 KiB
Python
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()
|