diff --git a/contributors/emails/itzhak.pan@gmail.com b/contributors/emails/itzhak.pan@gmail.com new file mode 100644 index 0000000000..ce22353e85 --- /dev/null +++ b/contributors/emails/itzhak.pan@gmail.com @@ -0,0 +1 @@ +yaozhen diff --git a/contributors/emails/keviea@gmail.com b/contributors/emails/keviea@gmail.com new file mode 100644 index 0000000000..16fd7a6a00 --- /dev/null +++ b/contributors/emails/keviea@gmail.com @@ -0,0 +1 @@ +bka9 diff --git a/contributors/emails/markmnl@fmsg.io b/contributors/emails/markmnl@fmsg.io new file mode 100644 index 0000000000..7373b15800 --- /dev/null +++ b/contributors/emails/markmnl@fmsg.io @@ -0,0 +1 @@ +markmnl diff --git a/cron/scheduler.py b/cron/scheduler.py index 79e6b39d1b..7c7d2b9c51 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1363,28 +1363,22 @@ def _resolve_single_delivery_target(job: dict, deliver_value: str) -> Optional[d platform_name, rest = deliver_value.split(":", 1) platform_key = platform_name.lower() - from tools.send_message_tool import _parse_target_ref + from tools.send_message_tool import ( + prepare_send_message_platforms, + resolve_send_target, + ) - parsed_chat_id, parsed_thread_id, is_explicit = _parse_target_ref(platform_key, rest) - if is_explicit: - chat_id, thread_id = parsed_chat_id, parsed_thread_id - else: - chat_id, thread_id = rest, None - - # Resolve human-friendly labels like "Alice (dm)" to real IDs. - try: - from gateway.channel_directory import resolve_channel_name - resolved = resolve_channel_name(platform_key, chat_id) - if resolved: - parsed_chat_id, parsed_thread_id, resolved_is_explicit = _parse_target_ref(platform_key, resolved) - if resolved_is_explicit: - chat_id = parsed_chat_id - if parsed_thread_id is not None: - thread_id = parsed_thread_id - else: - chat_id = resolved - except Exception: - pass + prepare_send_message_platforms() + chat_id, thread_id, resolution_error = resolve_send_target( + platform_key, rest + ) + if resolution_error: + logger.warning( + "Invalid cron delivery target '%s': %s", + deliver_value, + resolution_error, + ) + return None if ( thread_id is None diff --git a/gateway/platform_registry.py b/gateway/platform_registry.py index 15eeb8a172..54f3c4b960 100644 --- a/gateway/platform_registry.py +++ b/gateway/platform_registry.py @@ -173,10 +173,20 @@ class PlatformEntry: # Signature: # (target_ref: str) -> Optional[tuple[str, Optional[str]]] # - # If the callable returns None the target proceeds to the usual directory - # resolution / verbatim fallback path. + # If the callable returns None the target proceeds to channel-directory + # resolution. No opaque fallback is applied. parse_target_ref_fn: Optional[Callable[[str], Optional[tuple[str, Optional[str]]]]] = None + # Optional validation applied after parsing/normalization or + # channel-directory resolution. Return True to accept, False to reject, or + # a non-empty string to reject with that diagnostic. + validate_target_ref_fn: Optional[Callable[[str], bool | str]] = None + + # Optional whole-request handler for custom platform delivery. Receives + # (args, normalized_chat_id, platform_name, pconfig) and may be sync/async. + # Prefer standalone_sender_fn when the standard send contract is enough. + send_message_handler: Optional[Callable[[dict, str, str, Any], Any]] = None + # ── Standalone (out-of-process) sending ── # Optional: async coroutine that delivers a message without a live # gateway adapter. Called by ``tools/send_message_tool._send_via_adapter`` @@ -286,7 +296,8 @@ class PlatformRegistry: def unregister(self, name: str) -> bool: """Remove a platform entry. Returns True if it existed.""" self._deferred.pop(name, None) - return self._entries.pop(name, None) is not None + removed = self._entries.pop(name, None) is not None + return removed def get(self, name: str) -> Optional[PlatformEntry]: """Look up a platform entry by name.""" diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 8aead17dde..6ca0e79e16 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -1272,39 +1272,6 @@ class PluginContext: self.manifest.name, provider.name, ) - # -- send_message enricher registration --------------------------------- - - def register_send_message_enricher( - self, - platform_name: str, - handler: Callable, - schema_fragment: dict | None = None, - ) -> None: - """Register a send_message enricher for a plugin messaging platform. - - Lets plugin platforms extend ``send_message`` with custom target - prefixes, schema fields, and send handlers without modifying core code. - - Args: - platform_name: Platform key used in target strings - (e.g. ``"myplatform"`` → ``myplatform:chat_id``). - handler: Callable receiving ``(args, chat_id, platform_name, pconfig)`` - and returning a dict like ``{"success": True, "message_id": "..."}`` - or ``{"error": "..."}``. May be ``async def`` or a regular - function — the dispatcher detects coroutine functions via - ``inspect.iscoroutinefunction`` and awaits as needed. - schema_fragment: Optional dict of JSON-schema properties to merge - into ``send_message``'s parameter schema so the LLM sees the - custom fields. - """ - from tools.send_message_tool import register_send_message_enricher as _register_enricher - _register_enricher(platform_name, handler, schema_fragment=schema_fragment) - logger.debug( - "Plugin %s registered send_message enricher: %s", - self.manifest.name, - platform_name, - ) - # -- platform adapter registration --------------------------------------- def register_platform( @@ -1691,6 +1658,14 @@ class PluginManager: self._discovered = True return if force: + # Remove concrete and deferred platform registrations before + # clearing ownership metadata. Otherwise disabled plugins (or a + # profile switch in a long-lived process) leak their parser or + # send handler into the next discovery pass. + from gateway.platform_registry import platform_registry + + for platform_name in tuple(self._plugin_platform_names): + platform_registry.unregister(platform_name) self._plugins.clear() self._hooks.clear() self._middleware.clear() diff --git a/tests/tools/test_send_message_enrichers.py b/tests/tools/test_send_message_enrichers.py deleted file mode 100644 index 67df4cdf04..0000000000 --- a/tests/tools/test_send_message_enrichers.py +++ /dev/null @@ -1,332 +0,0 @@ -"""Tests for send_message plugin enrichers.""" - -import asyncio -from types import SimpleNamespace -from unittest.mock import AsyncMock, MagicMock - -import pytest - -from tools.send_message_tool import ( - SEND_MESSAGE_SCHEMA, - _SEND_MESSAGE_ENRICHERS, - _SEND_MESSAGE_SCHEMA_FRAGMENTS, - _parse_target_ref, - _send_to_platform, - get_send_message_schema, - register_send_message_enricher, -) - - -@pytest.fixture(autouse=True) -def _reset_enrichers(): - """Clear the enricher registry before and after every test.""" - _SEND_MESSAGE_ENRICHERS.clear() - _SEND_MESSAGE_SCHEMA_FRAGMENTS.clear() - yield - _SEND_MESSAGE_ENRICHERS.clear() - _SEND_MESSAGE_SCHEMA_FRAGMENTS.clear() - - -class TestParseTargetRef: - def test_enricher_target_parsing(self): - """Plugin-enriched platforms treat the target_ref as explicit.""" - register_send_message_enricher("myplatform", AsyncMock()) - chat_id, thread_id, is_explicit = _parse_target_ref("myplatform", "+123") - assert chat_id == "+123" - assert thread_id is None - assert is_explicit is True - - def test_builtin_precedence(self): - """Built-in platforms win over enrichers with the same name.""" - register_send_message_enricher("telegram", AsyncMock()) - chat_id, thread_id, is_explicit = _parse_target_ref("telegram", "777") - assert chat_id == "777" - assert thread_id is None - assert is_explicit is True - - def test_enricher_empty_target_ref(self): - """Empty target_ref for an enriched platform is not explicit.""" - register_send_message_enricher("myplatform", AsyncMock()) - chat_id, thread_id, is_explicit = _parse_target_ref("myplatform", "") - assert chat_id is None - assert thread_id is None - assert is_explicit is False - - -class TestPlatformEntryTargetParsing: - def test_parse_target_ref_fn_is_used(self): - """PlatformEntry.parse_target_ref_fn is consulted for explicit targets.""" - from gateway.platform_registry import PlatformEntry, platform_registry - - platform_name = "fmsg-parse-test" - - def _parser(ref): - if ref.startswith("@") and "@" in ref[1:]: - return ref, None - return None - - entry = PlatformEntry( - name=platform_name, - label="Fmsg parse test", - adapter_factory=lambda cfg: None, - check_fn=lambda: True, - parse_target_ref_fn=_parser, - ) - platform_registry.register(entry) - try: - chat_id, thread_id, is_explicit = _parse_target_ref( - platform_name, "@alice@example.com" - ) - assert chat_id == "@alice@example.com" - assert thread_id is None - assert is_explicit is True - finally: - platform_registry.unregister(platform_name) - - def test_parse_target_ref_fn_returns_none_falls_through(self): - """When parse_target_ref_fn returns None, target falls through to enricher.""" - from gateway.platform_registry import PlatformEntry, platform_registry - - platform_name = "fmsg-fallback-test" - - def _parser(ref): - return None - - entry = PlatformEntry( - name=platform_name, - label="Fmsg fallback test", - adapter_factory=lambda cfg: None, - check_fn=lambda: True, - parse_target_ref_fn=_parser, - ) - platform_registry.register(entry) - register_send_message_enricher(platform_name, AsyncMock()) - try: - chat_id, thread_id, is_explicit = _parse_target_ref( - platform_name, "anything" - ) - assert chat_id == "anything" - assert thread_id is None - assert is_explicit is True - finally: - platform_registry.unregister(platform_name) - - def test_parse_target_ref_fn_thread_id(self): - """parse_target_ref_fn can return a thread_id alongside chat_id.""" - from gateway.platform_registry import PlatformEntry, platform_registry - - platform_name = "threaded-parse-test" - - def _parser(ref): - if ":" in ref: - room, thread = ref.split(":", 1) - return room, thread - return None - - entry = PlatformEntry( - name=platform_name, - label="Threaded parse test", - adapter_factory=lambda cfg: None, - check_fn=lambda: True, - parse_target_ref_fn=_parser, - ) - platform_registry.register(entry) - try: - chat_id, thread_id, is_explicit = _parse_target_ref( - platform_name, "room-1:thread-99" - ) - assert chat_id == "room-1" - assert thread_id == "thread-99" - assert is_explicit is True - finally: - platform_registry.unregister(platform_name) - - def test_builtin_platform_ignores_parse_target_ref_fn(self): - """Built-in platform parsers win over a rogue PlatformEntry parser.""" - from gateway.platform_registry import PlatformEntry, platform_registry - - def _parser(ref): - return "SHOULD_NOT_RETURN", None - - entry = PlatformEntry( - name="telegram", - label="Telegram rogue", - adapter_factory=lambda cfg: None, - check_fn=lambda: True, - parse_target_ref_fn=_parser, - ) - platform_registry.register(entry) - try: - chat_id, thread_id, is_explicit = _parse_target_ref("telegram", "12345") - assert chat_id == "12345" - assert thread_id is None - assert is_explicit is True - finally: - platform_registry.unregister("telegram") - - -class TestSchemaMerge: - def test_schema_fragment_stored(self): - """Schema fragments are stored in _SEND_MESSAGE_SCHEMA_FRAGMENTS.""" - register_send_message_enricher( - "myplatform", - AsyncMock(), - schema_fragment={"voice": {"type": "string", "description": "Voice setting"}}, - ) - assert "myplatform" in _SEND_MESSAGE_SCHEMA_FRAGMENTS - assert _SEND_MESSAGE_SCHEMA_FRAGMENTS["myplatform"]["voice"]["type"] == "string" - - def test_get_send_message_schema_assembles_fragments(self): - """get_send_message_schema returns a copy with fragments merged.""" - register_send_message_enricher( - "myplatform", - AsyncMock(), - schema_fragment={"voice": {"type": "string", "description": "Voice setting"}}, - ) - schema = get_send_message_schema() - assert "voice" in schema["parameters"]["properties"] - assert schema["parameters"]["properties"]["voice"]["type"] == "string" - - def test_original_schema_not_mutated(self): - """The module-level SEND_MESSAGE_SCHEMA is never mutated.""" - register_send_message_enricher( - "myplatform", - AsyncMock(), - schema_fragment={"voice": {"type": "string", "description": "Voice setting"}}, - ) - assert "voice" not in SEND_MESSAGE_SCHEMA["parameters"]["properties"] - - def test_schema_fragment_without_registration(self): - """No fragment is added when schema_fragment is omitted.""" - register_send_message_enricher("myplatform", AsyncMock()) - assert "myplatform" not in _SEND_MESSAGE_SCHEMA_FRAGMENTS - - -class TestHandlerRouting: - def test_async_handler_invoked(self): - """The async enricher handler receives correct arguments.""" - handler = AsyncMock(return_value={"success": True, "message_id": "msg_1"}) - register_send_message_enricher("myplatform", handler) - - pconfig = SimpleNamespace(enabled=True, token="tok", extra={}) - result = asyncio.run( - _send_to_platform( - "myplatform", - pconfig, - "chat42", - "hello", - args={"target": "myplatform:chat42", "message": "hello", "intent": "greet"}, - ) - ) - - assert result == {"success": True, "message_id": "msg_1"} - handler.assert_awaited_once() - call_args = handler.await_args.args - assert call_args[0] == {"target": "myplatform:chat42", "message": "hello", "intent": "greet"} - assert call_args[1] == "chat42" - assert call_args[2] == "myplatform" - assert call_args[3] is pconfig - - def test_sync_handler_invoked(self): - """The sync enricher handler is called directly (not awaited).""" - handler = MagicMock(return_value={"success": True, "message_id": "msg_sync"}) - register_send_message_enricher("myplatform", handler) - - pconfig = SimpleNamespace(enabled=True, token="tok", extra={}) - result = asyncio.run( - _send_to_platform( - "myplatform", - pconfig, - "chat42", - "hello", - args={"target": "myplatform:chat42", "message": "hello"}, - ) - ) - - assert result == {"success": True, "message_id": "msg_sync"} - handler.assert_called_once() - call_args = handler.call_args.args - assert call_args[0] == {"target": "myplatform:chat42", "message": "hello"} - assert call_args[1] == "chat42" - assert call_args[2] == "myplatform" - assert call_args[3] is pconfig - - def test_async_handler_result(self): - """Async enricher result is returned verbatim.""" - handler = AsyncMock(return_value={"success": True, "message_id": "msg_1"}) - register_send_message_enricher("myplatform", handler) - - result = asyncio.run( - _send_to_platform( - "myplatform", - SimpleNamespace(enabled=True, token="tok", extra={}), - "chat42", - "hello", - ) - ) - assert result == {"success": True, "message_id": "msg_1"} - - def test_sync_handler_result(self): - """Sync enricher result is returned verbatim.""" - handler = MagicMock(return_value={"success": True, "message_id": "msg_sync"}) - register_send_message_enricher("myplatform", handler) - - result = asyncio.run( - _send_to_platform( - "myplatform", - SimpleNamespace(enabled=True, token="tok", extra={}), - "chat42", - "hello", - ) - ) - assert result == {"success": True, "message_id": "msg_sync"} - - def test_async_handler_error(self): - """Async enricher exceptions are caught and surfaced as error dicts.""" - handler = AsyncMock(side_effect=RuntimeError("boom")) - register_send_message_enricher("myplatform", handler) - - result = asyncio.run( - _send_to_platform( - "myplatform", - SimpleNamespace(enabled=True, token="tok", extra={}), - "chat42", - "hello", - ) - ) - assert "error" in result - assert "boom" in result["error"] - - def test_sync_handler_error(self): - """Sync enricher exceptions are caught and surfaced as error dicts.""" - handler = MagicMock(side_effect=RuntimeError("sync boom")) - register_send_message_enricher("myplatform", handler) - - result = asyncio.run( - _send_to_platform( - "myplatform", - SimpleNamespace(enabled=True, token="tok", extra={}), - "chat42", - "hello", - ) - ) - assert "error" in result - assert "sync boom" in result["error"] - - -class TestFallback: - def test_no_enricher_falls_through(self): - """Platforms without an enricher still reach _send_via_adapter.""" - result = asyncio.run( - _send_to_platform( - "unknownplatform", - SimpleNamespace(enabled=True, token="tok", extra={}), - "chat42", - "hello", - ) - ) - # _send_via_adapter will error because no adapter is registered; - # the exact error text is not important, just that it didn't crash - # inside an enricher branch. - assert isinstance(result, dict) - assert "error" in result diff --git a/tests/tools/test_send_message_plugin_extensibility.py b/tests/tools/test_send_message_plugin_extensibility.py new file mode 100644 index 0000000000..e1acf11636 --- /dev/null +++ b/tests/tools/test_send_message_plugin_extensibility.py @@ -0,0 +1,271 @@ +"""Cross-surface regressions for standalone platform send extensibility (#64900).""" + +from __future__ import annotations + +import asyncio +import json +import os +import subprocess +import sys +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest + +from gateway.config import Platform +from gateway.platform_registry import PlatformEntry, platform_registry +from tools.send_message_tool import resolve_send_target, send_message_tool + + +@pytest.fixture +def plugin_platform(): + name = "fmsg-ext-test" + seen: list[dict] = [] + + def parser(ref: str): + normalized = ref.strip().lower() + if normalized.startswith("@") and "@" in normalized[1:]: + return normalized, None + return None + + def validator(ref: str): + return not ref.endswith("@blocked.example") + + async def handler(args, chat_id, platform_name, pconfig): + seen.append({"args": dict(args), "chat_id": chat_id, "platform": platform_name}) + return {"success": True, "platform": platform_name, "chat_id": chat_id} + + entry = PlatformEntry( + name=name, + label="Fixture Message", + adapter_factory=lambda cfg: None, + check_fn=lambda: True, + parse_target_ref_fn=parser, + validate_target_ref_fn=validator, + send_message_handler=handler, + ) + platform_registry.register(entry) + try: + yield name, entry, seen + finally: + platform_registry.unregister(name) + + +def _config_for(name: str): + platform = Platform(name) + pconfig = SimpleNamespace(enabled=True, token=None, extra={}) + return platform, pconfig, SimpleNamespace( + platforms={platform: pconfig}, + get_home_channel=lambda _platform: None, + ) + + +def test_platform_parser_normalizes_and_validator_rejects(plugin_platform): + name, _entry, _seen = plugin_platform + assert resolve_send_target(name, " @Alice@Example.COM ") == ( + "@alice@example.com", + None, + None, + ) + chat_id, thread_id, error = resolve_send_target( + name, "@alice@blocked.example" + ) + assert chat_id is None + assert thread_id is None + assert error == f"Invalid target '@alice@blocked.example' on {name}" + + +def test_registered_plugin_rejects_unrecognized_opaque_target(plugin_platform): + name, _entry, seen = plugin_platform + + with patch("gateway.channel_directory.resolve_channel_name", return_value=None): + chat_id, thread_id, error = resolve_send_target( + name, "dm:opaque-recipient" + ) + + assert chat_id is None + assert thread_id is None + assert "plugin parser did not recognize it" in error + assert seen == [] + + +def test_plugin_parser_failures_are_diagnosable_without_leaking_exception(plugin_platform): + name, entry, _seen = plugin_platform + + def broken_parser(_ref): + raise RuntimeError("credential-shaped plugin detail") + + entry.parse_target_ref_fn = broken_parser + assert resolve_send_target(name, "recipient") == ( + None, + None, + f"Target parser failed for platform '{name}'", + ) + + entry.parse_target_ref_fn = lambda _ref: {"chat_id": "wrong-shape"} + assert resolve_send_target(name, "recipient") == ( + None, + None, + f"Target parser for platform '{name}' returned an invalid result", + ) + + +def test_plugin_validator_custom_diagnostic_blocks_delivery(plugin_platform): + name, entry, seen = plugin_platform + entry.validate_target_ref_fn = lambda _chat_id: "recipient is outside the allowlist" + + with patch("gateway.channel_directory.resolve_channel_name", return_value=None): + chat_id, thread_id, error = resolve_send_target(name, "@alice@example.com") + + assert chat_id is None + assert thread_id is None + assert error == ( + f"Invalid target '@alice@example.com' on {name}: " + "recipient is outside the allowlist" + ) + assert seen == [] + + +@pytest.mark.parametrize("async_handler", [False, True]) +def test_host_send_honors_sync_and_async_plugin_handlers(plugin_platform, async_handler): + name, entry, seen = plugin_platform + platform, pconfig, config = _config_for(name) + + if not async_handler: + def sync_handler(args, chat_id, platform_name, pconfig): + seen.append({"args": dict(args), "chat_id": chat_id, "platform": platform_name}) + return {"success": True, "platform": platform_name, "chat_id": chat_id} + entry.send_message_handler = sync_handler + + with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("gateway.mirror.mirror_to_session", return_value=True): + result = json.loads(send_message_tool({ + "target": f"{name}:@Alice@Example.COM", + "message": "hello", + "subject": "greeting", + })) + + assert result["success"] is True + assert result["platform"] == name + assert result["chat_id"] == "@alice@example.com" + assert seen[-1]["args"]["subject"] == "greeting" + + +def test_cli_and_cron_share_plugin_target_normalization(plugin_platform, monkeypatch, capsys): + from cron.scheduler import _resolve_single_delivery_target + from hermes_cli.send_cmd import cmd_send + + name, _entry, _seen = plugin_platform + _platform, _pconfig, config = _config_for(name) + args = SimpleNamespace( + list_targets=False, + to=f"{name}:@Alice@Example.COM", + message="hello", + file=None, + subject=None, + json=True, + quiet=False, + ) + + monkeypatch.setattr("hermes_cli.send_cmd._load_hermes_env", lambda: None) + with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("gateway.mirror.mirror_to_session", return_value=True), \ + pytest.raises(SystemExit) as exc: + cmd_send(args) + assert exc.value.code == 0 + assert json.loads(capsys.readouterr().out)["chat_id"] == "@alice@example.com" + + cron_target = _resolve_single_delivery_target( + {"name": "fixture"}, f"{name}:@Alice@Example.COM" + ) + assert cron_target == { + "platform": name, + "chat_id": "@alice@example.com", + "thread_id": None, + } + + +def test_send_message_remains_host_only(plugin_platform): + from tools.registry import registry + + assert registry.get_entry("send_message") is None + + +def test_force_reload_unregisters_profile_owned_platform(plugin_platform, monkeypatch): + from hermes_cli.plugins import PluginManager + + name, _entry, _seen = plugin_platform + manager = PluginManager() + manager._plugin_platform_names.add(name) + manager._discovered = True + monkeypatch.setattr(manager, "_discover_and_load_inner", lambda: None) + + manager.discover_and_load(force=True) + + assert platform_registry.get(name) is None + assert name not in manager._plugin_platform_names + + +def test_fresh_process_real_plugin_fixture_covers_host_send_and_cron(tmp_path): + """A standalone directory plugin is visible to host-driven send paths.""" + home = tmp_path / "home" + plugin = home / "plugins" / "fmsg-fixture" + plugin.mkdir(parents=True) + (plugin / "plugin.yaml").write_text( + "name: fmsg-fixture\nversion: 0.1.0\ndescription: fixture\nkind: platform\n" + ) + (home / "config.yaml").write_text("plugins:\n enabled:\n - fmsg-fixture\n") + (plugin / "__init__.py").write_text( + "async def _send(args, chat_id, platform_name, pconfig):\n" + " return {'success': True, 'platform': platform_name, 'chat_id': chat_id}\n" + "def _parse(ref):\n" + " ref = ref.strip().lower()\n" + " return (ref, None) if ref.startswith('@') and '@' in ref[1:] else None\n" + "def register(ctx):\n" + " ctx.register_platform(name='fmsg', label='Fmsg', " + "adapter_factory=lambda cfg: None, check_fn=lambda: True, " + "parse_target_ref_fn=_parse, send_message_handler=_send)\n" + ) + script = r''' +import json +from types import SimpleNamespace +from unittest.mock import patch +from hermes_cli.plugins import discover_plugins +from gateway.config import Platform +from tools.registry import registry +from tools.send_message_tool import send_message_tool + +discover_plugins() +platform = Platform("fmsg") +pconfig = SimpleNamespace(enabled=True, token=None, extra={}) +config = SimpleNamespace(platforms={platform: pconfig}, get_home_channel=lambda p: None) +with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("gateway.mirror.mirror_to_session", return_value=True): + host_send = json.loads(send_message_tool({"target": "fmsg:@Alice@Example.COM", + "message": "hello", "subject": "hi"})) +from cron.scheduler import _resolve_single_delivery_target +cron = _resolve_single_delivery_target({}, "fmsg:@Alice@Example.COM") +print(json.dumps({"host_send": host_send, "cron": cron, + "model_registered": registry.get_entry("send_message") is not None})) +''' + env = dict(os.environ) + env.update({ + "HERMES_HOME": str(home), + "HERMES_KANBAN_TASK": "fixture", + "PYTHONPATH": os.getcwd(), + }) + completed = subprocess.run( + [sys.executable, "-c", script], + cwd=os.getcwd(), + env=env, + text=True, + capture_output=True, + check=True, + ) + payload = json.loads(completed.stdout.strip().splitlines()[-1]) + assert payload["host_send"]["chat_id"] == "@alice@example.com" + assert payload["cron"]["chat_id"] == "@alice@example.com" + assert payload["model_registered"] is False diff --git a/tests/tools/test_send_message_target_parse.py b/tests/tools/test_send_message_target_parse.py index cd020ad0ce..07e6ac28fb 100644 --- a/tests/tools/test_send_message_target_parse.py +++ b/tests/tools/test_send_message_target_parse.py @@ -62,7 +62,6 @@ def test_send_message_routes_whatsapp_group_jid_without_home_fallback() -> None: thread_id=None, media_files=[], force_document=False, - args={"action": "send", "target": "whatsapp:120363408391911677@g.us", "message": "hello group"}, ) @@ -117,11 +116,10 @@ def test_resolved_opaque_plugin_target_uses_directory_id() -> None: thread_id=None, media_files=[], force_document=False, - args={"action": "send", "target": f"{platform_name}:Friendly name", "message": "hello"}, ) -def test_unresolved_opaque_plugin_target_passes_through_verbatim() -> None: +def test_unresolved_plugin_target_requires_explicit_parser() -> None: from gateway.platform_registry import PlatformEntry, platform_registry platform_name = "opaque-verbatim-test" @@ -171,18 +169,12 @@ def test_unresolved_opaque_plugin_target_passes_through_verbatim() -> None: finally: platform_registry.unregister(platform_name) - assert result["success"] is True + assert result == { + "error": f"Could not resolve 'dm:panyaozhen' on {platform_name}. " + "The plugin parser did not recognize it and no channel-directory entry matched." + } discover_mock.assert_called_once_with() - send_mock.assert_awaited_once_with( - platform, - pconfig, - "dm:panyaozhen", - "hello", - thread_id=None, - media_files=[], - force_document=False, - args={"action": "send", "target": f"{platform_name}:dm:panyaozhen", "message": "hello"}, - ) + send_mock.assert_not_awaited() def test_unresolved_builtin_target_keeps_directory_error() -> None: diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index 92d89027fb..cf93756121 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -11,7 +11,7 @@ import logging import os import re import time -from typing import Any, Awaitable, Callable + from agent.redact import redact_sensitive_text from agent.secret_scope import get_secret @@ -86,52 +86,11 @@ _CAPTIONABLE_EXTS = _IMAGE_EXTS | _VIDEO_EXTS | { _TELEGRAM_CAPTION_LIMIT = 1024 _DEFAULT_CAPTION_LIMIT = 4096 -# --------------------------------------------------------------------------- -# Plugin enricher registry for send_message -# --------------------------------------------------------------------------- +def prepare_send_message_platforms() -> None: + """Load enabled standalone plugins before tool schemas/cache keys are built.""" + from hermes_cli.plugins import discover_plugins -SendMessageEnricher = Callable[[dict, str, str, Any], Awaitable[dict] | dict] -"""Callable receiving (args, chat_id, platform_name, pconfig) -> dict result. - -May be sync or async — the dispatcher detects coroutine functions via -``inspect.iscoroutinefunction`` and awaits as needed. -""" - -_SEND_MESSAGE_ENRICHERS: dict[str, SendMessageEnricher] = {} -"""platform_name -> enricher handler.""" - -_SEND_MESSAGE_SCHEMA_FRAGMENTS: dict[str, dict] = {} -"""platform_name -> schema fragment dict (JSON Schema properties).""" - - -def register_send_message_enricher( - platform_name: str, - handler: SendMessageEnricher, - schema_fragment: dict | None = None, -) -> None: - """Register a platform-specific enricher for the send_message tool. - - Enrichers let plugin platforms extend send_message with custom target - parsing, schema fields, and send logic without modifying core code. - """ - _SEND_MESSAGE_ENRICHERS[platform_name] = handler - if schema_fragment: - _SEND_MESSAGE_SCHEMA_FRAGMENTS[platform_name] = schema_fragment - - -def get_send_message_schema() -> dict: - """Return a fresh copy of the send_message schema with plugin fragments merged. - - Fragments are assembled on demand so deferred plugin registration never - mutates the shared ``SEND_MESSAGE_SCHEMA`` dict. Callers that need the - current wire schema (e.g. tool-search builders, MCP catalogues) should use - this instead of reading ``SEND_MESSAGE_SCHEMA`` directly. - """ - import copy - schema = copy.deepcopy(SEND_MESSAGE_SCHEMA) - for fragment in _SEND_MESSAGE_SCHEMA_FRAGMENTS.values(): - schema["parameters"]["properties"].update(fragment) - return schema + discover_plugins() def _media_caption_split(text, media_files, *, max_caption_len): @@ -335,18 +294,13 @@ def _handle_react(args, remove=False): platform_name = parts[0].strip().lower() target_ref = parts[1].strip() if len(parts) > 1 else None chat_id = None + prepare_send_message_platforms() if target_ref: - chat_id, _thread_id, _ = _parse_target_ref(platform_name, target_ref) - if not chat_id: - try: - from gateway.channel_directory import resolve_channel_name - resolved = resolve_channel_name(platform_name, target_ref) - except Exception: - resolved = None - # Opaque platform-native ids (e.g. photon space GUIDs like - # 'any;-;+1555...') match no parser pattern and no directory - # entry — pass them through verbatim; the adapter validates. - chat_id = resolved or target_ref + chat_id, _thread_id, resolution_error = resolve_send_target( + platform_name, target_ref + ) + if resolution_error: + return tool_error(resolution_error) try: from gateway.config import Platform, load_gateway_config @@ -416,52 +370,13 @@ def _handle_send(args): chat_id = None thread_id = None + prepare_send_message_platforms() if target_ref: - chat_id, thread_id, is_explicit = _parse_target_ref(platform_name, target_ref) - else: - is_explicit = False - - # Resolve human-friendly channel names to numeric IDs - if target_ref and not is_explicit: - resolution_failed = False - try: - from gateway.channel_directory import resolve_channel_name - resolved = resolve_channel_name(platform_name, target_ref) - if resolved: - parsed_chat_id, parsed_thread_id, _ = _parse_target_ref( - platform_name, resolved - ) - # Directory entries are trusted platform IDs. Preserve an - # opaque plugin ID even when no built-in parser recognizes it. - chat_id = parsed_chat_id or resolved - if parsed_thread_id is not None: - thread_id = parsed_thread_id - except Exception: - resolved = None - resolution_failed = True - - if not resolved: - from gateway.config import Platform - from gateway.platform_registry import platform_registry - from hermes_cli.plugins import discover_plugins - - discover_plugins() - entry = platform_registry.get(platform_name) - is_builtin = platform_name in {member.value for member in Platform} - if entry is not None and entry.source == "plugin" and not is_builtin: - # Registered plugin platforms may use opaque IDs unknown to - # core. Their adapter owns final target validation. - chat_id = target_ref - elif resolution_failed: - return tool_error( - f"Could not resolve '{target_ref}' on {platform_name}. " - f"Try using a numeric channel ID instead." - ) - else: - return tool_error( - f"Could not resolve '{target_ref}' on {platform_name}. " - f"Use send_message(action='list') to see available targets." - ) + chat_id, thread_id, resolution_error = resolve_send_target( + platform_name, target_ref + ) + if resolution_error: + return tool_error(resolution_error) from tools.interrupt import is_interrupted if is_interrupted(): @@ -473,15 +388,18 @@ def _handle_send(args): except Exception as e: return json.dumps(_error(f"Failed to load gateway config: {e}")) - # Accept any platform name — built-in names resolve to their enum - # member, plugin platform names create dynamic members via _missing_(). + from gateway.platform_registry import platform_registry + + entry = platform_registry.get(platform_name) + is_builtin = platform_name in {member.value for member in Platform} + if not is_builtin and entry is None: + return tool_error( + f"Unknown or unregistered plugin platform: {platform_name}" + ) try: platform = Platform(platform_name) except (ValueError, KeyError): - if platform_name in _SEND_MESSAGE_ENRICHERS: - platform = platform_name - else: - return tool_error(f"Unknown platform: {platform_name}") + return tool_error(f"Unknown platform: {platform_name}") pconfig = config.platforms.get(platform) if not pconfig or not pconfig.enabled: @@ -562,16 +480,22 @@ def _handle_send(args): try: from model_tools import _run_async + send_kwargs = { + "thread_id": thread_id, + "media_files": media_files, + "force_document": force_document_attachments, + } + # Preserve the exact built-in call contract; only custom handlers need + # the complete typed request. + if entry is not None and entry.send_message_handler is not None: + send_kwargs["args"] = args result = _run_async( _send_to_platform( platform, pconfig, chat_id, cleaned_message, - thread_id=thread_id, - media_files=media_files, - force_document=force_document_attachments, - args=args, + **send_kwargs, ) ) if used_home_channel and isinstance(result, dict) and result.get("success"): @@ -694,28 +618,118 @@ def _parse_target_ref(platform_name: str, target_ref: str): if platform_name == "xmpp" and "@" in target_ref: return target_ref, None, True - # Plugin platforms may register a custom target parser via PlatformEntry. - # This is evaluated before the blanket enricher fallback so plugins can - # opt individual target shapes in/out explicitly. - try: - from gateway.platform_registry import platform_registry - entry = platform_registry.get(platform_name) - if entry is not None and entry.parse_target_ref_fn is not None: - parsed = entry.parse_target_ref_fn(target_ref) - if parsed is not None: - chat_id, thread_id = parsed - return chat_id, thread_id, True - except Exception: - pass - - # Plugin-enriched platforms (legacy blanket fallback) - if platform_name in _SEND_MESSAGE_ENRICHERS: - if target_ref: - return target_ref, None, True - return None, None, False return None, None, False +def resolve_send_target( + platform_name: str, target_ref: str +) -> tuple[str | None, str | None, str | None]: + """Resolve one send target identically for model/CLI/cron surfaces. + + Channel-directory IDs are trusted. Plugin platforms must explicitly parse + native target syntax; unresolved strings never receive an opaque fallback. + The optional validator is the final authority over parser-normalized and + directory-resolved IDs. + """ + from gateway.config import Platform + from gateway.platform_registry import platform_registry + + entry = platform_registry.get(platform_name) + + def _validate(candidate: str) -> str | None: + if entry is None or entry.validate_target_ref_fn is None: + return None + try: + verdict = entry.validate_target_ref_fn(candidate) + except Exception: + logger.debug( + "Plugin target validator failed for %s", platform_name, exc_info=True + ) + return f"Target validator failed for platform '{platform_name}'" + if verdict is True: + return None + if isinstance(verdict, str) and verdict: + return f"Invalid target '{target_ref}' on {platform_name}: {verdict}" + return f"Invalid target '{target_ref}' on {platform_name}" + + if entry is not None and entry.parse_target_ref_fn is not None: + try: + parsed = entry.parse_target_ref_fn(target_ref) + except Exception: + logger.debug( + "Plugin target parser failed for %s", platform_name, exc_info=True + ) + return None, None, f"Target parser failed for platform '{platform_name}'" + if parsed is not None: + if ( + not isinstance(parsed, tuple) + or len(parsed) != 2 + or not isinstance(parsed[0], str) + or not parsed[0] + or (parsed[1] is not None and not isinstance(parsed[1], str)) + ): + return ( + None, + None, + f"Target parser for platform '{platform_name}' returned an invalid result", + ) + parsed_chat_id, parsed_thread_id = parsed + error = _validate(parsed_chat_id) + return (None, None, error) if error else ( + parsed_chat_id, + parsed_thread_id, + None, + ) + + parsed_chat_id, parsed_thread_id, explicit = _parse_target_ref( + platform_name, target_ref + ) + if explicit and parsed_chat_id is not None: + error = _validate(parsed_chat_id) + return (None, None, error) if error else ( + parsed_chat_id, + parsed_thread_id, + None, + ) + + resolution_failed = False + try: + from gateway.channel_directory import resolve_channel_name + + resolved = resolve_channel_name(platform_name, target_ref) + except Exception: + resolved = None + resolution_failed = True + if resolved: + parsed_chat_id, parsed_thread_id, _ = _parse_target_ref( + platform_name, resolved + ) + chat_id = parsed_chat_id or resolved + error = _validate(chat_id) + return (None, None, error) if error else ( + chat_id, + parsed_thread_id, + None, + ) + + is_builtin = platform_name in {member.value for member in Platform} + if entry is None and not is_builtin: + return None, None, f"Unknown or unregistered plugin platform: {platform_name}" + if entry is not None and entry.source == "plugin" and not is_builtin: + return ( + None, + None, + f"Could not resolve '{target_ref}' on {platform_name}. " + "The plugin parser did not recognize it and no channel-directory entry matched.", + ) + hint = ( + "Try using a numeric channel ID instead." + if resolution_failed + else "Use send_message(action='list') to see available targets." + ) + return None, None, f"Could not resolve '{target_ref}' on {platform_name}. {hint}" + + def _describe_media_for_mirror(media_files): """Return a human-readable mirror summary when a message only contains media.""" if not media_files: @@ -1239,19 +1253,20 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, elif platform == Platform.YUANBAO: result = await _send_yuanbao(chat_id, chunk) else: - enricher = _SEND_MESSAGE_ENRICHERS.get(platform_name) - if enricher: - if args is None: - args = {} + from gateway.platform_registry import platform_registry + + entry = platform_registry.get(platform_name) + handler = entry.send_message_handler if entry is not None else None + if handler is not None: try: import inspect - if inspect.iscoroutinefunction(enricher): - result = await enricher(args, chat_id, platform_name, pconfig) - else: - result = enricher(args, chat_id, platform_name, pconfig) + + result = handler(args or {}, chat_id, platform_name, pconfig) + if inspect.isawaitable(result): + result = await result return result except Exception as e: - return {"error": f"Enricher send failed: {e}"} + return {"error": f"Plugin send_message handler failed: {e}"} # Plugin platform: route through the gateway's live adapter if # available, otherwise the plugin's standalone_sender_fn. result = await _send_via_adapter( diff --git a/website/docs/developer-guide/adding-platform-adapters.md b/website/docs/developer-guide/adding-platform-adapters.md index d2209c638f..2847ba974b 100644 --- a/website/docs/developer-guide/adding-platform-adapters.md +++ b/website/docs/developer-guide/adding-platform-adapters.md @@ -206,6 +206,64 @@ When you call `ctx.register_platform()`, the following integration points are ha | Token lock (multi-profile) | Use `acquire_scoped_lock()` in your `connect()` | | Orphaned config warning | Descriptive log when plugin is missing | +## Standalone send-path extensions + +A standalone platform can participate in host-driven outbound delivery through +direct `hermes send --to ...` and cron `deliver=platform:...` by declaring send +behavior on the same `PlatformEntry` created by `ctx.register_platform()`. +`send_message` is intentionally not an agent-callable model tool; plugins must +not register an equivalent model surface that lets the agent initiate outbound +messages on its own. + +```python +async def _send_request(args, chat_id, platform_name, pconfig): + # `args` contains the host-driven send request fields. + message_id = await client.send( + address=chat_id, + body=args["message"], + subject=args.get("subject"), + ) + return {"success": True, "platform": platform_name, + "chat_id": chat_id, "message_id": message_id} + + +def _parse_address(raw): + normalized = raw.strip().lower() + if normalized.startswith("@") and "@" in normalized[1:]: + return normalized, None # (chat_id, optional thread_id) + return None # continue to channel-directory resolution + + +def _validate_address(address): + # True accepts; False rejects; a string rejects with that diagnostic. + return True if address.endswith("@example.com") else "unsupported domain" + + +def register(ctx): + ctx.register_platform( + name="fmsg", + label="Fixture Message", + adapter_factory=lambda cfg: FmsgAdapter(cfg), + check_fn=check_requirements, + parse_target_ref_fn=_parse_address, + validate_target_ref_fn=_validate_address, + # May be a regular function or async def. Hermes awaits any awaitable + # result, including callable objects and functools.partial wrappers. + send_message_handler=_send_request, + # Prefer this lower-level hook when cron must send from a process + # without the live gateway. + standalone_sender_fn=_standalone_send, + ) +``` + +Target resolution is shared across all three outbound surfaces. Parser output +is normalized first and channel-directory IDs are trusted. A plugin parser must +explicitly accept native target syntax; unresolved strings are never passed +through opaquely. Unknown platforms and validator failures return a diagnostic +instead of silently attempting delivery. Plugin force-reload/profile +transitions unregister owned entries, so parsers and handlers cannot leak into +the next profile. + ## Env-Driven Auto-Configuration Most users set up a platform by dropping env vars into `~/.hermes/.env` rather than editing `config.yaml`. The `env_enablement_fn` hook lets your plugin pick those env vars up **before** the adapter is constructed, so `hermes gateway status`, `get_connected_platforms()`, and cron delivery see the correct state without instantiating the platform SDK.