diff --git a/CHANGELOG.md b/CHANGELOG.md index bace2d9577..0673a0517f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -327,6 +327,14 @@ This release closes that milestone with **181 merged pull requests**. ### Fixed +- **events:** Run-scoped reads no longer return 500 on the JSONL backend for a + run ID it cannot use as a filename. `GET + /api/threads/{thread_id}/runs/{run_id}/events`, `.../messages`, and + `.../workspace-changes` pass the URL's run ID to the event store unchecked; + with `run_events.backend: jsonl` an ID such as `run.1` raised `ValueError`, + while the memory and database stores return an empty result. JSONL reads and + deletes now treat such an ID as an unknown run; writes still reject it. + ([#6070]) - **scheduler:** Fixed-hour cron tasks no longer fire twice on the daylight-saving fall-back day. `croniter` returns both occurrences of an ambiguous wall-clock hour (the first with `fold=0`, the second with `fold=1`). For tasks where @@ -6268,4 +6276,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6015]: https://github.com/bytedance/deer-flow/pull/6015 [#6066]: https://github.com/bytedance/deer-flow/pull/6066 [#6069]: https://github.com/bytedance/deer-flow/pull/6069 +[#6070]: https://github.com/bytedance/deer-flow/pull/6070 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index b9046e9b9a..bc0c552602 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -332,6 +332,12 @@ ### 修复 +- **事件:** 在 JSONL 后端上,按 run 读取时,若 run ID 无法用作文件名,不再返回 + 500。`GET /api/threads/{thread_id}/runs/{run_id}/events`、`.../messages` 与 + `.../workspace-changes` 会把 URL 中的 run ID 原样传给事件存储;在 + `run_events.backend: jsonl` 下,`run.1` 这类 ID 会抛出 `ValueError`,而内存与 + 数据库存储返回空结果。现在 JSONL 的读取与删除把这类 ID 视为不存在的 run,写入 + 仍会拒绝它。([#6070]) - **调度器:** 固定小时的 cron 任务在夏令时回退(DST fall-back)当天不再重复运行两次。 `croniter` 会返回模糊本地时间的两个实例(首个为 `fold=0`,第二个为 `fold=1`)。 对于分和时字段不包含通配符的固定任务,现在会跳过第二个重复实例(`fold=1`),保持每天只运行一次 @@ -5316,3 +5322,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#6015]: https://github.com/bytedance/deer-flow/pull/6015 [#6066]: https://github.com/bytedance/deer-flow/pull/6066 [#6069]: https://github.com/bytedance/deer-flow/pull/6069 +[#6070]: https://github.com/bytedance/deer-flow/pull/6070 diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 3a13d27f10..42bc4960f0 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -90,6 +90,8 @@ run reads, and sequence recovery split on physical newlines. Do not use part of the record. Preserve existing UTF-8 files and the writer format. `tests/test_jsonl_event_store_unicode.py` covers Unicode values, reopening, idempotent writes, LF/CRLF, blank lines, and malformed records. +Reads and deletes treat a run ID writes reject as an unknown run (routes pass +URL IDs through); writes still raise. **Targeted run-event attribution** (`runtime/events/store/`): `RunEventStore.find_latest_ai_message_run_ids()` has a complete-or-error diff --git a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py index 76c3db829f..647410f689 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py +++ b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py @@ -94,10 +94,14 @@ class JsonlRunEventStore(RunEventStore): task = asyncio.create_task(operation(), name=f"jsonl-mutation:{thread_id}") return await self._await_owned_task(task) + @staticmethod + def _is_safe_id(value: str) -> bool: + return bool(value) and _SAFE_ID_PATTERN.match(value) is not None + @staticmethod def _validate_id(value: str, label: str) -> str: """Validate that an ID is safe for use in filesystem paths.""" - if not value or not _SAFE_ID_PATTERN.match(value): + if not JsonlRunEventStore._is_safe_id(value): raise ValueError(f"Invalid {label}: must be alphanumeric/dash/underscore, got {value!r}") return value @@ -109,6 +113,18 @@ class JsonlRunEventStore(RunEventStore): self._validate_id(run_id, "run_id") return self._thread_dir(thread_id) / f"{run_id}.jsonl" + def _existing_run_file(self, thread_id: str, run_id: str) -> Path | None: + """Return the run's file, or ``None`` when the run has none. + + Writes reject a run ID that is unsafe as a filename, so no file can hold + one; reads and deletes treat it as an unknown run, matching the memory + and database stores, rather than raising on a caller-supplied ID. + """ + if not self._is_safe_id(run_id): + return None + path = self._run_file(thread_id, run_id) + return path if path.exists() else None + def _next_seq(self, thread_id: str) -> int: self._seq_counters[thread_id] = self._seq_counters.get(thread_id, 0) + 1 return self._seq_counters[thread_id] @@ -159,8 +175,8 @@ class JsonlRunEventStore(RunEventStore): def _read_run_events(self, thread_id: str, run_id: str) -> list[dict]: """Read events for a specific run file (blocking I/O).""" - path = self._run_file(thread_id, run_id) - if not path.exists(): + path = self._existing_run_file(thread_id, run_id) + if path is None: return [] events = [] for line in path.read_text(encoding="utf-8").strip().split("\n"): @@ -180,8 +196,8 @@ class JsonlRunEventStore(RunEventStore): f.unlink() def _delete_run_file(self, thread_id: str, run_id: str) -> None: - path = self._run_file(thread_id, run_id) - if path.exists(): + path = self._existing_run_file(thread_id, run_id) + if path is not None: path.unlink() async def put(self, *, thread_id, run_id, event_type, category, content="", metadata=None, created_at=None): diff --git a/backend/tests/test_run_event_store.py b/backend/tests/test_run_event_store.py index 28383552f4..f56c8c426d 100644 --- a/backend/tests/test_run_event_store.py +++ b/backend/tests/test_run_event_store.py @@ -1285,6 +1285,44 @@ class TestJsonlRunEventStore: assert not (tmp_path / "jsonl" / "threads" / "t1" / "runs" / "r2.jsonl").exists() assert await s.count_messages("t1") == 1 + @pytest.mark.anyio + @pytest.mark.parametrize("run_id", ["", "run.1", "a b", "../decoy"]) + async def test_unsafe_run_id_reads_as_an_absent_run(self, tmp_path, run_id): + """A run ID no file can hold answers like the memory and DB stores' unknown run. + + Run-scoped routes pass the URL's ``run_id`` straight through, so raising + here surfaced as a 500 instead of the empty result the other backends give. + """ + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + s = JsonlRunEventStore(base_dir=tmp_path / "jsonl") + await s.put(thread_id="t1", run_id="r1", event_type="ai_message", category="message") + # ``../decoy`` would resolve here if the ID reached the filesystem. + decoy = tmp_path / "jsonl" / "threads" / "t1" / "decoy.jsonl" + decoy.write_text('{"seq": 9, "category": "message", "event_type": "ai_message"}\n', encoding="utf-8") + + assert await s.list_events("t1", run_id) == [] + assert await s.list_messages_by_run("t1", run_id) == [] + assert await s.get_last_visible_ai_seq_by_run("t1", [run_id, "r1"]) == {"r1": 1} + assert await s.delete_by_run("t1", run_id) == 0 + assert decoy.exists() + assert await s.count_messages("t1") == 1 + + @pytest.mark.anyio + async def test_unsafe_run_id_is_still_rejected_on_write(self, tmp_path): + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + s = JsonlRunEventStore(base_dir=tmp_path / "jsonl") + event = {"thread_id": "t1", "run_id": "run.1", "event_type": "ai_message", "category": "message"} + + with pytest.raises(ValueError, match="Invalid run_id"): + await s.put(**event) + with pytest.raises(ValueError, match="Invalid run_id"): + await s.put_if_absent(**event) + with pytest.raises(ValueError, match="Invalid run_id"): + await s.put_batch([event]) + assert await s.count_messages("t1") == 0 + class TestGetMessageSeqs: """Look up the thread-global seq of already-persisted messages by identity. diff --git a/backend/tests/test_run_events_endpoint.py b/backend/tests/test_run_events_endpoint.py index 3deb2e9bb2..10626cd168 100644 --- a/backend/tests/test_run_events_endpoint.py +++ b/backend/tests/test_run_events_endpoint.py @@ -111,6 +111,29 @@ async def test_list_run_events_redacts_historical_run_start_metadata(): assert events[0] is not stored_row +@pytest.mark.parametrize("run_id", ["run.1", "a%20b", "%2e%2e"]) +@pytest.mark.parametrize("route", ["events", "messages", "workspace-changes"]) +def test_run_scoped_reads_of_a_noncanonical_run_id_on_jsonl_match_the_memory_store(tmp_path, route, run_id): + """The URL's run_id reaches the event store unvalidated; JSONL must answer it as an unknown run, not a 500.""" + from _router_auth_helpers import make_authed_test_app + from fastapi.testclient import TestClient + + from app.gateway.routers import thread_runs + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + def get(store): + app = make_authed_test_app() + app.include_router(thread_runs.router) + app.state.run_event_store = store + with TestClient(app, raise_server_exceptions=False) as client: + return client.get(f"/api/threads/t1/runs/{run_id}/{route}") + + jsonl = get(JsonlRunEventStore(tmp_path)) + memory = get(MemoryRunEventStore()) + + assert (jsonl.status_code, jsonl.json()) == (200, memory.json()) + + @pytest.mark.anyio async def test_effective_memory_flows_from_injection_to_the_existing_debug_api(): """The production run-events route is the field-level consumer for M1.