PostgreSQL-Änderungsbenachrichtigung

version: "1.0"

name: change-analyzer
type: llm
model: connic/gpt-5.6-luna
description: "Analyzes PostgreSQL LISTEN/NOTIFY events and decides who to alert"

system_prompt: |
  You process database change notifications from PostgreSQL LISTEN/NOTIFY.
  The payload is JSON from a database trigger, typically containing table,
  operation, old row, new row, and a _postgres metadata object.

  Use _postgres.channel to distinguish notification sources if multiple
  channels are configured.

  Your responsibilities:
  1. Determine the significance of the change:
     - High: order status changed to "cancelled" or "refunded", user deleted,
       payment amount modified, permission escalation
     - Medium: order status changed, user profile updated, new customer created
     - Low: log entries, status timestamps updated, minor field changes
  2. Decide which teams/channels should be notified:
     - "ops-team": order and payment changes
     - "security-team": permission changes, user deletions
     - "sales-team": new customers, large orders
     - "none": low-significance changes (no notification)
  3. For significant changes, dispatch ONE notification by calling trigger_agent
     with agent_name "notification-dispatcher" and a non-empty payload (a JSON
     string) shaped exactly like:
       {"channel": "<the single most relevant team>", "title": "<short title>",
        "message": "<what changed and why it matters>", "severity": "<info|warning|critical>"}
     Fill every field from the actual change. Dispatch once, not per channel.
  4. Log high-significance events using notification_tools.log_audit_event for audit trailing.

  Tool argument shapes: classify_change takes changed_fields as a list of
  column-name strings (e.g. ["role", "email"]); log_audit_event takes
  affected_ids as a list of record-id strings (e.g. ["u_42"]). Pass lists, not
  the row object.

  Ignore changes to internal system columns (updated_at, version, etc.) unless
  accompanied by meaningful data changes.

temperature: 0.2
output_schema: change-analysis
timeout: 110

tools:
  - trigger_agent
  - notification_tools.classify_change
  - notification_tools.log_audit_event

version: "1.0"

name: notification-dispatcher
type: tool
description: "Formats and dispatches notifications to the appropriate channel"
tool_name: notification_tools.dispatch

"""Notification tools for database change events."""

from datetime import datetime, timezone
from typing import Any
from connic.tools import db_insert


async def log_audit_event(
    table: str,
    operation: str,
    significance: str,
    description: str,
    affected_ids: list[str] | None = None,
    channels_notified: list[str] | None = None,
) -> dict:
    """Log a high-significance database change event for audit trailing.

    Args:
        table: Database table where the change occurred.
        operation: INSERT, UPDATE, or DELETE.
        significance: high, medium, or low.
        description: Human-readable summary of what changed and why it matters.
        affected_ids: IDs of affected records.
        channels_notified: Which notification channels were alerted.

    Returns:
        The inserted audit event document.
    """
    result = await db_insert("audit_events", {
        "table": table,
        "operation": operation,
        "significance": significance,
        "description": description,
        "affected_ids": affected_ids or [],
        "channels_notified": channels_notified or [],
        "recorded_at": datetime.now(timezone.utc).isoformat(),
    })
    return result["inserted"][0] if result.get("inserted") else result


def classify_change(
    table: str,
    operation: str,
    changed_fields: list[str] | None = None,
) -> dict[str, Any]:
    """Classify the significance of a database change.

    Provides a rule-based initial classification that the LLM can
    override with more nuanced analysis.

    Args:
        table: Database table name
        operation: INSERT, UPDATE, or DELETE
        changed_fields: List of column names that were modified

    Returns:
        Dict with significance level and suggested channels
    """
    high_sig_tables = {"orders", "payments", "users", "permissions", "subscriptions"}
    sensitive_fields = {"status", "amount", "role", "permissions", "email", "password_hash"}

    significance = "low"
    channels = []

    if table in high_sig_tables:
        significance = "medium"
        if operation == "DELETE":
            significance = "high"
            channels.append("security-team")
        elif changed_fields:
            if any(f in sensitive_fields for f in changed_fields):
                significance = "high"

    if table in ("orders", "payments"):
        channels.append("ops-team")
    if table == "users" and operation == "INSERT":
        channels.append("sales-team")

    if not channels and significance != "low":
        channels.append("ops-team")

    return {
        "significance": significance,
        "suggested_channels": channels,
        "table": table,
        "operation": operation,
    }


def dispatch(
    payload: dict[str, Any] | None = None,
    context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Format a notification payload for webhook delivery.

    This runs as a tool agent (notification-dispatcher), so it receives the
    triggering agent's payload directly. The outbound HTTP webhook connector
    delivers the result to the configured endpoint (Slack, Teams, PagerDuty…).

    Args:
        payload: The notification to send, with keys: channel (ops-team,
            security-team, sales-team), title, message, and optional severity
            (info, warning, critical) and metadata (table, operation, ids).
        context: Auto-injected run context.

    Returns:
        Structured notification ready for outbound webhook delivery
    """
    payload = payload or {}
    return {
        "channel": payload.get("channel", "ops-team"),
        "title": payload.get("title", "Database change"),
        "message": payload.get("message", ""),
        "severity": payload.get("severity", "info"),
        "metadata": payload.get("metadata", {}),
    }

{
  "type": "object",
  "description": "Database change analysis result",
  "properties": {
    "table": {
      "type": "string",
      "description": "Source database table"
    },
    "operation": {
      "type": "string",
      "enum": ["INSERT", "UPDATE", "DELETE"]
    },
    "significance": {
      "type": "string",
      "enum": ["low", "medium", "high"],
      "description": "How significant this change is"
    },
    "notify_channels": {
      "type": "array",
      "items": {"type": "string"},
      "description": "Channels to notify (ops-team, security-team, sales-team, none)"
    },
    "summary": {
      "type": "string",
      "description": "Human-readable summary of what changed and why it matters"
    },
    "action_taken": {
      "type": "string",
      "enum": ["notified", "stored", "ignored"],
      "description": "What action was taken"
    }
  },
  "required": ["table", "operation", "significance", "action_taken"]
}

version: "1.0"

# The custom classification and audit tools are mocked so nothing is written to
# the database. trigger_agent runs for real (it cannot be mocked), so the
# dispatcher is exercised on significant changes.
defaults:
  runs: 1
  timeout_s: 115
  mocks: notification_mocks

tests:
  - name: high_significance_update
    payload: '{"table": "users", "operation": "DELETE", "old": {"id": "u_42", "email": "a@b.com", "role": "admin"}, "new": null, "_postgres": {"channel": "data_changes"}}'
    expected_result: output.significance != ""
    expected_tool_calls:
      - notification_tools.classify_change
      - notification_tools.log_audit_event

  - name: low_significance_change
    payload: '{"table": "logs", "operation": "INSERT", "old": null, "new": {"id": "log_1", "level": "info", "message": "request handled"}, "_postgres": {"channel": "data_changes"}}'
    expected_result: output.significance != ""
    expected_tool_calls:
      - notification_tools.classify_change

version: "1.0"

# notification-dispatcher is a tool agent that calls dispatch(channel, title,
# message, severity, metadata). The tool is pure, so it runs for real and the
# output is the webhook-ready notification.
defaults:
  runs: 1
  timeout_s: 60

tests:
  - name: formats_ops_notification
    payload: '{"channel": "ops-team", "title": "Order cancelled", "message": "Order o_99 was cancelled", "severity": "warning", "metadata": {"table": "orders", "operation": "UPDATE"}}'
    expected_result: output.channel == "ops-team"