patrickandClaude Sonnet 5.5
cdcf7f673a
Telegram: Plain-Text-Fallback bei Parse-Fehler, tag-sichere Kürzung auf 4096. Matrix: bytebewusste Kürzung. Downtime: lesender Abgleich nach dem POST, generische Fehlermeldung im Chat. Agent: Platzhalter für nicht ausgeführte Tool-Calls, Text im letzten Schritt erzwingen, Zeitreserve, Kompaktierung, Limit pro Antwort. CLAUDE.md: Fallbacks für Phase 6d. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
121 lines
4.3 KiB
Python
121 lines
4.3 KiB
Python
import os
|
|
|
|
from app import audit
|
|
from app.agent import AgentResult
|
|
from app.alerts import AlertRepository
|
|
from app.chat import MessageRef
|
|
from app.commands import CallbackHandler, CommandDeps, CommandHandler, IncomingMessage
|
|
from app.constants import (CHECKMK_DT_ENABLED, CHECKMK_DT_SECRET, CHECKMK_DT_USER,
|
|
CHECKMK_SITE, CHECKMK_URL, FLAG_ON, MATRIX_ALLOWLIST, TELEGRAM_ALLOWLIST)
|
|
from app.crypto import SecretBox
|
|
from app.db import Database
|
|
from app.checkmk import Downtime
|
|
from app.downtime import DowntimeError, DowntimeService, PendingStore
|
|
from app.downtime.client import iso_utc
|
|
from app.models import KIND_PROBLEM, Event
|
|
from app.mutes import MuteService
|
|
from app.settings import SettingsStore
|
|
|
|
TG_USER = "111"
|
|
TG_OTHER = "222"
|
|
MX_USER = "@alice:example.org"
|
|
CHAT = "chat1"
|
|
ROOM = "!room:example.org"
|
|
|
|
|
|
class FakeAgent:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
async def ask(self, question):
|
|
self.calls.append(("ask", question))
|
|
return AgentResult("Antwort")
|
|
|
|
async def explain(self, host, service):
|
|
self.calls.append(("explain", host, service))
|
|
return AgentResult("Erklärung")
|
|
|
|
|
|
class FakeDowntimeClient:
|
|
def __init__(self, fail=None):
|
|
self.calls, self.fail, self.closed = [], fail, False
|
|
|
|
async def create_service_downtime(self, host, service, start, end, comment):
|
|
if self.fail:
|
|
raise DowntimeError(self.fail)
|
|
self.calls.append((host, service, start, end, comment))
|
|
|
|
async def aclose(self):
|
|
self.closed = True
|
|
|
|
|
|
class FakeDowntimeReader:
|
|
def __init__(self, client):
|
|
self._client = client
|
|
self.closed = False
|
|
|
|
async def downtimes(self, host=None):
|
|
items = [Downtime(h, s, iso_utc(start), iso_utc(end), c, "auto") for h, s, start, end, c in self._client.calls]
|
|
if host:
|
|
items = [d for d in items if d.host.casefold() == host.casefold()]
|
|
return items
|
|
|
|
async def aclose(self):
|
|
self.closed = True
|
|
|
|
|
|
class Env:
|
|
def __init__(self, dt_enabled=True):
|
|
self.db = Database(":memory:")
|
|
self.store = SettingsStore(self.db, SecretBox(os.urandom(32)))
|
|
self.store.set(TELEGRAM_ALLOWLIST, TG_USER, "t")
|
|
self.store.set(MATRIX_ALLOWLIST, MX_USER, "t")
|
|
self.store.set(CHECKMK_URL, "https://cmk.example.de", "t")
|
|
self.store.set(CHECKMK_SITE, "prod", "t")
|
|
self.store.set(CHECKMK_DT_USER, "dtuser", "t")
|
|
self.store.set(CHECKMK_DT_SECRET, "dt-secret", "t", secret=True)
|
|
if dt_enabled:
|
|
self.store.set(CHECKMK_DT_ENABLED, FLAG_ON, "t")
|
|
self.alerts = AlertRepository(self.db)
|
|
self.mutes = MuteService(self.db)
|
|
self.client = FakeDowntimeClient()
|
|
self.clock = Clock()
|
|
self.pending = PendingStore(self.clock)
|
|
self.agent = FakeAgent()
|
|
self.downtime = DowntimeService(self.store, self.db, self.pending, self.allowlist,
|
|
lambda: self.client, clock=lambda: 1_800_000_000.0,
|
|
reader_factory=lambda: FakeDowntimeReader(self.client))
|
|
self.deps = CommandDeps(None, self.agent, mutes=self.mutes, lookup=self.alerts, downtime=self.downtime)
|
|
self.events = []
|
|
self.handler = CommandHandler(self.deps, self.allowlist, self.events.append)
|
|
self.callbacks = CallbackHandler(self.deps, self.allowlist, self.events.append)
|
|
|
|
def allowlist(self, backend):
|
|
return self.store.get(TELEGRAM_ALLOWLIST if backend == "telegram" else MATRIX_ALLOWLIST)
|
|
|
|
def alert(self, host="web01", service="CPU load", backend="telegram", target=CHAT, message_id="55"):
|
|
ids = self.alerts.insert_alerts([Event(1, host, service, KIND_PROBLEM, 2)], "g")
|
|
self.alerts.add_delivery(ids[0], MessageRef(backend, target, message_id))
|
|
return ids[0]
|
|
|
|
def audit_actions(self):
|
|
return [(e["user"], e["action"]) for e in reversed(audit.recent(self.db, 500))]
|
|
|
|
def downtime_audit(self):
|
|
return [a for _, a in self.audit_actions() if a.startswith("downtime.")]
|
|
|
|
|
|
class Clock:
|
|
now = 0.0
|
|
|
|
def __call__(self):
|
|
return self.now
|
|
|
|
|
|
def tg(text, sender=TG_USER, reply_to=None):
|
|
return IncomingMessage("telegram", sender, CHAT, text, reply_to)
|
|
|
|
|
|
def mx(text, sender=MX_USER, reply_to=None):
|
|
return IncomingMessage("matrix", sender, ROOM, text, reply_to)
|