Merge pull request #2094 from nesquena/stage-339

Release V — v0.51.46 (5-PR contributor batch — CSP report-only + logs panel polish + plugin slash commands + turn-journal crash-safe writer + lifecycle events)

CHANGELOG.md

@@ -2,6 +2,30 @@
 
 ## [Unreleased]
 
+## [v0.51.46] — 2026-05-11 — Release V (5-PR contributor batch — CSP report-only + logs panel polish + plugin slash commands + turn-journal crash-safe writer + lifecycle events)
+
+### Added
+
+- **PR #2059** by @ai-ag2026 — Append-only WebUI turn journal helper at `api/turn_journal.py` (new file, ~128 LOC). Writes one JSONL file per session under `_turn_journal/` and fsyncs `submitted`-turn events before the worker thread starts via `/api/chat/start` (after pending session state is saved and before `threading.Thread(...)` starts). `recovery_audit` extended to report non-terminal journal turns as `turn_journal_pending_turn` when the submitted user message is not present in the sidecar. Intentionally the minimal slice from `docs/rfcs/turn-journal.md` (RFC #2042): writer + reader + state derivation + audit-only reporting. No replay or repair yet.
+
+- **PR #2062** by @ai-ag2026 — Turn-journal lifecycle events on top of #2059's submitted-event writer. Records `worker_started` when the streaming worker begins, `assistant_started` before the final session save once an assistant message exists, `completed` after the final save, and `interrupted` on the provider-error path. `append_turn_journal_event_for_stream(...)` reuses the `turn_id` associated with the stream's submitted event. Still audit-only / journaling-only — does not replay turns or repair assistant output. The little WAL goblin remains on a leash.
+
+- **PR #2089** by @plerohellec — Plugin-defined slash commands now surface in the WebUI command picker and execute via a new `/api/commands/exec` route (closes #1935). `list_commands()` in `api/commands.py` merges `hermes_cli.plugins.get_plugin_commands()` into the `/api/commands` payload with `category: "Plugin"`; the frontend intercepts plugin commands in `static/messages.js` and `static/commands.js` to route through the plugin execution endpoint instead of falling through to the agent. Pre-fix the WebUI only learned slash commands from `hermes_cli.commands.COMMAND_REGISTRY` (commands.py:23), so plugin-registered commands were invisible to the picker, autocomplete, and routing — they fell through to the agent as raw text and the agent's response was about an unknown command. This is the WebUI half of the parity fix; the corresponding agent-side plumbing already existed in `hermes_cli/plugins.py:1424` (`get_plugin_commands()`).
+
+### Fixed
+
+- **PR #2085** by @bergeouss — Logs panel: clipboard `_copyText()` fallback + severity filter (closes #2081). Pre-fix `copyLogsAll()` called `navigator.clipboard.writeText()` directly with no fallback — failed silently on large payloads / non-secure contexts / unfocused pages, leaving users with a useless error toast. Now routes through `_copyText()` from `ui.js` which already has a `<textarea>` + `document.execCommand('copy')` fallback. Also adds a Severity dropdown (All / Errors / Warnings+) that filters the in-memory log cache without re-fetching — `errors.log` is ~90% WARNING tool noise so filtering down to ERROR/CRITICAL is a real triage time-saver. `copyLogsAll()` copies the FILTERED subset when a filter is active. 5 new i18n keys in all 9 locales.
+
+- **PR #2084** by @ai-ag2026 — `Content-Security-Policy-Report-Only` header (refs #1909). All WebUI responses now ship a CSP slice in report-only mode — non-enforcing, so the browser collects violations without blocking page behavior. Current UI allowances (`'unsafe-inline'` for scripts and styles, plus `https://cdn.jsdelivr.net` for the Prism/xterm/katex CDN assets that `static/index.html` loads with SRI hashes) are explicit so future tightening passes can replace them one constraint at a time. `object-src 'none'`, `base-uri 'self'`, and `frame-ancestors 'self'` are already enforced because they don't break the current UI. Server-side change only (`server.py` headers), zero client-side risk.
+
+### Stage-339 maintainer review (Opus advisor)
+
+- **`server.py:_CSP_REPORT_ONLY`** — Dropped `'unsafe-eval'` after Opus verified by grepping all production JS that nothing uses `eval()`, `new Function()`, or string-form `setTimeout`/`setInterval`. Keeping the allowance would have been a gratuitous privilege that defeats the purpose of the dry-run. ~1 LOC.
+
+- **`server.py:_CSP_REPORT_ONLY`** — Added `https://cdn.jsdelivr.net` to `script-src` and `style-src`. `static/index.html` loads Prism, xterm.js, and katex CSS from jsdelivr with SRI integrity hashes. Without the allowance, every page load would fire known-good CSP violations and drown out the real dry-run signal. ~2 LOC.
+
+- **`api/commands.py:execute_plugin_command`** — Sanitized the plugin error message. Previously returned `f"Plugin command error: {exc}"` which would leak paths / env / internal state from a `FileNotFoundError('/etc/something/secret.key')`-shape exception verbatim to the user-facing chat. Now returns only `type(exc).__name__`; the full traceback is logged at WARNING via `logger.warning(..., exc_info=exc)`. ~4 LOC.
+
 ## [v0.51.45] — 2026-05-11 — Release U (9-PR contributor batch — themes docs + gitignore policy + kanban parity + skill cache patching + fork lineage + sidebar spinner + custom provider slug + session recovery polish + compression anchor refactor)
 
 ### Added

api/commands.py

@@ -53,4 +53,72 @@ def list_commands(_registry=None) -> list[dict[str, Any]]:
             'cli_only': bool(cmd.cli_only),
             'gateway_only': bool(cmd.gateway_only),
         })
+
+    # Include plugin-registered slash commands
+    try:
+        from hermes_cli.plugins import get_plugin_commands
+        plugin_cmds = get_plugin_commands() or {}
+        existing_names = {c['name'] for c in out}
+        for cmd_name, cmd_info in plugin_cmds.items():
+            if cmd_name in existing_names or cmd_name in _NEVER_EXPOSE:
+                continue
+            out.append({
+                'name': cmd_name,
+                'description': str(cmd_info.get('description', 'Plugin command')),
+                'category': 'Plugin',
+                'aliases': [],
+                'args_hint': str(cmd_info.get('args_hint', '')),
+                'subcommands': [],
+                'cli_only': False,
+                'gateway_only': False,
+            })
+    except Exception:
+        pass
+
     return out
+
+
+def execute_plugin_command(command: str) -> str:
+    """Execute a plugin-registered slash command and return printable output.
+
+    Unknown commands raise ``KeyError`` so the HTTP layer can return 404.
+    Plugin handler failures are returned as output text instead of surfacing as
+    transport errors, matching Hermes' existing slash-command UX.
+    """
+
+    raw = str(command or "").strip()
+    if not raw:
+        raise ValueError("command is required")
+
+    cmd_text = raw[1:] if raw.startswith("/") else raw
+    cmd_parts = cmd_text.split(maxsplit=1)
+    cmd_base = (cmd_parts[0] if cmd_parts else "").strip().lower()
+    cmd_arg = cmd_parts[1] if len(cmd_parts) > 1 else ""
+    if not cmd_base:
+        raise ValueError("command is required")
+
+    try:
+        from hermes_cli.plugins import (
+            get_plugin_command_handler,
+            resolve_plugin_command_result,
+        )
+    except ImportError as exc:
+        raise RuntimeError("plugin command runtime unavailable") from exc
+
+    try:
+        handler = get_plugin_command_handler(cmd_base)
+    except Exception as exc:
+        raise RuntimeError(f"plugin command lookup failed: {exc}") from exc
+
+    if not handler:
+        raise KeyError(cmd_base)
+
+    try:
+        result = resolve_plugin_command_result(handler(cmd_arg))
+        return str(result or "(no output)")
+    except Exception as exc:
+        # Don't leak raw exception str (paths, env, internal state) to the
+        # user-facing chat. Type name is enough for the user to know what
+        # class of failure occurred; full traceback lives in the server log.
+        logger.warning("Plugin command %r failed", cmd_base, exc_info=exc)
+        return f"Plugin command error: {type(exc).__name__}"

api/routes.py

@@ -4539,6 +4539,22 @@ def handle_post(handler, parsed) -> bool:
     if parsed.path == "/api/clarify/respond":
         return _handle_clarify_respond(handler, body)
 
+    # ── Commands (POST) ──
+    if parsed.path == "/api/commands/exec":
+        from api.commands import execute_plugin_command
+
+        command = str(body.get("command", "") or "").strip()
+        if not command:
+            return bad(handler, "command is required")
+        try:
+            return j(handler, {"output": execute_plugin_command(command)})
+        except ValueError as e:
+            return bad(handler, str(e), 400)
+        except KeyError:
+            return bad(handler, "Plugin command not found", 404)
+        except RuntimeError as e:
+            return bad(handler, _sanitize_error(e), 500)
+
     # ── Skills (POST) ──
     if parsed.path == "/api/skills/save":
         return _handle_skill_save(handler, body)
@@ -6676,6 +6692,26 @@ def _start_chat_stream_for_session(
             model_provider=model_provider,
             stream_id=stream_id,
         )
+    diag.stage("turn_journal_submitted") if diag else None
+    journal_event = {}
+    try:
+        from api.turn_journal import append_turn_journal_event
+        journal_event = append_turn_journal_event(
+            s.session_id,
+            {
+                "event": "submitted",
+                "stream_id": stream_id,
+                "role": "user",
+                "content": msg,
+                "attachments": attachments,
+                "workspace": workspace,
+                "model": model,
+                "model_provider": model_provider,
+                "created_at": s.pending_started_at,
+            },
+        )
+    except Exception:
+        logger.warning("Failed to append submitted turn journal event", exc_info=True)
     diag.stage("set_last_workspace") if diag else None
     set_last_workspace(workspace)
     diag.stage("stream_registration") if diag else None
@@ -6697,6 +6733,7 @@ def _start_chat_stream_for_session(
         "stream_id": stream_id,
         "session_id": s.session_id,
         "pending_started_at": s.pending_started_at,
+        "turn_id": journal_event.get("turn_id"),
     }
     if normalized_model:
         response["effective_model"] = model

api/session_recovery.py

@@ -34,6 +34,13 @@ import sqlite3
 import threading
 from pathlib import Path
 
+from api.turn_journal import (
+    derive_turn_journal_states,
+    is_terminal_turn_event,
+    iter_turn_journal_session_ids,
+    read_turn_journal,
+)
+
 logger = logging.getLogger(__name__)
 
 
@@ -373,8 +380,9 @@ def _new_audit_item(
     recommendation: str,
     live_messages: int = -1,
     bak_messages: int = -1,
+    **extra,
 ) -> dict:
-    return {
+    item = {
         "session_id": session_id,
         "kind": kind,
         "category": category,
@@ -382,6 +390,8 @@ def _new_audit_item(
         "live_messages": live_messages,
         "bak_messages": bak_messages,
     }
+    item.update(extra)
+    return item
 
 
 def _read_index_session_ids(index_path: Path) -> set[str]:
@@ -487,6 +497,37 @@ def audit_session_recovery(session_dir: Path, state_db_path: Path | None = None)
             -1,
         ))
 
+    for session_id in iter_turn_journal_session_ids(session_dir):
+        journal = read_turn_journal(session_id, session_dir=session_dir)
+        states = derive_turn_journal_states(journal.get('events') or [])
+        live_path = session_dir / f"{session_id}.json"
+        live_messages = _msg_count(live_path)
+        existing_user_messages: set[str] = set()
+        try:
+            payload = json.loads(live_path.read_text(encoding='utf-8'))
+            if isinstance(payload, dict):
+                for message in payload.get('messages') or []:
+                    if isinstance(message, dict) and message.get('role') == 'user':
+                        existing_user_messages.add(str(message.get('content') or '').strip())
+        except (OSError, json.JSONDecodeError, ValueError):
+            pass
+        for turn_id, event in sorted(states.items()):
+            if is_terminal_turn_event(event):
+                continue
+            content = str(event.get('content') or '').strip()
+            if not content or content in existing_user_messages:
+                continue
+            items.append(_new_audit_item(
+                session_id,
+                "turn_journal_pending_turn",
+                "repairable",
+                "audit_only_pending_turn_journal",
+                live_messages,
+                -1,
+                turn_id=turn_id,
+                event=str(event.get('event') or ''),
+            ))
+
     summary = {"ok": len(live_paths), "repairable": 0, "unsafe_to_repair": 0}
     for item in items:
         category = item.get('category')

api/streaming.py

@@ -35,6 +35,7 @@ from api.config import (
 from api.helpers import redact_session_data, _redact_text
 from api.compression_anchor import visible_messages_for_anchor
 from api.metering import meter
+from api.turn_journal import append_turn_journal_event_for_stream
 
 # Global lock for os.environ writes. Per-session locks (_agent_lock) prevent
 # concurrent runs of the SAME session, but two DIFFERENT sessions can still
@@ -2016,6 +2017,15 @@ def _run_agent_streaming(
         provider=model_provider,
         ephemeral=bool(ephemeral),
     )
+    if not ephemeral:
+        try:
+            append_turn_journal_event_for_stream(
+                session_id,
+                stream_id,
+                {"event": "worker_started", "created_at": time.time()},
+            )
+        except Exception:
+            logger.debug("Failed to append worker_started turn journal event", exc_info=True)
     s = None
     _rt = {}
     old_cwd = None
@@ -3512,7 +3522,44 @@ def _run_agent_streaming(
                         # Older hermes-agent builds may not expose this helper.
                         # Better to leave context_length=0 than crash the save.
                         pass
+                if not ephemeral and s.messages:
+                    _latest_assistant_idx = next(
+                        (idx for idx in range(len(s.messages) - 1, -1, -1)
+                         if isinstance(s.messages[idx], dict) and s.messages[idx].get('role') == 'assistant'),
+                        None,
+                    )
+                    if _latest_assistant_idx is not None:
+                        _latest_assistant = s.messages[_latest_assistant_idx]
+                        try:
+                            append_turn_journal_event_for_stream(
+                                s.session_id,
+                                stream_id,
+                                {
+                                    "event": "assistant_started",
+                                    "created_at": float(_latest_assistant.get('timestamp') or time.time()),
+                                    "assistant_message_index": _latest_assistant_idx,
+                                },
+                            )
+                        except Exception:
+                            logger.debug("Failed to append assistant_started turn journal event", exc_info=True)
                 s.save()
+                if not ephemeral:
+                    try:
+                        append_turn_journal_event_for_stream(
+                            s.session_id,
+                            stream_id,
+                            {
+                                "event": "completed",
+                                "created_at": time.time(),
+                                "assistant_message_index": next(
+                                    (idx for idx in range(len(s.messages) - 1, -1, -1)
+                                     if isinstance(s.messages[idx], dict) and s.messages[idx].get('role') == 'assistant'),
+                                    None,
+                                ),
+                            },
+                        )
+                    except Exception:
+                        logger.debug("Failed to append completed turn journal event", exc_info=True)
             # Sync to state.db for /insights (opt-in setting)
             try:
                 from api.config import load_settings as _load_settings
@@ -3882,6 +3929,19 @@ def _run_agent_streaming(
                     s.save()
                 except Exception:
                     pass
+                if not ephemeral:
+                    try:
+                        append_turn_journal_event_for_stream(
+                            s.session_id,
+                            stream_id,
+                            {
+                                "event": "interrupted",
+                                "created_at": time.time(),
+                                "reason": _exc_type,
+                            },
+                        )
+                    except Exception:
+                        logger.debug("Failed to append interrupted turn journal event", exc_info=True)
         put('apperror', _error_payload)
     finally:
         # Stop the periodic checkpoint thread before the final recovery path.

api/turn_journal.py

@@ -0,0 +1,162 @@
+"""Crash-safe WebUI turn journal helpers.
+
+The journal is deliberately tiny: one JSONL file per session, append-only events,
+and read helpers that tolerate malformed lines. Recovery and repair can then
+reason about submitted turns without depending on in-memory stream state.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import time
+import uuid
+from pathlib import Path
+from typing import Iterable
+
+TURN_JOURNAL_DIR_NAME = "_turn_journal"
+_TERMINAL_EVENTS = {"completed", "interrupted"}
+_SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
+
+
+def _default_session_dir() -> Path:
+    from api.models import SESSION_DIR
+
+    return Path(SESSION_DIR)
+
+
+def _journal_path(session_id: str, session_dir: Path | None = None) -> Path:
+    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()
+    return root / TURN_JOURNAL_DIR_NAME / f"{sid}.jsonl"
+
+
+def _make_turn_id() -> str:
+    return f"{time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())}-{uuid.uuid4().hex[:12]}"
+
+
+def append_turn_journal_event(
+    session_id: str,
+    event: dict,
+    *,
+    session_dir: Path | None = None,
+) -> dict:
+    """Append one turn journal event and fsync it before returning.
+
+    The returned event is the exact payload written, with default ``version``,
+    ``session_id``, ``turn_id``, and ``created_at`` fields filled in.
+    """
+    if not isinstance(event, dict):
+        raise TypeError("event must be a dict")
+    event_name = str(event.get("event") or "").strip()
+    if not event_name:
+        raise ValueError("event is required")
+    payload = dict(event)
+    payload.setdefault("version", 1)
+    payload["session_id"] = str(session_id)
+    payload.setdefault("turn_id", _make_turn_id())
+    payload.setdefault("created_at", time.time())
+
+    path = _journal_path(session_id, session_dir=session_dir)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    line = json.dumps(payload, 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()
+        os.fsync(fh.fileno())
+    try:
+        dir_fd = os.open(path.parent, os.O_DIRECTORY)
+        try:
+            os.fsync(dir_fd)
+        finally:
+            os.close(dir_fd)
+    except OSError:
+        pass
+    return payload
+
+
+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)
+    events: list[dict] = []
+    malformed: list[dict] = []
+    try:
+        lines = path.read_text(encoding="utf-8").splitlines()
+    except FileNotFoundError:
+        return {"session_id": str(session_id), "events": [], "malformed": []}
+    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})
+            continue
+        if isinstance(event, dict):
+            events.append(event)
+        else:
+            malformed.append({"line": line_no, "raw": raw})
+    return {"session_id": str(session_id), "events": events, "malformed": malformed}
+
+
+def derive_turn_journal_states(events: Iterable[dict]) -> dict[str, dict]:
+    """Return the latest event per ``turn_id``."""
+    states: dict[str, dict] = {}
+    for event in events:
+        if not isinstance(event, dict):
+            continue
+        turn_id = str(event.get("turn_id") or "").strip()
+        if not turn_id:
+            continue
+        previous = states.get(turn_id)
+        if previous is None or float(event.get("created_at") or 0) >= float(previous.get("created_at") or 0):
+            states[turn_id] = event
+    return states
+
+
+def _latest_turn_id_for_stream(events: Iterable[dict], stream_id: str) -> str | None:
+    stream = str(stream_id or "").strip()
+    if not stream:
+        return None
+    latest: str | None = None
+    for event in events:
+        if not isinstance(event, dict):
+            continue
+        if str(event.get("stream_id") or "") != stream:
+            continue
+        turn_id = str(event.get("turn_id") or "").strip()
+        if turn_id:
+            latest = turn_id
+    return latest
+
+
+def append_turn_journal_event_for_stream(
+    session_id: str,
+    stream_id: str,
+    event: dict,
+    *,
+    session_dir: Path | None = None,
+) -> dict:
+    """Append a lifecycle event for the turn associated with ``stream_id``."""
+    payload = dict(event)
+    payload["stream_id"] = str(stream_id)
+    if not payload.get("turn_id"):
+        journal = read_turn_journal(session_id, session_dir=session_dir)
+        turn_id = _latest_turn_id_for_stream(journal.get("events") or [], stream_id)
+        if turn_id:
+            payload["turn_id"] = turn_id
+    return append_turn_journal_event(session_id, payload, session_dir=session_dir)
+
+
+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())
+
+
+def is_terminal_turn_event(event: dict) -> bool:
+    return str((event or {}).get("event") or "") in _TERMINAL_EVENTS

server.py

@@ -200,6 +200,27 @@ class Handler(BaseHTTPRequestHandler):
                 pass
     _ver_suffix = WEBUI_VERSION.removeprefix('v')
     server_version = ('HermesWebUI/' + _ver_suffix) if _ver_suffix != 'unknown' else 'HermesWebUI'
+    _CSP_REPORT_ONLY = (
+        "default-src 'self'; "
+        "base-uri 'self'; "
+        "object-src 'none'; "
+        "frame-ancestors 'self'; "
+        "script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; "
+        "style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; "
+        "img-src 'self' data: blob:; "
+        "font-src 'self' data:; "
+        "media-src 'self' data: blob:; "
+        "connect-src 'self' http://127.0.0.1:* http://localhost:* ws://127.0.0.1:* ws://localhost:*"
+    )
+
+    @classmethod
+    def csp_report_only_policy(cls) -> str:
+        return cls._CSP_REPORT_ONLY
+
+    def end_headers(self) -> None:
+        self.send_header("Content-Security-Policy-Report-Only", self.csp_report_only_policy())
+        super().end_headers()
+
     def log_message(self, fmt, *args): pass  # suppress default Apache-style log
 
     def log_request(self, code: str='-', size: str='-') -> None:

static/commands.js

@@ -79,6 +79,18 @@ function getMatchingCommands(prefix){
     matches.push(skill);
     seen.add(skill.name);
   }
+  // Include agent/plugin commands from /api/commands metadata
+  for(const cmd of (_agentCommandCache||[])){
+    const name=String(cmd&&cmd.name||'').toLowerCase();
+    if(!name.startsWith(q)||seen.has(name))continue;
+    if(cmd.cli_only)continue;
+    matches.push({
+      name,
+      desc:String(cmd&&cmd.description||'').trim()||'Agent command',
+      source:cmd.category==='Plugin'?'plugin':'agent',
+    });
+    seen.add(name);
+  }
   return matches;
 }
 
@@ -191,9 +203,10 @@ function _getSlashSubArgOptions(spec){
   return Promise.resolve([]);
 }
 
+let _agentCommandCacheReady=false;
 async function loadAgentCommandMetadata(force=false){
-  if(_agentCommandCache&&!force) return _agentCommandCache;
-  if(_agentCommandCachePromise&&!force) return _agentCommandCachePromise;
+  if(_agentCommandCacheReady&&!force)return _agentCommandCache||[];
+  if(_agentCommandCachePromise&&!force)return _agentCommandCachePromise;
   _agentCommandCachePromise=(async()=>{
     try{
       const data=await api('/api/commands');
@@ -201,6 +214,7 @@ async function loadAgentCommandMetadata(force=false){
     }catch(_){
       _agentCommandCache=[];
     }finally{
+      _agentCommandCacheReady=true;
       _agentCommandCachePromise=null;
     }
     return _agentCommandCache;
@@ -229,6 +243,16 @@ function cliOnlyCommandResponse(cmdName, meta){
   return `\`/${name}\` is a Hermes CLI-only command and cannot run inside the WebUI.${detail}${extra}`;
 }
 
+async function executeAgentPluginCommand(text,_meta){
+  const command=String(text||'').trim();
+  if(!command) throw new Error('command is required');
+  const data=await api('/api/commands/exec',{
+    method:'POST',
+    body:JSON.stringify({command})
+  });
+  return String(data&&data.output||'(no output)');
+}
+
 function _parseSlashAutocomplete(text){
   if(!text.startsWith('/')||text.indexOf('\n')!==-1) return null;
   const raw=text.slice(1);
@@ -1105,6 +1129,10 @@ function refreshSlashCommandDropdown(){
 function ensureSkillCommandsLoadedForAutocomplete(){
   if(_skillCommandCacheReady||_skillCommandLoadPromise)return;
   loadSkillCommands().then(()=>{refreshSlashCommandDropdown();});
+  // Also preload agent/plugin command metadata for autocomplete
+  if(!_agentCommandCacheReady&&!_agentCommandCachePromise){
+    loadAgentCommandMetadata().then(()=>{refreshSlashCommandDropdown();});
+  }
 }
 
 // ── Autocomplete dropdown ───────────────────────────────────────────────────

static/i18n.js

@@ -645,6 +645,11 @@ const LOCALES = {
     logs_no_mtime: 'not written yet',
     logs_truncated_hint: 'Showing the tail of a large log file; older bytes were skipped to keep memory bounded.',
     logs_copied: 'Logs copied',
+    logs_severity: 'Severity',
+    logs_severity_all: 'All',
+    logs_severity_errors: 'Errors',
+    logs_severity_warnings: 'Warnings+',
+    logs_filter_active: 'shown (filter active)',
 
     // Insights
     insights_title: 'Usage Analytics',
@@ -1735,6 +1740,11 @@ const LOCALES = {
     logs_no_mtime: '未書き込み',
     logs_truncated_hint: '大きなログファイルの末尾を表示しています。メモリ使用量を抑えるため、古いデータは省略されました。',
     logs_copied: 'ログをコピーしました',
+    logs_severity: 'Severity',  // TODO: translate
+    logs_severity_all: 'All',  // TODO: translate
+    logs_severity_errors: 'Errors',  // TODO: translate
+    logs_severity_warnings: 'Warnings+',  // TODO: translate
+    logs_filter_active: 'shown (filter active)',  // TODO: translate
 
     // Insights
     insights_title: '使用状況分析',
@@ -2639,6 +2649,11 @@ const LOCALES = {
     logs_no_mtime: 'not written yet',  // TODO: translate
     logs_truncated_hint: 'Showing the tail of a large log file; older bytes were skipped to keep memory bounded.',  // TODO: translate
     logs_copied: 'Logs copied',  // TODO: translate
+    logs_severity: 'Severity',  // TODO: translate
+    logs_severity_all: 'All',  // TODO: translate
+    logs_severity_errors: 'Errors',  // TODO: translate
+    logs_severity_warnings: 'Warnings+',  // TODO: translate
+    logs_filter_active: 'shown (filter active)',  // TODO: translate
     new_conversation: 'Новая беседа',
     filter_conversations: 'Фильтр бесед...',
     session_time_unknown: 'Неизвестно',
@@ -3663,6 +3678,11 @@ const LOCALES = {
     logs_no_mtime: 'not written yet',  // TODO: translate
     logs_truncated_hint: 'Showing the tail of a large log file; older bytes were skipped to keep memory bounded.',  // TODO: translate
     logs_copied: 'Logs copied',  // TODO: translate
+    logs_severity: 'Severity',  // TODO: translate
+    logs_severity_all: 'All',  // TODO: translate
+    logs_severity_errors: 'Errors',  // TODO: translate
+    logs_severity_warnings: 'Warnings+',  // TODO: translate
+    logs_filter_active: 'shown (filter active)',  // TODO: translate
     new_conversation: 'Nueva conversación',
     filter_conversations: 'Filtrar conversaciones...',
     session_time_unknown: 'Desconocido',
@@ -4670,6 +4690,11 @@ const LOCALES = {
     logs_no_mtime: 'not written yet',  // TODO: translate
     logs_truncated_hint: 'Showing the tail of a large log file; older bytes were skipped to keep memory bounded.',  // TODO: translate
     logs_copied: 'Logs copied',  // TODO: translate
+    logs_severity: 'Severity',  // TODO: translate
+    logs_severity_all: 'All',  // TODO: translate
+    logs_severity_errors: 'Errors',  // TODO: translate
+    logs_severity_warnings: 'Warnings+',  // TODO: translate
+    logs_filter_active: 'shown (filter active)',  // TODO: translate
     new_conversation: 'Neuer Chat',
     filter_conversations: 'Chats filtern...',
     scheduled_jobs: 'Geplante Aufgaben',
@@ -5710,6 +5735,11 @@ const LOCALES = {
     logs_no_mtime: '尚未写入',
     logs_truncated_hint: '此处显示的是日志文件的末尾内容。为节省内存，已省略较早的数据。',
     logs_copied: '日志已复制',
+    logs_severity: 'Severity',  // TODO: translate
+    logs_severity_all: 'All',  // TODO: translate
+    logs_severity_errors: 'Errors',  // TODO: translate
+    logs_severity_warnings: 'Warnings+',  // TODO: translate
+    logs_filter_active: 'shown (filter active)',  // TODO: translate
     new_conversation: '新建对话',
     filter_conversations: '筛选对话…',
     session_time_unknown: '未知',
@@ -7902,6 +7932,11 @@ const LOCALES = {
     logs_no_mtime: 'not written yet',  // TODO: translate
     logs_truncated_hint: 'Showing the tail of a large log file; older bytes were skipped to keep memory bounded.',  // TODO: translate
     logs_copied: 'Logs copied',  // TODO: translate
+    logs_severity: 'Severity',  // TODO: translate
+    logs_severity_all: 'All',  // TODO: translate
+    logs_severity_errors: 'Errors',  // TODO: translate
+    logs_severity_warnings: 'Warnings+',  // TODO: translate
+    logs_filter_active: 'shown (filter active)',  // TODO: translate
     new_conversation: 'Nova conversa',
     filter_conversations: 'Filtrar conversas...',
     session_time_unknown: 'Desconhecido',
@@ -8890,6 +8925,11 @@ const LOCALES = {
     logs_no_mtime: 'not written yet',  // TODO: translate
     logs_truncated_hint: 'Showing the tail of a large log file; older bytes were skipped to keep memory bounded.',  // TODO: translate
     logs_copied: 'Logs copied',  // TODO: translate
+    logs_severity: 'Severity',  // TODO: translate
+    logs_severity_all: 'All',  // TODO: translate
+    logs_severity_errors: 'Errors',  // TODO: translate
+    logs_severity_warnings: 'Warnings+',  // TODO: translate
+    logs_filter_active: 'shown (filter active)',  // TODO: translate
     new_conversation: '새 대화',
     filter_conversations: '대화 필터…',
     session_time_unknown: 'Unknown',

static/index.html

@@ -257,6 +257,12 @@
           <option value="500">500</option>
           <option value="1000">1000</option>
         </select>
+        <label class="logs-control-label" for="logsSeverityFilter" data-i18n="logs_severity">Severity</label>
+        <select id="logsSeverityFilter" onchange="_applyLogsSeverityFilter()">
+          <option value="all" data-i18n="logs_severity_all">All</option>
+          <option value="errors" data-i18n="logs_severity_errors">Errors</option>
+          <option value="warnings" data-i18n="logs_severity_warnings">Warnings+</option>
+        </select>
         <label class="logs-check-row"><input id="logsAutoRefresh" type="checkbox" checked onchange="_syncLogsAutoRefresh()"><span data-i18n="logs_auto_refresh">Auto-refresh (5s)</span></label>
         <label class="logs-check-row"><input id="logsWrap" type="checkbox" onchange="_syncLogsWrap()"><span data-i18n="logs_wrap">Wrap lines</span></label>
         <button type="button" class="logs-copy" id="logsCopyAll" onclick="copyLogsAll()" data-i18n="logs_copy_all">Copy all</button>

static/messages.js

@@ -166,6 +166,21 @@ async function send(){
         renderMessages();
         $('msg').value='';autoResize();hideCmdDropdown();return;
       }
+      if(_agentCmd&&_agentCmd.category==='Plugin'){
+        if(!S.session){await newSession();await renderSessionList();}
+        S.messages.push({role:'user',content:text,_ts:Date.now()/1000});
+        let _pluginOutput='(no output)';
+        try{
+          _pluginOutput=typeof executeAgentPluginCommand==='function'
+            ? await executeAgentPluginCommand(text,_agentCmd)
+            : 'Plugin command runtime unavailable in WebUI.';
+        }catch(e){
+          _pluginOutput=`Plugin command error: ${e&&e.message||e}`;
+        }
+        S.messages.push({role:'assistant',content:String(_pluginOutput||'(no output)'),_ts:Date.now()/1000});
+        renderMessages();
+        $('msg').value='';autoResize();hideCmdDropdown();return;
+      }
     }
   }
   if(!S.session){await newSession();await renderSessionList();}

static/panels.js

@@ -32,6 +32,7 @@ let _profilePreFormDetail = null;
 let _pendingSettingsTargetPanel = null; // destination selected while settings had unsaved changes
 let _logsAutoRefreshTimer = null;
 let _lastLogsLines = [];
+let _logsSeverityFilter = 'all';
 
 // Map of panel names → i18n keys for the app titlebar label.
 const APP_TITLEBAR_KEYS = {
@@ -2663,6 +2664,32 @@ function _selectedLogsTail() {
   return [100,200,500,1000].includes(value) ? value : 200;
 }
 
+function _severityForLine(line) {
+  const text = String(line || '').toUpperCase();
+  if (/\b(ERROR|CRITICAL|TRACEBACK)\b/.test(text)) return 'error';
+  if (/\b(WARNING|WARN)\b/.test(text)) return 'warning';
+  if (/\b(DEBUG)\b/.test(text)) return 'debug';
+  if (/\b(INFO)\b/.test(text)) return 'info';
+  return 'other';
+}
+
+function _filteredLogsLines() {
+  if (_logsSeverityFilter === 'all') return _lastLogsLines;
+  return _lastLogsLines.filter(line => {
+    const sev = _severityForLine(line);
+    if (_logsSeverityFilter === 'errors') return sev === 'error';
+    if (_logsSeverityFilter === 'warnings') return sev === 'warning' || sev === 'error';
+    return true;
+  });
+}
+
+function _applyLogsSeverityFilter() {
+  const el = $('logsSeverityFilter');
+  _logsSeverityFilter = (el && el.value) || 'all';
+  // Re-render from cached lines without re-fetching
+  _renderLogs({ lines: _lastLogsLines, hint: '', truncated: false, _fromFilter: true });
+}
+
 function _logLineSeverityClass(line) {
   const text = String(line || '').toUpperCase();
   if (/\b(WARNING|WARN)\b/.test(text)) return 'log-line-warning';
@@ -2710,14 +2737,19 @@ function _renderLogs(data) {
   const box = $('logsOutput');
   const status = $('logsStatus');
   if (!box) return;
-  const lines = Array.isArray(data && data.lines) ? data.lines : [];
-  _lastLogsLines = lines.slice();
+  const rawLines = Array.isArray(data && data.lines) ? data.lines : [];
+  // Only update cache when loading fresh data (not when re-rendering from filter)
+  if (data && !data._fromFilter) _lastLogsLines = rawLines.slice();
+  const displayLines = _filteredLogsLines();
   const hint = data && data.hint ? `<div class="logs-hint">${esc(data.hint)}</div>` : '';
   const truncated = data && data.truncated ? `<div class="logs-hint warn">${esc(t('logs_truncated_hint'))}</div>` : '';
-  if (!lines.length) {
-    box.innerHTML = `${hint}${truncated}<div class="logs-empty">${esc(t('logs_empty'))}</div>`;
+  const filterNote = _logsSeverityFilter !== 'all'
+    ? `<div class="logs-hint">${esc(displayLines.length + ' / ' + _lastLogsLines.length + ' ' + t('logs_filter_active'))}</div>`
+    : '';
+  if (!displayLines.length) {
+    box.innerHTML = `${hint}${truncated}${filterNote}<div class="logs-empty">${esc(t('logs_empty'))}</div>`;
   } else {
-    box.innerHTML = `${hint}${truncated}` + lines.map(line => {
+    box.innerHTML = `${hint}${truncated}${filterNote}` + displayLines.map(line => {
       const cls = _logLineSeverityClass(line);
       return `<div class="log-line ${cls}">${esc(line)}</div>`;
     }).join('');
@@ -2726,7 +2758,7 @@ function _renderLogs(data) {
   if (status) {
     const bytes = data && Number(data.total_bytes || 0);
     const when = data && data.mtime ? new Date(data.mtime * 1000).toLocaleString() : t('logs_no_mtime');
-    status.textContent = `${lines.length} / ${data.tail || _selectedLogsTail()} lines · ${bytes.toLocaleString()} bytes · ${when}`;
+    status.textContent = `${rawLines.length} / ${data.tail || _selectedLogsTail()} lines · ${bytes.toLocaleString()} bytes · ${when}`;
   }
 }
 
@@ -2754,9 +2786,10 @@ function _syncLogsAutoRefresh() {
 }
 
 async function copyLogsAll() {
-  const text = _lastLogsLines.join('\n');
+  const lines = _filteredLogsLines();
+  const text = lines.join('\n');
   try {
-    await navigator.clipboard.writeText(text);
+    await _copyText(text);
     showToast(t('logs_copied'));
   } catch(e) {
     showToast(t('copy_failed'), 'error');

tests/test_cli_only_slash_commands.py

@@ -170,14 +170,16 @@ def test_send_intercepts_cli_only_commands_before_agent_round_trip():
 
 
 def test_unknown_slash_commands_still_fall_through_to_agent():
-    """Only known cli_only commands should be intercepted."""
+    """Only explicitly supported metadata-backed commands should be intercepted."""
     intercept_idx = MESSAGES_JS.find("Slash command intercept")
     normal_send_idx = MESSAGES_JS.find("const activeSid=S.session.session_id", intercept_idx)
     intercept = MESSAGES_JS[intercept_idx:normal_send_idx]
 
     assert "if(_agentCmd&&_agentCmd.cli_only)" in intercept
+    assert "if(_agentCmd&&_agentCmd.category==='Plugin')" in intercept
     assert "if(_parsedCmd&&!_cmd)" in intercept
     assert "if(!_agentCmd" not in intercept
+    assert "if(_agentCmd){" not in intercept
     assert "else" not in intercept[intercept.find("if(_agentCmd&&_agentCmd.cli_only)") :]
 
 

tests/test_issue1909_csp_report_only.py

@@ -0,0 +1,35 @@
+"""Regression tests for #1909 CSP report-only security header."""
+
+from http.server import BaseHTTPRequestHandler
+
+from server import Handler
+
+
+def test_handler_adds_content_security_policy_report_only(monkeypatch):
+    sent_headers = []
+    handler = Handler.__new__(Handler)
+    handler.send_header = lambda key, value: sent_headers.append((key, value))
+    monkeypatch.setattr(BaseHTTPRequestHandler, "end_headers", lambda self: None)
+
+    Handler.end_headers(handler)
+
+    headers = dict(sent_headers)
+    assert "Content-Security-Policy-Report-Only" in headers
+    assert "Content-Security-Policy" not in headers
+    policy = headers["Content-Security-Policy-Report-Only"]
+    assert "default-src 'self'" in policy
+    assert "object-src 'none'" in policy
+    assert "frame-ancestors 'self'" in policy
+    assert "base-uri 'self'" in policy
+
+
+def test_csp_report_only_keeps_legacy_inline_allowances_for_current_ui():
+    policy = Handler.csp_report_only_policy()
+
+    assert "script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net" in policy
+    assert "style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net" in policy
+    # unsafe-eval was dropped after Opus stage-339 verification — no production
+    # JS uses eval(), new Function(), or string-form setTimeout/setInterval.
+    assert "'unsafe-eval'" not in policy
+    assert "img-src 'self' data: blob:" in policy
+    assert "connect-src 'self'" in policy

tests/test_logs_ui_static.py

@@ -93,7 +93,9 @@ def test_logs_panel_fetches_allowlisted_api_and_exposes_controls():
     assert "logsWrap" in INDEX
     assert "logsCopyAll" in INDEX
     assert "logsAutoRefresh" in INDEX
-    assert "navigator.clipboard.writeText" in PANELS
+    assert "logsSeverityFilter" in INDEX
+    copy_fn = _function_body(PANELS, "copyLogsAll")
+    assert "_copyText" in copy_fn
     assert "logs-copy" in INDEX
 
 

tests/test_pr1341_context_window_persistence.py

@@ -38,12 +38,11 @@ def test_streaming_persists_context_fields_on_session_before_save():
     # Save call follows shortly after
     save_call = src.find("\n                s.save()", block_start)
     assert save_call != -1, "s.save() not found after the post-merge marker"
-    # Limit bumped to 7000 in #1896 fix — the context_length fallback grew to
-    # accept config_context_length / provider / custom_providers kwargs and a
-    # legacy 2-arg fallback for older hermes-agent builds. The block is still
-    # focused: it's a single fallback resolver call with arg-prep scaffold and
-    # commentary explaining the failure mode it prevents.
-    assert save_call - block_start < 7000, (
+    # Limit bumped to 8200 by turn-journal lifecycle events: the block now also
+    # records `assistant_started` immediately before the durable final save.
+    # The context_length fallback is still a single focused resolver call with
+    # arg-prep scaffold and commentary explaining the failure mode it prevents.
+    assert save_call - block_start < 8200, (
         "s.save() should be close to the post-merge marker — block expanded unexpectedly. "
         "If you've added a new pre-save mutation block here, bump this limit."
     )

tests/test_turn_journal.py

@@ -0,0 +1,135 @@
+import json
+
+from api.session_recovery import audit_session_recovery
+from api.turn_journal import (
+    append_turn_journal_event,
+    derive_turn_journal_states,
+    read_turn_journal,
+)
+
+
+def _write_session(session_dir, sid, messages=None):
+    payload = {
+        "session_id": sid,
+        "title": "Turn journal test",
+        "messages": messages or [],
+    }
+    (session_dir / f"{sid}.json").write_text(json.dumps(payload), encoding="utf-8")
+
+
+def test_append_turn_journal_event_fsyncs_jsonl_and_preserves_payload(tmp_path):
+    event = append_turn_journal_event(
+        "sid-1",
+        {
+            "event": "submitted",
+            "turn_id": "turn-1",
+            "stream_id": "stream-1",
+            "role": "user",
+            "content": "hello",
+            "attachments": [{"name": "a.png", "path": "/tmp/a.png"}],
+        },
+        session_dir=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()
+    assert len(lines) == 1
+    assert json.loads(lines[0])["content"] == "hello"
+
+
+def test_read_turn_journal_tolerates_malformed_lines(tmp_path):
+    journal_dir = tmp_path / "_turn_journal"
+    journal_dir.mkdir()
+    (journal_dir / "sid-1.jsonl").write_text(
+        '{"event":"submitted","turn_id":"turn-1","session_id":"sid-1"}\n'
+        'not-json\n'
+        '{"event":"completed","turn_id":"turn-1","session_id":"sid-1"}\n',
+        encoding="utf-8",
+    )
+
+    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"}]
+
+
+def test_derive_turn_journal_states_keeps_latest_event_per_turn():
+    states = derive_turn_journal_states([
+        {"event": "submitted", "turn_id": "turn-1", "created_at": 1},
+        {"event": "worker_started", "turn_id": "turn-1", "created_at": 2},
+        {"event": "submitted", "turn_id": "turn-2", "created_at": 3},
+        {"event": "completed", "turn_id": "turn-1", "created_at": 4},
+    ])
+
+    assert states["turn-1"]["event"] == "completed"
+    assert states["turn-2"]["event"] == "submitted"
+
+
+def test_derive_turn_journal_states_uses_created_at_not_file_order():
+    states = derive_turn_journal_states([
+        {"event": "completed", "turn_id": "turn-1", "created_at": 20},
+        {"event": "submitted", "turn_id": "turn-1", "created_at": 10},
+    ])
+
+    assert states["turn-1"]["event"] == "completed"
+
+
+def test_audit_reports_pending_turn_journal_entry_when_user_message_absent(tmp_path):
+    _write_session(tmp_path, "sid-1", messages=[])
+    append_turn_journal_event(
+        "sid-1",
+        {
+            "event": "submitted",
+            "turn_id": "turn-1",
+            "stream_id": "stream-1",
+            "role": "user",
+            "content": "recover me",
+            "attachments": [],
+        },
+        session_dir=tmp_path,
+    )
+
+    report = audit_session_recovery(tmp_path)
+
+    assert report["status"] == "warn"
+    assert report["summary"]["repairable"] == 1
+    assert report["items"] == [
+        {
+            "session_id": "sid-1",
+            "kind": "turn_journal_pending_turn",
+            "category": "repairable",
+            "recommendation": "audit_only_pending_turn_journal",
+            "live_messages": 0,
+            "bak_messages": -1,
+            "turn_id": "turn-1",
+            "event": "submitted",
+        }
+    ]
+
+
+def test_audit_ignores_completed_or_already_materialized_turn_journal_entry(tmp_path):
+    _write_session(tmp_path, "sid-1", messages=[{"role": "user", "content": "already there"}])
+    append_turn_journal_event(
+        "sid-1",
+        {
+            "event": "submitted",
+            "turn_id": "turn-1",
+            "role": "user",
+            "content": "already there",
+        },
+        session_dir=tmp_path,
+    )
+    append_turn_journal_event(
+        "sid-1",
+        {"event": "completed", "turn_id": "turn-1"},
+        session_dir=tmp_path,
+    )
+
+    report = audit_session_recovery(tmp_path)
+
+    assert report["status"] == "ok"
+    assert report["items"] == []

tests/test_turn_journal_callsite.py

@@ -0,0 +1,25 @@
+from pathlib import Path
+
+
+def test_chat_start_appends_submitted_turn_journal_before_worker_thread_start():
+    src = Path("api/routes.py").read_text(encoding="utf-8")
+    save_idx = src.index("_prepare_chat_start_session_for_stream(")
+    append_idx = src.index("append_turn_journal_event(", save_idx)
+    thread_idx = src.index("threading.Thread(", append_idx)
+
+    assert save_idx < append_idx < thread_idx
+    assert '"event": "submitted"' in src[append_idx:thread_idx]
+    assert '"role": "user"' in src[append_idx:thread_idx]
+
+
+def test_chat_start_writes_turn_journal_after_session_lock_and_handles_failure():
+    src = Path("api/routes.py").read_text(encoding="utf-8")
+    lock_idx = src.index("with session_lock:")
+    append_idx = src.index("append_turn_journal_event(", lock_idx)
+    stream_registration_idx = src.index("STREAMS[stream_id] = stream", append_idx)
+    lock_block = src[lock_idx:append_idx]
+    append_block = src[append_idx:stream_registration_idx]
+
+    assert "append_turn_journal_event(" not in lock_block
+    assert "except Exception:" in append_block
+    assert "Failed to append submitted turn journal event" in append_block

tests/test_turn_journal_lifecycle.py

@@ -0,0 +1,38 @@
+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"

tests/test_turn_journal_lifecycle_callsite.py

@@ -0,0 +1,47 @@
+from pathlib import Path
+
+
+def test_streaming_appends_worker_started_before_running_phase():
+    src = Path("api/streaming.py").read_text(encoding="utf-8")
+    run_idx = src.index("def _run_agent_streaming(")
+    worker_idx = src.index('"event": "worker_started"', run_idx)
+    running_idx = src.index('update_active_run(stream_id, phase="running"', run_idx)
+
+    assert worker_idx < running_idx
+
+
+def test_streaming_appends_assistant_started_before_final_save():
+    src = Path("api/streaming.py").read_text(encoding="utf-8")
+    block_idx = src.index("if not ephemeral and s.messages:")
+    assistant_idx = src.index('"event": "assistant_started"', block_idx)
+    save_idx = src.index("s.save()", assistant_idx)
+
+    assert block_idx < assistant_idx < save_idx
+
+
+def test_streaming_assistant_started_uses_latest_assistant_message():
+    src = Path("api/streaming.py").read_text(encoding="utf-8")
+    block_idx = src.index("if not ephemeral and s.messages:")
+    assistant_idx = src.index('"event": "assistant_started"', block_idx)
+    block = src[block_idx:assistant_idx]
+
+    assert "range(len(s.messages) - 1, -1, -1)" in block
+    assert '"assistant_message_index": _latest_assistant_idx' in src[assistant_idx:src.index("s.save()", assistant_idx)]
+
+
+def test_streaming_appends_completed_after_final_save():
+    src = Path("api/streaming.py").read_text(encoding="utf-8")
+    assistant_idx = src.index('"event": "assistant_started"')
+    save_idx = src.index("s.save()", assistant_idx)
+    completed_idx = src.index('"event": "completed"', save_idx)
+
+    assert save_idx < completed_idx
+
+
+def test_streaming_appends_interrupted_on_provider_error_path():
+    src = Path("api/streaming.py").read_text(encoding="utf-8")
+    err_idx = src.index("err_str = str(e)")
+    interrupted_idx = src.index('"event": "interrupted"', err_idx)
+    apperror_idx = src.index("put('apperror'", interrupted_idx)
+
+    assert err_idx < interrupted_idx < apperror_idx