fix(telegram): block machine payload AI fanout

apps/api/tests/test_telegram_group_chat_provider_router.py

@@ -4,11 +4,14 @@ from unittest.mock import AsyncMock
 
 import pytest
 
-from src.services import chat_manager as chat_module
 from src.services import ai_rate_limiter as limiter_module
+from src.services import chat_manager as chat_module
 from src.services import telegram_gateway as gateway_module
 from src.services.chat_manager import ChatManager
-from src.services.telegram_gateway import TelegramGateway
+from src.services.telegram_gateway import (
+    TelegramGateway,
+    _sre_group_message_requires_ai_response,
+)
 
 
 class _ReceiptBackedCostLimiter:
@@ -27,6 +30,66 @@ class _ReceiptBackedCostLimiter:
         }
 
 
+@pytest.mark.parametrize(
+    "message",
+    [
+        r"new-item -itemtype directory -force -path \wooo\agent99\tmp",
+        r"set-content -encoding ascii -path \wooo\agent99\tmp\fingerprint.txt",
+        r"add-content -encoding ascii -path \wooo\agent99\tmp\events.log",
+        "kubectl get pods -n awoooi-prod",
+        "2026-07-19T00:26:31+08:00 ERROR provider timeout",
+        '{"event":"agent99_runtime_receipt","status":"pending"}',
+        "```powershell\nRestart-Service alertmanager\n```",
+    ],
+)
+def test_unaddressed_machine_payload_does_not_request_ai(message: str) -> None:
+    assert _sre_group_message_requires_ai_response(message) is False
+
+
+@pytest.mark.parametrize(
+    "message,reply_to_message",
+    [
+        ("kubectl get pods 為什麼失敗？", None),
+        ("@OpenClawAwoooI_Bot kubectl get pods 失敗", None),
+        ("set-content 這行是什麼意思", None),
+        ("目前有哪些異常服務", None),
+        (
+            "Restart-Service alertmanager",
+            {"from": {"is_bot": True, "username": "OpenClawAwoooI_Bot"}},
+        ),
+    ],
+)
+def test_explicit_dialogue_about_machine_payload_still_requests_ai(
+    message: str,
+    reply_to_message: dict | None,
+) -> None:
+    assert (
+        _sre_group_message_requires_ai_response(message, reply_to_message)
+        is True
+    )
+
+
+@pytest.mark.asyncio
+async def test_group_machine_payload_stops_before_context_or_provider(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(gateway_module.settings, "HERMES_NL_ENABLED", False)
+
+    def _chat_manager_forbidden():
+        raise AssertionError("machine payload must not build context or call AI")
+
+    monkeypatch.setattr(chat_module, "get_chat_manager", _chat_manager_forbidden)
+    gateway = object.__new__(TelegramGateway)
+
+    await gateway._handle_group_message(
+        r"new-item -itemtype directory -force -path \wooo\agent99\tmp",
+        user_id=99,
+        username="operator",
+        chat_id=-100123,
+        message_id=455,
+    )
+
+
 @pytest.mark.asyncio
 async def test_nemotron_group_mention_uses_structured_provider_status_router(
     monkeypatch: pytest.MonkeyPatch,