feat(telegram): create controlled action work items

2026-07-22 13:06:13 +08:00
parent 85853402ac
commit fb4f8e3d0c
5 changed files with 851 additions and 10 deletions
--- a/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py
+++ b/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py
@@ -155,9 +155,9 @@ def test_loader_returns_fixed_architecture_provider_order_and_agent99_bridge() -
        "active_or_completed_commitments": 68,
        "by_status": {
            "analysis_or_governance_complete": 6,
-            "source_implemented_runtime_pending": 27,
+            "source_implemented_runtime_pending": 28,
            "in_progress": 32,
-            "not_started_or_no_current_evidence": 3,
+            "not_started_or_no_current_evidence": 2,
            "superseded": 2,
        },
        "product_runtime_closed_commitments": 0,
@@ -170,6 +170,12 @@ def test_loader_returns_fixed_architecture_provider_order_and_agent99_bridge() -
    assert commitments["AIA-CONV-029"]["status"] == (
        "source_implemented_runtime_pending"
    )
+    assert commitments["AIA-CONV-030"]["status"] == (
+        "source_implemented_runtime_pending"
+    )
+    assert "fingerprint-deduplicated" in " ".join(
+        commitments["AIA-CONV-030"]["source_evidence"]
+    )
    assert "Host112" in commitments["AIA-CONV-049"]["title"]
    assert commitments["AIA-CONV-049"]["status"] == (
        "source_implemented_runtime_pending"
--- a/apps/api/tests/test_telegram_controlled_action_work_item.py
+++ b/apps/api/tests/test_telegram_controlled_action_work_item.py
@@ -0,0 +1,309 @@
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import UUID
+
+import pytest
+
+from src.services import chat_manager as chat_module
+from src.services import telegram_controlled_action_work_item as work_item_module
+from src.services import telegram_gateway as gateway_module
+from src.services.chat_manager import ChatManager
+from src.services.telegram_controlled_action_work_item import (
+    TelegramControlledActionWorkItemService,
+    persist_controlled_action_work_item,
+)
+from src.services.telegram_gateway import TelegramGateway
+
+
+class _Registry:
+    def resolve_identity(self, name: str) -> dict:
+        if name == "alertmanager":
+            return {
+                "resolution_status": "resolved",
+                "canonical_id": "service:alertmanager",
+                "asset_domain": "docker_container",
+                "executor": "host_ansible_executor",
+                "verifier": "container_runtime_independent_verifier",
+                "controlled_apply_allowed": False,
+                "drift_work_item_id": None,
+            }
+        return {
+            "resolution_status": "asset_identity_unresolved",
+            "normalized_identity": name,
+            "canonical_id": None,
+            "asset_domain": "unknown",
+            "executor": None,
+            "verifier": None,
+            "controlled_apply_allowed": False,
+            "drift_work_item_id": "AIA-ASSET-DRIFT-UNKNOWN001",
+        }
+
+
+class _MemoryStore:
+    def __init__(self) -> None:
+        self.records: dict[str, tuple[str, dict]] = {}
+
+    async def __call__(self, record: dict) -> dict:
+        fingerprint = record["fingerprint"]
+        existing = self.records.get(fingerprint)
+        if existing is not None:
+            return {"receipt_id": existing[0], "created": False}
+        receipt_id = f"work-item-receipt-{len(self.records) + 1}"
+        self.records[fingerprint] = (receipt_id, record)
+        return {"receipt_id": receipt_id, "created": True}
+
+
+def _inbound_receipt(*, suffix: str = "1") -> dict[str, str]:
+    return {
+        "event_id": f"event-{suffix}",
+        "trace_id": f"telegram-message:event-{suffix}",
+        "run_id": "9e39daf8-83e4-5fe8-a3d2-aebaa969a0d1",
+    }
+
+
+def _chat_identity(*, verified: bool = True) -> dict[str, str]:
+    return {
+        "status": "verified" if verified else "unverified",
+        "room": "awoooi_sre_war_room",
+    }
+
+
+@pytest.mark.asyncio
+async def test_duplicate_controlled_action_reuses_one_durable_work_item() -> None:
+    store = _MemoryStore()
+    service = TelegramControlledActionWorkItemService(
+        registry=_Registry(),
+        persist_work_item=store,
+    )
+
+    first = await service.create_from_verified_message(
+        message="請重啟 alertmanager",
+        intent="controlled_action_request",
+        inbound_receipt=_inbound_receipt(suffix="1"),
+        canonical_chat_identity=_chat_identity(),
+    )
+    duplicate = await service.create_from_verified_message(
+        message="請重啟 alertmanager",
+        intent="controlled_action_request",
+        inbound_receipt=_inbound_receipt(suffix="2"),
+        canonical_chat_identity=_chat_identity(),
+    )
+
+    assert first.accepted is True
+    assert first.created is True
+    assert duplicate.accepted is True
+    assert duplicate.created is False
+    assert duplicate.deduplicated is True
+    assert duplicate.work_item_id == first.work_item_id
+    assert duplicate.receipt_id == first.receipt_id
+    assert len(store.records) == 1
+    record = next(iter(store.records.values()))[1]
+    assert record["kind"] == "codex_development_work_item"
+    assert record["canonical_asset_identity"]["canonical_id"] == (
+        "service:alertmanager"
+    )
+    assert record["raw_message_recorded"] is False
+    assert set(record["policy"].values()) == {False}
+
+
+@pytest.mark.asyncio
+async def test_unverified_chat_identity_creates_no_work_item() -> None:
+    store = _MemoryStore()
+    service = TelegramControlledActionWorkItemService(
+        registry=_Registry(),
+        persist_work_item=store,
+    )
+
+    result = await service.create_from_verified_message(
+        message="請重啟 alertmanager",
+        intent="controlled_action_request",
+        inbound_receipt=_inbound_receipt(),
+        canonical_chat_identity=_chat_identity(verified=False),
+    )
+
+    assert result.accepted is False
+    assert result.status == "identity_or_ingress_unverified"
+    assert result.work_item_id == "none"
+    assert store.records == {}
+
+
+@pytest.mark.asyncio
+async def test_unknown_asset_creates_only_fail_closed_drift_item() -> None:
+    store = _MemoryStore()
+    service = TelegramControlledActionWorkItemService(
+        registry=_Registry(),
+        persist_work_item=store,
+    )
+
+    result = await service.create_from_verified_message(
+        message="請重啟 mystery-service",
+        intent="controlled_action_request",
+        inbound_receipt=_inbound_receipt(),
+        canonical_chat_identity=_chat_identity(),
+    )
+
+    assert result.accepted is True
+    assert result.status == "asset_identity_unresolved"
+    assert result.candidate_created is False
+    assert result.work_item_id == "AIA-ASSET-DRIFT-UNKNOWN001"
+    record = next(iter(store.records.values()))[1]
+    assert record["kind"] == "asset_identity_drift_work_item"
+    assert record["canonical_asset_identity"]["executor"] is None
+    assert record["canonical_asset_identity"]["verifier"] is None
+    assert record["policy"]["fallback_auto_allowed"] is False
+    assert record["policy"]["runtime_mutation_allowed"] is False
+
+
+@pytest.mark.asyncio
+async def test_default_store_uses_idempotent_internal_command_event(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    event_id = UUID("bfa2fc3e-3fcb-4a2f-adf9-3c8f0ec7844c")
+    inserted = SimpleNamespace(fetchone=lambda: (event_id,))
+    db = SimpleNamespace(execute=AsyncMock(return_value=inserted))
+
+    @asynccontextmanager
+    async def _db_context(project_id: str):
+        assert project_id == "awoooi"
+        yield db
+
+    from src.db import base as base_module
+
+    monkeypatch.setattr(base_module, "get_db_context", _db_context)
+    record = {
+        "kind": "codex_development_work_item",
+        "status": "candidate_recorded",
+        "work_item_id": "CODEX-SRE-1234",
+        "fingerprint": "a" * 43,
+        "run_id": "9e39daf8-83e4-5fe8-a3d2-aebaa969a0d1",
+        "policy": {"runtime_mutation_allowed": False},
+        "raw_message_recorded": False,
+    }
+
+    receipt = await persist_controlled_action_work_item(record)
+
+    assert receipt == {"receipt_id": str(event_id), "created": True}
+    sql = str(db.execute.await_args.args[0])
+    assert "'internal'" in sql
+    assert "'command'" in sql
+    assert "ON CONFLICT" in sql
+    params = db.execute.await_args.args[1]
+    assert len(params["content_hash"]) == 64
+    assert "請重啟" not in params["source_envelope"]
+
+
+@pytest.mark.asyncio
+async def test_gateway_controlled_action_stops_before_rag_provider_or_runtime(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(gateway_module.settings, "HERMES_NL_ENABLED", False)
+    manager = ChatManager()
+    monkeypatch.setattr(chat_module, "get_chat_manager", lambda: manager)
+    monkeypatch.setattr(
+        manager,
+        "get_rag_context",
+        AsyncMock(side_effect=AssertionError("controlled action must not read RAG")),
+    )
+    monkeypatch.setattr(
+        manager,
+        "get_system_context",
+        AsyncMock(
+            side_effect=AssertionError("controlled action must not read runtime")
+        ),
+    )
+    call_openclaw = AsyncMock(
+        side_effect=AssertionError("controlled action must not call provider")
+    )
+    call_nemotron = AsyncMock(
+        side_effect=AssertionError("controlled action must not call provider")
+    )
+    monkeypatch.setattr(manager, "_call_openclaw", call_openclaw)
+    monkeypatch.setattr(manager, "_call_nemotron", call_nemotron)
+
+    store = _MemoryStore()
+    service = TelegramControlledActionWorkItemService(
+        registry=_Registry(),
+        persist_work_item=store,
+    )
+    monkeypatch.setattr(
+        work_item_module,
+        "get_telegram_controlled_action_work_item_service",
+        lambda: service,
+    )
+
+    gateway = object.__new__(TelegramGateway)
+    monkeypatch.setattr(
+        gateway,
+        "mirror_sre_group_message_received",
+        AsyncMock(return_value=_inbound_receipt()),
+    )
+    send_as_openclaw = AsyncMock(return_value={"ok": True})
+    monkeypatch.setattr(gateway, "send_as_openclaw", send_as_openclaw)
+
+    await gateway._handle_group_message(
+        "@OpenClawAwoooI_Bot 請重啟 alertmanager",
+        user_id=99,
+        username="operator",
+        chat_id=-100123,
+        message_id=459,
+        update_id=9003,
+    )
+
+    manager.get_rag_context.assert_not_awaited()
+    manager.get_system_context.assert_not_awaited()
+    call_openclaw.assert_not_awaited()
+    call_nemotron.assert_not_awaited()
+    send_as_openclaw.assert_awaited_once()
+    outbound = send_as_openclaw.await_args.kwargs["text"]
+    assert "work_item=<code>CODEX-SRE-" in outbound
+    assert "status=<code>candidate_recorded</code>" in outbound
+    assert "provider=none" in outbound
+    assert "paid_provider=false" in outbound
+    assert "executor_invoked=false" in outbound
+    assert "Agent99=false" in outbound
+    assert "runtime_mutation=false" in outbound
+    assert len(store.records) == 1
+
+
+@pytest.mark.asyncio
+async def test_sre_group_ingress_uses_channel_hub_text_contract(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    @asynccontextmanager
+    async def _db_context(project_id: str):
+        assert project_id == "awoooi"
+        yield object()
+
+    from src.db import base as base_module
+    from src.services import channel_hub
+
+    event_id = UUID("5ca599f7-9b18-47ea-a3e6-33ee71d067a1")
+    mirror = AsyncMock(return_value=event_id)
+    monkeypatch.setattr(base_module, "get_db_context", _db_context)
+    monkeypatch.setattr(channel_hub, "mirror_inbound_event", mirror)
+    monkeypatch.setattr(gateway_module.settings, "SRE_GROUP_CHAT_ID", "-100123")
+
+    gateway = object.__new__(TelegramGateway)
+    receipt = await gateway.mirror_sre_group_message_received(
+        update_id=9004,
+        text="請重啟 alertmanager",
+        user_id=99,
+        username="operator",
+        chat_id=-100123,
+        message_id=460,
+        intent="controlled_action_request",
+    )
+
+    assert receipt is not None
+    assert receipt["event_id"] == str(event_id)
+    assert mirror.await_args.kwargs["content_type"] == "text"
+    envelope = mirror.await_args.kwargs["source_envelope"]
+    assert envelope["telegram_sre_group_message"]["intent"] == (
+        "controlled_action_request"
+    )
+    assert (
+        envelope["telegram_sre_group_message"]["executor_invocation_allowed"] is False
+    )