From 0ab5ab8c5a1d66e5590f32b32849d15435c985b1 Mon Sep 17 00:00:00 2001 From: oliver Date: Mon, 10 Aug 2026 22:47:26 +0800 Subject: [PATCH] Send throttled WhatsApp progress during long tool turns. Wire channel inbound on_progress/on_tool_ui to outbound queue so group users see wait signals for CLI/UME work without sharing per-speaker session context. Co-authored-by: Cursor --- .../application/gateway/inbound_service.py | 42 +++- .../application/gateway/whatsapp_progress.py | 227 ++++++++++++++++++ runtime/orchestration/group_ingest.py | 6 + runtime/workspaces/ops/ROLE_SYSTEM.en.md | 1 + runtime/workspaces/ops/ROLE_SYSTEM.md | 1 + tests/test_whatsapp_turn_progress.py | 222 +++++++++++++++++ 6 files changed, 497 insertions(+), 2 deletions(-) create mode 100644 runtime/application/gateway/whatsapp_progress.py create mode 100644 tests/test_whatsapp_turn_progress.py diff --git a/runtime/application/gateway/inbound_service.py b/runtime/application/gateway/inbound_service.py index dd7d3f8c..a252b3bb 100644 --- a/runtime/application/gateway/inbound_service.py +++ b/runtime/application/gateway/inbound_service.py @@ -915,8 +915,9 @@ def _enqueue_whatsapp_inbound_reply( reply_text: str, reply_attachments: list[dict[str, Any]] | None, reply_metadata: dict[str, Any] | None, + kind: str = "inbound_reply", ) -> str: - """Persist final WhatsApp reply on the outbound queue (sidecar poller delivers).""" + """Persist WhatsApp reply/progress on the outbound queue (sidecar poller delivers).""" from runtime.scheduler.whatsapp_mentions import encode_whatsapp_outbound_source text = str(reply_text or "").strip() @@ -948,7 +949,7 @@ def _enqueue_whatsapp_inbound_reply( if str(meta.get(k) or "").strip() or meta.get(k) is True } source = encode_whatsapp_outbound_source( - kind="inbound_reply", + kind=str(kind or "inbound_reply").strip() or "inbound_reply", mention_jids=mention_jids, mention_names=mention_names, mention_text_ready=bool(mention_jids), @@ -1436,6 +1437,41 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: ) job_lang = str(turn_job.get("lang") or lang or "zh") job_inbound = turn_job.get("inbound") or inbound + progress_cb = None + tool_ui_cb = None + if wa_queue_delivery: + from runtime.application.gateway.whatsapp_progress import ( + WhatsappTurnProgressPublisher, + whatsapp_turn_progress_enabled, + ) + + if whatsapp_turn_progress_enabled(): + + def _enqueue_progress( + progress_text: str, + progress_meta: dict[str, Any] | None, + *, + _inbound: Any = job_inbound, + ) -> None: + _enqueue_whatsapp_inbound_reply( + store, + inbound=_inbound, + account_id=account_id, + tenant_id=str(tenant_id or ""), + reply_text=progress_text, + reply_attachments=None, + reply_metadata=progress_meta, + kind="inbound_progress", + ) + + publisher = WhatsappTurnProgressPublisher( + enqueue=_enqueue_progress, + lang=job_lang, + is_group=bool(getattr(job_inbound, "is_group", False)), + inbound=job_inbound, + ) + progress_cb = publisher.on_progress + tool_ui_cb = publisher.on_tool_ui try: turn_result = gw.handle_turn( msg=StandardMessage( @@ -1451,6 +1487,8 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: lang=job_lang, executor=manager, specialist_executor_factory=specialist_factory, + on_progress=progress_cb, + on_tool_ui=tool_ui_cb, ) channel_turn_uuid = str(turn_result.turn_uuid or "").strip() turn_reply = str(turn_result.reply_text or "").strip() diff --git a/runtime/application/gateway/whatsapp_progress.py b/runtime/application/gateway/whatsapp_progress.py new file mode 100644 index 00000000..4a9bde18 --- /dev/null +++ b/runtime/application/gateway/whatsapp_progress.py @@ -0,0 +1,227 @@ +"""Rate-limited WhatsApp interim progress during long channel turns.""" + +from __future__ import annotations + +import os +import re +import threading +import time +from typing import Any, Callable + +_TOOLS_DONE_RE = re.compile(r"tools done\s*\((\d+)\s*ms\)", re.IGNORECASE) + +# Tools that commonly exceed ~10s on production WA turns. +_LONG_TOOL_LABELS_ZH: dict[str, str] = { + "execmanagedne": "正在设备上执行命令，请稍候…", + "listclitargets": "正在枚举 CLI 目标，请稍候…", + "umealarmxlsxreport": "正在生成告警 Excel，请稍候…", + "queryumealarms": "正在查询 UME 告警，请稍候…", + "queryumealarmsraw": "正在查询 UME 告警，请稍候…", + "aggregateumealarms": "正在汇总 UME 告警，请稍候…", + "aggregateumealarmsraw": "正在汇总 UME 告警，请稍候…", + "runumediagnostics": "正在跑 UME 诊断，请稍候…", + "sqlqueryume": "正在执行 UME SQL，请稍候…", + "findtopologypaths": "正在查拓扑路径，请稍候…", + "writexlsx": "正在写 Excel，请稍候…", + "getmanagedne": "正在读取纳管网元信息，请稍候…", + "getumene": "正在读取 UME 网元详情，请稍候…", +} + +_LONG_TOOL_LABELS_EN: dict[str, str] = { + "execmanagedne": "Running device CLI, please wait…", + "listclitargets": "Listing CLI targets, please wait…", + "umealarmxlsxreport": "Building alarm Excel, please wait…", + "queryumealarms": "Querying UME alarms, please wait…", + "queryumealarmsraw": "Querying UME alarms, please wait…", + "aggregateumealarms": "Aggregating UME alarms, please wait…", + "aggregateumealarmsraw": "Aggregating UME alarms, please wait…", + "runumediagnostics": "Running UME diagnostics, please wait…", + "sqlqueryume": "Running UME SQL, please wait…", + "findtopologypaths": "Finding topology paths, please wait…", + "writexlsx": "Writing Excel, please wait…", + "getmanagedne": "Loading managed NE, please wait…", + "getumene": "Loading UME NE detail, please wait…", +} + + +def whatsapp_turn_progress_enabled() -> bool: + raw = str(os.environ.get("OCLAW_WHATSAPP_TURN_PROGRESS") or "").strip().lower() + if not raw: + return True + return raw not in {"0", "false", "no", "off"} + + +def progress_min_interval_sec() -> float: + raw = str(os.environ.get("OCLAW_WHATSAPP_PROGRESS_MIN_INTERVAL_SEC") or "").strip() + try: + return max(3.0, min(float(raw), 120.0)) + except Exception: + return 12.0 + + +def normalize_tool_key(name: str) -> str: + s = str(name or "").strip() + if not s: + return "" + low = s.lower() + if low.startswith("mcp__"): + # mcp__netx__execManagedNe -> execmanagedne + parts = low.split("__") + if len(parts) >= 3: + low = parts[-1] + elif low.startswith("netx_"): + low = low[len("netx_") :] + return low.replace("_", "") + + +def humanize_long_tool(*, tool_name: str, lang: str = "zh") -> str | None: + key = normalize_tool_key(tool_name) + if not key: + return None + table = _LONG_TOOL_LABELS_EN if str(lang or "").startswith("en") else _LONG_TOOL_LABELS_ZH + # Keys in tables are already underscore-free lower names. + return table.get(key) + + +def should_forward_progress_text(text: str) -> bool: + """Filter noisy think/retry ticks; keep meaningful wait signals.""" + t = str(text or "").strip() + if not t: + return False + low = t.lower() + if low in {"oclaw: running…", "oclaw: running...", "oclaw: finalize…", "oclaw: finalize..."}: + return False + if "think (" in low or low.startswith("oclaw: think"): + return False + if "retry-empty" in low or "retry-native-tool-calls" in low: + return False + m = _TOOLS_DONE_RE.search(t) + if m: + try: + return int(m.group(1)) >= 8000 + except Exception: + return False + # Specialist / other explicit progress lines + if low.startswith("oclaw:"): + return True + return len(t) >= 8 + + +def humanize_progress_text(*, text: str, lang: str = "zh") -> str: + t = str(text or "").strip() + m = _TOOLS_DONE_RE.search(t) + if m: + if str(lang or "").startswith("en"): + return "Tools finished; composing the reply…" + return "工具已完成，正在整理回复…" + if t.lower().startswith("oclaw:"): + body = t.split(":", 1)[-1].strip() + if str(lang or "").startswith("en"): + return body or t + # Keep short Chinese-friendly wait copy for unknown oclaw:* lines. + return f"处理中：{body}" if body else "处理中，请稍候…" + return t + + +def build_whatsapp_group_progress_metadata(*, inbound: Any) -> dict[str, Any]: + """@sender without quoting (avoids sticky quote spam on interim ticks).""" + from runtime.orchestration.group_ingest import build_whatsapp_group_reply_metadata + + meta = build_whatsapp_group_reply_metadata(inbound=inbound) + for key in ( + "quote_remote_jid", + "quote_stanza_id", + "quote_participant", + "quote_text", + "quote_push_name", + ): + meta.pop(key, None) + return meta + + +class WhatsappTurnProgressPublisher: + """Enqueue throttled interim WhatsApp texts during a channel turn.""" + + def __init__( + self, + *, + enqueue: Callable[[str, dict[str, Any] | None], None], + lang: str = "zh", + is_group: bool = False, + inbound: Any = None, + min_interval_sec: float | None = None, + enabled: bool | None = None, + clock: Callable[[], float] | None = None, + ) -> None: + self._enqueue = enqueue + self._lang = str(lang or "zh") + self._is_group = bool(is_group) + self._inbound = inbound + self._min_interval = float(min_interval_sec if min_interval_sec is not None else progress_min_interval_sec()) + self._enabled = bool(whatsapp_turn_progress_enabled() if enabled is None else enabled) + self._clock = clock or time.monotonic + self._lock = threading.Lock() + self._last_sent_at = 0.0 + self._last_text = "" + self._sent_count = 0 + + @property + def sent_count(self) -> int: + return int(self._sent_count) + + def on_progress(self, text: str) -> None: + if not self._enabled: + return + if not should_forward_progress_text(text): + return + msg = humanize_progress_text(text=text, lang=self._lang) + self._maybe_send(msg) + + def on_tool_ui(self, event: str, payload: dict[str, Any] | None) -> None: + if not self._enabled: + return + if str(event or "").strip() != "tool_use_call": + return + pl = payload if isinstance(payload, dict) else {} + label = humanize_long_tool(tool_name=str(pl.get("tool_name") or ""), lang=self._lang) + if not label: + return + self._maybe_send(label) + + def _reply_metadata(self) -> dict[str, Any] | None: + if not self._is_group or self._inbound is None: + return None + try: + return build_whatsapp_group_progress_metadata(inbound=self._inbound) + except Exception: + return None + + def _maybe_send(self, text: str) -> None: + msg = str(text or "").strip() + if not msg: + return + with self._lock: + now = float(self._clock()) + if msg == self._last_text and self._sent_count > 0: + return + if self._sent_count > 0 and (now - self._last_sent_at) < self._min_interval: + return + try: + self._enqueue(msg, self._reply_metadata()) + except Exception: + return + self._last_sent_at = now + self._last_text = msg + self._sent_count += 1 + + +__all__ = [ + "WhatsappTurnProgressPublisher", + "build_whatsapp_group_progress_metadata", + "humanize_long_tool", + "humanize_progress_text", + "normalize_tool_key", + "progress_min_interval_sec", + "should_forward_progress_text", + "whatsapp_turn_progress_enabled", +] diff --git a/runtime/orchestration/group_ingest.py b/runtime/orchestration/group_ingest.py index 654377fb..34b95c5c 100644 --- a/runtime/orchestration/group_ingest.py +++ b/runtime/orchestration/group_ingest.py @@ -353,6 +353,12 @@ def _parse_group_policy_dict(raw: Any) -> GroupPolicyConfig | None: def resolve_group_policy(*, account: dict[str, Any] | None = None) -> GroupPolicyConfig: + """Resolve group mention/session policy. + + Default ``session_scope`` is ``user_in_chat`` (per speaker in a group). Override via + account ``group_policy.session_scope`` or env ``AIA_WHATSAPP_GROUP_SESSION_SCOPE=chat`` + when a shared group transcript is intentionally desired. + """ cfg = (account or {}).get("config") if isinstance(cfg, dict): gp = _parse_group_policy_dict(cfg.get("group_policy")) diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.en.md b/runtime/workspaces/ops/ROLE_SYSTEM.en.md index 53643dcd..93fb470b 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.en.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.en.md @@ -36,6 +36,7 @@ You are the ops specialist (network operations expert). ## WhatsApp interaction (mandatory) - Short ops intents follow `ops-netx-ume-playbook` WhatsApp recipes; target **≤3 tool calls** per user message. For Excel exports prefer `ume_alarm_xlsx_report`. - Spreadsheet delivery: `ume_alarm_xlsx_report` or `write_xlsx(deliverable=true)` — never claim a file was sent without deliverable marking. +- Group chats default to **per-speaker session isolation** (members do not share dialogue memory within the same group). - Call `listCliTargets` at most once per session and reuse ids; batch `execManagedNe` commands; default `read_timeout_sec=60` — on timeout raise it, no blind retries. - Replies like `YES` / `confirm` / `继续` / `please continue`: continue the previous unfinished task — do **not** re-ask for confirmation or restart the query. - On `tool_invalid_arguments`, fix args using the returned `example`; on timeout hints, raise `read_timeout_sec` or shrink commands. diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.md b/runtime/workspaces/ops/ROLE_SYSTEM.md index 028d0653..0be2d237 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.md @@ -27,6 +27,7 @@ ## WhatsApp 交互（强制） - 短句优先走 `ops-netx-ume-playbook` 的「WhatsApp 短指令配方」，控制在 ≤3 次工具调用；要 Excel 时优先 `ume_alarm_xlsx_report`。 - 用户要表格/Excel：`ume_alarm_xlsx_report` 或 `write_xlsx(deliverable=true)`；禁止只写文件不投递。 +- 群聊默认按**发言人隔离会话**（同群不同人互不串上下文）；勿假设「群共享一个对话记忆」。 - `listCliTargets` 每会话最多查一次并复用 id；`execManagedNe` 合并 commands，超时调 `read_timeout_sec`（默认 60），禁止盲重试。 - 用户回复 `YES` / `confirm` / `确认` / `可以` / `继续` / `please continue`：直接承接上一未完成任务继续执行，**不要**再问一遍确认或重开查询。 - 工具返回 `tool_invalid_arguments` 时按返回的 `example` 修正参数；返回超时 hint 时提高 `read_timeout_sec` 或减命令，禁止相同参数重试。 diff --git a/tests/test_whatsapp_turn_progress.py b/tests/test_whatsapp_turn_progress.py new file mode 100644 index 00000000..98132f2e --- /dev/null +++ b/tests/test_whatsapp_turn_progress.py @@ -0,0 +1,222 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path +from typing import Any +from unittest import mock + +from runtime.application.gateway import inbound_service as inbound_mod +from runtime.application.gateway.channel_turn_gate import reset_channel_turn_gate_for_tests +from runtime.application.gateway.whatsapp_progress import ( + WhatsappTurnProgressPublisher, + humanize_long_tool, + humanize_progress_text, + normalize_tool_key, + should_forward_progress_text, +) +from svc.persistence.sqlite_store import SqliteStore + + +class WhatsappProgressHelpersTests(unittest.TestCase): + def test_normalize_tool_key(self) -> None: + self.assertEqual(normalize_tool_key("mcp__netx__execManagedNe"), "execmanagedne") + self.assertEqual(normalize_tool_key("netx_exec_managed_ne"), "execmanagedne") + self.assertEqual(normalize_tool_key("ume_alarm_xlsx_report"), "umealarmxlsxreport") + + def test_humanize_long_tool(self) -> None: + zh = humanize_long_tool(tool_name="mcp__netx__execManagedNe", lang="zh") + en = humanize_long_tool(tool_name="mcp__netx__execManagedNe", lang="en") + self.assertIsNotNone(zh) + self.assertIsNotNone(en) + self.assertIn("设备", str(zh)) + self.assertIn("CLI", str(en)) + self.assertIsNone(humanize_long_tool(tool_name="read_file", lang="zh")) + + def test_should_forward_progress_text(self) -> None: + self.assertFalse(should_forward_progress_text("oclaw: running…")) + self.assertFalse(should_forward_progress_text("oclaw: think (1)…")) + self.assertFalse(should_forward_progress_text("oclaw: tools done (120ms)")) + self.assertTrue(should_forward_progress_text("oclaw: tools done (12000ms)")) + self.assertTrue(should_forward_progress_text("oclaw: image specialist (legacy multimodal HTTP)…")) + + def test_humanize_progress_text(self) -> None: + self.assertIn("整理", humanize_progress_text(text="oclaw: tools done (9000ms)", lang="zh")) + self.assertIn("composing", humanize_progress_text(text="oclaw: tools done (9000ms)", lang="en").lower()) + + +class WhatsappTurnProgressPublisherTests(unittest.TestCase): + def test_tool_call_emits_and_throttles(self) -> None: + sent: list[tuple[str, dict[str, Any] | None]] = [] + clock = {"t": 100.0} + + def enqueue(text: str, meta: dict[str, Any] | None) -> None: + sent.append((text, meta)) + + pub = WhatsappTurnProgressPublisher( + enqueue=enqueue, + lang="zh", + is_group=False, + min_interval_sec=10.0, + enabled=True, + clock=lambda: clock["t"], + ) + pub.on_progress("oclaw: think (1)…") + self.assertEqual(sent, []) + + pub.on_tool_ui("tool_use_call", {"tool_name": "mcp__netx__execManagedNe"}) + self.assertEqual(len(sent), 1) + self.assertIn("设备", sent[0][0]) + + clock["t"] = 105.0 + pub.on_tool_ui("tool_use_call", {"tool_name": "mcp__netx__queryUmeAlarms"}) + self.assertEqual(len(sent), 1) # throttled + + clock["t"] = 111.0 + pub.on_tool_ui("tool_use_call", {"tool_name": "mcp__netx__queryUmeAlarms"}) + self.assertEqual(len(sent), 2) + + pub.on_progress("oclaw: tools done (15000ms)") + self.assertEqual(len(sent), 2) + clock["t"] = 122.0 + pub.on_progress("oclaw: tools done (15000ms)") + self.assertEqual(len(sent), 3) + self.assertIn("整理", sent[2][0]) + + def test_group_metadata_mentions_without_quote(self) -> None: + sent: list[tuple[str, dict[str, Any] | None]] = [] + + class _Inbound: + external_user_id = "628100000@s.whatsapp.net" + external_chat_id = "120363011111111111@g.us" + text = "@bot hello" + metadata = { + "raw": { + "id": "stanza1", + "participant": "628100000@s.whatsapp.net", + "pushName": "Ops", + } + } + + pub = WhatsappTurnProgressPublisher( + enqueue=lambda t, m: sent.append((t, m)), + lang="zh", + is_group=True, + inbound=_Inbound(), + min_interval_sec=1.0, + enabled=True, + ) + pub.on_tool_ui("tool_use_call", {"tool_name": "ume_alarm_xlsx_report"}) + self.assertEqual(len(sent), 1) + meta = sent[0][1] or {} + self.assertTrue(meta.get("mention_jids")) + self.assertNotIn("quote_stanza_id", meta) + + +class _FakeTurn: + def __init__(self, text: str = "final answer") -> None: + self.reply_text = text + self.turn_uuid = "turn-1" + + +class WhatsappInboundProgressWiringTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.db = Path(self._tmp.name) / "wa_progress.sqlite" + self.store = SqliteStore(str(self.db)) + tenant = self.store.create_tenant("Team") + self.tenant_id = str(tenant["id"]) + user = self.store.create_user(tenant_id=self.tenant_id, display_name="ops", role="administrator") + self.user_id = str(user["id"]) + self.store.upsert_user_channel_account( + tenant_id=self.tenant_id, + user_id=self.user_id, + channel="whatsapp", + account_id="wa-default", + name="wa-default", + config={}, + is_active=True, + ) + self.store.upsert_channel_identity_v2( + tenant_id=self.tenant_id, + channel="whatsapp", + account_id="wa-default", + external_user_id="628100000@s.whatsapp.net", + user_id=self.user_id, + ) + reset_channel_turn_gate_for_tests() + + def tearDown(self) -> None: + self._tmp.cleanup() + + def _payload(self) -> dict: + return { + "channel": "whatsapp", + "account_id": "wa-default", + "user_id": "628100000@s.whatsapp.net", + "chat_id": "120363011111111111@g.us", + "text": "@bot run cli", + "is_group": True, + "mentions": ["bot@s.whatsapp.net"], + "metadata": { + "bot_jid": "bot@s.whatsapp.net", + "mentions_bot": True, + "group_name": "AI nms", + "raw": { + "id": "stanza-progress", + "participant": "628100000@s.whatsapp.net", + "pushName": "Egista", + }, + }, + } + + def test_handle_turn_progress_enqueues_inbound_progress(self) -> None: + captured: dict[str, Any] = {} + + def _handle_turn(**kwargs: Any) -> _FakeTurn: + captured.update(kwargs) + on_tool_ui = kwargs.get("on_tool_ui") + if callable(on_tool_ui): + on_tool_ui("tool_use_call", {"tool_name": "mcp__netx__execManagedNe"}) + return _FakeTurn("cli done") + + with mock.patch.multiple( + inbound_mod, + get_assistant_store=mock.MagicMock(return_value=self.store), + _build_admin_gateway_executor=mock.MagicMock(return_value=object()), + _resolve_channel_dispatch=mock.MagicMock(return_value=("expert", "ops", "zh")), + ), mock.patch("runtime.gateway.OclawGateway") as gw_cls, mock.patch( + "runtime.orchestration.group_ingest.should_process_group_inbound", + return_value=True, + ), mock.patch( + "runtime.application.gateway.whatsapp_inbound_access.handle_whatsapp_access", + return_value=None, + ), mock.patch.dict( + "os.environ", + {"OCLAW_WHATSAPP_TURN_PROGRESS": "1", "OCLAW_WHATSAPP_INBOUND_QUEUE_DELIVERY": "1"}, + clear=False, + ): + gw = gw_cls.return_value + gw.handle_turn.side_effect = lambda **kw: _handle_turn(**kw) + out = inbound_mod.process_inbound_payload(self._payload()) + + self.assertEqual(out.get("delivery"), "queued") + self.assertTrue(callable(captured.get("on_progress"))) + self.assertTrue(callable(captured.get("on_tool_ui"))) + pending = self.store.list_pending_channel_outbound_messages( + channel="whatsapp", account_id="wa-default", limit=10 + ) + kinds = [json.loads(str(p.get("source") or "{}")).get("kind") for p in pending] + self.assertIn("inbound_progress", kinds) + self.assertIn("inbound_reply", kinds) + progress_rows = [p for p, k in zip(pending, kinds) if k == "inbound_progress"] + self.assertTrue(progress_rows) + self.assertIn("设备", str(progress_rows[0].get("text") or "")) + progress_src = json.loads(str(progress_rows[0].get("source") or "{}")) + self.assertTrue(progress_src.get("mention_jids")) + self.assertFalse(progress_src.get("quote_stanza_id")) + + +if __name__ == "__main__": + unittest.main()