Stage 374: PR #2424 — feat(runtime): add default-off RuntimeAdapter seam (HERMES_WEBUI_RUNTIME_ADAPTER=legacy-journal) by @Michaelyklam (refs #1925)

Co-authored-by: Michael Lam <michael@example.local>

api/routes.py

@@ -7764,16 +7764,56 @@ def _handle_chat_start(handler, body, diag=None):
             requested_model,
             requested_provider,
         )
-        response = _start_chat_stream_for_session(
-            s,
-            msg=msg,
-            attachments=attachments,
-            workspace=workspace,
-            model=model,
-            model_provider=model_provider,
-            normalized_model=normalized_model,
-            diag=diag,
+        from api.runtime_adapter import (
+            LegacyJournalRuntimeAdapter,
+            StartRunRequest,
+            runtime_adapter_enabled,
         )
+
+        if runtime_adapter_enabled():
+            def _legacy_start_run(request: StartRunRequest) -> dict:
+                return _start_chat_stream_for_session(
+                    s,
+                    msg=request.message,
+                    attachments=request.attachments,
+                    workspace=request.workspace or workspace,
+                    model=request.model or model,
+                    model_provider=request.provider or model_provider,
+                    normalized_model=normalized_model,
+                    diag=diag,
+                )
+
+            adapter = LegacyJournalRuntimeAdapter(start_run_delegate=_legacy_start_run)
+            result = adapter.start_run(
+                StartRunRequest(
+                    session_id=s.session_id,
+                    message=msg,
+                    attachments=attachments,
+                    workspace=workspace,
+                    profile=getattr(s, "profile", None),
+                    provider=model_provider,
+                    model=model,
+                    source="webui",
+                    metadata={"route": "/api/chat/start"},
+                )
+            )
+            response = dict(result.payload)
+            response.setdefault("stream_id", result.stream_id)
+            response.setdefault("session_id", result.session_id)
+            response.setdefault("run_id", result.run_id)
+            response.setdefault("status", result.status)
+            response.setdefault("active_controls", result.active_controls)
+        else:
+            response = _start_chat_stream_for_session(
+                s,
+                msg=msg,
+                attachments=attachments,
+                workspace=workspace,
+                model=model,
+                model_provider=model_provider,
+                normalized_model=normalized_model,
+                diag=diag,
+            )
         status = int(response.pop("_status", 200) or 200)
         diag.stage("response_write") if diag else None
         return j(handler, response, status=status)

api/runtime_adapter.py

@@ -0,0 +1,223 @@
+"""RuntimeAdapter seam for WebUI-owned run execution.
+
+This is the #1925 Slice 2 seam only.  The default WebUI chat path remains the
+legacy direct route; enabling ``HERMES_WEBUI_RUNTIME_ADAPTER=legacy-journal``
+routes through this protocol-translator facade over the same legacy execution
+path plus the Slice 1 run journal.  This module intentionally does not own
+AIAgent instances, cancellation flags, approval callbacks, clarify callbacks, or
+new long-lived queues.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import os
+from pathlib import Path
+from typing import Any, Callable, Iterable, Protocol
+
+_RUNTIME_ADAPTER_ENV = "HERMES_WEBUI_RUNTIME_ADAPTER"
+_RUNTIME_ADAPTER_DIRECT = "legacy-direct"
+_RUNTIME_ADAPTER_JOURNAL = "legacy-journal"
+_VALID_RUNTIME_ADAPTER_MODES = {_RUNTIME_ADAPTER_DIRECT, _RUNTIME_ADAPTER_JOURNAL}
+
+
+@dataclass(frozen=True)
+class StartRunRequest:
+    session_id: str
+    message: str
+    attachments: list[dict[str, Any]] = field(default_factory=list)
+    workspace: str | None = None
+    profile: str | None = None
+    provider: str | None = None
+    model: str | None = None
+    toolsets: list[str] = field(default_factory=list)
+    source: str = "webui"
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class RunStartResult:
+    run_id: str
+    session_id: str
+    stream_id: str
+    status: str = "started"
+    started_at: float | None = None
+    cursor: str | None = None
+    active_controls: list[str] = field(default_factory=list)
+    payload: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class RunEventStream:
+    run_id: str
+    events: list[dict[str, Any]] = field(default_factory=list)
+    cursor: str | None = None
+    last_event_id: str | None = None
+
+
+@dataclass(frozen=True)
+class RunStatus:
+    run_id: str
+    session_id: str | None = None
+    status: str = "unknown"
+    last_event_id: str | None = None
+    terminal_state: str | None = None
+    active_controls: list[str] = field(default_factory=list)
+    pending_approval_id: str | None = None
+    pending_clarify_id: str | None = None
+
+
+@dataclass(frozen=True)
+class ControlResult:
+    accepted: bool
+    status: str = "accepted"
+    event_id: str | None = None
+    safe_message: str | None = None
+
+
+class RuntimeAdapter(Protocol):
+    def start_run(self, request: StartRunRequest) -> RunStartResult: ...
+    def observe_run(self, run_id: str, *, cursor: str | None = None) -> RunEventStream: ...
+    def get_run(self, run_id: str) -> RunStatus: ...
+    def cancel_run(self, run_id: str) -> ControlResult: ...
+    def respond_approval(self, run_id: str, approval_id: str, choice: str) -> ControlResult: ...
+    def respond_clarify(self, run_id: str, clarify_id: str, response: str) -> ControlResult: ...
+
+
+def runtime_adapter_mode(environ: dict[str, str] | None = None) -> str:
+    """Return the configured adapter mode, defaulting safely to legacy-direct."""
+    source = os.environ if environ is None else environ
+    raw = str(source.get(_RUNTIME_ADAPTER_ENV, _RUNTIME_ADAPTER_DIRECT) or "").strip().lower()
+    return raw if raw in _VALID_RUNTIME_ADAPTER_MODES else _RUNTIME_ADAPTER_DIRECT
+
+
+def runtime_adapter_enabled(environ: dict[str, str] | None = None) -> bool:
+    return runtime_adapter_mode(environ) == _RUNTIME_ADAPTER_JOURNAL
+
+
+def _cursor_to_after_seq(cursor: str | None) -> int | None:
+    if cursor in (None, ""):
+        return None
+    try:
+        text = str(cursor)
+        if ":" in text:
+            text = text.rsplit(":", 1)[-1]
+        return max(0, int(text))
+    except (TypeError, ValueError):
+        return 0
+
+
+def _active_control_result(value: Any) -> ControlResult:
+    accepted = bool(value)
+    return ControlResult(
+        accepted=accepted,
+        status="accepted" if accepted else "not-active",
+        safe_message=None if accepted else "Legacy control did not accept the request.",
+    )
+
+
+class LegacyJournalRuntimeAdapter:
+    """Protocol-translator facade over the current legacy streaming path.
+
+    Delegates keep Slice 2 honest: this adapter has no worker thread, AIAgent
+    cache, cancellation registry, approval queue, or clarify queue of its own.
+    """
+
+    def __init__(
+        self,
+        *,
+        start_run_delegate: Callable[[StartRunRequest], dict[str, Any]] | None = None,
+        cancel_delegate: Callable[[str], Any] | None = None,
+        approval_delegate: Callable[[str, str, str], Any] | None = None,
+        clarify_delegate: Callable[[str, str, str], Any] | None = None,
+        live_stream_lookup: Callable[[str], bool] | None = None,
+        session_dir: Path | None = None,
+    ):
+        self._start_run_delegate = start_run_delegate
+        self._cancel_delegate = cancel_delegate
+        self._approval_delegate = approval_delegate
+        self._clarify_delegate = clarify_delegate
+        self._live_stream_lookup = live_stream_lookup or (lambda _run_id: False)
+        self._session_dir = Path(session_dir) if session_dir is not None else None
+
+    def start_run(self, request: StartRunRequest) -> RunStartResult:
+        if self._start_run_delegate is None:
+            raise NotImplementedError("LegacyJournalRuntimeAdapter.start_run requires a legacy delegate")
+        payload = dict(self._start_run_delegate(request) or {})
+        stream_id = str(payload.get("stream_id") or payload.get("run_id") or "")
+        run_id = str(payload.get("run_id") or stream_id)
+        session_id = str(payload.get("session_id") or request.session_id)
+        active_controls = payload.get("active_controls")
+        if not isinstance(active_controls, list):
+            active_controls = ["cancel"] if stream_id else []
+        return RunStartResult(
+            run_id=run_id,
+            session_id=session_id,
+            stream_id=stream_id,
+            status=str(payload.get("status") or "started"),
+            started_at=payload.get("started_at"),
+            cursor=payload.get("cursor"),
+            active_controls=active_controls,
+            payload=payload,
+        )
+
+    def observe_run(self, run_id: str, *, cursor: str | None = None) -> RunEventStream:
+        from api.run_journal import find_run_summary, read_run_events
+
+        summary = find_run_summary(run_id, session_dir=self._session_dir)
+        if not summary:
+            return RunEventStream(run_id=run_id, events=[], cursor=cursor, last_event_id=None)
+        journal = read_run_events(
+            str(summary.get("session_id") or ""),
+            run_id,
+            after_seq=_cursor_to_after_seq(cursor),
+            session_dir=self._session_dir,
+        )
+        events = list(journal.get("events") or [])
+        last_event_id = events[-1].get("event_id") if events else summary.get("last_event_id")
+        return RunEventStream(
+            run_id=run_id,
+            events=events,
+            cursor=str(events[-1].get("seq")) if events else cursor,
+            last_event_id=last_event_id,
+        )
+
+    def get_run(self, run_id: str) -> RunStatus:
+        from api.run_journal import find_run_summary
+
+        live = bool(self._live_stream_lookup(run_id))
+        summary = find_run_summary(run_id, session_dir=self._session_dir)
+        if live:
+            return RunStatus(
+                run_id=run_id,
+                session_id=str((summary or {}).get("session_id") or "") or None,
+                status="running",
+                last_event_id=(summary or {}).get("last_event_id"),
+                terminal_state=None,
+                active_controls=["cancel"],
+            )
+        if summary:
+            terminal_state = summary.get("terminal_state")
+            return RunStatus(
+                run_id=run_id,
+                session_id=str(summary.get("session_id") or "") or None,
+                status=str(terminal_state or "unknown"),
+                last_event_id=summary.get("last_event_id"),
+                terminal_state=terminal_state,
+                active_controls=[],
+            )
+        return RunStatus(run_id=run_id)
+
+    def cancel_run(self, run_id: str) -> ControlResult:
+        if self._cancel_delegate is None:
+            return ControlResult(False, status="unsupported", safe_message="Cancel is not wired for this adapter.")
+        return _active_control_result(self._cancel_delegate(run_id))
+
+    def respond_approval(self, run_id: str, approval_id: str, choice: str) -> ControlResult:
+        if self._approval_delegate is None:
+            return ControlResult(False, status="unsupported", safe_message="Approval is delegated to the legacy path.")
+        return _active_control_result(self._approval_delegate(run_id, approval_id, choice))
+
+    def respond_clarify(self, run_id: str, clarify_id: str, response: str) -> ControlResult:
+        if self._clarify_delegate is None:
+            return ControlResult(False, status="unsupported", safe_message="Clarify is delegated to the legacy path.")
+        return _active_control_result(self._clarify_delegate(run_id, clarify_id, response))

tests/test_runtime_adapter_seam.py

@@ -0,0 +1,121 @@
+import importlib
+import queue
+
+
+def test_runtime_adapter_interface_and_legacy_journal_methods_exist():
+    runtime = importlib.import_module("api.runtime_adapter")
+
+    required = (
+        "start_run",
+        "observe_run",
+        "get_run",
+        "cancel_run",
+        "respond_approval",
+        "respond_clarify",
+    )
+    for name in required:
+        assert hasattr(runtime.RuntimeAdapter, name)
+        assert hasattr(runtime.LegacyJournalRuntimeAdapter, name)
+
+    assert runtime.runtime_adapter_mode({}) == "legacy-direct"
+    assert runtime.runtime_adapter_enabled({}) is False
+    assert runtime.runtime_adapter_mode({"HERMES_WEBUI_RUNTIME_ADAPTER": "legacy-journal"}) == "legacy-journal"
+    assert runtime.runtime_adapter_enabled({"HERMES_WEBUI_RUNTIME_ADAPTER": "legacy-journal"}) is True
+    assert runtime.runtime_adapter_mode({"HERMES_WEBUI_RUNTIME_ADAPTER": "sidecar"}) == "legacy-direct"
+
+
+def test_legacy_journal_adapter_start_run_delegates_without_owning_runtime_state():
+    runtime = importlib.import_module("api.runtime_adapter")
+    calls = []
+
+    def start_delegate(request):
+        calls.append(request)
+        return {
+            "stream_id": "stream-123",
+            "session_id": request.session_id,
+            "status": "started",
+            "active_controls": ["cancel"],
+        }
+
+    adapter = runtime.LegacyJournalRuntimeAdapter(start_run_delegate=start_delegate)
+    request = runtime.StartRunRequest(
+        session_id="s1",
+        message="hello",
+        attachments=[{"name": "a.txt"}],
+        workspace="/tmp/work",
+        profile="default",
+        provider="openai-codex",
+        model="gpt-5.5",
+        toolsets=["terminal"],
+        source="webui",
+        metadata={"k": "v"},
+    )
+
+    result = adapter.start_run(request)
+
+    assert calls == [request]
+    assert result.session_id == "s1"
+    assert result.stream_id == "stream-123"
+    assert result.run_id == "stream-123"
+    assert result.status == "started"
+    assert result.active_controls == ["cancel"]
+
+
+def test_legacy_journal_adapter_observe_and_get_run_use_journal_and_live_state(tmp_path):
+    runtime = importlib.import_module("api.runtime_adapter")
+    run_journal = importlib.import_module("api.run_journal")
+
+    run_journal.append_run_event("s1", "r1", "token", {"text": "a"}, session_dir=tmp_path)
+    run_journal.append_run_event("s1", "r1", "done", {"ok": True}, session_dir=tmp_path)
+
+    adapter = runtime.LegacyJournalRuntimeAdapter(
+        session_dir=tmp_path,
+        live_stream_lookup=lambda run_id: run_id == "live-run",
+    )
+
+    replay = adapter.observe_run("r1", cursor="0")
+    assert [event["type"] for event in replay.events] == ["token", "done"]
+    assert replay.last_event_id == "r1:2"
+
+    completed = adapter.get_run("r1")
+    assert completed.run_id == "r1"
+    assert completed.session_id == "s1"
+    assert completed.status == "completed"
+    assert completed.terminal_state == "completed"
+    assert completed.last_event_id == "r1:2"
+
+    live = adapter.get_run("live-run")
+    assert live.run_id == "live-run"
+    assert live.status == "running"
+    assert live.active_controls == ["cancel"]
+
+
+def test_legacy_journal_adapter_controls_delegate_to_existing_handlers():
+    runtime = importlib.import_module("api.runtime_adapter")
+    calls = []
+    adapter = runtime.LegacyJournalRuntimeAdapter(
+        cancel_delegate=lambda run_id: calls.append(("cancel", run_id)) or True,
+        approval_delegate=lambda run_id, approval_id, choice: calls.append(("approval", run_id, approval_id, choice)) or True,
+        clarify_delegate=lambda run_id, clarify_id, response: calls.append(("clarify", run_id, clarify_id, response)) or True,
+    )
+
+    assert adapter.cancel_run("r1").accepted is True
+    assert adapter.respond_approval("r1", "a1", "once").accepted is True
+    assert adapter.respond_clarify("r1", "c1", "answer").accepted is True
+    assert calls == [
+        ("cancel", "r1"),
+        ("approval", "r1", "a1", "once"),
+        ("clarify", "r1", "c1", "answer"),
+    ]
+
+
+def test_chat_start_route_selects_adapter_only_when_flag_enabled():
+    routes = importlib.import_module("api.routes")
+    src = (routes.Path(__file__).parent.parent / "api" / "routes.py").read_text(encoding="utf-8")
+    start_idx = src.index("def _handle_chat_start")
+    start_body = src[start_idx:src.index("def _resolve_chat_workspace_with_recovery", start_idx)]
+
+    assert "runtime_adapter_enabled()" in start_body
+    assert "LegacyJournalRuntimeAdapter" in start_body
+    assert "_start_chat_stream_for_session(" in start_body
+    assert "HERMES_WEBUI_RUNTIME_ADAPTER" not in start_body, "route should use runtime_adapter_enabled(), not inline env checks"