From 16c06ee2b0a0b7227b4e3506dc7c0293e2fd53af Mon Sep 17 00:00:00 2001 From: Safwan Erooth Date: Sun, 26 Jul 2026 18:07:30 +0400 Subject: [PATCH] feat(gateway): bridge adapters to runtime routing --- huf/ai/gateway_service.py | 15 ++- huf/ai/gateway_webhook.py | 133 +++++++++++++++++++++++++++ huf/ai/tests/test_gateway_service.py | 34 +++++++ huf/ai/tests/test_gateway_webhook.py | 86 +++++++++++++++++ huf/huf/doctype/gateway/gateway.json | 4 +- huf/huf/doctype/gateway/gateway.py | 10 ++ 6 files changed, 278 insertions(+), 4 deletions(-) create mode 100644 huf/ai/gateway_webhook.py create mode 100644 huf/ai/tests/test_gateway_webhook.py diff --git a/huf/ai/gateway_service.py b/huf/ai/gateway_service.py index 9b4763661..c03c56eaa 100644 --- a/huf/ai/gateway_service.py +++ b/huf/ai/gateway_service.py @@ -238,15 +238,26 @@ def process_gateway_event(event_name: str) -> dict: event.db_set("status", "Running") if event.target_type == "Agent": from huf.ai.agent_integration import run_agent_sync + from huf.ai.gateway_webhook import send_gateway_reply result = run_agent_sync( agent_name=event.target_agent, prompt=event.message_text, channel_id=f"gateway:{gateway.name}", external_id=event.thread_id or event.conversation_id or event.sender_id, + now=True, ) - event.db_set({"agent_run": result.get("agent_run_id"), "status": "Queued"}) - return {"event_name": event.name, "status": "Queued", "agent_run_id": result.get("agent_run_id")} + response = result.get("response") if isinstance(result, dict) else None + if not response or not str(response).strip(): + raise frappe.ValidationError(_("Gateway agent run completed without a text response.")) + delivery = send_gateway_reply(gateway, event, str(response)) + event.db_set({"agent_run": result.get("agent_run_id"), "status": "Succeeded"}) + return { + "event_name": event.name, + "status": "Succeeded", + "agent_run_id": result.get("agent_run_id"), + "provider_message_id": delivery.provider_message_id, + } if event.target_type == "Flow": from huf.ai.flow_engine import create_flow_run diff --git a/huf/ai/gateway_webhook.py b/huf/ai/gateway_webhook.py new file mode 100644 index 000000000..6966ece93 --- /dev/null +++ b/huf/ai/gateway_webhook.py @@ -0,0 +1,133 @@ +"""Runtime bridge between Gateway routing and provider-native adapters. + +The module is intentionally lazy about provider imports: the Gateway foundation +can remain independently reviewable while regional adapter packages are merged +on top of the Adapter SDK. A configured gateway still fails closed if its +adapter package or required credentials are unavailable. +""" + +from __future__ import annotations + +from importlib import import_module +from typing import Any + +import frappe +from frappe import _ + + +_ADAPTER_CLASSES = { + "VK": ("huf.ai.gateway_adapters.vk", "VKGatewayAdapter"), + "WeCom": ("huf.ai.gateway_adapters.wecom", "WeComGatewayAdapter"), +} + + +def _adapter_class_for_provider(provider: str): + try: + module_name, class_name = _ADAPTER_CLASSES[provider] + except KeyError as exc: + raise frappe.ValidationError(_("No installed Gateway Adapter supports this channel.")) from exc + try: + return getattr(import_module(module_name), class_name) + except (ImportError, AttributeError) as exc: + raise frappe.ValidationError( + _("The Gateway Adapter package for this channel is not installed.") + ) from exc + + +def _gateway_credentials(gateway) -> dict[str, str]: + if not gateway.integration_settings: + raise frappe.ValidationError(_("This gateway needs a connected integration for its credentials.")) + settings = frappe.get_doc("Integration Settings", gateway.integration_settings) + credentials = {} + for row in settings.credentials or []: + if row.key: + credentials[row.key] = row.get_password("value") or "" + return credentials + + +def get_gateway_adapter(gateway): + """Instantiate the configured adapter without exposing stored credentials.""" + return _adapter_class_for_provider(gateway.provider)(_gateway_credentials(gateway)) + + +def _inbound_request(): + from huf.ai.gateway_adapters.types import GatewayInboundRequest + + request = frappe.request + if not request: + raise frappe.ValidationError(_("A provider HTTP request is required.")) + return GatewayInboundRequest( + body=request.get_data(), + headers=dict(request.headers), + query=dict(request.args), + method=request.method, + ) + + +def _event_context(event) -> dict[str, Any]: + return { + "sender_id": event.sender_id, + "conversation_id": event.conversation_id, + "thread_id": event.thread_id, + "message_text": event.message_text, + "is_room": event.is_room, + "mentioned": event.mentioned, + } + + +def _text_response(value: str) -> None: + """Return a provider challenge as raw text rather than Frappe API JSON.""" + frappe.local.response.update( + {"type": "txt", "doctype": "gateway-callback", "result": value} + ) + + +@frappe.whitelist(allow_guest=True, methods=["GET", "POST"]) +def handle_gateway_webhook(gateway_name: str) -> dict | None: + """Verify a native provider callback, then hand only normalized data to Gateway. + + The URL contains a Gateway document name, never a credential. Native + verification happens before persistence or queueing, and no provider payload + is normalized until that verification succeeds. + """ + from huf.ai.gateway_service import ingest_gateway_event + + try: + gateway = frappe.get_doc("Gateway", gateway_name) + except frappe.DoesNotExistError: + return {"success": False, "error": "Unknown gateway"} + if not gateway.is_enabled: + return {"success": False, "error": "Gateway is disabled"} + + adapter = get_gateway_adapter(gateway) + request = _inbound_request() + if request.method == "GET" and gateway.provider == "WeCom": + _text_response(adapter.verify_url(request)) + return None + if not adapter.verify_inbound(request): + return {"success": False, "error": "Provider verification failed"} + + event = adapter.normalize_inbound(request) + result = ingest_gateway_event( + gateway.name, + event.provider_event_id, + _event_context(event), + verified_sender=True, + raw_payload=dict(event.raw_payload), + ) + return {"success": True, **result} + + +def send_gateway_reply(gateway, event, text: str): + """Deliver one completed Agent response through the same verified gateway.""" + from huf.ai.gateway_adapters.types import GatewayReply + + adapter = get_gateway_adapter(gateway) + return adapter.send_reply( + GatewayReply( + conversation_id=event.conversation_id, + text=text, + thread_id=event.thread_id or None, + reply_to_provider_message_id=event.thread_id or None, + ) + ) diff --git a/huf/ai/tests/test_gateway_service.py b/huf/ai/tests/test_gateway_service.py index 3d0806225..838a996f6 100644 --- a/huf/ai/tests/test_gateway_service.py +++ b/huf/ai/tests/test_gateway_service.py @@ -1,6 +1,7 @@ """Focused unit tests for the provider-neutral Gateway foundation.""" from types import SimpleNamespace +import sys import unittest from unittest.mock import MagicMock, patch from datetime import datetime @@ -186,3 +187,36 @@ def test_room_requires_room_sender_and_mention_when_configured(self, mock_frappe admitted, reason = gateway_service._admission(gateway(), {"sender_id": "42", "conversation_id": "room:1", "is_room": True, "mentioned": False}) assert admitted is False assert reason == "Room messages must mention the gateway" + + +class TestGatewayReplyDelivery(unittest.TestCase): + @patch("huf.ai.gateway_service.frappe") + @patch("huf.ai.gateway_webhook.send_gateway_reply") + def test_agent_result_is_delivered_through_the_gateway(self, mock_send, mock_frappe): + event = MagicMock( + name="GATEWAY-EVENT-0004", + status="Queued", + gateway="Support VK", + target_type="Agent", + target_agent="Support Agent", + message_text="hello", + thread_id="123", + conversation_id="2000000001", + sender_id="42", + ) + configured_gateway = gateway(name="Support VK", execution_user="gateway-bot") + mock_frappe.get_doc.side_effect = [event, configured_gateway] + mock_run = MagicMock(return_value={"agent_run_id": "AR-001", "response": "Hello back"}) + mock_send.return_value = SimpleNamespace(provider_message_id="vk-message-1") + + with patch.dict(sys.modules, {"huf.ai.agent_integration": SimpleNamespace(run_agent_sync=mock_run)}): + result = gateway_service.process_gateway_event(event.name) + + assert result == { + "event_name": event.name, + "status": "Succeeded", + "agent_run_id": "AR-001", + "provider_message_id": "vk-message-1", + } + assert mock_run.call_args.kwargs["now"] is True + mock_send.assert_called_once_with(configured_gateway, event, "Hello back") diff --git a/huf/ai/tests/test_gateway_webhook.py b/huf/ai/tests/test_gateway_webhook.py new file mode 100644 index 000000000..1648687f5 --- /dev/null +++ b/huf/ai/tests/test_gateway_webhook.py @@ -0,0 +1,86 @@ +"""Focused tests for the Gateway Adapter runtime bridge.""" + +from types import SimpleNamespace +import unittest +from unittest.mock import MagicMock, patch + +from huf.ai import gateway_webhook + + +class TestGatewayWebhook(unittest.TestCase): + @patch("huf.ai.gateway_webhook.frappe") + def test_credentials_are_read_from_password_rows(self, mock_frappe): + row = MagicMock(key="community_token") + row.get_password.return_value = "secret-value" + settings = SimpleNamespace(credentials=[row]) + configured_gateway = SimpleNamespace(integration_settings="VK-0001") + mock_frappe.get_doc.return_value = settings + + assert gateway_webhook._gateway_credentials(configured_gateway) == {"community_token": "secret-value"} + row.get_password.assert_called_once_with("value") + + @patch("huf.ai.gateway_webhook.ingest_gateway_event", create=True) + @patch("huf.ai.gateway_webhook.get_gateway_adapter") + @patch("huf.ai.gateway_webhook._inbound_request") + @patch("huf.ai.gateway_webhook.frappe") + def test_verified_event_is_normalized_before_ingress( + self, mock_frappe, mock_request, mock_adapter, mock_ingest + ): + configured_gateway = SimpleNamespace(name="Support VK", provider="VK", is_enabled=1) + fake_request = SimpleNamespace(method="POST") + normalized = SimpleNamespace( + provider_event_id="event-1", + sender_id="42", + conversation_id="2000000001", + thread_id="123", + message_text="hello", + is_room=True, + mentioned=False, + raw_payload={"type": "message_new"}, + ) + adapter = MagicMock() + adapter.verify_inbound.return_value = True + adapter.normalize_inbound.return_value = normalized + mock_frappe.get_doc.return_value = configured_gateway + mock_request.return_value = fake_request + mock_adapter.return_value = adapter + + with patch("huf.ai.gateway_service.ingest_gateway_event", return_value={"event_name": "GE-1"}) as ingest: + result = gateway_webhook.handle_gateway_webhook("Support VK") + + assert result == {"success": True, "event_name": "GE-1"} + adapter.normalize_inbound.assert_called_once_with(fake_request) + assert ingest.call_args.args[:2] == ("Support VK", "event-1") + assert ingest.call_args.kwargs["verified_sender"] is True + + @patch("huf.ai.gateway_webhook.get_gateway_adapter") + @patch("huf.ai.gateway_webhook._inbound_request") + @patch("huf.ai.gateway_webhook.frappe") + def test_unverified_request_is_never_normalized(self, mock_frappe, mock_request, mock_adapter): + mock_frappe.get_doc.return_value = SimpleNamespace(name="Support VK", provider="VK", is_enabled=1) + mock_request.return_value = SimpleNamespace(method="POST") + adapter = MagicMock() + adapter.verify_inbound.return_value = False + mock_adapter.return_value = adapter + + assert gateway_webhook.handle_gateway_webhook("Support VK") == { + "success": False, + "error": "Provider verification failed", + } + adapter.normalize_inbound.assert_not_called() + + @patch("huf.ai.gateway_webhook.get_gateway_adapter") + @patch("huf.ai.gateway_webhook._inbound_request") + @patch("huf.ai.gateway_webhook._text_response") + @patch("huf.ai.gateway_webhook.frappe") + def test_wecom_get_challenge_returns_raw_provider_echo( + self, mock_frappe, mock_text_response, mock_request, mock_adapter + ): + mock_frappe.get_doc.return_value = SimpleNamespace(name="Support WeCom", provider="WeCom", is_enabled=1) + mock_request.return_value = SimpleNamespace(method="GET") + adapter = MagicMock() + adapter.verify_url.return_value = "decrypted-echo" + mock_adapter.return_value = adapter + + assert gateway_webhook.handle_gateway_webhook("Support WeCom") is None + mock_text_response.assert_called_once_with("decrypted-echo") diff --git a/huf/huf/doctype/gateway/gateway.json b/huf/huf/doctype/gateway/gateway.json index 5b5de35ba..a7ac58db7 100644 --- a/huf/huf/doctype/gateway/gateway.json +++ b/huf/huf/doctype/gateway/gateway.json @@ -8,9 +8,9 @@ "field_order": ["gateway_name", "provider", "is_enabled", "integration_settings", "execution_user", "description", "section_admission", "direct_policy", "room_policy", "room_sender_policy", "mention_required", "pairing_ttl_minutes", "section_default_route", "default_target_type", "default_agent", "default_flow", "section_status", "last_event_at", "last_error"], "fields": [ {"fieldname":"gateway_name","fieldtype":"Data","in_list_view":1,"label":"Gateway name","reqd":1,"unique":1}, - {"fieldname":"provider","fieldtype":"Select","in_list_view":1,"label":"Channel","options":"Telegram\nSlack\nEmail\nWhatsApp","reqd":1}, + {"fieldname":"provider","fieldtype":"Select","in_list_view":1,"label":"Channel","options":"Telegram\nSlack\nEmail\nWhatsApp\nVK\nWeCom","reqd":1}, {"default":"1","fieldname":"is_enabled","fieldtype":"Check","in_list_view":1,"label":"Enabled"}, - {"description":"Optional existing integration that owns this channel's credentials. This keeps sending tools separate from inbound routing.","fieldname":"integration_settings","fieldtype":"Link","label":"Connected integration","options":"Integration Settings"}, + {"description":"Existing integration that owns this channel's credentials. VK and WeCom gateways require it; use its Password credential rows for the adapter's documented keys.","fieldname":"integration_settings","fieldtype":"Link","label":"Connected integration","options":"Integration Settings"}, {"description":"The least-privileged Huf user used when this gateway starts an Agent or Flow. Never use Administrator.","fieldname":"execution_user","fieldtype":"Link","label":"Run as user","options":"User"}, {"fieldname":"description","fieldtype":"Small Text","label":"What this gateway is for"}, {"fieldname":"section_admission","fieldtype":"Section Break","label":"Admission policy"}, diff --git a/huf/huf/doctype/gateway/gateway.py b/huf/huf/doctype/gateway/gateway.py index e61962a6a..22cb334a1 100644 --- a/huf/huf/doctype/gateway/gateway.py +++ b/huf/huf/doctype/gateway/gateway.py @@ -20,3 +20,13 @@ def validate(self): frappe.throw(_("Choose a default flow or clear the default route.")) if self.direct_policy == "Open": frappe.throw(_("Public direct-message gateways are not available in this release.")) + + expected_service = {"VK": "vk", "WeCom": "wecom"}.get(self.provider) + if expected_service and self.is_enabled: + if not self.integration_settings: + frappe.throw(_("Enabled {0} gateways need a connected integration.").format(self.provider)) + integration = frappe.get_doc("Integration Settings", self.integration_settings) + if integration.service != expected_service: + frappe.throw( + _("The connected integration must use the {0} service.").format(expected_service) + )