460 lines
16 KiB
Python
460 lines
16 KiB
Python
"""Append-only WebUI run event journal helpers.
|
|
|
|
This is the first #1925 journal/replay slice. It mirrors SSE events emitted by
|
|
the existing in-process streaming path without changing execution ownership.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
RUN_JOURNAL_DIR_NAME = "_run_journal"
|
|
_SAFE_ID_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
|
|
_WRITER_LOCKS: dict[tuple[str, str, str], threading.Lock] = {}
|
|
_WRITER_LOCKS_GUARD = threading.Lock()
|
|
_TERMINAL_SSE_EVENTS = {"done", "cancel", "apperror", "error", "stream_end"}
|
|
_FSYNC_MODE_ENV = "HERMES_WEBUI_RUN_JOURNAL_FSYNC"
|
|
_FSYNC_MODE_EAGER = "eager"
|
|
_FSYNC_MODE_TERMINAL_ONLY = "terminal-only"
|
|
|
|
|
|
def _default_session_dir() -> Path:
|
|
from api.models import SESSION_DIR
|
|
|
|
return Path(SESSION_DIR)
|
|
|
|
|
|
def _validate_id(value: str, field: str) -> str:
|
|
cleaned = str(value or "").strip()
|
|
if not cleaned or "/" in cleaned or "\\" in cleaned or not _SAFE_ID_RE.fullmatch(cleaned):
|
|
raise ValueError(f"invalid {field}")
|
|
return cleaned
|
|
|
|
|
|
def _run_path(session_id: str, run_id: str, session_dir: Path | None = None) -> Path:
|
|
sid = _validate_id(session_id, "session_id")
|
|
rid = _validate_id(run_id, "run_id")
|
|
root = Path(session_dir) if session_dir is not None else _default_session_dir()
|
|
return root / RUN_JOURNAL_DIR_NAME / sid / f"{rid}.jsonl"
|
|
|
|
|
|
def _lock_for(path: Path) -> threading.Lock:
|
|
key = (str(path.parent), path.name, str(os.getpid()))
|
|
with _WRITER_LOCKS_GUARD:
|
|
lock = _WRITER_LOCKS.get(key)
|
|
if lock is None:
|
|
lock = threading.Lock()
|
|
_WRITER_LOCKS[key] = lock
|
|
return lock
|
|
|
|
|
|
def _read_jsonl(path: Path) -> tuple[list[dict], list[dict]]:
|
|
events: list[dict] = []
|
|
malformed: list[dict] = []
|
|
try:
|
|
lines = path.read_text(encoding="utf-8").splitlines()
|
|
except FileNotFoundError:
|
|
return events, malformed
|
|
for line_no, raw in enumerate(lines, start=1):
|
|
if not raw.strip():
|
|
continue
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
malformed.append({"line": line_no, "raw": raw})
|
|
continue
|
|
if isinstance(parsed, dict):
|
|
events.append(parsed)
|
|
else:
|
|
malformed.append({"line": line_no, "raw": raw})
|
|
return events, malformed
|
|
|
|
|
|
def _parse_run_journal_event_id(raw: str | None) -> tuple[str | None, int | None]:
|
|
raw = str(raw or "").strip()
|
|
if not raw:
|
|
return None, None
|
|
if ":" in raw:
|
|
run_id, tail = raw.rsplit(":", 1)
|
|
else:
|
|
run_id, tail = None, raw
|
|
try:
|
|
seq = max(0, int(tail))
|
|
except (TypeError, ValueError):
|
|
return run_id or None, None
|
|
return run_id or None, seq
|
|
|
|
|
|
def _next_seq(path: Path) -> int:
|
|
events, _malformed = _read_jsonl(path)
|
|
seqs = [int(event.get("seq") or 0) for event in events if isinstance(event.get("seq"), int)]
|
|
return (max(seqs) + 1) if seqs else 1
|
|
|
|
|
|
def _terminal_state_for_event(event_name: str, payload) -> str | None:
|
|
name = str(event_name or "")
|
|
if name == "done" or name == "stream_end":
|
|
if isinstance(payload, dict):
|
|
explicit_state = str(payload.get("terminal_state") or "").strip().lower()
|
|
if explicit_state in {"tool_limit_reached"}:
|
|
return explicit_state
|
|
return "completed"
|
|
if name == "cancel":
|
|
return "interrupted-by-user"
|
|
if name in {"apperror", "error"}:
|
|
err_type = str((payload or {}).get("type") or "").strip().lower() if isinstance(payload, dict) else ""
|
|
if err_type == "tool_limit_reached":
|
|
return "tool_limit_reached"
|
|
if err_type in {"cancelled", "canceled"}:
|
|
return "interrupted-by-user"
|
|
if err_type == "interrupted":
|
|
return "interrupted-by-crash"
|
|
return "errored"
|
|
return None
|
|
|
|
|
|
def _run_journal_fsync_mode() -> str:
|
|
raw = os.environ.get(_FSYNC_MODE_ENV, _FSYNC_MODE_TERMINAL_ONLY)
|
|
mode = str(raw or "").strip().lower()
|
|
if mode in {_FSYNC_MODE_EAGER, _FSYNC_MODE_TERMINAL_ONLY}:
|
|
return mode
|
|
return _FSYNC_MODE_TERMINAL_ONLY
|
|
|
|
|
|
def _should_fsync_event(terminal_state: str | None) -> bool:
|
|
if _run_journal_fsync_mode() == _FSYNC_MODE_EAGER:
|
|
return True
|
|
return bool(terminal_state)
|
|
|
|
|
|
def _fsync_parent_dir(path: Path) -> None:
|
|
try:
|
|
dir_fd = os.open(path.parent, getattr(os, "O_DIRECTORY", 0))
|
|
try:
|
|
os.fsync(dir_fd)
|
|
finally:
|
|
os.close(dir_fd)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _event_created_at(event: dict, *, fallback: float = 0.0) -> float:
|
|
try:
|
|
return float(event.get("created_at") or fallback)
|
|
except (TypeError, ValueError):
|
|
return fallback
|
|
|
|
|
|
def _session_run_journal_entries(session_id: str, *, session_dir: Path | None = None) -> list[dict]:
|
|
sid = _validate_id(session_id, "session_id")
|
|
root = Path(session_dir) if session_dir is not None else _default_session_dir()
|
|
session_root = root / RUN_JOURNAL_DIR_NAME / sid
|
|
if not session_root.exists():
|
|
return []
|
|
entries: list[dict] = []
|
|
for path in sorted(session_root.glob("*.jsonl")):
|
|
run_id = path.stem
|
|
try:
|
|
run_id = _validate_id(run_id, "run_id")
|
|
except ValueError:
|
|
continue
|
|
events, malformed = _read_jsonl(path)
|
|
ordered_events = sorted(
|
|
[event for event in events if isinstance(event, dict)],
|
|
key=lambda event: (
|
|
int(event.get("seq") or 0),
|
|
_event_created_at(event),
|
|
str(event.get("event_id") or ""),
|
|
),
|
|
)
|
|
summary = _summary_from_events(sid, run_id, ordered_events)
|
|
created_at = min((
|
|
_event_created_at(event)
|
|
for event in ordered_events
|
|
), default=path.stat().st_mtime if path.exists() else 0.0)
|
|
entries.append(
|
|
{
|
|
"session_id": sid,
|
|
"run_id": run_id,
|
|
"path": str(path),
|
|
"created_at": created_at,
|
|
"event_count": len(ordered_events),
|
|
"events": ordered_events,
|
|
"summary": summary,
|
|
"malformed": malformed,
|
|
}
|
|
)
|
|
entries.sort(key=lambda entry: (entry["created_at"], entry["run_id"]))
|
|
return entries
|
|
|
|
|
|
def append_run_event(
|
|
session_id: str,
|
|
run_id: str,
|
|
event_name: str,
|
|
payload=None,
|
|
*,
|
|
session_dir: Path | None = None,
|
|
seq: int | None = None,
|
|
created_at: float | None = None,
|
|
) -> dict:
|
|
"""Append one durable run event and fsync it according to the journal policy."""
|
|
path = _run_path(session_id, run_id, session_dir=session_dir)
|
|
payload = payload if payload is not None else {}
|
|
event_name = str(event_name or "").strip()
|
|
if not event_name:
|
|
raise ValueError("event_name is required")
|
|
with _lock_for(path):
|
|
assigned_seq = int(seq) if seq is not None else _next_seq(path)
|
|
terminal_state = _terminal_state_for_event(event_name, payload)
|
|
event = {
|
|
"version": 1,
|
|
"event_id": f"{run_id}:{assigned_seq}",
|
|
"seq": assigned_seq,
|
|
"run_id": str(run_id),
|
|
"session_id": str(session_id),
|
|
"event": event_name,
|
|
"type": event_name,
|
|
"created_at": float(created_at if created_at is not None else time.time()),
|
|
"terminal": bool(terminal_state),
|
|
"terminal_state": terminal_state,
|
|
"payload": payload,
|
|
}
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
created_file = not path.exists()
|
|
line = json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n"
|
|
fd = os.open(path, os.O_CREAT | os.O_APPEND | os.O_WRONLY, 0o600)
|
|
with os.fdopen(fd, "a", encoding="utf-8") as fh:
|
|
fh.write(line)
|
|
fh.flush()
|
|
if _should_fsync_event(terminal_state):
|
|
os.fsync(fh.fileno())
|
|
if created_file:
|
|
_fsync_parent_dir(path)
|
|
return event
|
|
|
|
|
|
class RunJournalWriter:
|
|
"""Stateful writer for one WebUI stream/run."""
|
|
|
|
def __init__(self, session_id: str, run_id: str, *, session_dir: Path | None = None):
|
|
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)
|
|
with self._lock:
|
|
self._next_seq = _next_seq(self._path)
|
|
|
|
def append_sse_event(self, event_name: str, payload=None) -> dict:
|
|
with self._lock:
|
|
seq = self._next_seq
|
|
self._next_seq += 1
|
|
return append_run_event(
|
|
self.session_id,
|
|
self.run_id,
|
|
event_name,
|
|
payload or {},
|
|
session_dir=self.session_dir,
|
|
seq=seq,
|
|
)
|
|
|
|
|
|
def read_run_events(
|
|
session_id: str,
|
|
run_id: str,
|
|
*,
|
|
after_seq: int | None = None,
|
|
max_seq: int | None = None,
|
|
session_dir: Path | None = None,
|
|
) -> dict:
|
|
path = _run_path(session_id, run_id, session_dir=session_dir)
|
|
events, malformed = _read_jsonl(path)
|
|
if after_seq is not None:
|
|
events = [event for event in events if int(event.get("seq") or 0) > int(after_seq)]
|
|
if max_seq is not None:
|
|
events = [event for event in events if int(event.get("seq") or 0) <= int(max_seq)]
|
|
return {
|
|
"session_id": str(session_id),
|
|
"run_id": str(run_id),
|
|
"events": events,
|
|
"malformed": malformed,
|
|
}
|
|
|
|
|
|
def _summary_from_events(session_id: str, run_id: str, events: Iterable[dict]) -> dict:
|
|
ordered = [event for event in events if isinstance(event, dict)]
|
|
last = ordered[-1] if ordered else None
|
|
terminal_events = [event for event in ordered if event.get("terminal")]
|
|
terminal = next(
|
|
(event for event in reversed(terminal_events) if event.get("event") != "stream_end"),
|
|
terminal_events[-1] if terminal_events else None,
|
|
)
|
|
status = terminal.get("terminal_state") if terminal else ("running" if ordered else "unknown")
|
|
return {
|
|
"session_id": str(session_id),
|
|
"run_id": str(run_id),
|
|
"stream_id": str(run_id),
|
|
"event_count": len(ordered),
|
|
"last_seq": int((last or {}).get("seq") or 0),
|
|
"last_event_id": (last or {}).get("event_id"),
|
|
"terminal": bool(terminal),
|
|
"terminal_state": status,
|
|
"last_event": (last or {}).get("event"),
|
|
}
|
|
|
|
|
|
def latest_run_summary(session_id: str, run_id: str, *, session_dir: Path | None = None) -> dict:
|
|
journal = read_run_events(session_id, run_id, session_dir=session_dir)
|
|
return _summary_from_events(session_id, run_id, journal.get("events") or [])
|
|
|
|
|
|
def find_run_summary(run_id: str, *, session_dir: Path | None = None) -> dict | None:
|
|
rid = _validate_id(run_id, "run_id")
|
|
root = Path(session_dir) if session_dir is not None else _default_session_dir()
|
|
journal_root = root / RUN_JOURNAL_DIR_NAME
|
|
for path in journal_root.glob(f"*/{rid}.jsonl"):
|
|
session_id = path.parent.name
|
|
events, _malformed = _read_jsonl(path)
|
|
summary = _summary_from_events(session_id, rid, events)
|
|
summary["path"] = str(path)
|
|
return summary
|
|
return None
|
|
|
|
|
|
def read_session_run_events(
|
|
session_id: str,
|
|
*,
|
|
after_event_id: str | None = None,
|
|
session_dir: Path | None = None,
|
|
) -> dict:
|
|
"""Replay durable run-journal rows for one session after an opaque cursor."""
|
|
sid = _validate_id(session_id, "session_id")
|
|
cursor_run_id, cursor_seq = _parse_run_journal_event_id(after_event_id)
|
|
raw_cursor = str(after_event_id or "").strip()
|
|
if raw_cursor and cursor_run_id is not None:
|
|
try:
|
|
cursor_run_id = _validate_id(cursor_run_id, "run_id")
|
|
except ValueError:
|
|
cursor_seq = None
|
|
if raw_cursor:
|
|
try:
|
|
if int(raw_cursor.rsplit(":", 1)[-1]) < 0:
|
|
cursor_seq = None
|
|
except (TypeError, ValueError):
|
|
pass
|
|
if raw_cursor and (cursor_run_id is None or cursor_seq is None):
|
|
return {
|
|
"session_id": sid,
|
|
"cursor_run_id": cursor_run_id,
|
|
"cursor_seq": cursor_seq,
|
|
"status": "cursor_invalid",
|
|
"events": [],
|
|
}
|
|
runs = _session_run_journal_entries(sid, session_dir=session_dir)
|
|
if not raw_cursor:
|
|
return {
|
|
"session_id": sid,
|
|
"cursor_run_id": None,
|
|
"cursor_seq": None,
|
|
"status": "ok",
|
|
"events": [],
|
|
"runs": runs,
|
|
}
|
|
run_lookup = {entry["run_id"]: entry for entry in runs}
|
|
cursor_entry = run_lookup.get(cursor_run_id or "")
|
|
if cursor_entry is None:
|
|
summary = find_run_summary(cursor_run_id or "", session_dir=session_dir) if cursor_run_id else None
|
|
status = "cursor_run_missing"
|
|
summary_sid = str(summary.get("session_id") or "").strip() if summary else ""
|
|
if summary_sid and summary_sid != sid:
|
|
status = "cursor_session_mismatch"
|
|
return {
|
|
"session_id": sid,
|
|
"cursor_run_id": cursor_run_id,
|
|
"cursor_seq": cursor_seq,
|
|
"status": status,
|
|
"events": [],
|
|
}
|
|
replay_events: list[dict] = []
|
|
started = False
|
|
for entry in runs:
|
|
if not started:
|
|
if entry["run_id"] != cursor_entry["run_id"]:
|
|
continue
|
|
started = True
|
|
events = entry["events"]
|
|
if entry["run_id"] == cursor_entry["run_id"] and cursor_seq is not None:
|
|
events = [event for event in events if int(event.get("seq") or 0) > int(cursor_seq)]
|
|
replay_events.extend(events)
|
|
return {
|
|
"session_id": sid,
|
|
"cursor_run_id": cursor_run_id,
|
|
"cursor_seq": cursor_seq,
|
|
"status": "ok",
|
|
"events": replay_events,
|
|
"runs": runs,
|
|
}
|
|
|
|
|
|
def delete_run_journal(session_id: str, *, session_dir: Path | None = None) -> bool:
|
|
"""Remove the entire per-session run-journal directory (``_run_journal/{sid}/``).
|
|
|
|
The run journal stores one directory per session containing a ``{rid}.jsonl``
|
|
file per run, so removing the session's directory clears every run's full
|
|
request/response payloads. Invalid/empty ids and a missing directory are a
|
|
no-op so callers can invoke this unconditionally on delete. Returns ``True``
|
|
if a directory was removed, ``False`` otherwise.
|
|
"""
|
|
import shutil
|
|
|
|
sid = str(session_id or "").strip()
|
|
# Reject path-traversal ids: the regex below permits dots, so a bare "." or
|
|
# ".." would resolve `root / RUN_JOURNAL_DIR_NAME / sid` to the journal ROOT
|
|
# (or its parent) and rmtree the wrong directory. The route call site only
|
|
# passes real sids, but this is a public helper — guard it directly.
|
|
if sid in (".", "..") or not sid or "/" in sid or "\\" in sid or not _SAFE_ID_RE.fullmatch(sid):
|
|
return False
|
|
root = Path(session_dir) if session_dir is not None else _default_session_dir()
|
|
session_journal_dir = root / RUN_JOURNAL_DIR_NAME / sid
|
|
if not session_journal_dir.exists():
|
|
return False
|
|
shutil.rmtree(session_journal_dir, ignore_errors=True)
|
|
return not session_journal_dir.exists()
|
|
|
|
|
|
def stale_interrupted_event(session_id: str, run_id: str, *, after_seq: int | None = None) -> dict | None:
|
|
summary = latest_run_summary(session_id, run_id)
|
|
if summary.get("terminal") or not summary.get("event_count"):
|
|
return None
|
|
seq = int(summary.get("last_seq") or 0) + 1
|
|
if after_seq is not None and seq <= int(after_seq):
|
|
return None
|
|
payload = {
|
|
"type": "interrupted",
|
|
"recovery_control": True,
|
|
"message": "The live worker stopped before this run finished.",
|
|
"hint": "The transcript was restored to the last journaled event. Start a new turn if you still need the task to continue.",
|
|
"session_id": session_id,
|
|
"stream_id": run_id,
|
|
"journal_last_seq": summary.get("last_seq"),
|
|
}
|
|
return {
|
|
"version": 1,
|
|
"event_id": f"{run_id}:{seq}",
|
|
"seq": seq,
|
|
"run_id": run_id,
|
|
"session_id": session_id,
|
|
"event": "apperror",
|
|
"type": "apperror",
|
|
"created_at": time.time(),
|
|
"terminal": True,
|
|
"terminal_state": "lost-worker-bookkeeping",
|
|
"payload": payload,
|
|
"synthetic": True,
|
|
}
|