From 7b6a67f82e6209e71561d442f4c811f01ae4e2ed Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:01:20 -0700 Subject: [PATCH] refactor(gateway): extract duplicated StreamConsumerConfig setup into _build_stream_consumer_config (preserves both sites' divergent fallback semantics via parameter) --- gateway/run.py | 153 ++++++++++++++++++++++++++----------------------- 1 file changed, 81 insertions(+), 72 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 8ebac1d942a..159ce304a71 100644 --- a/gateway/run.py +++ b/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,