diff --git a/CHANGELOG.md b/CHANGELOG.md index a6173ff86..455c8cd53 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/api/turn_journal.py b/api/turn_journal.py index 3dba35bdf..e05363cac 100644 --- a/api/turn_journal.py +++ b/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: diff --git a/tests/test_turn_journal.py b/tests/test_turn_journal.py index 4cae1c8f9..477074c7c 100644 --- a/tests/test_turn_journal.py +++ b/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"] diff --git a/tests/test_turn_journal_lifecycle.py b/tests/test_turn_journal_lifecycle.py index 656714d83..575f5c648 100644 --- a/tests/test_turn_journal_lifecycle.py +++ 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())}"