fix(events): preserve JSONL sequences after run deletion and restart (#6546)

* fix(events): preserve JSONL sequences after run deletion and restart

* fix(events): match JSONL watermark permissions to run files

---------

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

README.md

@@ -2484,6 +2484,12 @@ DeerFlow can be used as an embedded Python library without running the full HTTP
 
 For database-backed run events, deleting a run preserves its thread's sequence watermark. Thread deletion removes that watermark once no events remain, allowing a recreated thread to restart at sequence 1. Owner-scoped deletion preserves the watermark when another owner's events remain.
 
+Single-process JSONL event storage also retains the thread sequence watermark
+across run deletion and restarts, so clients using `after_seq` do not miss later
+messages. Keep `runs/.seq-watermark` with the run files when backing up this
+backend. New or replaced watermarks inherit the deleted run file's permission bits.
+Deleting the complete thread removes the watermark and resets allocation.
+
 Thread IDs may be supplied by callers and do not have to be UUIDs. Explicit
 IDs must contain 1–64 ASCII letters, digits, hyphens, or underscores
 (`^[A-Za-z0-9_-]{1,64}$`). DeerFlow generates a UUID only when `thread_id` is

backend/packages/harness/deerflow/runtime/AGENTS.md

@@ -120,6 +120,17 @@ 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.
 
+**JSONL sequence watermark:** before deleting a non-empty run, save the thread's
+allocation floor with atomic replacement of `runs/.seq-watermark`, off-loop under
+the existing mutation fence. Recovery takes the maximum of that floor and surviving
+run files, including legacy directories without a watermark. Copy the deleted run's
+permission bits onto the temporary watermark before publication, preserving both
+shared-read and restrictive modes. Failed permission setup or publication must leave
+the run intact; unreadable/corrupt watermarks fail closed. Complete
+thread deletion removes the watermark only after deleting all run files, including
+when no events remain. This does not add multi-process JSONL support. Coverage:
+`tests/test_jsonl_event_store_seq_watermark.py` and the mutation cancellation suite.
+
 **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

@@ -33,6 +33,7 @@ import json
 import logging
 import os
 import re
+import tempfile
 import weakref
 from collections.abc import Callable, Coroutine, Iterator
 from datetime import UTC, datetime
@@ -134,7 +135,7 @@ class JsonlRunEventStore(RunEventStore):
 
     def _compute_max_seq(self, thread_id: str) -> int:
         """Scan all run files for a thread and return the current max seq (blocking I/O)."""
-        max_seq = 0
+        max_seq = self._read_seq_watermark(thread_id)
         thread_dir = self._thread_dir(thread_id)
         if thread_dir.exists():
             for f in thread_dir.glob("*.jsonl"):
@@ -142,6 +143,37 @@ class JsonlRunEventStore(RunEventStore):
                     max_seq = max(max_seq, record.get("seq", 0))
         return max_seq
 
+    def _read_seq_watermark(self, thread_id: str) -> int:
+        path = self._thread_dir(thread_id) / ".seq-watermark"
+        try:
+            seq = int(path.read_text(encoding="utf-8"))
+        except FileNotFoundError:
+            return 0
+        if seq < 0:
+            raise ValueError(f"Invalid JSONL sequence watermark in {path}")
+        return seq
+
+    def _save_seq_watermark(self, thread_id: str, seq: int, run_id: str) -> None:
+        """Publish the allocation floor before deleting records (blocking I/O)."""
+        if seq <= self._read_seq_watermark(thread_id):
+            return
+        path = self._thread_dir(thread_id) / ".seq-watermark"
+        run_mode = self._run_file(thread_id, run_id).stat().st_mode & 0o777
+        temporary: Path | None = None
+        try:
+            with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent, prefix=".seq-", suffix=".tmp", delete=False) as file:
+                temporary = Path(file.name)
+                file.write(str(seq))
+            # Match the run's access policy before atomically publishing the floor.
+            temporary.chmod(run_mode)
+            temporary.replace(path)
+        finally:
+            if temporary is not None:
+                try:
+                    temporary.unlink(missing_ok=True)
+                except OSError:
+                    logger.warning("Could not remove JSONL watermark temporary file %s", temporary, exc_info=True)
+
     async def _ensure_seq_loaded(self, thread_id: str) -> None:
         """Load max seq from existing files into the in-memory counter (non-blocking)."""
         if thread_id in self._seq_counters:
@@ -189,6 +221,8 @@ class JsonlRunEventStore(RunEventStore):
         if thread_dir.exists():
             for f in thread_dir.glob("*.jsonl"):
                 f.unlink()
+            # Reset only after every run file has been removed successfully.
+            (thread_dir / ".seq-watermark").unlink(missing_ok=True)
 
     def _delete_run_file(self, thread_id: str, run_id: str) -> None:
         path = self._existing_run_file(thread_id, run_id)
@@ -477,6 +511,9 @@ class JsonlRunEventStore(RunEventStore):
         async def mutate():
             events = await asyncio.to_thread(self._read_run_events, thread_id, run_id)
             count = len(events)
+            if count:
+                await self._ensure_seq_loaded(thread_id)
+                await asyncio.to_thread(self._save_seq_watermark, thread_id, self._seq_counters[thread_id], run_id)
             await asyncio.to_thread(self._delete_run_file, thread_id, run_id)
             return count
 

backend/tests/test_jsonl_event_store_cancellation.py

@@ -41,6 +41,36 @@ async def _checkpoint():
     await asyncio.sleep(0)
 
 
+@pytest.mark.anyio
+async def test_cancelled_watermark_publication_drains_before_thread_delete(tmp_path, monkeypatch):
+    store = JsonlRunEventStore(tmp_path)
+    await store.put(**_event(content="baseline"))
+    paused = _PausedIO(store._save_seq_watermark)
+    monkeypatch.setattr(store, "_save_seq_watermark", paused)
+    pending = asyncio.create_task(store.delete_by_run("t1", "r1"))
+    deletion = None
+    try:
+        await asyncio.wait_for(paused.entered.wait(), 5)
+        pending.cancel()
+        await _checkpoint()
+        pending.cancel()
+        deletion = asyncio.create_task(store.delete_by_thread("t1"))
+        await _checkpoint()
+        assert not pending.done()
+        assert not deletion.done()
+        paused.release.set()
+        with pytest.raises(asyncio.CancelledError):
+            await pending
+        assert await deletion == 0
+        assert not (tmp_path / "threads" / "t1" / "runs" / ".seq-watermark").exists()
+        new = await JsonlRunEventStore(tmp_path).put(**_event("r2"))
+        assert new["seq"] == 1
+    finally:
+        paused.release.set()
+        await asyncio.gather(pending, *([deletion] if deletion is not None else []), return_exceptions=True)
+        await asyncio.wait_for(paused.finished.wait(), 5)
+
+
 @pytest.mark.anyio
 @pytest.mark.parametrize("method", ["put", "put_if_absent", "put_batch"])
 @pytest.mark.parametrize("cancellations", [1, 3])

backend/tests/test_jsonl_event_store_seq_watermark.py

@@ -0,0 +1,185 @@
+"""Durable cursor allocation after JSONL run deletion and store reopening."""
+
+import os
+from pathlib import Path
+
+import pytest
+
+from deerflow.runtime.events.store.jsonl import JsonlRunEventStore
+
+
+def event(run_id):
+    return {"thread_id": "t1", "run_id": run_id, "event_type": "human_message", "category": "message", "content": run_id}
+
+
+async def write(store, run_id, method):
+    if method == "put_batch":
+        return (await store.put_batch([event(run_id)]))[0]
+    if method == "put_if_absent":
+        return (await store.put_if_absent(**event(run_id)))[0]
+    return await store.put(**event(run_id))
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("method", ["put", "put_batch", "put_if_absent"])
+@pytest.mark.parametrize("reopen_before_delete", [False, True])
+@pytest.mark.parametrize("delete_all", [False, True])
+async def test_run_deletion_preserves_cursor_across_reopen(tmp_path, method, reopen_before_delete, delete_all):
+    store = JsonlRunEventStore(tmp_path)
+    await write(store, "r1", method)
+    last = await write(store, "r2", method)
+    assert last["seq"] == 2
+    if reopen_before_delete:
+        store = JsonlRunEventStore(tmp_path)
+    assert await store.delete_by_run("t1", "r2") == 1
+    if delete_all:
+        assert await store.delete_by_run("t1", "r1") == 1
+    reopened = JsonlRunEventStore(tmp_path)
+    new = await write(reopened, "r3", method)
+    assert new["seq"] == 3
+    assert [row["seq"] for row in await reopened.list_messages("t1", after_seq=last["seq"])] == [3]
+
+
+@pytest.mark.anyio
+async def test_thread_deletion_resets_watermark_after_all_runs_deleted(tmp_path):
+    store = JsonlRunEventStore(tmp_path)
+    await store.put(**event("r1"))
+    await store.delete_by_run("t1", "r1")
+    assert await JsonlRunEventStore(tmp_path).delete_by_thread("t1") == 0
+    recreated = await JsonlRunEventStore(tmp_path).put(**event("r2"))
+    assert recreated["seq"] == 1
+
+
+@pytest.mark.anyio
+async def test_missing_run_deletion_does_not_create_storage(tmp_path):
+    store = JsonlRunEventStore(tmp_path)
+    assert await store.delete_by_run("t1", "missing") == 0
+    assert not (tmp_path / "threads").exists()
+
+
+@pytest.mark.anyio
+async def test_legacy_directory_and_surviving_higher_seq(tmp_path):
+    runs = tmp_path / "threads" / "t1" / "runs"
+    runs.mkdir(parents=True)
+    (runs / "legacy.jsonl").write_text('{"seq":9,"run_id":"legacy","category":"message"}\n', encoding="utf-8")
+    store = JsonlRunEventStore(tmp_path)
+    assert await store.delete_by_run("t1", "legacy") == 1
+    assert (await store.put(**event("r1")))["seq"] == 10
+    # The persisted floor is 9; recovery must also consider newer run records.
+    assert (await JsonlRunEventStore(tmp_path).put(**event("r2")))["seq"] == 11
+
+
+@pytest.mark.anyio
+async def test_deleting_older_run_cannot_lower_watermark(tmp_path):
+    store = JsonlRunEventStore(tmp_path)
+    await store.put(**event("r1"))
+    await store.put(**event("r2"))
+    await store.delete_by_run("t1", "r2")
+    await JsonlRunEventStore(tmp_path).delete_by_run("t1", "r1")
+    assert (await JsonlRunEventStore(tmp_path).put(**event("r3")))["seq"] == 3
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("invalid", ["garbage", "-1"])
+async def test_invalid_watermark_fails_closed(tmp_path, invalid):
+    runs = tmp_path / "threads" / "t1" / "runs"
+    runs.mkdir(parents=True)
+    (runs / ".seq-watermark").write_text(invalid, encoding="utf-8")
+    with pytest.raises(ValueError):
+        await JsonlRunEventStore(tmp_path).put(**event("r1"))
+    assert not (runs / "r1.jsonl").exists()
+
+
+@pytest.mark.anyio
+async def test_failed_watermark_publish_keeps_previous_floor_and_run(tmp_path, monkeypatch):
+    store = JsonlRunEventStore(tmp_path)
+    await store.put(**event("r1"))
+    await store.delete_by_run("t1", "r1")
+    await store.put(**event("r2"))
+    replace = Path.replace
+
+    def fail_publish(path, target):
+        if Path(target).name == ".seq-watermark":
+            raise OSError("synthetic watermark publication failure")
+        return replace(path, target)
+
+    monkeypatch.setattr(Path, "replace", fail_publish)
+    with pytest.raises(OSError, match="publication failure"):
+        await store.delete_by_run("t1", "r2")
+    assert len(await store.list_events("t1", "r2")) == 1
+    runs = tmp_path / "threads" / "t1" / "runs"
+    assert (runs / ".seq-watermark").read_text(encoding="utf-8") == "1"
+    assert not list(runs.glob(".seq-*.tmp"))
+    assert (await JsonlRunEventStore(tmp_path).put(**event("r3")))["seq"] == 3
+
+
+@pytest.mark.anyio
+@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits are not supported on Windows")
+@pytest.mark.parametrize("run_mode", [0o600, 0o640, 0o644])
+@pytest.mark.parametrize("replace_existing", [False, True])
+async def test_watermark_matches_run_permissions_before_publication(tmp_path, monkeypatch, run_mode, replace_existing):
+    store = JsonlRunEventStore(tmp_path)
+    if replace_existing:
+        await store.put(**event("r0"))
+        await store.delete_by_run("t1", "r0")
+    last = await store.put(**event("r1"))
+    runs = tmp_path / "threads" / "t1" / "runs"
+    (runs / "r1.jsonl").chmod(run_mode)
+    replace = Path.replace
+
+    def check_permissions(path, target):
+        if Path(target).name == ".seq-watermark":
+            assert path.stat().st_mode & 0o777 == run_mode
+        return replace(path, target)
+
+    monkeypatch.setattr(Path, "replace", check_permissions)
+    assert await store.delete_by_run("t1", "r1") == 1
+    watermark = runs / ".seq-watermark"
+    assert watermark.stat().st_mode & 0o777 == run_mode
+    assert watermark.read_text(encoding="utf-8") == str(last["seq"])
+    assert not (runs / "r1.jsonl").exists()
+    assert (await JsonlRunEventStore(tmp_path).put(**event("r2")))["seq"] == last["seq"] + 1
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("replace_existing", [False, True])
+async def test_failed_watermark_permissions_keep_previous_floor_and_run(tmp_path, monkeypatch, replace_existing):
+    store = JsonlRunEventStore(tmp_path)
+    if replace_existing:
+        await store.put(**event("r1"))
+        await store.delete_by_run("t1", "r1")
+    last = await store.put(**event("r2"))
+    chmod = Path.chmod
+
+    def fail_permissions(path, mode, **kwargs):
+        if path.name.startswith(".seq-") and path.suffix == ".tmp":
+            raise OSError("synthetic watermark permission failure")
+        return chmod(path, mode, **kwargs)
+
+    monkeypatch.setattr(Path, "chmod", fail_permissions)
+    with pytest.raises(OSError, match="permission failure"):
+        await store.delete_by_run("t1", "r2")
+    assert len(await store.list_events("t1", "r2")) == 1
+    runs = tmp_path / "threads" / "t1" / "runs"
+    watermark = runs / ".seq-watermark"
+    if replace_existing:
+        assert watermark.read_text(encoding="utf-8") == "1"
+    else:
+        assert not watermark.exists()
+    assert not list(runs.glob(".seq-*.tmp"))
+    assert (await JsonlRunEventStore(tmp_path).put(**event("r3")))["seq"] == last["seq"] + 1
+
+
+@pytest.mark.anyio
+async def test_failed_run_unlink_keeps_safe_allocation_floor(tmp_path, monkeypatch):
+    store = JsonlRunEventStore(tmp_path)
+    await store.put(**event("r1"))
+
+    def fail_delete(*args):
+        raise OSError("synthetic unlink failure")
+
+    monkeypatch.setattr(store, "_delete_run_file", fail_delete)
+    with pytest.raises(OSError, match="unlink failure"):
+        await store.delete_by_run("t1", "r1")
+    assert len(await store.list_messages("t1")) == 1
+    assert (await JsonlRunEventStore(tmp_path).put(**event("r2")))["seq"] == 2