fix(events): preserve JSONL framing after interrupted appends (#6520)

* fix(events): preserve JSONL framing after interrupted appends

* docs(changelog): link #6520

* docs(runtime): cite #6520 for JSONL tail recovery

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -804,6 +804,15 @@ This release closes that milestone with **439 merged pull requests**.
 
 ### Fixed
 
+- **runtime:** The JSONL event store no longer loses events after a torn final
+  line. A write interrupted mid-record left the file without a trailing newline,
+  so the next append was glued onto the partial record and both became one
+  unparseable line; a truncated multibyte character also made the whole file
+  fail to decode, hiding every intact record and breaking sequence recovery.
+  Appends now insert a separator when the existing file does not end in a
+  newline, reads decode each physical line on its own and skip only the broken
+  one, and a failed batch append still truncates back to the original size.
+  ([#6520])
 - **channels:** `/goal <objective>` and `/goal clear` from an IM channel work
   again when Gateway auth is enabled (the default). Both sent their write with
   the internal auth token alone, and the Gateway's CSRF check, which does not
@@ -9334,5 +9343,6 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6506]: https://github.com/bytedance/deer-flow/pull/6506
 [#6512]: https://github.com/bytedance/deer-flow/pull/6512
 [#6519]: https://github.com/bytedance/deer-flow/pull/6519
+[#6520]: https://github.com/bytedance/deer-flow/pull/6520
 [#6537]: https://github.com/bytedance/deer-flow/pull/6537
 [#6543]: https://github.com/bytedance/deer-flow/pull/6543

CHANGELOG_zh.md

@@ -694,6 +694,10 @@
 
 ### 修复
 
+- **运行时：** JSONL 事件存储在末行写坏后不再丢失事件。写入中途被打断会让文件缺少结尾换行，下一次追加会接在残缺记录后面，
+  两者合成一行无法解析；被截断的多字节字符还会让整个文件解码失败，所有完好记录都读不出来，序号恢复也随之失效。现在追加前
+  若文件末尾不是换行会先补一个分隔符，读取时逐个物理行单独解码、只跳过损坏的那一行，批量追加失败时仍会截回原始大小。
+  ([#6520])
 - **渠道：** 在启用 Gateway 认证（默认配置）时，从 IM 渠道发送的 `/goal <目标>` 和
   `/goal clear` 恢复正常。此前这两个写请求只携带内部认证令牌，而 Gateway 的 CSRF 检查不会
   豁免内部认证，因此返回 403，渠道回复“Failed to set goal.”或“Failed to clear goal.”。
@@ -7555,4 +7559,5 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6506]: https://github.com/bytedance/deer-flow/pull/6506
 [#6512]: https://github.com/bytedance/deer-flow/pull/6512
 [#6519]: https://github.com/bytedance/deer-flow/pull/6519
+[#6520]: https://github.com/bytedance/deer-flow/pull/6520
 [#6537]: https://github.com/bytedance/deer-flow/pull/6537

backend/docs/runtime-guidance-details.md

@@ -2,6 +2,19 @@
 
 The runtime AGENTS.md indexes these detailed contracts. Preserve their behavior when changing the corresponding code.
 
+**JSONL tail recovery** (`runtime/events/store/jsonl.py`, #6520): every append
+path goes through `_append_records`. A nonempty file whose last byte is not LF
+gets a separator written first, so a crash that left either valid JSON without
+its newline or a partial record cannot glue the next event onto that tail.
+`_iter_records` reads physical byte lines and decodes each one on its own;
+`JSONDecodeError` and `UnicodeDecodeError` skip only that line, so an
+incomplete UTF-8/JSON tail cannot hide intact records or block sequence
+recovery. Existing bytes, Unicode separators inside JSON strings, and
+per-thread mutation ownership stay as they are. Batch rollback restores the
+pre-append byte size, including removal of a newly inserted separator. Tests
+live in `tests/test_jsonl_event_store_tail_recovery.py`. This does not add
+multi-process support or crash-atomic batches.
+
 **Message feed seq stamping** (#4666): a checkpoint carries no position of its
 own and loses messages to summarization, so a client merging a `values` frame
 with the seq-ordered `run_events` feed cannot place a checkpoint-kept message

backend/packages/harness/deerflow/runtime/events/store/jsonl.py

@@ -21,7 +21,9 @@ thread since messages from multiple runs need unified seq ordering.
 
 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.
+their record. Decode one physical line at a time so an incomplete UTF-8 tail
+cannot hide earlier records. Writers separate unterminated tails from new
+records without discarding existing bytes.
 """
 
 from __future__ import annotations
@@ -29,9 +31,10 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+import os
 import re
 import weakref
-from collections.abc import Callable, Coroutine
+from collections.abc import Callable, Coroutine, Iterator
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -135,12 +138,8 @@ 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().split("\n"):
-                    try:
-                        record = json.loads(line)
-                        max_seq = max(max_seq, record.get("seq", 0))
-                    except json.JSONDecodeError:
-                        logger.debug("Skipping malformed JSONL line in %s", f)
+                for record in self._iter_records(f):
+                    max_seq = max(max_seq, record.get("seq", 0))
         return max_seq
 
     async def _ensure_seq_loaded(self, thread_id: str) -> None:
@@ -152,9 +151,18 @@ class JsonlRunEventStore(RunEventStore):
 
     def _write_record(self, record: dict) -> None:
         path = self._run_file(record["thread_id"], record["run_id"])
-        path.parent.mkdir(parents=True, exist_ok=True)
-        with open(path, "a", encoding="utf-8") as f:
-            f.write(json.dumps(record, default=str, ensure_ascii=False) + "\n")
+        self._append_records(path, [record])
+
+    def _iter_records(self, path: Path) -> Iterator[dict]:
+        """Read physical UTF-8 lines, skipping malformed or interrupted records."""
+        with open(path, "rb") as f:
+            for line in f:
+                if not line.strip():
+                    continue
+                try:
+                    yield json.loads(line.decode("utf-8"))
+                except (json.JSONDecodeError, UnicodeDecodeError):
+                    logger.debug("Skipping malformed JSONL line in %s", path)
 
     def _read_thread_events(self, thread_id: str) -> list[dict]:
         """Read all events for a thread, sorted by seq (blocking I/O)."""
@@ -163,13 +171,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().split("\n"):
-                if not line:
-                    continue
-                try:
-                    events.append(json.loads(line))
-                except json.JSONDecodeError:
-                    logger.debug("Skipping malformed JSONL line in %s", f)
+            events.extend(self._iter_records(f))
         events.sort(key=lambda e: e.get("seq", 0))
         return events
 
@@ -178,14 +180,7 @@ class JsonlRunEventStore(RunEventStore):
         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"):
-            if not line:
-                continue
-            try:
-                events.append(json.loads(line))
-            except json.JSONDecodeError:
-                logger.debug("Skipping malformed JSONL line in %s", path)
+        events = list(self._iter_records(path))
         events.sort(key=lambda e: e.get("seq", 0))
         return events
 
@@ -308,8 +303,14 @@ class JsonlRunEventStore(RunEventStore):
     def _append_records(self, path: Path, records: list[dict[str, Any]]) -> None:
         path.parent.mkdir(parents=True, exist_ok=True)
         lines = "".join(json.dumps(r, default=str, ensure_ascii=False) + "\n" for r in records)
-        with open(path, "a", encoding="utf-8") as f:
-            f.write(lines)
+        with open(path, "a+b") as f:
+            if f.tell():
+                f.seek(-1, os.SEEK_END)
+                if f.read(1) != b"\n":
+                    # A crash can leave either valid JSON without its newline
+                    # or a partial record. Never concatenate new JSON onto it.
+                    f.write(b"\n")
+            f.write(lines.encode("utf-8"))
 
     def _append_record_groups(self, groups: list[tuple[Path, list[dict[str, Any]]]]) -> None:
         """Append run groups and restore their original sizes if one fails."""

backend/tests/test_jsonl_event_store_tail_recovery.py

@@ -0,0 +1,102 @@
+"""Appending after an incomplete JSONL tail must not swallow later events."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from deerflow.runtime.events.store.jsonl import JsonlRunEventStore
+
+TAILS = {
+    "lf": b"\n",
+    "crlf": b"\r\n",
+    "no-newline": b"",
+    "partial-json": b'\n{"seq": 999, "content": "unfinished',
+    "partial-utf8": b'\n{"seq": 999, "content": "\xe4\xb8',
+}
+
+
+def _seed(tmp_path, tail):
+    path = tmp_path / "threads/t1/runs/r1.jsonl"
+    path.parent.mkdir(parents=True)
+    record = {
+        "thread_id": "t1",
+        "run_id": "r1",
+        "event_type": "llm.ai.response",
+        "category": "message",
+        "content": {"type": "ai", "id": "old-message", "content": "你好\u2028world"},
+        "metadata": {},
+        "seq": 41,
+        "created_at": "2026-01-01T00:00:00+00:00",
+    }
+    path.write_bytes(json.dumps(record, ensure_ascii=False).encode("utf-8") + tail)
+    return path, record
+
+
+def _event(run_id="r1", event_type="run.end"):
+    return {"thread_id": "t1", "run_id": run_id, "event_type": event_type, "category": "message", "content": "next message"}
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("tail", TAILS.values(), ids=TAILS)
+@pytest.mark.parametrize("write_method", ["put", "put_batch", "put_if_absent"])
+async def test_append_after_incomplete_tail_preserves_history_and_sequence(tmp_path, tail, write_method):
+    path, old = _seed(tmp_path, tail)
+    original_bytes = path.read_bytes()
+    store = JsonlRunEventStore(tmp_path)
+
+    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
+        after_write = path.read_bytes()
+        existing, created_again = await store.put_if_absent(**_event())
+        assert not created_again
+        assert existing == saved
+        assert path.read_bytes() == after_write
+    else:
+        saved = await store.put(**_event())
+
+    assert saved["seq"] == 42
+    assert path.read_bytes().startswith(original_bytes), "recovery must not rewrite or discard existing bytes"
+    # The new record needs its own physical line, even after an invalid tail.
+    assert json.loads(path.read_bytes().split(b"\n")[-2]) == saved
+    assert await store.list_events("t1", "r1") == [old, saved]
+    assert await store.list_messages("t1") == [old, saved]
+    assert await store.get_message_seqs("t1", ["message:old-message"]) == {"message:old-message": 41}
+
+    restarted = JsonlRunEventStore(tmp_path)
+    following = await restarted.put(**_event(event_type="run.following"))
+    assert following["seq"] == 43
+    assert await restarted.list_events("t1", "r1") == [old, saved, following]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("tail", [TAILS["no-newline"], TAILS["partial-json"], TAILS["partial-utf8"]], ids=["no-newline", "partial-json", "partial-utf8"])
+async def test_batch_failure_restores_original_tail_bytes_before_retry(tmp_path, monkeypatch, tail):
+    path, old = _seed(tmp_path, tail)
+    original_bytes = path.read_bytes()
+    new_path = path.with_name("r2.jsonl")
+    store = JsonlRunEventStore(tmp_path)
+    append = store._append_records
+
+    def fail_second_group(path, records):
+        append(path, records)
+        if path == new_path:
+            raise OSError("second group failed after writing")
+
+    monkeypatch.setattr(store, "_append_records", fail_second_group)
+    with pytest.raises(OSError, match="second group failed"):
+        await store.put_batch([_event(), _event("r2")])
+
+    assert path.read_bytes() == original_bytes
+    assert not new_path.exists()
+    assert await store.list_messages("t1") == [old]
+
+    monkeypatch.setattr(store, "_append_records", append)
+    saved = await store.put_batch([_event(), _event("r2")])
+    assert await store.list_messages("t1") == [old, *saved]
+    assert saved[0]["seq"] > old["seq"]
+    assert saved[1]["seq"] == saved[0]["seq"] + 1