mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-30 19:09:28 +00:00
refactor(gateway): extract duplicated StreamConsumerConfig setup into _build_stream_consumer_config (preserves both sites' divergent fallback semantics via parameter)
This commit is contained in:
teknium1
Teknium
parent
326764e255
commit
7b6a67f82e
1 changed files with 81 additions and 72 deletions
153
gateway/run.py
153
gateway/run.py
|
|
@ -20330,6 +20330,75 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
return url.rstrip("/")
|
||||
return None
|
||||
|
||||
def _build_stream_consumer_config(
|
||||
self,
|
||||
source: "SessionSource",
|
||||
scfg: Any,
|
||||
adapter: Any,
|
||||
*,
|
||||
on_missing_cursor: str,
|
||||
) -> "tuple[Any, Optional[Callable[[], None]]]":
|
||||
"""Build the shared ``StreamConsumerConfig`` and the optional
|
||||
Telegram pause-typing closure used by both agent-run paths.
|
||||
|
||||
``on_missing_cursor`` controls how platforms whose adapter sets
|
||||
``SUPPORTS_MESSAGE_EDITING = False`` are handled — both semantics
|
||||
are preserved verbatim from the pre-refactor call sites:
|
||||
|
||||
- ``"fallback"`` (proxy path): stream anyway with an empty cursor.
|
||||
- ``"raise"`` (in-process agent path): raise ``RuntimeError`` so
|
||||
the caller's ``except`` skips streaming entirely.
|
||||
|
||||
Returns ``(consumer_cfg, pause_typing_before_finalize)``.
|
||||
"""
|
||||
from gateway.stream_consumer import StreamConsumerConfig
|
||||
|
||||
_pause_typing_before_finalize = None
|
||||
if source.platform == Platform.TELEGRAM and hasattr(adapter, "pause_typing_for_chat"):
|
||||
def _pause_typing_before_finalize(
|
||||
_adapter=adapter,
|
||||
_chat_id=source.chat_id,
|
||||
) -> None:
|
||||
_adapter.pause_typing_for_chat(_chat_id)
|
||||
# Platforms that don't support editing sent messages
|
||||
# (e.g. QQ, WeChat) should skip streaming entirely —
|
||||
# without edit support, the consumer sends a partial
|
||||
# first message that can never be updated, resulting in
|
||||
# duplicate messages (partial + final).
|
||||
# (The proxy path instead opts into a cursorless fallback
|
||||
# via on_missing_cursor="fallback".)
|
||||
_adapter_supports_edit = getattr(adapter, "SUPPORTS_MESSAGE_EDITING", True)
|
||||
if not _adapter_supports_edit and on_missing_cursor == "raise":
|
||||
raise RuntimeError("skip streaming for non-editable platform")
|
||||
_effective_cursor = scfg.cursor if _adapter_supports_edit else ""
|
||||
# Some Matrix clients render the streaming cursor
|
||||
# as a visible tofu/white-box artifact. Keep
|
||||
# streaming text on Matrix, but suppress the cursor.
|
||||
_buffer_only = False
|
||||
if source.platform == Platform.MATRIX:
|
||||
_effective_cursor = ""
|
||||
_buffer_only = True
|
||||
# Fresh-final applies to Telegram only — other
|
||||
# platforms either edit in place cheaply (Discord,
|
||||
# Slack) or don't have the timestamp-on-edit /
|
||||
# edit-timestamp-stays-stale problem.
|
||||
# (Ported from openclaw/openclaw#72038.)
|
||||
_fresh_final_secs = (
|
||||
float(getattr(scfg, "fresh_final_after_seconds", 0.0) or 0.0)
|
||||
if source.platform == Platform.TELEGRAM
|
||||
else 0.0
|
||||
)
|
||||
_consumer_cfg = StreamConsumerConfig(
|
||||
edit_interval=scfg.edit_interval,
|
||||
buffer_threshold=scfg.buffer_threshold,
|
||||
cursor=_effective_cursor,
|
||||
buffer_only=_buffer_only,
|
||||
fresh_final_after_seconds=_fresh_final_secs,
|
||||
transport=scfg.transport or "edit",
|
||||
chat_type=getattr(source, "chat_type", "") or "",
|
||||
)
|
||||
return _consumer_cfg, _pause_typing_before_finalize
|
||||
|
||||
async def _run_agent_via_proxy(
|
||||
self,
|
||||
message: str,
|
||||
|
|
@ -20439,39 +20508,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
if _streaming_enabled:
|
||||
try:
|
||||
from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
|
||||
from gateway.stream_consumer import GatewayStreamConsumer
|
||||
_adapter = self._adapter_for_source(source)
|
||||
if _adapter:
|
||||
_pause_typing_before_finalize = None
|
||||
if source.platform == Platform.TELEGRAM and hasattr(_adapter, "pause_typing_for_chat"):
|
||||
def _pause_typing_before_finalize(
|
||||
_adapter=_adapter,
|
||||
_chat_id=source.chat_id,
|
||||
) -> None:
|
||||
_adapter.pause_typing_for_chat(_chat_id)
|
||||
_adapter_supports_edit = getattr(_adapter, "SUPPORTS_MESSAGE_EDITING", True)
|
||||
_effective_cursor = _scfg.cursor if _adapter_supports_edit else ""
|
||||
_buffer_only = False
|
||||
if source.platform == Platform.MATRIX:
|
||||
_effective_cursor = ""
|
||||
_buffer_only = True
|
||||
# Fresh-final applies to Telegram only — other
|
||||
# platforms either edit in place cheaply (Discord,
|
||||
# Slack) or don't have the timestamp-on-edit
|
||||
# problem. (Ported from openclaw/openclaw#72038.)
|
||||
_fresh_final_secs = (
|
||||
float(getattr(_scfg, "fresh_final_after_seconds", 0.0) or 0.0)
|
||||
if source.platform == Platform.TELEGRAM
|
||||
else 0.0
|
||||
)
|
||||
_consumer_cfg = StreamConsumerConfig(
|
||||
edit_interval=_scfg.edit_interval,
|
||||
buffer_threshold=_scfg.buffer_threshold,
|
||||
cursor=_effective_cursor,
|
||||
buffer_only=_buffer_only,
|
||||
fresh_final_after_seconds=_fresh_final_secs,
|
||||
transport=_scfg.transport or "edit",
|
||||
chat_type=getattr(source, "chat_type", "") or "",
|
||||
_consumer_cfg, _pause_typing_before_finalize = (
|
||||
self._build_stream_consumer_config(
|
||||
source, _scfg, _adapter,
|
||||
on_missing_cursor="fallback",
|
||||
)
|
||||
)
|
||||
_stream_consumer = GatewayStreamConsumer(
|
||||
adapter=_adapter,
|
||||
|
|
@ -22021,49 +22065,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_want_interim_consumer = _want_interim_messages
|
||||
if _want_stream_deltas or _want_interim_consumer:
|
||||
try:
|
||||
from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
|
||||
from gateway.stream_consumer import GatewayStreamConsumer
|
||||
_adapter = self._adapter_for_source(source)
|
||||
if _adapter:
|
||||
_pause_typing_before_finalize = None
|
||||
if source.platform == Platform.TELEGRAM and hasattr(_adapter, "pause_typing_for_chat"):
|
||||
def _pause_typing_before_finalize(
|
||||
_adapter=_adapter,
|
||||
_chat_id=source.chat_id,
|
||||
) -> None:
|
||||
_adapter.pause_typing_for_chat(_chat_id)
|
||||
# Platforms that don't support editing sent messages
|
||||
# (e.g. QQ, WeChat) should skip streaming entirely —
|
||||
# without edit support, the consumer sends a partial
|
||||
# first message that can never be updated, resulting in
|
||||
# duplicate messages (partial + final).
|
||||
_adapter_supports_edit = getattr(_adapter, "SUPPORTS_MESSAGE_EDITING", True)
|
||||
if not _adapter_supports_edit:
|
||||
raise RuntimeError("skip streaming for non-editable platform")
|
||||
_effective_cursor = _scfg.cursor
|
||||
# Some Matrix clients render the streaming cursor
|
||||
# as a visible tofu/white-box artifact. Keep
|
||||
# streaming text on Matrix, but suppress the cursor.
|
||||
_buffer_only = False
|
||||
if source.platform == Platform.MATRIX:
|
||||
_effective_cursor = ""
|
||||
_buffer_only = True
|
||||
# Fresh-final applies to Telegram only — other
|
||||
# platforms either edit in place cheaply or don't
|
||||
# have the edit-timestamp-stays-stale problem.
|
||||
# (Ported from openclaw/openclaw#72038.)
|
||||
_fresh_final_secs = (
|
||||
float(getattr(_scfg, "fresh_final_after_seconds", 0.0) or 0.0)
|
||||
if source.platform == Platform.TELEGRAM
|
||||
else 0.0
|
||||
)
|
||||
_consumer_cfg = StreamConsumerConfig(
|
||||
edit_interval=_scfg.edit_interval,
|
||||
buffer_threshold=_scfg.buffer_threshold,
|
||||
cursor=_effective_cursor,
|
||||
buffer_only=_buffer_only,
|
||||
fresh_final_after_seconds=_fresh_final_secs,
|
||||
transport=_scfg.transport or "edit",
|
||||
chat_type=getattr(source, "chat_type", "") or "",
|
||||
_consumer_cfg, _pause_typing_before_finalize = (
|
||||
self._build_stream_consumer_config(
|
||||
source, _scfg, _adapter,
|
||||
on_missing_cursor="raise",
|
||||
)
|
||||
)
|
||||
_stream_consumer = GatewayStreamConsumer(
|
||||
adapter=_adapter,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue