From c540399a5e44def10f3e2a76ae6c5f4799188075 Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Mon, 5 Oct 2026 11:23:55 +0900 Subject: [PATCH] fix(gateway): scope per-run reads to the run's data identity (#6282) * fix(gateway): scope per-run message and event reads to the run's data identity start_run stamps run-event rows with the raw trusted owner id, but GET /runs/{rid}/messages and /runs/{rid}/events called the event store with user_id=AUTO, which resolves to the internal caller's make_safe_user_id form. On the database event store, owners whose id contains characters outside [A-Za-z0-9_-] (IM channel ids such as "feishu:owner-777") therefore got an empty list. The regenerate source-run scan had the same mismatch and fell back or returned 409. Pass _run_scope_user_id() explicitly, as the thread message routes already do (#5448), and add user_id to RunEventStore.list_messages_by_run() so every backend accepts it like list_messages()/list_events(). Browser and API sessions keep their per-user filter. * fix(gateway): scope workspace-changes and archive reads to the run's data identity The same raw-owner mismatch reached two more run-scoped routes. On the SQL stores, GET /runs/{rid}/workspace-changes reported no changes because its event lookup used user_id=AUTO, and GET /runs/{rid}/artifacts/archive answered 404 because both its run-row lookup and its run.delivery event lookup did. Pass _run_scope_user_id() to both, and let get_workspace_changes_response() forward an explicit user_id to the event store. * docs(changelog): reference #6282 in the run-scope read fix entries --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 12 ++ CHANGELOG_zh.md | 10 + backend/app/gateway/routers/thread_runs.py | 24 ++- .../harness/deerflow/runtime/AGENTS.md | 7 + .../deerflow/runtime/events/store/base.py | 3 + .../deerflow/runtime/events/store/jsonl.py | 2 +- .../deerflow/runtime/events/store/memory.py | 2 +- .../harness/deerflow/workspace_changes/api.py | 4 + backend/tests/test_run_event_store.py | 35 +++ backend/tests/test_run_events_endpoint.py | 10 +- .../tests/test_thread_regenerate_prepare.py | 2 +- .../test_thread_run_messages_pagination.py | 3 + .../tests/test_thread_run_query_validation.py | 2 + .../tests/test_thread_runs_internal_scope.py | 203 +++++++++++++++++- backend/tests/test_workspace_changes.py | 3 +- 15 files changed, 306 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a672add8ed..97932d5aa3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -466,6 +466,17 @@ This release closes that milestone with **301 merged pull requests**. ### Fixed +- **gateway:** Per-run reads now return the rows of IM-channel owners. + `start_run` stamps run rows and run events with the raw trusted owner id (for + example `feishu:owner-777`), but several run-scoped routes filtered by the + internal caller's normalized id, so on the SQL stores any owner id containing + characters outside `[A-Za-z0-9_-]` matched nothing: + `GET /api/threads/{id}/runs/{rid}/messages` and `/events` returned an empty + list, `/workspace-changes` reported no changes, `/artifacts/archive` answered + 404, and the regenerate source-run lookup fell back or failed with 409. These + reads now use the same data identity as the thread message routes (#5448), + and every event store accepts `user_id` on `list_messages_by_run()`. Browser + and API sessions keep their per-user filter. ([#6282]) - **runtime:** A multi-worker run that finished successfully is no longer reclaimed as an orphan `error` while its worker is still finalizing. With an event store, the worker records the terminal status in memory first and @@ -7722,3 +7733,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6238]: https://github.com/bytedance/deer-flow/pull/6238 [#6255]: https://github.com/bytedance/deer-flow/pull/6255 [#6263]: https://github.com/bytedance/deer-flow/pull/6263 +[#6282]: https://github.com/bytedance/deer-flow/pull/6282 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 8b5317bc2b..487c16f4aa 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -423,6 +423,15 @@ ### 修复 +- **网关：** 单次运行的读取现在能返回 IM 渠道所有者的数据。`start_run` 用原始 + 受信所有者 ID（例如 `feishu:owner-777`）标记运行行与运行事件，但多个运行级路由 + 按内部调用方规范化后的 ID 过滤，因此在 SQL 存储上，只要所有者 ID 含有 + `[A-Za-z0-9_-]` 以外的字符就匹配不到任何数据： + `GET /api/threads/{id}/runs/{rid}/messages` 与 `/events` 返回空列表， + `/workspace-changes` 报告没有变更，`/artifacts/archive` 返回 404，重新生成的 + 源运行查找退回兜底路径或返回 409。这些读取现在使用与线程消息路由（#5448）相同 + 的数据身份，所有事件存储的 `list_messages_by_run()` 都接受 `user_id`。浏览器与 + API 会话仍保留按用户过滤。([#6282]) - **运行时：** 多 worker 部署中已成功结束的运行，不会再在其 worker 仍在收尾时被 当作孤儿运行回收为 `error`。配置了事件存储时，worker 先在内存中记录终态，直到 journal 刷新、交付回执、工作区扫描与时长 checkpoint 写入完成后才写入运行存储。 @@ -6445,3 +6454,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6238]: https://github.com/bytedance/deer-flow/pull/6238 [#6255]: https://github.com/bytedance/deer-flow/pull/6255 [#6263]: https://github.com/bytedance/deer-flow/pull/6263 +[#6282]: https://github.com/bytedance/deer-flow/pull/6282 diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index 6a49c69bf5..74afd2e662 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -604,8 +604,9 @@ async def _find_target_run_id( source_human: Any, request: Request, ) -> str: + user_id = await _run_scope_user_id(request, thread_id) event_store = get_run_event_store(request) - rows = await event_store.list_messages(thread_id, limit=REGENERATE_HISTORY_SCAN_LIMIT) + rows = await event_store.list_messages(thread_id, limit=REGENERATE_HISTORY_SCAN_LIMIT, user_id=user_id) for row in reversed(rows): if row.get("event_type") not in {"ai_message", "llm.ai.response"}: continue @@ -619,7 +620,6 @@ async def _find_target_run_id( return source_run_id run_mgr = get_run_manager(request) - user_id = await _run_scope_user_id(request, thread_id) records = await run_mgr.list_by_thread(thread_id, user_id=user_id, limit=10) fallback_record = next( (record for record in records if record.status == RunStatus.success and _run_last_ai_matches_message(record, target_message)), @@ -1159,10 +1159,12 @@ async def _require_run_visible_to_scope(run_id: str, thread_id: str, request: Re """Gate run-scoped sub-resource reads and writes (events, messages, join, stream, cancel, artifact archive). - These routes query or mutate by ``(thread_id, run_id)`` without a - per-user filter of their own. For trusted internal callers on threads - without established ownership, that let an internal caller acting for - owner A read or cancel owner B's run by id (#5448 review P1 follow-up). + These routes query or mutate by ``(thread_id, run_id)``; events and + messages add only the ``_run_scope_user_id`` data filter, which matches + row stamps rather than gating the run itself. For trusted internal + callers on threads without established ownership, a missing gate let an + internal caller acting for owner A read or cancel owner B's run by id + (#5448 review P1 follow-up). The run's own stamp must therefore match the acting owner's raw value (or the legacy ``"default"`` stamp); every other caller and every established-ownership thread keeps its existing semantics. @@ -1638,6 +1640,7 @@ async def list_run_messages( Response: { data: [...], has_more: bool } """ await _require_run_visible_to_scope(run_id, thread_id, request) + user_id = await _run_scope_user_id(request, thread_id) event_store = get_run_event_store(request) rows = await event_store.list_messages_by_run( thread_id, @@ -1645,6 +1648,7 @@ async def list_run_messages( limit=limit + 1, before_seq=before_seq, after_seq=after_seq, + user_id=user_id, ) data, has_more = trim_run_message_page(rows, limit=limit, after_seq=after_seq) @@ -1718,7 +1722,8 @@ def _presented_files_from_delivery(events: list[dict]) -> list[str]: async def _archive_presented_paths(thread_id: ThreadId, run_id: str, request: Request) -> list[str]: - run = await get_run_store(request).get(run_id) + user_id = await _run_scope_user_id(request, thread_id) + run = await get_run_store(request).get(run_id, user_id=user_id) if run is None or run.get("thread_id") != thread_id or run.get("operation_kind", "run") != "run": raise HTTPException(status_code=404, detail=f"Run {run_id} not found") if run.get("status") in {RunStatus.pending.value, RunStatus.running.value}: @@ -1729,6 +1734,7 @@ async def _archive_presented_paths(thread_id: ThreadId, run_id: str, request: Re run_id, event_types=["run.delivery"], limit=2, + user_id=user_id, ) return _presented_files_from_delivery(events) @@ -1825,6 +1831,7 @@ async def list_run_events( task's persisted steps without the run-wide ``limit`` truncating the tail (#3779). """ await _require_run_visible_to_scope(run_id, thread_id, request) + user_id = await _run_scope_user_id(request, thread_id) event_store = get_run_event_store(request) types = event_types.split(",") if event_types else None events = await event_store.list_events( @@ -1834,6 +1841,7 @@ async def list_run_events( task_id=task_id, limit=limit, after_seq=after_seq, + user_id=user_id, ) return [ { @@ -1857,6 +1865,7 @@ async def get_run_workspace_changes( ) -> dict: """Return workspace/output file changes recorded for one run.""" await _require_run_visible_to_scope(run_id, thread_id, request) + user_id = await _run_scope_user_id(request, thread_id) event_store = get_run_event_store(request) return await get_workspace_changes_response( event_store, @@ -1864,6 +1873,7 @@ async def get_run_workspace_changes( run_id, include_files=include_files, include_diff=include_diff, + user_id=user_id, ) diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 10304622de..7926da44a9 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -132,6 +132,13 @@ writer enter the supposedly stable snapshot. The default and JSONL paths share t `normalize_message_ids()` and `match_ai_message_run_id()` helpers from `events/store/base.py`. Database owner filtering is inherited on every page. +**Run-event read identity**: `list_messages`, `list_events` and +`list_messages_by_run` accept `user_id` on every backend (DB filters; +memory/JSONL accept it for parity). `start_run` stamps rows with the raw trusted +owner, but `AUTO` resolves to the internal user's `make_safe_user_id` form, so +Gateway thread/run reads (including run-row lookups) must pass +`_run_scope_user_id()` explicitly. + **Event-store mutation fence** (`runtime/events/store/`): every thread mutation — `put`, `put_batch`, `put_if_absent`, `delete_by_thread`, `delete_by_run` — shares one serialization domain: the per-thread `asyncio` lock, plus (on PostgreSQL) the diff --git a/backend/packages/harness/deerflow/runtime/events/store/base.py b/backend/packages/harness/deerflow/runtime/events/store/base.py index afd54be0a1..29bdfac302 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/base.py +++ b/backend/packages/harness/deerflow/runtime/events/store/base.py @@ -218,6 +218,7 @@ class RunEventStore(abc.ABC): limit: int = 50, before_seq: int | None = None, after_seq: int | None = None, + user_id: str | None | _AutoSentinel = AUTO, ) -> list[dict]: """Return displayable messages (category=message) for a specific run, ordered by seq ascending. @@ -228,6 +229,8 @@ class RunEventStore(abc.ABC): When both cursors are supplied, return the first ``limit`` records in the exclusive window ``after_seq < seq < before_seq`` (ascending). + ``user_id`` follows the same explicit-caller semantics as + :meth:`list_messages`. """ @abc.abstractmethod diff --git a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py index 4a79a0bad1..139a227fa7 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py +++ b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py @@ -399,7 +399,7 @@ class JsonlRunEventStore(RunEventStore): events = [e for e in events if e.get("seq", 0) > after_seq] return events[:limit] - async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None): + async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None, user_id: str | None | _AutoSentinel = AUTO): events = await asyncio.to_thread(self._read_run_events, thread_id, run_id) filtered = [e for e in events if e.get("category") == "message"] if before_seq is not None: diff --git a/backend/packages/harness/deerflow/runtime/events/store/memory.py b/backend/packages/harness/deerflow/runtime/events/store/memory.py index 8b8475e821..ade56ca4c9 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/memory.py +++ b/backend/packages/harness/deerflow/runtime/events/store/memory.py @@ -164,7 +164,7 @@ class MemoryRunEventStore(RunEventStore): run_events = [e for e in run_events if e.get("seq", 0) > after_seq] return run_events[:limit] - async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None): + async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None, user_id: str | None | _AutoSentinel = AUTO): # Per-run, messages-only, seq-sorted: the seq window is a contiguous # slice located with bisect (O(log m_run)) over only this run's # messages, instead of re-scanning the whole thread's event log. diff --git a/backend/packages/harness/deerflow/workspace_changes/api.py b/backend/packages/harness/deerflow/workspace_changes/api.py index ac4bddbfbe..16948fc528 100644 --- a/backend/packages/harness/deerflow/workspace_changes/api.py +++ b/backend/packages/harness/deerflow/workspace_changes/api.py @@ -2,6 +2,8 @@ from __future__ import annotations from typing import Any +from deerflow.runtime.user_context import AUTO, _AutoSentinel + from .types import WORKSPACE_CHANGES_EVENT_TYPE, WORKSPACE_CHANGES_METADATA_KEY EMPTY_SUMMARY = { @@ -22,12 +24,14 @@ async def get_workspace_changes_response( *, include_files: bool = True, include_diff: bool = True, + user_id: str | None | _AutoSentinel = AUTO, ) -> dict[str, Any]: events = await event_store.list_events( thread_id, run_id, event_types=[WORKSPACE_CHANGES_EVENT_TYPE], limit=10, + user_id=user_id, ) if not events: return _empty_response() diff --git a/backend/tests/test_run_event_store.py b/backend/tests/test_run_event_store.py index f56c8c426d..c7b749db6b 100644 --- a/backend/tests/test_run_event_store.py +++ b/backend/tests/test_run_event_store.py @@ -393,6 +393,41 @@ class TestListMessagesByRun: assert messages[0]["run_id"] == "r1" assert messages[0]["category"] == "message" + @pytest.mark.anyio + @pytest.mark.parametrize("backend", ["memory", "jsonl", "db"]) + async def test_explicit_user_id_follows_list_messages_semantics(self, tmp_path, backend): + """Run-scoped Gateway reads pass an explicit data identity on every backend. + + Only the SQL store isolates by user; the others accept and ignore it, + as they do for ``list_messages`` and ``list_events``. + """ + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + from deerflow.runtime.events.store.db import DbRunEventStore + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + from deerflow.runtime.user_context import reset_current_user, set_current_user + + if backend == "db": + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'test.db'}", sqlite_dir=str(tmp_path)) + s = DbRunEventStore(get_session_factory()) + elif backend == "jsonl": + s = JsonlRunEventStore(base_dir=tmp_path / "jsonl") + else: + s = MemoryRunEventStore() + try: + token = set_current_user(type("Owner", (), {"id": "feishu:owner"})()) + try: + await s.put(thread_id="t1", run_id="r1", event_type="ai_message", category="message") + finally: + reset_current_user(token) + + assert len(await s.list_messages_by_run("t1", "r1", user_id="feishu:owner")) == 1 + assert len(await s.list_messages_by_run("t1", "r1", user_id=None)) == 1 + other = await s.list_messages_by_run("t1", "r1", user_id="someone-else") + assert len(other) == (0 if backend == "db" else 1) + finally: + if backend == "db": + await close_engine() + # -- count_messages -- diff --git a/backend/tests/test_run_events_endpoint.py b/backend/tests/test_run_events_endpoint.py index 10626cd168..3afa4e32e1 100644 --- a/backend/tests/test_run_events_endpoint.py +++ b/backend/tests/test_run_events_endpoint.py @@ -35,8 +35,8 @@ async def test_list_run_events_forwards_task_id_and_after_seq(): calls: dict = {} class FakeStore: - async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None): - calls.update(thread_id=thread_id, run_id=run_id, event_types=event_types, task_id=task_id, limit=limit, after_seq=after_seq) + async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None, user_id=None): + calls.update(thread_id=thread_id, run_id=run_id, event_types=event_types, task_id=task_id, limit=limit, after_seq=after_seq, user_id=user_id) return [{"seq": 1, "event_type": "subagent.step"}] class FakeState: @@ -48,6 +48,7 @@ async def test_list_run_events_forwards_task_id_and_after_seq(): class FakeRequest: app = FakeApp() _deerflow_test_bypass_auth = True + cookies: dict = {} result = await list_run_events( thread_id="t1", @@ -63,6 +64,7 @@ async def test_list_run_events_forwards_task_id_and_after_seq(): assert calls["task_id"] == "task-A" assert calls["after_seq"] == 7 assert calls["event_types"] == ["subagent.start", "subagent.step", "subagent.end"] + assert calls["user_id"] is None @pytest.mark.anyio @@ -80,7 +82,7 @@ async def test_list_run_events_redacts_historical_run_start_metadata(): } class FakeStore: - async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None): + async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None, user_id=None): return [stored_row] class FakeState: @@ -92,6 +94,7 @@ async def test_list_run_events_redacts_historical_run_start_metadata(): class FakeRequest: app = FakeApp() _deerflow_test_bypass_auth = True + cookies: dict = {} events = await list_run_events( thread_id="legacy-thread", @@ -175,6 +178,7 @@ async def test_effective_memory_flows_from_injection_to_the_existing_debug_api() class FakeRequest_: app = FakeApp() _deerflow_test_bypass_auth = True + cookies: dict = {} events = await list_run_events( thread_id="t1", diff --git a/backend/tests/test_thread_regenerate_prepare.py b/backend/tests/test_thread_regenerate_prepare.py index 94e66b31c3..8121ed6348 100644 --- a/backend/tests/test_thread_regenerate_prepare.py +++ b/backend/tests/test_thread_regenerate_prepare.py @@ -181,7 +181,7 @@ class FakeEventStore: def __init__(self, rows): self.rows = rows - async def list_messages(self, thread_id, *, limit=50, before_seq=None, after_seq=None): + async def list_messages(self, thread_id, *, limit=50, before_seq=None, after_seq=None, user_id=None): return self.rows[-limit:] diff --git a/backend/tests/test_thread_run_messages_pagination.py b/backend/tests/test_thread_run_messages_pagination.py index f6669bbb55..8e62190000 100644 --- a/backend/tests/test_thread_run_messages_pagination.py +++ b/backend/tests/test_thread_run_messages_pagination.py @@ -164,6 +164,7 @@ def test_after_seq_forwarded_to_event_store(): limit=51, # default limit(50) + 1 before_seq=None, after_seq=5, + user_id=None, ) @@ -181,6 +182,7 @@ def test_before_seq_forwarded_to_event_store(): limit=51, before_seq=10, after_seq=None, + user_id=None, ) @@ -198,6 +200,7 @@ def test_custom_limit_forwarded_to_event_store(): limit=11, # 10 + 1 before_seq=None, after_seq=None, + user_id=None, ) diff --git a/backend/tests/test_thread_run_query_validation.py b/backend/tests/test_thread_run_query_validation.py index 9ff0e1b9f3..794bcac553 100644 --- a/backend/tests/test_thread_run_query_validation.py +++ b/backend/tests/test_thread_run_query_validation.py @@ -86,6 +86,7 @@ def test_read_endpoints_accept_positive_limits_and_hit_store(): limit=2, before_seq=None, after_seq=None, + user_id=None, ) app.state.run_event_store.list_events.assert_awaited_once_with( "thread-1", @@ -94,6 +95,7 @@ def test_read_endpoints_accept_positive_limits_and_hit_store(): task_id=None, limit=1, after_seq=None, + user_id=None, ) diff --git a/backend/tests/test_thread_runs_internal_scope.py b/backend/tests/test_thread_runs_internal_scope.py index f6aea056e7..a04daecde8 100644 --- a/backend/tests/test_thread_runs_internal_scope.py +++ b/backend/tests/test_thread_runs_internal_scope.py @@ -16,6 +16,7 @@ import asyncio from types import SimpleNamespace from uuid import UUID +import anyio import pytest from fastapi import FastAPI, HTTPException, Request, Response from fastapi.testclient import TestClient @@ -27,10 +28,15 @@ from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL, AUTH_SOURCE_SESSION from app.gateway.authz import AuthContext, Permissions from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, get_internal_user from app.gateway.routers import thread_runs +from deerflow.persistence.engine import close_engine, get_session_factory, init_engine +from deerflow.persistence.run.sql import RunRepository from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore +from deerflow.runtime.events.store.db import DbRunEventStore from deerflow.runtime.events.store.memory import MemoryRunEventStore from deerflow.runtime.runs.manager import RunManager from deerflow.runtime.runs.store.memory import MemoryRunStore +from deerflow.runtime.user_context import reset_current_user, set_current_user +from deerflow.workspace_changes.types import WORKSPACE_CHANGES_EVENT_TYPE THREAD_ID = "thread-scope" BROWSER_USER_ID = UUID("00000000-0000-0000-0000-00000000000a") @@ -49,7 +55,11 @@ _STUB_PERMISSIONS: list[str] = [ class _ScopeAuthMiddleware(BaseHTTPMiddleware): - """Stamp the same state trio production ``AuthMiddleware`` stamps.""" + """Stamp the state trio and user ContextVar production ``AuthMiddleware`` sets. + + The ContextVar is what ``user_id=AUTO`` repository reads resolve against, + so SQL-backed stores see the same identity they would in the Gateway. + """ def __init__(self, app, *, user, auth_source: str) -> None: super().__init__(app) @@ -60,7 +70,11 @@ class _ScopeAuthMiddleware(BaseHTTPMiddleware): request.state.user = self._user request.state.auth_source = self._auth_source request.state.auth = AuthContext(user=self._user, permissions=list(_STUB_PERMISSIONS)) - return await call_next(request) + token = set_current_user(self._user) + try: + return await call_next(request) + finally: + reset_current_user(token) class _PermissiveThreadStore: @@ -778,3 +792,188 @@ def test_token_usage_unfiltered_on_established_ownership_for_internal_callers() assert response.status_code == 200 body = response.json() assert body["total_tokens"] == 166 # both stamps fold when ownership is established + + +# --------------------------------------------------------------------------- +# SQL-backed run-event reads +# --------------------------------------------------------------------------- +# +# ``MemoryRunEventStore`` ignores ``user_id``, so the identity mismatch only +# surfaces on the SQL store: ``start_run`` stamps run-event rows with the raw +# trusted owner, while an ``AUTO`` read filters by the internal user's +# ``make_safe_user_id``-normalized id and matches nothing. + + +@pytest.fixture() +def db_event_store(tmp_path): + async def _init() -> DbRunEventStore: + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'run_events.db'}", sqlite_dir=str(tmp_path)) + return DbRunEventStore(get_session_factory()) + + yield anyio.run(_init) + anyio.run(close_engine) + + +def _as_owner(owner: str, write) -> None: + """Run ``write`` under ``owner``'s user context, as ``start_run`` does.""" + + async def _run() -> None: + token = set_current_user(SimpleNamespace(id=owner)) + try: + await write() + finally: + reset_current_user(token) + + anyio.run(_run) + + +def _seed_owner_stamped_message(event_store: DbRunEventStore, run_id: str, message_id: str, *, owner: str) -> None: + _as_owner( + owner, + lambda: event_store.put( + thread_id=THREAD_ID, + run_id=run_id, + event_type="llm.ai.response", + category="message", + content={"type": "ai", "id": message_id, "content": message_id, "additional_kwargs": {}}, + metadata={}, + ), + ) + + +@pytest.mark.parametrize("ownership", ["established", "missing_meta"]) +def test_internal_caller_reads_raw_owner_stamped_run_events(db_event_store: DbRunEventStore, ownership: str) -> None: + """Run messages/events resolve the same data identity as thread messages.""" + thread_store = _PermissiveThreadStore() if ownership == "established" else MemoryThreadMetaStore(InMemoryStore()) + run_store = _RecordingRunStore() + _seed_run(run_store, RUN_OWNER, user_id=OWNER_RAW) + _seed_owner_stamped_message(db_event_store, RUN_OWNER, "msg-owner", owner=OWNER_RAW) + + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + event_store=db_event_store, + thread_store=thread_store, + ) + headers = {INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW} + base = f"/api/threads/{THREAD_ID}/runs/{RUN_OWNER}" + + with client: + messages = client.get(base + "/messages", headers=headers) + events = client.get(base + "/events", headers=headers) + + assert messages.status_code == 200 + assert [row["content"]["id"] for row in messages.json()["data"]] == ["msg-owner"] + assert events.status_code == 200 + assert [event["content"]["id"] for event in events.json()] == ["msg-owner"] + + +def test_browser_session_run_events_keep_per_user_filter(db_event_store: DbRunEventStore) -> None: + """Browser sessions still read only rows stamped with their own id.""" + run_store = _RecordingRunStore() + _seed_run(run_store, RUN_OWNER, user_id=OWNER_RAW) + _seed_owner_stamped_message(db_event_store, RUN_OWNER, "msg-owner", owner=OWNER_RAW) + + client = _make_app( + user=_browser_user(), + auth_source=AUTH_SOURCE_SESSION, + run_store=run_store, + event_store=db_event_store, + ) + base = f"/api/threads/{THREAD_ID}/runs/{RUN_OWNER}" + + with client: + messages = client.get(base + "/messages") + events = client.get(base + "/events") + + assert messages.status_code == 200 + assert messages.json()["data"] == [] + assert events.status_code == 200 + assert events.json() == [] + + +def test_regenerate_target_lookup_reads_raw_owner_stamped_events(db_event_store: DbRunEventStore) -> None: + """The regenerate source-run scan finds the AI message by its raw stamp. + + The run store is empty, so only the event-store scan can resolve the run. + """ + _seed_owner_stamped_message(db_event_store, RUN_OWNER, "msg-owner", owner=OWNER_RAW) + user = _internal_user(OWNER_RAW) + request = _helper_request( + user=user, + auth_source=AUTH_SOURCE_INTERNAL, + run_store=_RecordingRunStore(), + event_store=db_event_store, + owner_header=OWNER_RAW, + ) + request.app.state.thread_store = _PermissiveThreadStore() + + async def _lookup() -> str: + token = set_current_user(user) + try: + return await thread_runs._find_target_run_id( + THREAD_ID, + "msg-owner", + {"content": "msg-owner"}, + {"additional_kwargs": {}}, + request, + ) + finally: + reset_current_user(token) + + assert anyio.run(_lookup) == RUN_OWNER + + +@pytest.mark.parametrize("ownership", ["established", "null_owner"]) +def test_internal_caller_reads_raw_owner_stamped_run_outputs(db_event_store: DbRunEventStore, ownership: str) -> None: + """Workspace changes and the archive manifest resolve the raw owner stamp. + + Both read SQL rows stamped by ``start_run``: the archive manifest also + looks the run row up itself, so it needs the SQL run store here. + """ + if ownership == "established": + thread_store = _PermissiveThreadStore() + else: + thread_store = MemoryThreadMetaStore(InMemoryStore()) + asyncio.run(thread_store.create(THREAD_ID, assistant_id=None, user_id=None)) + run_store = RunRepository(get_session_factory()) + + async def _seed() -> None: + await run_store.put(RUN_OWNER, thread_id=THREAD_ID, status="success") + await db_event_store.put( + thread_id=THREAD_ID, + run_id=RUN_OWNER, + event_type="run.delivery", + category="outputs", + content={"presented": 1, "by_tool": {"present_files": ["/mnt/user-data/outputs/a.txt"]}}, + ) + await db_event_store.put( + thread_id=THREAD_ID, + run_id=RUN_OWNER, + event_type=WORKSPACE_CHANGES_EVENT_TYPE, + category="outputs", + content={"summary": {"created": 1}, "files": []}, + ) + + _as_owner(OWNER_RAW, _seed) + + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + event_store=db_event_store, + thread_store=thread_store, + ) + headers = {INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW} + base = f"/api/threads/{THREAD_ID}/runs/{RUN_OWNER}" + + with client: + changes = client.get(base + "/workspace-changes", headers=headers) + manifest = client.get(base + "/artifacts/archive", headers=headers) + + assert changes.status_code == 200 + assert changes.json()["available"] is True + assert changes.json()["summary"]["created"] == 1 + assert manifest.status_code == 200, manifest.text + assert manifest.json()["file_count"] == 1 diff --git a/backend/tests/test_workspace_changes.py b/backend/tests/test_workspace_changes.py index a2932c45ed..5c9f366248 100644 --- a/backend/tests/test_workspace_changes.py +++ b/backend/tests/test_workspace_changes.py @@ -703,7 +703,7 @@ async def test_workspace_changes_route_forwards_include_files_flag(): calls: dict = {} class FakeStore: - async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None): + async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None, user_id=None): calls.update(thread_id=thread_id, run_id=run_id, event_types=event_types) return [ { @@ -734,6 +734,7 @@ async def test_workspace_changes_route_forwards_include_files_flag(): class FakeRequest: app = FakeApp() _deerflow_test_bypass_auth = True + cookies: dict = {} response = await get_run_workspace_changes( thread_id="thread-1",