Release v0.51.255 — Release HW (stage-r3) (#3594)

## Release v0.51.255 — Release HW (stage-r3)

Backend hardening — single PR.

### Fixed
| PR | Author | Fix |
|----|--------|-----|
| #3561 | @rodboev | Turn journal (crash-recovery backbone) writes **pid-scoped shards** (`{sid}~{pid}.jsonl`) instead of one shared `{sid}.jsonl`, so concurrent processes (e.g. a self-restart overlap) can't interleave-corrupt large JSON lines. `read_turn_journal` merges all shards + the legacy file and sorts by `created_at` — recovery unchanged, backward-compatible. |

### Gate
- Full pytest suite: **7593 passed, 0 failed**
- ruff: CLEAN · 18 turn-journal tests pass
- Codex (regression): **SAFE TO SHIP** — verified legacy+shard merge (no data loss on upgrade), `~` separator can't collide with a session id, the cross-shard `created_at` sort doesn't break recovery (it derives state by timestamp; stream lookup keys by unique `stream_id`), and no reader/writer bypasses `_journal_path`.
- *Non-blocking note:* old `{sid}~{oldpid}.jsonl` shards aren't pruned, so the journal dir can grow across restarts — storage hygiene, not a core-flow regression. Worth a follow-up cleanup (e.g. drop shards with no live pid on session delete).

Co-authored-by: rodboev <rodboev@users.noreply.github.com>

CHANGELOG.md

@@ -3,6 +3,11 @@
 
 ## [Unreleased]
 
+## [v0.51.255] — 2026-06-04 — Release HW (stage-r3 — pid-scoped turn-journal shards for multi-process safety)
+
+### Fixed
+- The turn journal (the crash-recovery backbone for session state) now writes to **pid-scoped shards** (`{sid}~{pid}.jsonl`) instead of a single shared `{sid}.jsonl`, so two processes writing concurrently (e.g. during a self-restart overlap) can't interleave-corrupt each other's large JSON lines. `read_turn_journal` merges all shards plus the legacy file and sorts events by `created_at`, so recovery is unchanged and fully backward-compatible. Terminal events are now flagged explicitly. (@rodboev)
+
 ## [v0.51.254] — 2026-06-04 — Release HV (stage-r2 — mobile/cancel/scroll fixes + custom-provider model dedup)
 
 ### Fixed

api/turn_journal.py

@@ -36,7 +36,7 @@ def _journal_path(session_id: str, session_dir: Path | None = None) -> Path:
     if not sid or "/" in sid or "\\" in sid or not _SESSION_ID_RE.fullmatch(sid):
         raise ValueError("invalid session_id")
     root = Path(session_dir) if session_dir is not None else _default_session_dir()
-    return root / TURN_JOURNAL_DIR_NAME / f"{sid}.jsonl"
+    return root / TURN_JOURNAL_DIR_NAME / f"{sid}~{os.getpid()}.jsonl"
 
 
 def _make_turn_id() -> str:
@@ -84,6 +84,8 @@ def append_turn_journal_event(
     payload["session_id"] = str(session_id)
     payload.setdefault("turn_id", _make_turn_id())
     payload.setdefault("created_at", time.time())
+    if event_name in _TERMINAL_EVENTS:
+        payload.setdefault("terminal", True)
 
     path = _journal_path(session_id, session_dir=session_dir)
     path.parent.mkdir(parents=True, exist_ok=True)
@@ -108,26 +110,46 @@ def append_turn_journal_event(
 
 
 def read_turn_journal(session_id: str, *, session_dir: Path | None = None) -> dict:
-    """Read a session journal, returning valid events plus malformed lines."""
-    path = _journal_path(session_id, session_dir=session_dir)
+    """Read a session journal, merging all pid-scoped shards and returning valid events plus malformed lines."""
+    sid = str(session_id or "").strip()
+    if not sid or "/" in sid or "\\" in sid or not _SESSION_ID_RE.fullmatch(sid):
+        raise ValueError("invalid session_id")
+    root = Path(session_dir) if session_dir is not None else _default_session_dir()
+    journal_dir = root / TURN_JOURNAL_DIR_NAME
     events: list[dict] = []
     malformed: list[dict] = []
-    try:
-        lines = path.read_text(encoding="utf-8").splitlines()
-    except FileNotFoundError:
+    # Collect pid-scoped shards ({sid}~{pid}.jsonl) plus legacy ({sid}.jsonl).
+    # The ~ separator cannot appear in session IDs (_SESSION_ID_RE allows only [A-Za-z0-9_.-]),
+    # so the glob is unambiguous even for dotted-numeric session IDs like "sess.123".
+    shards: list[Path] = list(journal_dir.glob(f"{sid}~*.jsonl")) if journal_dir.exists() else []
+    legacy = journal_dir / f"{sid}.jsonl"
+    if legacy.exists():
+        shards.append(legacy)
+    if not shards:
         return {"session_id": str(session_id), "events": [], "malformed": []}
-    for line_no, raw in enumerate(lines, start=1):
-        if not raw.strip():
-            continue
+    for shard in shards:
         try:
-            event = json.loads(raw)
-        except json.JSONDecodeError:
-            malformed.append({"line": line_no, "raw": raw})
+            lines = shard.read_text(encoding="utf-8").splitlines()
+        except FileNotFoundError:
             continue
-        if isinstance(event, dict):
-            events.append(event)
-        else:
-            malformed.append({"line": line_no, "raw": raw})
+        for line_no, raw in enumerate(lines, start=1):
+            if not raw.strip():
+                continue
+            try:
+                event = json.loads(raw)
+            except json.JSONDecodeError:
+                malformed.append({"line": line_no, "raw": raw, "shard": shard.name})
+                continue
+            if isinstance(event, dict):
+                events.append(event)
+            else:
+                malformed.append({"line": line_no, "raw": raw, "shard": shard.name})
+    def _safe_ts(e):
+        try:
+            return float(e.get("created_at") or 0)
+        except (ValueError, TypeError):
+            return 0.0
+    events.sort(key=_safe_ts)
     return {"session_id": str(session_id), "events": events, "malformed": malformed}
 
 
@@ -207,7 +229,17 @@ def iter_turn_journal_session_ids(session_dir: Path) -> list[str]:
     journal_dir = Path(session_dir) / TURN_JOURNAL_DIR_NAME
     if not journal_dir.exists():
         return []
-    return sorted(path.stem for path in journal_dir.glob("*.jsonl") if path.is_file())
+    session_ids: set[str] = set()
+    for path in journal_dir.glob("*.jsonl"):
+        if not path.is_file():
+            continue
+        stem = path.stem  # e.g. "sid-1~12345" or "sid-1"
+        tilde = stem.find("~")
+        if tilde > 0:
+            session_ids.add(stem[:tilde])
+        else:
+            session_ids.add(stem)
+    return sorted(session_ids)
 
 
 def is_terminal_turn_event(event: dict) -> bool:

tests/test_turn_journal.py

@@ -1,10 +1,12 @@
 import json
+import os
 
 import api.turn_journal as turn_journal
 from api.session_recovery import audit_session_recovery
 from api.turn_journal import (
     append_turn_journal_event,
     derive_turn_journal_states,
+    iter_turn_journal_session_ids,
     read_turn_journal,
 )
 
@@ -35,9 +37,10 @@ def test_append_turn_journal_event_fsyncs_jsonl_and_preserves_payload(tmp_path):
     assert event["version"] == 1
     assert event["session_id"] == "sid-1"
     assert event["created_at"] > 0
-    journal_path = tmp_path / "_turn_journal" / "sid-1.jsonl"
-    assert journal_path.exists()
-    lines = journal_path.read_text(encoding="utf-8").splitlines()
+    journal_dir = tmp_path / "_turn_journal"
+    shards = list(journal_dir.glob(f"sid-1~{os.getpid()}.jsonl"))
+    assert len(shards) == 1, f"expected one pid-scoped shard, found: {list(journal_dir.iterdir())}"
+    lines = shards[0].read_text(encoding="utf-8").splitlines()
     assert len(lines) == 1
     assert json.loads(lines[0])["content"] == "hello"
 
@@ -55,7 +58,9 @@ def test_read_turn_journal_tolerates_malformed_lines(tmp_path):
     result = read_turn_journal("sid-1", session_dir=tmp_path)
 
     assert [event["event"] for event in result["events"]] == ["submitted", "completed"]
-    assert result["malformed"] == [{"line": 2, "raw": "not-json"}]
+    assert len(result["malformed"]) == 1
+    assert result["malformed"][0]["line"] == 2
+    assert result["malformed"][0]["raw"] == "not-json"
 
 
 def test_append_turn_journal_event_locks_around_write_and_fsync(tmp_path, monkeypatch):
@@ -201,3 +206,69 @@ def test_derive_turn_journal_states_no_collision_when_single_terminal():
 
     assert states['turn-normal']['event'] == 'completed'
     assert collisions == []
+
+
+def test_terminal_field_set_on_completed_event(tmp_path):
+    event = append_turn_journal_event(
+        "sid-term",
+        {"event": "completed", "turn_id": "turn-1"},
+        session_dir=tmp_path,
+    )
+    assert event.get("terminal") is True
+
+
+def test_terminal_field_set_on_interrupted_event(tmp_path):
+    event = append_turn_journal_event(
+        "sid-term-int",
+        {"event": "interrupted", "turn_id": "turn-1"},
+        session_dir=tmp_path,
+    )
+    assert event.get("terminal") is True
+
+
+def test_terminal_field_not_set_on_non_terminal_event(tmp_path):
+    event = append_turn_journal_event(
+        "sid-nterm",
+        {"event": "submitted", "turn_id": "turn-1", "content": "hi"},
+        session_dir=tmp_path,
+    )
+    assert "terminal" not in event
+
+
+def test_read_turn_journal_merges_pid_shards(tmp_path, monkeypatch):
+    journal_dir = tmp_path / "_turn_journal"
+    journal_dir.mkdir()
+
+    # Simulate two worker processes writing separate shards
+    monkeypatch.setattr(os, "getpid", lambda: 1001)
+    append_turn_journal_event(
+        "sid-merge",
+        {"event": "submitted", "turn_id": "turn-1", "created_at": 1.0},
+        session_dir=tmp_path,
+    )
+    monkeypatch.setattr(os, "getpid", lambda: 1002)
+    append_turn_journal_event(
+        "sid-merge",
+        {"event": "completed", "turn_id": "turn-1", "created_at": 2.0},
+        session_dir=tmp_path,
+    )
+
+    result = read_turn_journal("sid-merge", session_dir=tmp_path)
+
+    assert len(result["events"]) == 2
+    assert result["events"][0]["event"] == "submitted"
+    assert result["events"][1]["event"] == "completed"
+
+
+def test_iter_turn_journal_session_ids_deduplicates_pid_shards(tmp_path):
+    journal_dir = tmp_path / "_turn_journal"
+    journal_dir.mkdir()
+
+    # Create two pid-scoped shards for the same session, plus a legacy file for another session
+    (journal_dir / "sess-a~1001.jsonl").write_text("", encoding="utf-8")
+    (journal_dir / "sess-a~1002.jsonl").write_text("", encoding="utf-8")
+    (journal_dir / "sess-b.jsonl").write_text("", encoding="utf-8")
+
+    ids = iter_turn_journal_session_ids(tmp_path)
+
+    assert ids == ["sess-a", "sess-b"]

tests/test_turn_journal_lifecycle.py

@@ -50,4 +50,6 @@ def test_append_turn_journal_event_skips_directory_fsync_without_o_directory(tmp
     )
 
     assert event["event"] == "submitted"
-    assert (tmp_path / "_turn_journal" / "sid-windows.jsonl").is_file()
+    journal_dir = tmp_path / "_turn_journal"
+    shards = list(journal_dir.glob(f"sid-windows~{os.getpid()}.jsonl"))
+    assert len(shards) == 1, f"expected one pid-scoped shard, found: {list(journal_dir.iterdir())}"