mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-28 06:00:38 +00:00
fix(events): read a non-filename run ID as an unknown run on JSONL (#6070)
* fix(events): read a non-filename run ID as an unknown run on JSONL The run-scoped routes (events, messages, workspace-changes) pass the URL's run_id straight to the event store. With run_events.backend: jsonl, an ID that cannot be a filename (run.1, a%20b, %2e%2e) raised ValueError and surfaced as a 500, while the memory and database stores return an empty result. JSONL reads and deletes now treat such an ID as an unknown run through one helper, _existing_run_file(), which returns before building any path. Writes still reject it. * docs(changelog): reference #6070 for the JSONL run ID fix --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
48aec0bc84
commit
777961f84e
@ -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
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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.
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user