"""Optional progressive WeChat replies during long streams (Hermes GatewayStream subset).""" from __future__ import annotations import time from typing import cast from butler.utilities.env_parse import env_truthy, int_env, float_env from butler.defaults.env_defaults import ( GATEWAY_PROGRESSIVE_STREAM_DEFAULT, GATEWAY_PROGRESSIVE_MIN_CHARS, GATEWAY_PROGRESSIVE_INTERVAL_SECONDS, ) def progressive_stream_enabled() -> bool: return cast(bool, env_truthy("BUTLER_GATEWAY_PROGRESSIVE_STREAM", default=GATEWAY_PROGRESSIVE_STREAM_DEFAULT)) def progressive_stream_min_chars() -> int: try: result: int = int_env("BUTLER_GATEWAY_PROGRESSIVE_MIN_CHARS", GATEWAY_PROGRESSIVE_MIN_CHARS, min=80) return result except ValueError: return int(GATEWAY_PROGRESSIVE_MIN_CHARS) def progressive_stream_interval_seconds() -> float: try: result: float = float_env("BUTLER_GATEWAY_PROGRESSIVE_INTERVAL", GATEWAY_PROGRESSIVE_INTERVAL_SECONDS, min=15.0) return result except ValueError: return float(GATEWAY_PROGRESSIVE_INTERVAL_SECONDS) def format_progressive_chunk(preview: str) -> str: body = (preview or "").strip() if len(body) > 320: body = body[:317] + "..." if not body: return "" return f"⏳ 处理中…\n{body}" def maybe_schedule_progressive_reply(bridge: object | None, preview: str) -> None: """Rate-limited supplementary message while the main reply is still generating.""" if bridge is None or not progressive_stream_enabled(): return text = format_progressive_chunk(preview) if not text: return now = time.monotonic() last = float(getattr(bridge, "_progressive_last_at", 0.0) or 0.0) if now - last < progressive_stream_interval_seconds(): return chars = int(getattr(bridge, "_stream_chars", 0) or 0) if chars < progressive_stream_min_chars(): return schedule = getattr(bridge, "schedule_supplementary_reply", None) if not callable(schedule): return if schedule(text, kind="progressive_stream"): bridge._progressive_last_at = now # type: ignore[attr-defined]