feat(qqbot): wire native tool-approval UX via inline keyboards

Makes the in-tree QQ inline keyboards actually light up when the agent
blocks on a dangerous-command approval. Matches the cross-adapter
gateway contract already implemented by Discord, Telegram, Slack,
Matrix, and Feishu.

Gateway/run.py's _approval_notify_sync checks type(adapter).send_exec_approval
and falls back to a text prompt when it's missing. Without this wiring,
QQ users stared at plain '/approve' text even though the adapter shipped
button primitives.

### send_exec_approval(chat_id, command, session_key, description, metadata)

Matches the signature the gateway calls with. Builds an ApprovalRequest
(command_preview, description, timeout) and delegates to send_approval_request.
Uses the last inbound msg_id as reply_to so QQ accepts the passive
message. The 'metadata' parameter is accepted for contract parity but
intentionally unused — QQ doesn't have thread_id/DM-targeting overrides.

### send_update_prompt(chat_id, prompt, default, session_key, metadata)

Signature updated to match the cross-adapter contract used by
'hermes update --gateway' watcher. Renders a 'Update Needs Your Input'
prompt with the optional default hint and a Yes/No keyboard. Replaces
the earlier 3-arg helper that wasn't wired anywhere.

### Default interaction dispatcher

_default_interaction_dispatch() auto-registered as the adapter's
interaction callback in __init__. Routes:

- approve:<session_key>:<decision> → tools.approval.resolve_gateway_approval
  Button → choice mapping:
    allow-once  → 'once'
    allow-always → 'always'
    deny        → 'deny'
  (QQ's 3-button mobile layout deliberately collapses 'session' + 'always'
  into one button; /approve session text fallback remains available.)
- update_prompt:<answer> → atomic write of y/n to ~/.hermes/.update_response
  (the detached 'hermes update --gateway' watcher polls this file)
- anything else → logged and dropped

Resolve exceptions are caught and logged — never propagate into the WS
loop. Callers can override via set_interaction_callback() to route
clicks elsewhere or pass None to drop them entirely.

### Net effect

QQ users now get native tap-to-approve UX on dangerous-command prompts
and update-confirmation prompts, without having to type /approve or /deny
as text. The adapter hooks into tools.approval the same way every other
button-capable platform does.

### Tests

14 new tests cover:
- Default callback installed on __init__
- send_exec_approval / send_update_prompt exist as class methods (so the
  gateway's type-probe detects them)
- allow-once/always/deny each map to the correct resolve choice
- update_prompt:y / update_prompt:n each write atomically to the response
  file (via monkeypatched get_hermes_home)
- Unknown button_data / empty button_data / resolve exceptions are harmless
- send_exec_approval honours last_msg_id reply-to and accepts metadata
- send_update_prompt delegates with correct content + keyboard

Full qqbot suite: 144 passed (72 pre-existing + 72 from this salvage arc).
Also ran tools/test_approval.py alongside — no regressions (276 passed
combined).

Co-authored-by: WideLee <limkuan24@gmail.com>
This commit is contained in:
WideLee 2026-05-07 07:47:14 -07:00 committed by Teknium
parent a1fe5f473d
commit 4de3ef38b1
2 changed files with 455 additions and 7 deletions

View file

@ -232,6 +232,14 @@ class QQAdapter(BasePlatformAdapter):
Callable[[InteractionEvent], Awaitable[None]]
] = None
# Default interaction dispatcher: routes approval-button clicks to
# tools.approval.resolve_gateway_approval() and update-prompt clicks
# to ~/.hermes/.update_response. Set here so the cross-adapter gateway
# contract (send_exec_approval / send_update_prompt) works out of the
# box; callers can override with set_interaction_callback(None) or
# register a custom handler.
self._interaction_callback = self._default_interaction_dispatch
# ------------------------------------------------------------------
# Properties
# ------------------------------------------------------------------
@ -963,6 +971,101 @@ class QQAdapter(BasePlatformAdapter):
f"{resp.text[:200]}"
)
# Mapping from QQ keyboard button decisions → the ``choice`` vocabulary
# accepted by ``tools.approval.resolve_gateway_approval``. QQ's 3-button
# layout (mobile-space constraint) collapses "session" and "always" into
# a single "always" button; users wanting session-only approval can fall
# back to the ``/approve session`` text command.
_APPROVAL_BUTTON_TO_CHOICE = {
"allow-once": "once",
"allow-always": "always",
"deny": "deny",
}
async def _default_interaction_dispatch(
self,
event: InteractionEvent,
) -> None:
"""Route ``INTERACTION_CREATE`` button clicks to the right subsystem.
- ``approve:<session_key>:<decision>``
:func:`tools.approval.resolve_gateway_approval`
(unblocks the agent thread waiting on a dangerous-command approval).
- ``update_prompt:<answer>``
writes the answer to ``~/.hermes/.update_response`` for the
detached ``hermes update --gateway`` process to consume.
- Anything else is logged at DEBUG and ignored.
Installed as the adapter's default interaction callback in
``__init__``. Callers can replace via
:meth:`set_interaction_callback` to route clicks elsewhere (or pass
``None`` to drop them entirely).
"""
button_data = event.button_data
if not button_data:
return
approval = parse_approval_button_data(button_data)
if approval is not None:
session_key, decision = approval
choice = self._APPROVAL_BUTTON_TO_CHOICE.get(decision)
if choice is None:
logger.warning(
"[%s] Unknown approval decision %r (session=%s)",
self._log_tag, decision, session_key,
)
return
try:
# Import lazily to keep the adapter importable in tests that
# don't exercise the approval subsystem.
from tools.approval import resolve_gateway_approval
count = resolve_gateway_approval(session_key, choice)
logger.info(
"[%s] Button resolved %d approval(s) for session %s "
"(choice=%s, operator=%s)",
self._log_tag, count, session_key, choice,
event.operator_openid,
)
except Exception as exc:
logger.error(
"[%s] resolve_gateway_approval failed for session %s: %s",
self._log_tag, session_key, exc,
)
return
update_answer = parse_update_prompt_button_data(button_data)
if update_answer is not None:
self._write_update_response(update_answer, event.operator_openid)
return
logger.debug(
"[%s] Unrecognised button_data %r from interaction %s",
self._log_tag, button_data, event.id,
)
@staticmethod
def _write_update_response(answer: str, operator: str = "") -> None:
"""Atomically write the update-prompt answer to ``.update_response``.
Mirrors the Discord / Telegram / Feishu adapters: the detached
``hermes update --gateway`` watcher polls this file for a ``y``/``n``
response to its interactive prompts (stash-restore, config migration).
Writes via ``tmp + rename`` so a partial write can't fool the reader.
"""
try:
from hermes_constants import get_hermes_home
home = get_hermes_home()
response_path = home / ".update_response"
tmp = response_path.with_suffix(".tmp")
tmp.write_text(answer)
tmp.replace(response_path)
logger.info(
"QQ update prompt answered %r by %s",
answer, operator or "(unknown)",
)
except Exception as exc:
logger.error("Failed to write update response: %s", exc)
async def _handle_c2c_message(
self,
d: Dict[str, Any],
@ -2391,22 +2494,78 @@ class QQAdapter(BasePlatformAdapter):
reply_to=reply_to,
)
# ------------------------------------------------------------------
# Cross-adapter gateway contract — send_exec_approval + send_update_prompt
# ------------------------------------------------------------------
#
# These mirror the signatures that gateway/run.py detects on the adapter
# class (e.g. type(adapter).send_exec_approval, type(adapter).send_update_prompt)
# for button-based approval / update-confirm UX. Discord, Telegram, Slack,
# Matrix, and Feishu already implement the same contract.
async def send_exec_approval(
self,
chat_id: str,
command: str,
session_key: str,
description: str = "dangerous command",
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Send a button-based exec-approval prompt for a dangerous command.
Called by ``gateway/run.py``'s ``_approval_notify_sync`` when the
agent is blocked waiting for approval. Button clicks resolve via
:func:`tools.approval.resolve_gateway_approval` dispatched by the
adapter's interaction callback (:meth:`_default_interaction_dispatch`).
"""
del metadata # QQ doesn't have thread_id / DM targeting overrides.
# Use the reply-to message for passive-message context when we have one.
# QQ requires a msg_id on outbound messages to a user we've never
# seen; the last inbound msg_id is the natural choice.
msg_id = self._last_msg_id.get(chat_id)
req = ApprovalRequest(
session_key=session_key,
title=f"Execute this command?",
description=description,
command_preview=command,
timeout_sec=self._APPROVAL_TIMEOUT_SECONDS,
)
return await self.send_approval_request(
chat_id, req, reply_to=msg_id,
)
_APPROVAL_TIMEOUT_SECONDS = 300 # matches gateway's default gateway_timeout
async def send_update_prompt(
self,
chat_id: str,
content: str,
reply_to: Optional[str] = None,
prompt: str,
default: str = "",
session_key: str = "",
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Send a Yes/No update-confirmation prompt with inline buttons.
Button clicks surface as ``INTERACTION_CREATE`` with
``button_data = 'update_prompt:y'`` or ``'update_prompt:n'``.
Matches the cross-adapter contract used by
``gateway/run.py``'s ``hermes update --gateway`` watcher. Button
clicks surface as ``INTERACTION_CREATE`` with
``button_data = 'update_prompt:y'`` or ``'update_prompt:n'``;
the adapter's interaction callback writes the answer to
``~/.hermes/.update_response`` so the detached update process
can read it.
"""
del session_key, metadata # present for contract parity only.
default_hint = f" (default: {default})" if default else ""
content = f"⚕ **Update Needs Your Input**\n\n{prompt}{default_hint}"
msg_id = self._last_msg_id.get(chat_id)
return await self.send_with_keyboard(
chat_id,
content,
build_update_prompt_keyboard(),
reply_to=reply_to,
reply_to=msg_id,
)
def _build_text_body(

View file

@ -1287,14 +1287,16 @@ class TestAdapterInteractionDispatch:
})
@pytest.mark.asyncio
async def test_no_callback_is_harmless(self):
async def test_explicit_no_callback_is_harmless(self):
adapter = self._make_adapter()
async def fake_ack(interaction_id, code=0):
pass
adapter._acknowledge_interaction = fake_ack # type: ignore[assignment]
# No callback set — default None.
# Explicitly clear the default callback. With no callback set,
# _on_interaction should still ACK and not raise.
adapter.set_interaction_callback(None)
await adapter._on_interaction({
"id": "i-3",
"chat_type": 2,
@ -1518,3 +1520,290 @@ class TestMergeQuoteInto:
from gateway.platforms.qqbot.adapter import QQAdapter
merged = QQAdapter._merge_quote_into("hi there", "[Quoted]:\nctx")
assert merged == "[Quoted]:\nctx\n\nhi there"
# ---------------------------------------------------------------------------
# Gateway-contract approval UX — send_exec_approval + default dispatcher
# ---------------------------------------------------------------------------
class TestDefaultInteractionDispatch:
"""Verify the adapter's default INTERACTION_CREATE router."""
def _make_adapter(self):
from gateway.platforms.qqbot.adapter import QQAdapter
return QQAdapter(_make_config(app_id="a", client_secret="b"))
def test_default_callback_installed_on_init(self):
"""Fresh adapter has a working default interaction callback."""
adapter = self._make_adapter()
assert adapter._interaction_callback is not None
assert adapter._interaction_callback == adapter._default_interaction_dispatch
def test_send_exec_approval_is_a_class_method(self):
"""gateway/run.py uses ``type(adapter).send_exec_approval`` to detect support."""
from gateway.platforms.qqbot.adapter import QQAdapter
assert getattr(QQAdapter, "send_exec_approval", None) is not None
assert getattr(QQAdapter, "send_update_prompt", None) is not None
@pytest.mark.asyncio
async def test_approval_click_once_maps_to_once(self):
"""'allow-once' button → resolve_gateway_approval(session, 'once')."""
adapter = self._make_adapter()
resolve_calls = []
def fake_resolve(session_key, choice, resolve_all=False):
resolve_calls.append((session_key, choice, resolve_all))
return 1
# Patch the *module-level* function that _default_interaction_dispatch
# imports lazily.
import tools.approval
orig = tools.approval.resolve_gateway_approval
tools.approval.resolve_gateway_approval = fake_resolve
try:
from gateway.platforms.qqbot.keyboards import parse_interaction_event
event = parse_interaction_event({
"id": "i",
"chat_type": 2,
"user_openid": "u-42",
"data": {"resolved": {"button_data": "approve:sess-abc:allow-once"}},
})
await adapter._default_interaction_dispatch(event)
finally:
tools.approval.resolve_gateway_approval = orig
assert resolve_calls == [("sess-abc", "once", False)]
@pytest.mark.asyncio
async def test_approval_click_always_maps_to_always(self):
adapter = self._make_adapter()
resolve_calls = []
def fake_resolve(session_key, choice, resolve_all=False):
resolve_calls.append((session_key, choice, resolve_all))
return 1
import tools.approval
orig = tools.approval.resolve_gateway_approval
tools.approval.resolve_gateway_approval = fake_resolve
try:
from gateway.platforms.qqbot.keyboards import parse_interaction_event
event = parse_interaction_event({
"id": "i", "chat_type": 2, "user_openid": "u",
"data": {"resolved": {"button_data": "approve:s:allow-always"}},
})
await adapter._default_interaction_dispatch(event)
finally:
tools.approval.resolve_gateway_approval = orig
assert resolve_calls == [("s", "always", False)]
@pytest.mark.asyncio
async def test_approval_click_deny_maps_to_deny(self):
adapter = self._make_adapter()
resolve_calls = []
def fake_resolve(session_key, choice, resolve_all=False):
resolve_calls.append((session_key, choice, resolve_all))
return 1
import tools.approval
orig = tools.approval.resolve_gateway_approval
tools.approval.resolve_gateway_approval = fake_resolve
try:
from gateway.platforms.qqbot.keyboards import parse_interaction_event
event = parse_interaction_event({
"id": "i", "chat_type": 2, "user_openid": "u",
"data": {"resolved": {"button_data": "approve:s:deny"}},
})
await adapter._default_interaction_dispatch(event)
finally:
tools.approval.resolve_gateway_approval = orig
assert resolve_calls == [("s", "deny", False)]
@pytest.mark.asyncio
async def test_update_prompt_click_writes_response_file(self, tmp_path, monkeypatch):
"""update_prompt:y click writes 'y' to ~/.hermes/.update_response."""
adapter = self._make_adapter()
hermes_home = tmp_path / "hermes_home"
hermes_home.mkdir()
monkeypatch.setattr(
"hermes_constants.get_hermes_home",
lambda: hermes_home,
)
from gateway.platforms.qqbot.keyboards import parse_interaction_event
event = parse_interaction_event({
"id": "i", "chat_type": 2, "user_openid": "u-1",
"data": {"resolved": {"button_data": "update_prompt:y"}},
})
await adapter._default_interaction_dispatch(event)
response = hermes_home / ".update_response"
assert response.exists()
assert response.read_text() == "y"
@pytest.mark.asyncio
async def test_update_prompt_click_no_writes_n(self, tmp_path, monkeypatch):
adapter = self._make_adapter()
hermes_home = tmp_path / "hermes_home"
hermes_home.mkdir()
monkeypatch.setattr(
"hermes_constants.get_hermes_home",
lambda: hermes_home,
)
from gateway.platforms.qqbot.keyboards import parse_interaction_event
event = parse_interaction_event({
"id": "i", "chat_type": 2, "user_openid": "u",
"data": {"resolved": {"button_data": "update_prompt:n"}},
})
await adapter._default_interaction_dispatch(event)
response = hermes_home / ".update_response"
assert response.read_text() == "n"
@pytest.mark.asyncio
async def test_unknown_button_data_is_harmless(self):
"""Unrecognised button_data is logged and dropped — no exception."""
adapter = self._make_adapter()
from gateway.platforms.qqbot.keyboards import parse_interaction_event
event = parse_interaction_event({
"id": "i", "chat_type": 2, "user_openid": "u",
"data": {"resolved": {"button_data": "some:unknown:format"}},
})
# Must not raise.
await adapter._default_interaction_dispatch(event)
@pytest.mark.asyncio
async def test_empty_button_data_is_harmless(self):
adapter = self._make_adapter()
from gateway.platforms.qqbot.keyboards import InteractionEvent
await adapter._default_interaction_dispatch(InteractionEvent(id="i"))
@pytest.mark.asyncio
async def test_resolve_exception_is_swallowed(self):
"""If resolve_gateway_approval raises, we log but don't propagate."""
adapter = self._make_adapter()
def bad_resolve(session_key, choice, resolve_all=False):
raise RuntimeError("boom")
import tools.approval
orig = tools.approval.resolve_gateway_approval
tools.approval.resolve_gateway_approval = bad_resolve
try:
from gateway.platforms.qqbot.keyboards import parse_interaction_event
event = parse_interaction_event({
"id": "i", "chat_type": 2, "user_openid": "u",
"data": {"resolved": {"button_data": "approve:s:deny"}},
})
# Must not raise.
await adapter._default_interaction_dispatch(event)
finally:
tools.approval.resolve_gateway_approval = orig
class TestSendExecApproval:
"""Verify the gateway contract: QQAdapter.send_exec_approval(...)."""
def _make_adapter(self):
from gateway.platforms.qqbot.adapter import QQAdapter
return QQAdapter(_make_config(app_id="a", client_secret="b"))
@pytest.mark.asyncio
async def test_delegates_to_send_approval_request(self):
adapter = self._make_adapter()
calls = []
async def fake_send_approval(chat_id, req, reply_to=None):
from gateway.platforms.base import SendResult
calls.append({"chat_id": chat_id, "req": req, "reply_to": reply_to})
return SendResult(success=True, message_id="m-1")
adapter.send_approval_request = fake_send_approval # type: ignore[assignment]
# Seed last-msg-id so the reply_to path is exercised.
adapter._last_msg_id["user-1"] = "inbound-42"
result = await adapter.send_exec_approval(
chat_id="user-1",
command="rm -rf /tmp/demo",
session_key="sess:abc",
description="delete temp dir",
)
assert result.success
assert len(calls) == 1
req = calls[0]["req"]
assert req.session_key == "sess:abc"
assert req.command_preview == "rm -rf /tmp/demo"
assert req.description == "delete temp dir"
assert calls[0]["reply_to"] == "inbound-42"
@pytest.mark.asyncio
async def test_accepts_metadata_arg(self):
"""Gateway always passes metadata=…; the adapter must accept + ignore it."""
adapter = self._make_adapter()
async def fake_send_approval(chat_id, req, reply_to=None):
from gateway.platforms.base import SendResult
return SendResult(success=True)
adapter.send_approval_request = fake_send_approval # type: ignore[assignment]
# Should not raise even when metadata is a dict with unknown keys.
await adapter.send_exec_approval(
chat_id="u", command="ls", session_key="s",
metadata={"thread_id": "ignored", "anything": "else"},
)
class TestSendUpdatePrompt:
"""Verify the cross-adapter send_update_prompt signature + behaviour."""
def _make_adapter(self):
from gateway.platforms.qqbot.adapter import QQAdapter
return QQAdapter(_make_config(app_id="a", client_secret="b"))
@pytest.mark.asyncio
async def test_delegates_to_send_with_keyboard(self):
adapter = self._make_adapter()
captured = {}
async def fake_swk(chat_id, content, keyboard, reply_to=None):
from gateway.platforms.base import SendResult
captured["chat_id"] = chat_id
captured["content"] = content
captured["keyboard"] = keyboard
captured["reply_to"] = reply_to
return SendResult(success=True, message_id="mid")
adapter.send_with_keyboard = fake_swk # type: ignore[assignment]
adapter._last_msg_id["u1"] = "prev-msg"
result = await adapter.send_update_prompt(
chat_id="u1", prompt="Continue with update?",
default="y", session_key="ignored", metadata={"x": 1},
)
assert result.success
assert "Continue with update?" in captured["content"]
assert "default: y" in captured["content"]
assert captured["reply_to"] == "prev-msg"
# Keyboard has the Yes/No buttons.
dd = captured["keyboard"].to_dict()
datas = [b["action"]["data"] for b in dd["content"]["rows"][0]["buttons"]]
assert datas == ["update_prompt:y", "update_prompt:n"]
@pytest.mark.asyncio
async def test_empty_default_has_no_hint(self):
adapter = self._make_adapter()
async def fake_swk(chat_id, content, keyboard, reply_to=None):
from gateway.platforms.base import SendResult
assert "default:" not in content
return SendResult(success=True)
adapter.send_with_keyboard = fake_swk # type: ignore[assignment]
await adapter.send_update_prompt(chat_id="u", prompt="ok?")