diff --git a/plugins/mika_runtime/__init__.py b/plugins/mika_runtime/__init__.py index 22ed9a9..c0217d9 100644 --- a/plugins/mika_runtime/__init__.py +++ b/plugins/mika_runtime/__init__.py @@ -9,6 +9,7 @@ from plugins.mika_runtime.tools import ( TODOIST_API_SCHEMA, handle_calcom_api, handle_cronjob_create, + handle_gateway_platform_action_intercept, handle_integrations_status, handle_notion_api, handle_skill_create, @@ -60,3 +61,7 @@ def register(ctx) -> None: handler=handle_skill_create, emoji="🧩", ) + ctx.register_hook( + "pre_gateway_dispatch", + handle_gateway_platform_action_intercept, + ) diff --git a/plugins/mika_runtime/tools.py b/plugins/mika_runtime/tools.py index 2c619d3..83166e3 100644 --- a/plugins/mika_runtime/tools.py +++ b/plugins/mika_runtime/tools.py @@ -3,7 +3,12 @@ from __future__ import annotations import json +import asyncio +import logging import os +import re +import threading +import unicodedata from pathlib import Path from typing import Any, Dict, Iterable, Tuple from urllib import error, parse, request @@ -17,6 +22,7 @@ DEFAULT_NOTION_VERSION = "2022-06-28" DEFAULT_CALCOM_VERSION = "2026-02-25" MAX_RESPONSE_CHARS = 20000 USER_AGENT = "domco-mika-runtime/0.1" +logger = logging.getLogger(__name__) AGENT_INSTANCE_ENV_NAMES = ( "MIKA_AGENT_INSTANCE_ID", "HERMES_AGENT_INSTANCE_ID", @@ -27,6 +33,10 @@ INTERNAL_SECRET_ENV_NAMES = ( "HERMES_INTERNAL_FUNCTION_SECRET", "INTERNAL_FUNCTION_SECRET", ) +GATEWAY_ACTION_INTERCEPT_ENV_NAMES = ( + "MIKA_GATEWAY_ACTION_INTERCEPT", + "HERMES_GATEWAY_ACTION_INTERCEPT", +) INTEGRATIONS_STATUS_SCHEMA = { "name": "mika_integrations_status", @@ -165,6 +175,89 @@ def _json_response(payload: dict[str, Any]) -> str: return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) +def _is_truthy_env_default_true(*names: str) -> bool: + for name in names: + value = os.environ.get(name) + if value is None: + continue + lowered = str(value).strip().lower() + if lowered in {"0", "false", "no", "off", "disabled"}: + return False + if lowered in {"1", "true", "yes", "on", "enabled"}: + return True + return True + + +def _normalize_intent_text(value: Any) -> str: + text = str(value or "").strip().lower() + text = unicodedata.normalize("NFKD", text) + text = "".join(ch for ch in text if not unicodedata.combining(ch)) + return re.sub(r"\s+", " ", text) + + +_TEMPORAL_RE = re.compile( + r"(" + r"\bdaqui\s+(?:a\s+)?\d+\s*(?:min(?:uto)?s?|h(?:ora)?s?|dias?|semanas?)\b" + r"|\b(?:hoje|amanha|depois de amanha)\b" + r"|\b(?:todo|toda|todos|todas|diariamente|semanalmente|mensalmente|anualmente)\b" + r"|\b(?:segunda|terca|quarta|quinta|sexta|sabado|domingo)(?:-feira)?s?\b" + r"|\b(?:dia util|dias uteis|fim de semana)\b" + r"|\b(?:as|às)\s*\d{1,2}(?::\d{2}|h\d{0,2})?\b" + r"|\b\d{1,2}(?::\d{2}|h\d{0,2})\b" + r"|\b(?:cron|cronjob|automacao|automatizacao|lembrete|reminder|schedule)\b" + r")", + re.IGNORECASE, +) + +_CRON_INTENT_RE = re.compile( + r"(" + r"\bme\s+lemb(?:ra|re)\b" + r"|\blemb(?:ra|re)(?:-me)?\b" + r"|\bme\s+avis(?:a|e)\b" + r"|\bavis(?:a|e)(?:-me)?\b" + r"|\bagend(?:e|ar)\b" + r"|\bme\s+agenda\b" + r"|\bagenda\s+(?:um|uma|isso|para|pra)\b" + r"|\bprogram(?:a|e|ar)\b" + r"|\bautomatiz(?:a|e|ar)\b" + r"|\bcria(?:r)?\s+(?:um\s+|uma\s+)?(?:cronjob|lembrete|automacao)\b" + r"|\b(?:todo|toda|todos|todas)\b.*\b(?:manda|envia|me\s+manda|me\s+envia|resum[ao])\b" + r")", + re.IGNORECASE, +) + +_SKILL_INTENT_RE = re.compile( + r"(" + r"\bcria(?:r)?\s+(?:uma\s+|um\s+)?skill\b" + r"|\bcrie\s+(?:uma\s+|um\s+)?skill\b" + r"|\bnova\s+skill\b" + r"|\badicion(?:a|e|ar)\s+(?:uma\s+|um\s+)?skill\b" + r"|\bensina(?:r)?\b.*\b(?:skill|quando eu mandar|workflow|processo)\b" + r"|\bsalv(?:a|e|ar)\b.*\b(?:skill|workflow|processo)\b" + r")", + re.IGNORECASE, +) + + +def detect_gateway_platform_action(text: Any) -> str | None: + """Return a deterministic Mika platform action for explicit user intents.""" + normalized = _normalize_intent_text(text) + if not normalized: + return None + + # Plain slash commands should continue to Hermes/skills dispatch. + if normalized.startswith("/") and not _SKILL_INTENT_RE.search(normalized): + return None + + if _SKILL_INTENT_RE.search(normalized): + return "skill" + + if _CRON_INTENT_RE.search(normalized) and _TEMPORAL_RE.search(normalized): + return "cronjob" + + return None + + def _load_manifest() -> dict[str, Any]: if not INTEGRATIONS_MANIFEST_PATH.exists(): return {"integrations": []} @@ -751,3 +844,169 @@ def handle_skill_create(args: dict[str, Any], **_: Any) -> str: if synced_count is not None: msg += f" Skills ativas sincronizadas: {synced_count}." return msg + + +def _source_is_authorized_for_gateway_intercept(gateway: Any, source: Any) -> bool: + checker = getattr(gateway, "_is_user_authorized", None) + if not callable(checker): + return False + try: + return bool(checker(source)) + except Exception: + logger.debug("gateway auth check failed for Mika intercept", exc_info=True) + return False + + +def _gateway_thread_metadata(gateway: Any, event: Any) -> dict[str, Any] | None: + builder = getattr(gateway, "_thread_metadata_for_source", None) + if not callable(builder): + return None + try: + return builder(event.source, getattr(event, "message_id", None)) + except TypeError: + try: + return builder(event.source) + except Exception: + return None + except Exception: + return None + + +def _gateway_reply_anchor(gateway: Any, event: Any) -> str | None: + resolver = getattr(gateway, "_reply_anchor_for_event", None) + if callable(resolver): + try: + return resolver(event) + except Exception: + pass + message_id = getattr(event, "message_id", None) + return str(message_id) if message_id is not None else None + + +def _schedule_gateway_reply( + *, + loop: asyncio.AbstractEventLoop, + adapter: Any, + chat_id: str, + content: str, + reply_to: str | None, + metadata: dict[str, Any] | None, +) -> None: + async def _send() -> None: + await adapter.send( + chat_id=chat_id, + content=content, + reply_to=reply_to, + metadata=metadata, + ) + + future = asyncio.run_coroutine_threadsafe(_send(), loop) + + def _log_failure(done: Any) -> None: + try: + done.result() + except Exception: + logger.warning("failed to send Mika platform action reply", exc_info=True) + + future.add_done_callback(_log_failure) + + +def _run_gateway_platform_action( + *, + action: str, + natural_language_input: str, + loop: asyncio.AbstractEventLoop, + adapter: Any, + chat_id: str, + reply_to: str | None, + metadata: dict[str, Any] | None, +) -> None: + try: + if action == "cronjob": + content = handle_cronjob_create({ + "natural_language_input": natural_language_input, + }) + elif action == "skill": + content = handle_skill_create({ + "natural_language_input": natural_language_input, + }) + else: + content = "Erro: ação da plataforma não reconhecida." + except Exception as exc: + logger.warning("Mika platform action intercept failed", exc_info=True) + label = "automação" if action == "cronjob" else "skill" + content = f"Erro ao criar {label}: {exc}" + + _schedule_gateway_reply( + loop=loop, + adapter=adapter, + chat_id=chat_id, + content=content, + reply_to=reply_to, + metadata=metadata, + ) + + +def handle_gateway_platform_action_intercept( + *, + event: Any, + gateway: Any, + session_store: Any = None, +) -> dict[str, str] | None: + """Pre-gateway hook that makes Mika platform actions deterministic. + + The LLM can still use cronjob_create/skill_create as tools, but explicit + Telegram requests are also routed directly to Supabase before the model + runs. That keeps Supabase as the source of truth and avoids a natural + language "ok, vou lembrar" response that never persisted anything. + """ + del session_store + + if not _is_truthy_env_default_true(*GATEWAY_ACTION_INTERCEPT_ENV_NAMES): + return None + + if bool(getattr(event, "internal", False)): + return None + + source = getattr(event, "source", None) + if source is None or bool(getattr(source, "is_bot", False)): + return None + + text = str(getattr(event, "text", "") or "").strip() + action = detect_gateway_platform_action(text) + if not action: + return None + + if not _source_is_authorized_for_gateway_intercept(gateway, source): + return None + + adapters = getattr(gateway, "adapters", {}) or {} + adapter = adapters.get(getattr(source, "platform", None)) + chat_id = str(getattr(source, "chat_id", "") or "").strip() + if adapter is None or not chat_id: + logger.warning("Mika intercept could not find adapter/chat_id for action=%s", action) + return None + + try: + loop = asyncio.get_running_loop() + except RuntimeError: + logger.warning("Mika intercept has no running event loop") + return None + + thread = threading.Thread( + target=_run_gateway_platform_action, + kwargs={ + "action": action, + "natural_language_input": text, + "loop": loop, + "adapter": adapter, + "chat_id": chat_id, + "reply_to": _gateway_reply_anchor(gateway, event), + "metadata": _gateway_thread_metadata(gateway, event), + }, + name=f"mika-platform-action-{action}", + daemon=True, + ) + thread.start() + + return {"action": "skip", "reason": f"mika_{action}_handled"} diff --git a/tests/test_mika_runtime_tools.py b/tests/test_mika_runtime_tools.py index db828d4..c4f0299 100644 --- a/tests/test_mika_runtime_tools.py +++ b/tests/test_mika_runtime_tools.py @@ -6,6 +6,7 @@ import sys import tempfile import types import unittest +import asyncio from unittest import mock @@ -34,6 +35,28 @@ class FakeResponse: class MikaRuntimePlatformActionTests(unittest.TestCase): + def test_detect_gateway_platform_action_for_cronjob_and_skill(self) -> None: + self.assertEqual( + tools.detect_gateway_platform_action( + "Mika, me lembra daqui 3 minutos de validar salvamento na plataforma" + ), + "cronjob", + ) + self.assertEqual( + tools.detect_gateway_platform_action( + "Mika, todo dia às 9h me manda um resumo da minha agenda" + ), + "cronjob", + ) + self.assertEqual( + tools.detect_gateway_platform_action( + "Mika, cria uma skill chamada teste-go-live que responda skill ativa" + ), + "skill", + ) + self.assertIsNone(tools.detect_gateway_platform_action("/teste_go_live")) + self.assertIsNone(tools.detect_gateway_platform_action("Qual é minha agenda hoje?")) + def test_cronjob_create_posts_platform_contract(self) -> None: captured = {} @@ -135,5 +158,100 @@ class MikaRuntimePlatformActionTests(unittest.TestCase): self.assertIn("endpoint da plataforma não configurado", result) +class FakeGatewaySendResult: + success = True + message_id = "sent-1" + + +class FakeGatewayAdapter: + def __init__(self) -> None: + self.sent: list[dict[str, object]] = [] + + async def send(self, chat_id, content, reply_to=None, metadata=None): + self.sent.append({ + "chat_id": chat_id, + "content": content, + "reply_to": reply_to, + "metadata": metadata, + }) + return FakeGatewaySendResult() + + +class FakeGateway: + def __init__(self, adapter: FakeGatewayAdapter) -> None: + self.adapters = {"telegram": adapter} + + def _is_user_authorized(self, source) -> bool: + return True + + def _thread_metadata_for_source(self, source, reply_to_message_id=None): + return {"thread_id": "topic-1", "reply": reply_to_message_id} + + def _reply_anchor_for_event(self, event): + return event.message_id + + +class MikaGatewayInterceptTests(unittest.IsolatedAsyncioTestCase): + async def test_gateway_intercept_handles_cronjob_without_llm_dispatch(self) -> None: + adapter = FakeGatewayAdapter() + gateway = FakeGateway(adapter) + event = types.SimpleNamespace( + text="Mika, me lembra daqui 3 minutos de validar salvamento", + message_id="msg-1", + internal=False, + source=types.SimpleNamespace( + platform="telegram", + chat_id="chat-1", + is_bot=False, + ), + ) + + with mock.patch.object( + tools, + "handle_cronjob_create", + return_value="Automação criada e sincronizada: daqui 3 minutos.", + ) as create: + result = tools.handle_gateway_platform_action_intercept( + event=event, + gateway=gateway, + session_store=None, + ) + for _ in range(100): + if adapter.sent: + break + await asyncio.sleep(0.01) + + self.assertEqual(result, {"action": "skip", "reason": "mika_cronjob_handled"}) + create.assert_called_once_with({ + "natural_language_input": "Mika, me lembra daqui 3 minutos de validar salvamento", + }) + self.assertEqual(adapter.sent[0]["chat_id"], "chat-1") + self.assertEqual(adapter.sent[0]["reply_to"], "msg-1") + self.assertIn("Automação criada", str(adapter.sent[0]["content"])) + + async def test_gateway_intercept_ignores_non_platform_action(self) -> None: + adapter = FakeGatewayAdapter() + gateway = FakeGateway(adapter) + event = types.SimpleNamespace( + text="/teste_go_live", + message_id="msg-2", + internal=False, + source=types.SimpleNamespace( + platform="telegram", + chat_id="chat-1", + is_bot=False, + ), + ) + + result = tools.handle_gateway_platform_action_intercept( + event=event, + gateway=gateway, + session_store=None, + ) + + self.assertIsNone(result) + self.assertEqual(adapter.sent, []) + + if __name__ == "__main__": unittest.main()