From f770ecc0b83968ce0a04158a580d872dc34d698a Mon Sep 17 00:00:00 2001 From: Undermoon1412 <80385295+Undermoon1412@users.noreply.github.com> Date: Mon, 14 Sep 2026 18:24:51 +0800 Subject: [PATCH] fix(events): preserve Unicode separators in JSONL event records (#5429) Signed-off-by: Undermoon1412 Co-authored-by: Undermoon1412 --- README.md | 4 + .../harness/deerflow/runtime/AGENTS.md | 7 ++ .../deerflow/runtime/events/store/jsonl.py | 10 +- .../tests/test_jsonl_event_store_unicode.py | 111 ++++++++++++++++++ 4 files changed, 129 insertions(+), 3 deletions(-) create mode 100644 backend/tests/test_jsonl_event_store_unicode.py diff --git a/README.md b/README.md index 43500d174..973bc41a3 100644 --- a/README.md +++ b/README.md @@ -334,6 +334,10 @@ For persistent deployments, configure `database.backend` as `sqlite` or LangGraph Store, and DeerFlow application data. The deprecated `checkpointer` section, when present, overrides the first two for backward compatibility. +For lightweight single-process event persistence, `run_events.backend: jsonl` +keeps Unicode message content intact, including line and paragraph separators. +Existing valid JSONL records remain readable without rewriting the files. + The unified nginx endpoint is same-origin by default and does not emit browser CORS headers. If you run a split-origin or port-forwarded browser client, set `GATEWAY_CORS_ORIGINS` to comma-separated exact origins such as `http://localhost:3000`; the Gateway then applies the CORS allowlist and matching CSRF origin checks. Browser login uses `HttpOnly` session cookies. The login page offers a "keep me signed in" option that extends the browser session when the request is HTTPS (including trusted `X-Forwarded-Proto: https`) or localhost HTTP. The localhost exception uses the direct request `Host` and ignores forwarded host headers. Public HTTP deployments, including many temporary sandbox URLs, fall back to session cookies by default. DeerFlow never stores the password in browser storage; the UI may remember only the email address. diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 9d90badce..95af28db8 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -135,6 +135,13 @@ their per-execution parent-loop proxy, preserving separate events when two different delegated agents promote the same tool. The active catalog is fixed for one graph execution, so the claim needs no persisted catalog hash. +**JSONL record boundaries** (`runtime/events/store/jsonl.py`): thread reads, +run reads, and sequence recovery split on physical newlines. Do not use +`str.splitlines()`: U+0085/U+2028/U+2029 inside valid JSON strings must remain +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. + **Targeted run-event attribution** (`runtime/events/store/`): `RunEventStore.find_latest_ai_message_run_ids()` has a complete-or-error contract. Its default implementation walks `list_messages()` backward in diff --git a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py index 6d1a642c2..23ea0520d 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py +++ b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py @@ -18,6 +18,10 @@ writes within a single process to prevent interleaved JSONL lines. Known trade-off: ``list_messages()`` must scan all run files for a thread since messages from multiple runs need unified seq ordering. ``list_events()`` reads only one file -- the fast path. + +Read records using physical newline boundaries, not ``str.splitlines()``: +Unicode line separators are valid JSON string content and must stay inside +their record. ``read_text`` normalizes CRLF before the LF split. """ from __future__ import annotations @@ -75,7 +79,7 @@ class JsonlRunEventStore(RunEventStore): thread_dir = self._thread_dir(thread_id) if thread_dir.exists(): for f in thread_dir.glob("*.jsonl"): - for line in f.read_text(encoding="utf-8").strip().splitlines(): + for line in f.read_text(encoding="utf-8").strip().split("\n"): try: record = json.loads(line) max_seq = max(max_seq, record.get("seq", 0)) @@ -103,7 +107,7 @@ class JsonlRunEventStore(RunEventStore): if not thread_dir.exists(): return events for f in sorted(thread_dir.glob("*.jsonl")): - for line in f.read_text(encoding="utf-8").strip().splitlines(): + for line in f.read_text(encoding="utf-8").strip().split("\n"): if not line: continue try: @@ -119,7 +123,7 @@ class JsonlRunEventStore(RunEventStore): if not path.exists(): return [] events = [] - for line in path.read_text(encoding="utf-8").strip().splitlines(): + for line in path.read_text(encoding="utf-8").strip().split("\n"): if not line: continue try: diff --git a/backend/tests/test_jsonl_event_store_unicode.py b/backend/tests/test_jsonl_event_store_unicode.py new file mode 100644 index 000000000..df5066036 --- /dev/null +++ b/backend/tests/test_jsonl_event_store_unicode.py @@ -0,0 +1,111 @@ +"""JSONL framing must not split Unicode separators inside event values (#5420).""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + +SEPARATORS = [chr(0x85), chr(0x2028), chr(0x2029)] +SEPARATOR_IDS = ["next-line", "line-separator", "paragraph-separator"] + + +@pytest.mark.anyio +@pytest.mark.parametrize("separator", SEPARATORS, ids=SEPARATOR_IDS) +@pytest.mark.parametrize("write_method", ["put", "put_batch", "put_if_absent"]) +async def test_unicode_event_values_round_trip(tmp_path: Path, separator: str, write_method: str): + store = JsonlRunEventStore(base_dir=tmp_path) + content = {"type": "ai", "id": "message-1", "content": f"before{separator}after"} + event = { + "thread_id": "thread-1", + "run_id": "run-1", + "event_type": "llm.ai.response", + "category": "message", + "content": content, + "metadata": {"note": f"first{separator}second"}, + } + if write_method == "put_batch": + saved = (await store.put_batch([event]))[0] + elif write_method == "put_if_absent": + saved, created = await store.put_if_absent(**event) + assert created + else: + saved = await store.put(**event) + + # The writer already emits valid JSON: this must be repaired on the read side. + raw = (tmp_path / "threads/thread-1/runs/run-1.jsonl").read_bytes() + assert separator.encode("utf-8") in raw + assert json.loads(raw) == saved + assert await store.list_messages("thread-1") == [saved] + assert await store.list_events("thread-1", "run-1") == [saved] + assert await store.list_messages_by_run("thread-1", "run-1") == [saved] + assert await store.count_messages("thread-1") == 1 + assert await store.get_message_seqs("thread-1", ["message:message-1"]) == {"message:message-1": saved["seq"]} + assert await store.find_latest_ai_message_run_ids("thread-1", {"message-1"}) == {"message-1": "run-1"} + + +@pytest.mark.anyio +@pytest.mark.parametrize("separator", SEPARATORS, ids=SEPARATOR_IDS) +async def test_reopen_recovers_seq_from_existing_unicode_record(tmp_path: Path, separator: str): + # Seed a pre-existing file directly: changing only the writer cannot fix it. + path = tmp_path / "threads/thread-1/runs/run-1.jsonl" + path.parent.mkdir(parents=True) + record = { + "thread_id": "thread-1", + "run_id": "run-1", + "event_type": "llm.ai.response", + "category": "message", + "content": f"old{separator}message", + "metadata": {}, + "seq": 41, + "created_at": "2026-01-01T00:00:00+00:00", + } + path.write_text(json.dumps(record, ensure_ascii=False) + "\n", encoding="utf-8") + + reopened = JsonlRunEventStore(base_dir=tmp_path) + following = await reopened.put(thread_id="thread-1", run_id="run-2", event_type="llm.ai.response", category="message", content="next message") + + assert following["seq"] == 42 + assert await reopened.list_messages("thread-1") == [record, following] + assert await reopened.list_messages("thread-1", after_seq=41) == [following] + + +@pytest.mark.anyio +@pytest.mark.parametrize("separator", SEPARATORS, ids=SEPARATOR_IDS) +async def test_unicode_metadata_does_not_duplicate_idempotent_event(tmp_path: Path, separator: str): + store = JsonlRunEventStore(base_dir=tmp_path) + event = { + "thread_id": "thread-1", + "run_id": "run-1", + "event_type": "run.end", + "category": "lifecycle", + "content": "completed", + "metadata": {"note": f"first{separator}second"}, + } + saved, created = await store.put_if_absent(**event) + assert created + + reopened = JsonlRunEventStore(base_dir=tmp_path) + existing, created_again = await reopened.put_if_absent(**event) + assert not created_again + assert existing == saved + assert await reopened.list_events("thread-1", "run-1") == [saved] + + +@pytest.mark.anyio +@pytest.mark.parametrize("newline", ["\n", "\r\n"], ids=["lf", "crlf"]) +async def test_physical_lines_keep_controls_and_skip_malformed_records(tmp_path: Path, newline: str): + store = JsonlRunEventStore(base_dir=tmp_path) + saved = await store.put(thread_id="thread-1", run_id="run-1", event_type="llm.ai.response", category="message", content="ordinary\n你好\r\ntext") + path = tmp_path / "threads/thread-1/runs/run-1.jsonl" + # Blank lines, malformed JSON, and a final record without a trailing newline. + path.write_bytes(newline.join(["", "not-json", " ", json.dumps(saved, ensure_ascii=False)]).encode("utf-8")) + + reopened = JsonlRunEventStore(base_dir=tmp_path) + assert await reopened.list_messages("thread-1") == [saved] + assert await reopened.list_events("thread-1", "run-1") == [saved] + following = await reopened.put(thread_id="thread-1", run_id="run-2", event_type="llm.ai.response", category="message", content="next message") + assert following["seq"] == saved["seq"] + 1