Merge master (exp-v0.52.348) into stage/7749-terminal-handoff

# Conflicts:
#	CHANGELOG.md

CHANGELOG.md

@@ -31,6 +31,11 @@
   and the settled scene was never saved. The cleanup now waits briefly for the pane's own open
   stream, then checks `/api/chat/stream/status` once, and falls back to the existing
   interrupted-stream recovery if that check fails. (#7749 by @franksong2702)
+- **A session's run journal can no longer be written out of order.** The journal writer reserved a
+  sequence number under the per-path lock, released it, and then appended, so two concurrent writers
+  could land sequence N+1 on disk before N and the replay reader would stop at the gap
+  (`replay_noncontiguous`). Sequence allocation and the physical append now happen under the same
+  existing lock. (#7751 by @franksong2702)
 - **A stopped chat can no longer publish or reuse its agent after Stop.** A worker that was
   cancelled could still publish its cached Agent or invoke it after Stop landed, and its late
   cleanup could close or evict the same Agent object a successor turn had just picked up. The initial

api/run_journal.py

@@ -451,8 +451,6 @@ class RunJournalWriter:
         self.session_id = _validate_id(session_id, "session_id")
         self.run_id = _validate_id(run_id, "run_id")
         self.session_dir = Path(session_dir) if session_dir is not None else None
-        self._path = _run_path(self.session_id, self.run_id, session_dir=self.session_dir)
-        self._lock = _lock_for(self._path)
 
     def append_sse_event(self, event_name: str, payload=None) -> dict | None:
         # Live-UI-only telemetry (metering) has no recovery value in the journal:
@@ -464,18 +462,15 @@ class RunJournalWriter:
         # the offline-gap coverage and replay-cursor contiguity checks rely on.
         if str(event_name or "").strip() in REPLAY_SKIPPED_SSE_EVENTS:
             return None
-        # Draw from the shared module-level seq cache under the per-path lock so
-        # this writer and any direct append_run_event() call on the same path
-        # agree on one monotonic, gapless sequence.
-        with self._lock:
-            seq = _reserve_next_seq(self._path)
+        # Allocate the sequence inside the same per-path transaction that writes
+        # the row. Reserving here, then releasing the lock before append, lets a
+        # concurrent writer put a higher sequence on disk first.
         return append_run_event(
             self.session_id,
             self.run_id,
             event_name,
             payload or {},
             session_dir=self.session_dir,
-            seq=seq,
         )
 
 

docs/rfcs/webui-run-state-consistency-contract.md

@@ -64,6 +64,16 @@ Session finalization still owns the session lock: the returned-error path alread
 holds it, while initial registration and the exception path acquire it. Do not
 reacquire this non-reentrant lock from a branch that already owns it.
 
+## Run-journal sequence publication
+
+Within one WebUI process, auto-numbered appends to the same journal allocate
+sequence numbers and write their rows under the same per-path lock.
+`RunJournalWriter` delegates both operations to `append_run_event`; it must not
+reserve a sequence and release the lock before the physical append. Otherwise
+individually valid rows can reach disk out of order and the session replay
+reader must reject them as noncontiguous. This does not change caller-supplied
+sequence semantics, cross-process ownership, or failed-write recovery.
+
 ## Inactive compression continuation recovery
 
 The Agent profile's SQLite compression lineage owns the canonical continuation,

tests/test_run_journal_writer_ordering.py

@@ -0,0 +1,81 @@
+"""Sequence allocation and physical append must share one journal lock.
+
+Only scheduling at the real append boundary is controlled. Writers, sequence
+allocation, on-disk JSONL and the session replay reader are production code.
+"""
+from concurrent.futures import ThreadPoolExecutor
+import threading
+
+import pytest
+
+from api import run_journal
+
+
+@pytest.mark.parametrize("competitor", ["same_writer", "other_writer", "free_function"])
+def test_writer_sequence_follows_physical_append_order(tmp_path, monkeypatch, competitor):
+    sid, rid = "writer_order_session", "writer_order_run"
+    writer = run_journal.RunJournalWriter(sid, rid, session_dir=tmp_path)
+    other = run_journal.RunJournalWriter(sid, rid, session_dir=tmp_path)
+    reached = threading.Event()
+    release = threading.Event()
+    real_append = run_journal.append_run_event
+
+    def pause_first_append(session_id, run_id, event_name, payload=None, **kwargs):
+        if payload == {"text": "paused"}:
+            reached.set()
+            assert release.wait(5), "test did not release the paused append"
+        return real_append(session_id, run_id, event_name, payload, **kwargs)
+
+    monkeypatch.setattr(run_journal, "append_run_event", pause_first_append)
+
+    def append_competitor():
+        if competitor == "free_function":
+            return run_journal.append_run_event(
+                sid, rid, "token", {"text": "overtaking"}, session_dir=tmp_path
+            )
+        target = writer if competitor == "same_writer" else other
+        return target.append_sse_event("token", {"text": "overtaking"})
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        first = pool.submit(writer.append_sse_event, "token", {"text": "paused"})
+        try:
+            assert reached.wait(5), "writer did not reach the append boundary"
+            second = pool.submit(append_competitor).result(timeout=5)
+        finally:
+            release.set()
+        first_result = first.result(timeout=5)
+
+    journal = run_journal.read_run_events(sid, rid, session_dir=tmp_path)
+    # Deliberately do not sort: the reader requires physical contiguous order.
+    assert [row["seq"] for row in journal["events"]] == [1, 2]
+    assert [row["payload"]["text"] for row in journal["events"]] == ["overtaking", "paused"]
+    assert second["event_id"] == f"{rid}:1"
+    assert first_result["event_id"] == f"{rid}:2"
+    replay = run_journal.read_session_run_events(
+        sid, after_event_id=second["event_id"], session_dir=tmp_path
+    )
+    assert replay["status"] == "ok"
+    assert [row["event_id"] for row in replay["events"]] == [first_result["event_id"]]
+
+
+def test_writer_rejected_empty_name_does_not_reserve_a_sequence(tmp_path):
+    writer = run_journal.RunJournalWriter("name_session", "name_run", session_dir=tmp_path)
+    with pytest.raises(ValueError, match="event_name is required"):
+        writer.append_sse_event("  ", {"text": "rejected"})
+    event = writer.append_sse_event("token", {"text": "accepted"})
+    assert event["seq"] == 1
+    assert event["event_id"] == "name_run:1"
+
+
+def test_metering_skip_still_does_not_reserve_or_write(tmp_path):
+    writer = run_journal.RunJournalWriter("meter_session", "meter_run", session_dir=tmp_path)
+    assert writer.append_sse_event("metering", {"tps": 10}) is None
+    assert not (tmp_path / "_run_journal" / "meter_session" / "meter_run.jsonl").exists()
+    assert writer.append_sse_event("token", {"text": "kept"})["seq"] == 1
+
+
+def test_unused_writer_does_not_allocate_a_registry_lock(tmp_path):
+    parent = str(tmp_path / run_journal.RUN_JOURNAL_DIR_NAME / "unused_session")
+    writer = run_journal.RunJournalWriter("unused_session", "unused_run", session_dir=tmp_path)
+    assert writer.append_sse_event("metering", {"tps": 10}) is None
+    assert not any(key[0] == parent for key in run_journal._WRITER_LOCKS)