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 <cursoragent@cursor.com>

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()

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",
+]

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"))

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.

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` 或减命令，禁止相同参数重试。

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()