fix(steer): revalidate the active-run backend on the cache-only path
The cache-only Steer path only special-cased backend == "gateway" and sent every other value — including a missing, empty, or unknown backend — to the cached local agent. Another runtime could own the run while guidance was enqueued on a stale local agent. Fail closed instead: the in-process worker now registers its active run with the explicit local backend tag (WEBUI_LOCAL_CHAT_BACKEND, the same "legacy" value webui_chat_backend_mode() returns), and cache-only Steer only enqueues agent.steer() on that tag. Gateway keeps its gateway_steer_queued outcome; a missing, empty, or foreign backend returns stream_dead without steering. Extends the positive-ownership regression with backend cases and adds a worker-side check that the local backend tag is actually registered.
This commit is contained in:
1 parent
20c16e9167
commit
26277f933b
3 files changed
+60
-4
No files matched your search
+4
-1
@@ -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:
|
||||
|
||||
+10
-2
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in new issue
Block a user