nesquena-hermesandrodboev
8066d384ca
## 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>
56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
import os
|
|
|
|
from api.turn_journal import (
|
|
append_turn_journal_event,
|
|
append_turn_journal_event_for_stream,
|
|
derive_turn_journal_states,
|
|
)
|
|
|
|
|
|
def test_append_turn_journal_event_for_stream_reuses_submitted_turn_id(tmp_path):
|
|
submitted = append_turn_journal_event(
|
|
"sid-1",
|
|
{"event": "submitted", "turn_id": "turn-1", "stream_id": "stream-1", "content": "hello"},
|
|
session_dir=tmp_path,
|
|
)
|
|
|
|
worker = append_turn_journal_event_for_stream(
|
|
"sid-1",
|
|
"stream-1",
|
|
{"event": "worker_started"},
|
|
session_dir=tmp_path,
|
|
)
|
|
|
|
assert submitted["turn_id"] == "turn-1"
|
|
assert worker["turn_id"] == "turn-1"
|
|
states, _ = derive_turn_journal_states([submitted, worker])
|
|
assert states["turn-1"]["event"] == "worker_started"
|
|
|
|
|
|
def test_append_turn_journal_event_for_stream_falls_back_to_new_turn_for_missing_stream(tmp_path):
|
|
event = append_turn_journal_event_for_stream(
|
|
"sid-1",
|
|
"stream-missing",
|
|
{"event": "interrupted", "reason": "no submitted event found"},
|
|
session_dir=tmp_path,
|
|
)
|
|
|
|
assert event["stream_id"] == "stream-missing"
|
|
assert event["turn_id"]
|
|
assert event["event"] == "interrupted"
|
|
|
|
|
|
def test_append_turn_journal_event_skips_directory_fsync_without_o_directory(tmp_path, monkeypatch):
|
|
monkeypatch.delattr(os, "O_DIRECTORY", raising=False)
|
|
|
|
event = append_turn_journal_event(
|
|
"sid-windows",
|
|
{"event": "submitted", "content": "hello"},
|
|
session_dir=tmp_path,
|
|
)
|
|
|
|
assert event["event"] == "submitted"
|
|
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())}"
|