diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 35ced6fd..c759c68d 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -153,6 +153,9 @@ _WEBUI_GATEWAY_BASE_URL_ENV = "HERMES_WEBUI_GATEWAY_BASE_URL" _WEBUI_GATEWAY_API_KEY_ENV = "HERMES_WEBUI_GATEWAY_API_KEY" _WEBUI_GATEWAY_USE_RUNS_API_ENV = "HERMES_WEBUI_GATEWAY_USE_RUNS_API" _GATEWAY_CHAT_BACKENDS = {"gateway", "api_server", "api-server"} +# Backend tag of the in-process WebUI runtime. Local workers register their +# active run with it; cache-only Steer only enqueues on this explicit value. +WEBUI_LOCAL_CHAT_BACKEND = "legacy" def _gateway_model_field(model: str | None) -> str: @@ -272,7 +275,7 @@ def webui_chat_backend_mode(config_data=None, environ: dict[str, str] | None = N ).strip().lower() if raw in _GATEWAY_CHAT_BACKENDS: return "gateway" - return "legacy" + return WEBUI_LOCAL_CHAT_BACKEND def webui_gateway_chat_enabled(config_data=None, environ: dict[str, str] | None = None) -> bool: diff --git a/api/streaming.py b/api/streaming.py index 88e5c6e0..901a61ea 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -67,6 +67,7 @@ from api.helpers import ( ) from api.compression_anchor import is_context_compression_marker, visible_messages_for_anchor from api.compression_recovery import stamp_compression_exhausted_recovery +from api.gateway_chat import WEBUI_LOCAL_CHAT_BACKEND from api.metering import meter from api.run_journal import RunJournalWriter from api.todo_state import attach_todo_state, emit_todo_state @@ -9192,6 +9193,7 @@ def _run_agent_streaming( model=model, provider=model_provider, ephemeral=bool(ephemeral), + backend=WEBUI_LOCAL_CHAT_BACKEND, ) try: run_journal = RunJournalWriter(session_id, stream_id) @@ -13691,6 +13693,11 @@ def _handle_chat_steer(handler, body: dict) -> bool: # stream owner AND the active-run session must equal the requesting # session. Missing metadata is ambiguous and fails closed (stream_dead); # an unfenced cache object is never steered on absent ownership. + # + # The active-run backend is revalidated the same way. With no registered + # worker, the backend tag is the only proof that an in-process runtime + # owns this run: Gateway resolves to its own outcome, the explicit local + # tag may enqueue, and a missing, empty, or foreign backend fails closed. result = {"accepted": False, "fallback": "stream_dead", "stream_id": None} with _cfg.STREAMS_LOCK: if active_stream_id in _cfg.STREAMS: @@ -13700,11 +13707,12 @@ def _handle_chat_steer(handler, body: dict) -> bool: owned_stream = bool(owner) and owner == sid owned_run = bool(run.get("session_id")) and run["session_id"] == sid if owned_stream and owned_run and run.get("phase") != "cancelling": - if run.get("backend") == "gateway": + backend = run.get("backend") + if backend == "gateway": # Gateway owns transport; a local cache object is never steered. result = {"accepted": False, "fallback": "gateway_steer_queued", "stream_id": active_stream_id} - else: + elif backend == WEBUI_LOCAL_CHAT_BACKEND: try: accepted = bool(agent.steer(text)) except Exception as exc: diff --git a/tests/test_steer_active_run_identity.py b/tests/test_steer_active_run_identity.py index 50a25359..9b7e651c 100644 --- a/tests/test_steer_active_run_identity.py +++ b/tests/test_steer_active_run_identity.py @@ -121,6 +121,8 @@ def test_cache_only_mismatch_does_not_close_an_agent_owned_by_another_run(scene) @pytest.mark.parametrize("ownership", [ "both", "missing-owner", "missing-run-session", "empty-run-session", "missing-both", "missing-run-entry", + "local-backend", "missing-backend", "empty-backend", "foreign-backend", + "gateway-backend", ]) def test_cache_only_steer_requires_positive_stream_and_run_ownership(scene, ownership): """Cache-only Steer must prove ownership, not merely fail to refute it. @@ -129,6 +131,11 @@ def test_cache_only_steer_requires_positive_stream_and_run_ownership(scene, owne missing stream owner or a missing active-run session is ambiguous and must fail closed with ``stream_dead`` without calling ``agent.steer()``. Only both identities present and equal to the requesting session may enqueue. + + The active-run backend is revalidated the same way: only the explicit + local backend tag registered by ``_run_agent_streaming`` may enqueue on the + cached local agent. Gateway keeps its own terminal outcome; a missing, + empty, or foreign backend is ambiguous and fails closed. """ agent, other, _ = scene agent.session_id = "original" @@ -141,12 +148,25 @@ def test_cache_only_steer_requires_positive_stream_and_run_ownership(scene, owne config.ACTIVE_RUNS["run"]["session_id"] = "" elif ownership == "missing-run-entry": config.ACTIVE_RUNS.pop("run") + elif ownership == "local-backend": + config.ACTIVE_RUNS["run"]["backend"] = streaming.WEBUI_LOCAL_CHAT_BACKEND + elif ownership == "missing-backend": + config.ACTIVE_RUNS["run"].pop("backend") + elif ownership == "empty-backend": + config.ACTIVE_RUNS["run"]["backend"] = "" + elif ownership == "foreign-backend": + config.ACTIVE_RUNS["run"]["backend"] = "foreign" + elif ownership == "gateway-backend": + config.ACTIVE_RUNS["run"]["backend"] = "gateway" before = dict(config.SESSION_AGENT_CACHE) assert "run" in config.STREAMS result = steer() - if ownership == "both": + if ownership in ("both", "local-backend"): assert result == {"accepted": True, "fallback": None, "stream_id": "run"} agent.steer.assert_called_once_with("updated guidance") + elif ownership == "gateway-backend": + assert result == {"accepted": False, "fallback": "gateway_steer_queued", "stream_id": "run"} + agent.steer.assert_not_called() else: assert result == {"accepted": False, "fallback": "stream_dead", "stream_id": None} agent.steer.assert_not_called() @@ -157,6 +177,31 @@ def test_cache_only_steer_requires_positive_stream_and_run_ownership(scene, owne assert "run" in config.STREAMS +def test_local_worker_registers_the_backend_cache_only_steer_requires(monkeypatch): + """The in-process worker must tag its active run with the local backend. + + Cache-only Steer only enqueues on that explicit tag, so a worker that + registered without it would make every cache-only Steer fail closed. + """ + captured = {} + + class _Registered(Exception): + pass + + def register(stream_id, **metadata): + captured[stream_id] = metadata + raise _Registered() + + monkeypatch.setattr(streaming, "peek_stream", lambda stream_id: queue.Queue()) + monkeypatch.setattr(streaming, "register_active_run", register) + with pytest.raises(_Registered): + streaming._run_agent_streaming("original", "hi", "m", "/tmp", "run") + assert captured["run"]["session_id"] == "original" + assert captured["run"]["backend"] == streaming.WEBUI_LOCAL_CHAT_BACKEND + assert streaming.WEBUI_LOCAL_CHAT_BACKEND + assert streaming.WEBUI_LOCAL_CHAT_BACKEND != "gateway" + + def test_http_response_is_written_after_stream_lock_release(scene, monkeypatch): from api import helpers