From 51a0fca24ab75502113d52449decc24ea16d063a Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 29 Jun 2026 22:17:19 -0400 Subject: [PATCH] fix(#5092): continue WebUI goals after gateway turns --- api/gateway_chat.py | 53 ++++++++ api/routes.py | 4 +- tests/test_goal_command_webui.py | 57 +++++++++ tests/test_webui_gateway_chat_backend.py | 156 ++++++++++++++++++++++- 4 files changed, 266 insertions(+), 4 deletions(-) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 0a93eb4e..9329e6f2 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -12,6 +12,8 @@ from typing import Any from api.config import ( CANCEL_FLAGS, + PENDING_GOAL_CONTINUATION, + STREAM_GOAL_RELATED, STREAMS, STREAMS_LOCK, STREAM_LAST_EVENT_ID, @@ -503,6 +505,7 @@ def _run_gateway_chat_streaming( attachments=None, *, model_provider=None, + goal_related=False, ): """Bridge a WebUI chat turn through Hermes Gateway's API server. @@ -886,6 +889,55 @@ def _run_gateway_chat_streaming( s.model = model s.model_provider = model_provider s.save() + try: + from api.goals import evaluate_goal_after_turn, has_active_goal + from api.profiles import get_hermes_home_for_profile + + profile_home = get_hermes_home_for_profile(getattr(s, "profile", None)) + if goal_related and has_active_goal(session_id, profile_home=profile_home): + put_gateway_event("goal", { + "session_id": session_id, + "state": "evaluating", + "message": "Evaluating goal progress…", + "message_key": "goal_evaluating_progress", + }) + decision = evaluate_goal_after_turn( + session_id, + assistant_text, + user_initiated=True, + profile_home=profile_home, + ) or {} + goal_message = str(decision.get("message") or "").strip() + if goal_message: + put_gateway_event("goal", { + "session_id": session_id, + "state": "continuing" if decision.get("should_continue") else "idle", + "message": goal_message, + "message_key": decision.get("message_key") or ( + "goal_continuing" if goal_message else "" + ), + "message_args": decision.get("message_args") or [], + "decision": decision, + }) + if decision.get("should_continue"): + continuation_prompt = str(decision.get("continuation_prompt") or "").strip() + if continuation_prompt: + PENDING_GOAL_CONTINUATION.add(session_id) + put_gateway_event("goal_continue", { + "session_id": session_id, + "continuation_prompt": continuation_prompt, + "text": continuation_prompt, + "message": goal_message, + "message_key": decision.get("message_key") or "goal_continuing", + "message_args": decision.get("message_args") or [], + "decision": decision, + }) + except Exception as goal_exc: + logger.debug( + "Gateway goal continuation hook failed for session %s: %s", + session_id, + goal_exc, + ) from api.streaming import _session_payload_with_full_messages gateway_session_payload = _session_payload_with_full_messages(s, tool_calls=[]) put_gateway_event("done", {"session": redact_session_data(gateway_session_payload), "usage": usage}) @@ -917,6 +969,7 @@ def _run_gateway_chat_streaming( _cleanup_gateway_pending_mirror(session_id) with STREAMS_LOCK: CANCEL_FLAGS.pop(stream_id, None) + STREAM_GOAL_RELATED.pop(stream_id, None) STREAM_PARTIAL_TEXT.pop(stream_id, None) STREAM_REASONING_TEXT.pop(stream_id, None) STREAM_LIVE_TOOL_CALLS.pop(stream_id, None) diff --git a/api/routes.py b/api/routes.py index 01825de1..9495e511 100644 --- a/api/routes.py +++ b/api/routes.py @@ -17624,9 +17624,7 @@ def _start_chat_stream_for_session( diag.stage("worker_thread_start") if diag else None backend_is_gateway = webui_gateway_chat_enabled(get_config()) worker_target = _run_gateway_chat_streaming if backend_is_gateway else _run_agent_streaming - worker_kwargs = {"model_provider": model_provider} - if not backend_is_gateway: - worker_kwargs["goal_related"] = goal_related + worker_kwargs = {"model_provider": model_provider, "goal_related": goal_related} if moa_config and not backend_is_gateway: worker_kwargs["moa_config"] = moa_config thr = threading.Thread( diff --git a/tests/test_goal_command_webui.py b/tests/test_goal_command_webui.py index 4a70cbde..2389c08b 100644 --- a/tests/test_goal_command_webui.py +++ b/tests/test_goal_command_webui.py @@ -2,6 +2,7 @@ import io import json +import threading from pathlib import Path from types import SimpleNamespace @@ -376,6 +377,62 @@ def test_routes_register_goal_endpoint_and_kickoff_stream(): assert "_start_chat_stream_for_session" in ROUTES_PY +def test_chat_start_forwards_goal_related_to_gateway_worker(monkeypatch, tmp_path): + from api import routes + import api.turn_journal as turn_journal + + class FakeSession: + session_id = "sid-goal-related-gateway" + active_stream_id = None + pending_started_at = 0.0 + title = "Goal Chat" + profile = "default" + + captured = {} + + class FakeThread: + def __init__(self, *, target, args, kwargs, daemon): + captured["target"] = target + captured["args"] = args + captured["kwargs"] = kwargs + captured["daemon"] = daemon + + def start(self): + captured["started"] = True + + def fake_prepare(session, **kwargs): + session.pending_started_at = 123.0 + session.title = "Goal Chat" + + monkeypatch.setattr(routes, "_get_session_agent_lock", lambda *args, **kwargs: threading.Lock()) + monkeypatch.setattr(routes, "_active_stream_blocks_chat_start", lambda *args, **kwargs: False) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda *args, **kwargs: None) + monkeypatch.setattr(routes, "_prepare_chat_start_session_for_stream", fake_prepare) + monkeypatch.setattr(routes, "_is_hidden_empty_session", lambda *args, **kwargs: False) + monkeypatch.setattr(routes, "publish_session_list_changed", lambda *args, **kwargs: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda *args, **kwargs: None) + monkeypatch.setattr(turn_journal, "append_turn_journal_event", lambda *args, **kwargs: {}) + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda *args, **kwargs: True) + monkeypatch.setattr(routes.threading, "Thread", FakeThread) + monkeypatch.setattr(routes.uuid, "uuid4", lambda: SimpleNamespace(hex="goal-stream-id")) + + response = routes._start_chat_stream_for_session( + FakeSession(), + msg="continue the goal", + attachments=[], + workspace=str(tmp_path), + model="gpt-5.5", + model_provider="openai-codex", + goal_related=True, + ) + + assert response["stream_id"] == "goal-stream-id" + assert captured["target"] is routes._run_gateway_chat_streaming + assert captured["kwargs"]["goal_related"] is True + assert captured["kwargs"]["model_provider"] == "openai-codex" + assert captured["started"] is True + + def test_streaming_post_turn_goal_hook_surfaces_and_continues(): assert "evaluate_goal_after_turn" in STREAMING_PY assert "put('goal'" in STREAMING_PY diff --git a/tests/test_webui_gateway_chat_backend.py b/tests/test_webui_gateway_chat_backend.py index 81fce5a9..fd5c3176 100644 --- a/tests/test_webui_gateway_chat_backend.py +++ b/tests/test_webui_gateway_chat_backend.py @@ -9,7 +9,7 @@ import urllib.error import api.gateway_chat as gateway_chat import api.models as models import api.streaming as streaming -from api.config import STREAMS, create_stream_channel +from api.config import PENDING_GOAL_CONTINUATION, STREAMS, create_stream_channel from api.models import new_session from api.gateway_chat import ( _gateway_http_error_event, @@ -490,6 +490,160 @@ def test_gateway_chat_worker_reads_reasoning_content_deltas_from_chat_completion assert reasoning_events == ["Let me ", "think"] +def test_gateway_chat_worker_emits_goal_continue_for_goal_related_turn(tmp_path, monkeypatch): + session_dir = tmp_path / "sessions" + session_dir.mkdir() + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") + monkeypatch.setattr(models, "SESSIONS", OrderedDict()) + + class FakeResponse: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def __iter__(self): + yield b'data: {"choices":[{"delta":{"content":"goal "}}]}\n\n' + yield b'data: {"choices":[{"delta":{"content":"reply"}}]}\n\n' + yield b'data: [DONE]\n\n' + + monkeypatch.setenv("HERMES_WEBUI_GATEWAY_BASE_URL", "http://gateway.local") + monkeypatch.setattr(gateway_chat.urllib.request, "urlopen", lambda req, timeout=0: FakeResponse()) + + from api import goals as webui_goals + + monkeypatch.setattr(webui_goals, "has_active_goal", lambda *args, **kwargs: True) + monkeypatch.setattr( + webui_goals, + "evaluate_goal_after_turn", + lambda *args, **kwargs: { + "should_continue": True, + "continuation_prompt": "continue the goal", + "message": "Continuing goal", + "message_key": "goal_continuing", + "message_args": ["one step remains"], + }, + ) + + s = new_session() + stream_id = "stream-gateway-goal-continue" + s.active_stream_id = stream_id + s.pending_user_message = "finish it" + s.pending_attachments = [] + s.pending_started_at = 123 + s.save() + channel = create_stream_channel() + subscriber = channel.subscribe() + STREAMS[stream_id] = channel + PENDING_GOAL_CONTINUATION.discard(s.session_id) + + gateway_chat._run_gateway_chat_streaming( + s.session_id, + "finish it", + "test-model", + str(tmp_path), + stream_id, + [], + goal_related=True, + ) + + saved = models.get_session(s.session_id) + events = [] + while not subscriber.empty(): + events.append(subscriber.get_nowait()) + event_names = [item[0] for item in events] + + assert event_names.count("goal") == 2 + assert "goal_continue" in event_names + assert "done" in event_names + assert "stream_end" in event_names + assert event_names.index("goal_continue") < event_names.index("done") + assert event_names.index("done") < event_names.index("stream_end") + assert s.session_id in PENDING_GOAL_CONTINUATION + + goal_continue_event = next(item for item in events if item[0] == "goal_continue") + assert goal_continue_event[1]["continuation_prompt"] == "continue the goal" + assert goal_continue_event[1]["message"] == "Continuing goal" + assert goal_continue_event[1]["message_key"] == "goal_continuing" + assert saved.messages[-1]["role"] == "assistant" + assert saved.messages[-1]["content"] == "goal reply" + + +def test_gateway_chat_worker_skips_goal_judge_for_non_goal_turn(tmp_path, monkeypatch): + session_dir = tmp_path / "sessions" + session_dir.mkdir() + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") + monkeypatch.setattr(models, "SESSIONS", OrderedDict()) + + class FakeResponse: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def __iter__(self): + yield b'data: {"choices":[{"delta":{"content":"plain reply"}}]}\n\n' + yield b'data: [DONE]\n\n' + + monkeypatch.setenv("HERMES_WEBUI_GATEWAY_BASE_URL", "http://gateway.local") + monkeypatch.setattr(gateway_chat.urllib.request, "urlopen", lambda req, timeout=0: FakeResponse()) + + from api import goals as webui_goals + + has_goal_calls = [] + judge_calls = [] + + monkeypatch.setattr( + webui_goals, + "has_active_goal", + lambda *args, **kwargs: has_goal_calls.append((args, kwargs)) or True, + ) + monkeypatch.setattr( + webui_goals, + "evaluate_goal_after_turn", + lambda *args, **kwargs: judge_calls.append((args, kwargs)), + ) + + s = new_session() + stream_id = "stream-gateway-no-goal" + s.active_stream_id = stream_id + s.pending_user_message = "plain turn" + s.pending_attachments = [] + s.pending_started_at = 123 + s.save() + channel = create_stream_channel() + subscriber = channel.subscribe() + STREAMS[stream_id] = channel + PENDING_GOAL_CONTINUATION.discard(s.session_id) + + gateway_chat._run_gateway_chat_streaming( + s.session_id, + "plain turn", + "test-model", + str(tmp_path), + stream_id, + [], + goal_related=False, + ) + + events = [] + while not subscriber.empty(): + events.append(subscriber.get_nowait()) + event_names = [item[0] for item in events] + + assert "goal" not in event_names + assert "goal_continue" not in event_names + assert "done" in event_names + assert "stream_end" in event_names + assert has_goal_calls == [] + assert judge_calls == [] + assert s.session_id not in PENDING_GOAL_CONTINUATION + + def test_gateway_chat_worker_normalizes_prefill_slice_before_system_prefix(tmp_path, monkeypatch): session_dir = tmp_path / "sessions" session_dir.mkdir()