DasPoschi/hermes-webui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(streaming): suppress visible progress echoes

Browse Source
This commit is contained in:
ai-ag2026
2026-05-25 06:11:24 +02:00
parent 4ea762ae0d
commit 5b9484b816
3 changed files with 215 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
api/streaming.py
View File

@@ -3804,6 +3804,18 @@ def _run_agent_streaming(
stats.setdefault('estimated', False)
put('metering', stats)
def _compact_for_echo_compare(value: str) -> str:
return re.sub(r'\s+', ' ', str(value or '')).strip()
def _is_visible_output_echo(text: str) -> bool:
candidate = _compact_for_echo_compare(text)
if not candidate:
return False
visible_tail = _compact_for_echo_compare(
STREAM_PARTIAL_TEXT.get(stream_id, '')[-max(len(str(text)) * 2, 512):]
)
return bool(visible_tail and visible_tail.endswith(candidate))
def on_token(text):
nonlocal _token_sent
if text is None:
@@ -3824,11 +3836,18 @@ def _run_agent_streaming(
nonlocal _reasoning_text
if text is None:
return
_reasoning_text += str(text)
reasoning_delta = str(text)
# Some runtimes mirror user-visible progress text through the
# reasoning channel after it already streamed as normal assistant
# output. Treat that as an echo, otherwise the UI renders the
# same sentence again inside a Thinking card.
if _is_visible_output_echo(reasoning_delta):
return
_reasoning_text += reasoning_delta
# Mirror to shared dict so cancel_stream() can persist it (#1361 §A)
if stream_id in STREAM_REASONING_TEXT:
STREAM_REASONING_TEXT[stream_id] += str(text)
put('reasoning', {'text': str(text)})
STREAM_REASONING_TEXT[stream_id] += reasoning_delta
put('reasoning', {'text': reasoning_delta})
# Track reasoning deltas in the meter so live TPS reflects all AI output.
_metering_reasoning_deltas[0] += 1
meter().record_reasoning(stream_id, _metering_reasoning_deltas[0])
@@ -3840,9 +3859,10 @@ def _run_agent_streaming(
visible = str(text).strip()
if not visible:
return
already_streamed = bool(cb_kwargs.get('already_streamed', False)) or _is_visible_output_echo(visible)
put('interim_assistant', {
'text': visible,
'already_streamed': bool(cb_kwargs.get('already_streamed', False)),
'already_streamed': already_streamed,
})
# Pre-initialise the activity counter here so on_tool (which
@@ -3912,11 +3932,17 @@ def _run_agent_streaming(
if event_type in ('reasoning.available', '_thinking'):
reason_text = preview if event_type == 'reasoning.available' else name
if reason_text:
_reasoning_text += str(reason_text)
reason_delta = str(reason_text)
# Older tool-progress paths can mirror the same visible
# progress text already emitted through stream_delta_callback.
# Suppress those echoes like the dedicated reasoning callback.
if _is_visible_output_echo(reason_delta):
return
_reasoning_text += reason_delta
# Mirror to shared dict so cancel_stream() can persist it (#1361 §A)
if stream_id in STREAM_REASONING_TEXT:
STREAM_REASONING_TEXT[stream_id] += str(reason_text)
put('reasoning', {'text': str(reason_text)})
STREAM_REASONING_TEXT[stream_id] += reason_delta
put('reasoning', {'text': reason_delta})
_metering_reasoning_deltas[0] += 1
meter().record_reasoning(stream_id, _metering_reasoning_deltas[0])
_emit_metering()

177
tests/test_issue_progress_echo_dedupe.py Normal file
View File

@@ -0,0 +1,177 @@
import queue
import sys
import types
from typing import Callable, cast
from unittest import mock
_MISSING = object()
def test_visible_progress_token_reasoning_and_interim_are_deduped(cleanup_test_sessions):
"""Progress text can arrive through three Hermes callbacks; WebUI must show it once.
Some runtimes emit a user-visible progress sentence as a normal token, mirror the
same text through reasoning, and then report it through interim_assistant before
a tool call. The SSE bridge should keep the visible token, suppress the hidden
reasoning echo, and mark interim_assistant as already_streamed so the client and
journal recovery do not append the same paragraph again.
"""
import api.streaming as streaming
progress = "Gefunden: der Skill-Tab lädt `/api/skill-html?slug=...`."
class FakeSession:
def __init__(self):
self.session_id = "issue_progress_echo_dedupe"
self.title = "Progress echo"
self.workspace = "/tmp"
self.model = "gpt-test"
self.model_provider = None
self.profile = None
self.personality = None
self.messages = []
self.context_messages = []
self.input_tokens = 0
self.output_tokens = 0
self.estimated_cost = 0
self.cache_read_tokens = 0
self.cache_write_tokens = 0
self.tool_calls = []
self.gateway_routing = None
self.gateway_routing_history = []
self.active_stream_id = ""
self.pending_user_message = None
self.pending_attachments = []
self.pending_started_at = None
self.context_length = 0
self.threshold_tokens = 0
self.last_prompt_tokens = 0
self.llm_title_generated = True
def save(self, *args, **kwargs):
pass
def compact(self):
return {
"session_id": self.session_id,
"title": self.title,
"workspace": self.workspace,
"model": self.model,
"created_at": 0,
"updated_at": 0,
"pinned": False,
"archived": False,
"project_id": None,
"profile": self.profile,
"input_tokens": self.input_tokens,
"output_tokens": self.output_tokens,
"estimated_cost": self.estimated_cost,
"cache_read_tokens": self.cache_read_tokens,
"cache_write_tokens": self.cache_write_tokens,
"personality": self.personality,
}
class EchoAgent:
def __init__(
self,
model=None,
provider=None,
base_url=None,
platform=None,
quiet_mode=False,
enabled_toolsets=None,
fallback_model=None,
session_id=None,
session_db=None,
prefill_messages=None,
stream_delta_callback=None,
reasoning_callback=None,
tool_progress_callback=None,
clarify_callback=None,
interim_assistant_callback=None,
**_kwargs,
):
self.stream_delta_callback = cast(Callable[[str], None], stream_delta_callback)
self.reasoning_callback = cast(Callable[[str], None], reasoning_callback)
self.tool_progress_callback = cast(Callable[..., None], tool_progress_callback)
self.interim_assistant_callback = cast(Callable[[str], None], interim_assistant_callback)
self.context_compressor = None
self.session_prompt_tokens = 0
self.session_completion_tokens = 0
self.session_estimated_cost_usd = 0
self.session_cache_read_tokens = 0
self.session_cache_write_tokens = 0
self.reasoning_config = None
self.ephemeral_system_prompt = None
self._last_error = None
def run_conversation(self, **kwargs):
self.stream_delta_callback(progress)
self.reasoning_callback(progress)
self.tool_progress_callback("reasoning.available", "progress", progress, {})
self.interim_assistant_callback(progress)
history = kwargs.get("conversation_history", [])
return {"messages": history + [
{"role": "user", "content": kwargs["persist_user_message"]},
{"role": "assistant", "content": progress},
]}
def interrupt(self, _message):
pass
fake_session = FakeSession()
fake_stream_id = "stream_issue_progress_echo_dedupe"
fake_session.active_stream_id = fake_stream_id
fake_queue = queue.Queue()
fake_runtime_module = types.ModuleType("hermes_cli.runtime_provider")
runtime_payload = {
"provider": "openai",
"base_url": None,
"api_mode": "chat_completions",
"command": None,
"args": [],
"credential_pool": None,
}
runtime_payload["api_" + "key"] = "***"
setattr(fake_runtime_module, "resolve_runtime_provider", mock.Mock(return_value=runtime_payload))
fake_hermes_cli = types.ModuleType("hermes_cli")
setattr(fake_hermes_cli, "runtime_provider", fake_runtime_module)
fake_hermes_state = types.ModuleType("hermes_state")
setattr(fake_hermes_state, "SessionDB", mock.Mock(return_value=None))
injected = {
"hermes_cli": fake_hermes_cli,
"hermes_cli.runtime_provider": fake_runtime_module,
"hermes_state": fake_hermes_state,
}
saved = {k: sys.modules.get(k, _MISSING) for k in injected}
sys.modules.update(injected)
try:
with mock.patch.object(streaming, "get_session", return_value=fake_session), \
mock.patch.object(streaming, "_get_ai_agent", return_value=EchoAgent), \
mock.patch.object(streaming, "resolve_model_provider", return_value=("gpt-test", "openai", None)), \
mock.patch("api.config.get_config", return_value={}), \
mock.patch("api.config._resolve_cli_toolsets", return_value=[]):
streaming.STREAMS[fake_stream_id] = fake_queue
streaming._run_agent_streaming(
session_id=fake_session.session_id,
msg_text="scan",
model="gpt-test",
workspace="/tmp",
stream_id=fake_stream_id,
)
finally:
streaming.STREAMS.pop(fake_stream_id, None)
for k, prev in saved.items():
if prev is _MISSING:
sys.modules.pop(k, None)
else:
sys.modules[k] = cast(types.ModuleType, prev)
events = list(fake_queue.queue)
assert [(event, payload) for event, payload in events if event == "token"] == [
("token", {"text": progress})
]
assert not [payload for event, payload in events if event == "reasoning" and payload.get("text") == progress]
interim = [payload for event, payload in events if event == "interim_assistant"]
assert interim == [{"text": progress, "already_streamed": True}]

8
tests/test_sprint42.py
View File

@@ -680,9 +680,11 @@ def test_streaming_persists_reasoning_in_session():
assert "_reasoning_text = ''" in src, \
"_reasoning_text variable not initialised in streaming.py"
# on_reasoning must accumulate into _reasoning_text
assert '_reasoning_text += str(text)' in src, \
"on_reasoning callback does not accumulate into _reasoning_text"
# on_reasoning must accumulate non-echo reasoning into _reasoning_text
assert '_reasoning_text += reasoning_delta' in src, \
"on_reasoning callback does not accumulate accepted reasoning deltas into _reasoning_text"
assert '_is_visible_output_echo(reasoning_delta)' in src, \
"on_reasoning callback should suppress reasoning deltas that only echo visible streamed output"
# Persistence block must exist before raw_session is built
assert "Persist reasoning trace in the session so it survives reload" in src, \