fix(events): preserve Unicode separators in JSONL event records (#5429)

Signed-off-by: Undermoon1412 <Undermoon1412@users.noreply.github.com>
Co-authored-by: Undermoon1412 <Undermoon1412@users.noreply.github.com>

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.

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

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:

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