From 9fec727cf741dffae489119cf9783374b1b316e4 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Sun, 17 May 2026 02:49:35 +0000 Subject: [PATCH] =?UTF-8?q?Stage=20374:=20PR=20#2424=20=E2=80=94=20feat(ru?= =?UTF-8?q?ntime):=20add=20default-off=20RuntimeAdapter=20seam=20(HERMES?= =?UTF-8?q?=5FWEBUI=5FRUNTIME=5FADAPTER=3Dlegacy-journal)=20by=20@Michaely?= =?UTF-8?q?klam=20(refs=20#1925)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Michael Lam --- api/routes.py | 58 ++++++-- api/runtime_adapter.py | 223 +++++++++++++++++++++++++++++ tests/test_runtime_adapter_seam.py | 121 ++++++++++++++++ 3 files changed, 393 insertions(+), 9 deletions(-) create mode 100644 api/runtime_adapter.py create mode 100644 tests/test_runtime_adapter_seam.py diff --git a/api/routes.py b/api/routes.py index 217659a8..677a6677 100644 --- a/api/routes.py +++ b/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) diff --git a/api/runtime_adapter.py b/api/runtime_adapter.py new file mode 100644 index 00000000..f9140593 --- /dev/null +++ b/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)) diff --git a/tests/test_runtime_adapter_seam.py b/tests/test_runtime_adapter_seam.py new file mode 100644 index 00000000..296abd64 --- /dev/null +++ b/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"