Skip to content
← Back to Blog
ml-systemsobservabilitypythonanomaly-detectionincident-response

Notification Adjudication in My Ops Intelligence Agent: Canonical Events, Cheap Arbitration, and a Sender That Refuses to Spam

Daniel Anthony Romitelli Jr. · July 28, 2026

I didn’t want “more alerts.” I wanted fewer, better ones.

When I added the Ops Intelligence Agent to a recruitment platform Operations Dashboard repo, the raw ingredients for noise were already sitting there: live operational telemetry over SignalR, a dashboard that makes it easy to stare at problems, and a set of services that can fail in correlated ways.

So the feature I built, and the one I’m happiest with, is notification adjudication. It’s a pipeline that takes heterogeneous events, maps them into a canonical shape, scores them quickly, arbitrates overlaps, and then dispatches to Microsoft Teams with the kind of defensive engineering that stops humans from muting the channel.

Think of incoming telemetry as a busy kitchen pass. Tickets arrive from every station at once, and the job is to consolidate what’s actually one dish, prioritize what’s burning, and send a single actionable callout, rather than forwarding every ticket to the head chef.

Adjudication is a product in its own right

A naive implementation is:

  • every “bad-looking” event triggers a Teams message
  • every detector gets its own message format
  • retries are “try again later” and hope the channel survives

That fails for two reasons:

  1. Heterogeneous inputs explode your surface area. Give every event type its own notification logic and you don’t have a system. You have a pile of special cases.

  2. Correlated failures create alert storms. If an indexer fails, you often see multiple symptoms near-simultaneously: failed run, degraded status, throttling, and so on. Humans read that as spam.

So I treat notifications as the output of a small adjudication pipeline:

  • normalize everything into a canonical event schema
  • do fast, local scoring (thresholds plus heuristics) to bucket confidence and severity
  • collapse overlaps via cheap arbitration (the “multi-detector consensus” step)
  • dispatch with guardrails (idempotency, backoff, rate limiting)

The repo has the beginnings of the agent structure to support this: an event processor that “consumes events from SignalR and processes them for anomaly detection” (the event processor module), analysis tooling with an explicit anomaly_threshold = 2.5 standard deviations (the analysis tools module), and a Teams notifier built as an async HTTP client with a 30s timeout and a webhook URL pulled from TEAMS_DEVOPS_WEBHOOK_URL (the teams notifier module).

How it works under the hood

At a high level, the agent breaks into these parts:

  • an ingestion/processing layer (EventProcessor) that receives events and emits “insights” and “anomalies” via callbacks
  • analysis tooling (AnalysisTools) that keeps in-memory metric history and an error history, and is configured with a standard-deviation-based anomaly threshold
  • a notification API surface that can test whether Teams is configured (/test returns 503 if no webhook URL)
  • a Teams sender (TeamsNotifier) that owns the webhook and the HTTP client lifecycle

Here’s how those pieces line up conceptually, with adjudication as the spine:

The non-obvious design choice is that I’m not optimizing for perfect classification. I’m optimizing for:

  • low latency for first-seen issues
  • high resistance to duplicates
  • clear operator-facing messages

Stage 1: telemetry normalization (canonical schema)

The repo shows multiple sources and shapes of operational data: dashboard panels, AI Search metrics, worker health, and more. The AI Search detail view, for one, defines a SearchMetrics structure with fields like failed_indexer_runs_24h, throttled_queries_24h, and service_status (the AISearchDetailView module).

To adjudicate across sources, I need a canonical event representation.

There is no explicit canonical event dataclass or model for the agent in the repo. What follows is a runnable Python module that defines a conservative canonical shape using only standard library types, with comments marking where the real system would map known fields like failed_indexer_runs_24h.

# canonical_event.py
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict, Optional


@dataclass(frozen=True)
class CanonicalEvent:
    """Canonical event shape used for notification adjudication.

    Note: The actual canonical schema isn't published here.
    This file defines a minimal, conservative shape suitable for normalization.

    Real mappings in this codebase would likely normalize fields seen in:
    - src/components/details/AISearchDetailView.tsx (SearchMetrics)
    - ops-intelligence-agent/agent/event_processor.py (incoming event stream)
    """

    timestamp: datetime
    source: str
    kind: str
    subject: str
    attributes: Dict[str, Any]
    raw: Optional[Dict[str, Any]] = None


if __name__ == "__main__":
    # Example instance; real values would come from SignalR-consumed events.
    evt = CanonicalEvent(
        timestamp=datetime.utcnow(),
        source="ai_search",
        kind="indexer",
        subject="atlas-candidates",
        attributes={"service_status": "degraded"},
        raw=None,
    )
    print(evt)

What surprised me here is how quickly “just pass through the JSON” becomes a trap. Add a second producer and you’re debugging shapes instead of incidents.

Stage 2: fast scoring/rules (thresholds, heuristics, confidence buckets)

One number is explicit: AnalysisTools.anomaly_threshold = 2.5 # Standard deviations (the analysis tools module). That says a lot about the intent. Anomaly scoring is based on deviation from recent history.

I am not reproducing the rest of that configuration or the full scoring method bodies here.

Instead, here’s a runnable scoring shell that shows how I combine:

  • an anomaly score (z-score-like) gated by the known 2.5 threshold
  • recency, so newer events score higher
  • source weighting, left as configuration
# scoring.py
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional


@dataclass(frozen=True)
class ScoreResult:
    score: float
    is_anomalous: bool


def score_event(
    *,
    anomaly_score: Optional[float],
    event_time: datetime,
    now: datetime,
    anomaly_threshold: float,
) -> ScoreResult:
    """Compact scoring function.

    From the repo:
    - anomaly_threshold is set to 2.5 standard deviations in AnalysisTools.

    Note:
    - The production scoring formula isn't published here.
    - This function demonstrates the shape of the combination logic.
    """

    # Recency factor: decays to 0 after 24h (hours default appears in ops_agent tool stubs).
    age = now - event_time
    recency = max(0.0, 1.0 - (age / timedelta(hours=24)))

    if anomaly_score is None:
        return ScoreResult(score=recency, is_anomalous=False)

    is_anom = anomaly_score >= anomaly_threshold
    # Combine anomaly and recency.
    combined = float(anomaly_score) * recency

    return ScoreResult(score=combined, is_anomalous=is_anom)


if __name__ == "__main__":
    now = datetime.utcnow()
    r = score_event(
        anomaly_score=3.0,
        event_time=now - timedelta(minutes=3),
        now=now,
        anomaly_threshold=2.5,
    )
    print(r)

The non-obvious detail is that recency is doing incident hygiene here rather than serving math purity. It biases the pipeline toward telling humans about what’s happening now instead of what looked strange hours ago.

Stage 3: multi-detector consensus (cheap arbitration to collapse duplicates)

The agent codebase is structured around multiple outputs. EventProcessor is initialized with on_insight and on_anomaly callbacks (the event processor module), and the services import notify_insight / notify_anomaly.

Which is exactly the setup where duplicates happen: two different detectors can report the same underlying incident.

There is no arbitration module in the agent. What follows is a runnable dedup/arbitration strategy that’s consistent with the repo’s needs (collapsing overlapping alerts).

Key strategy: derive a deduplication key from canonical fields (source, kind, subject) plus a short time window.

# dedup.py
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from typing import Iterable, List, Dict, Tuple


@dataclass(frozen=True)
class AlertCandidate:
    timestamp: datetime
    source: str
    kind: str
    subject: str
    summary: str


def dedup_key(c: AlertCandidate) -> str:
    """Deduplication key strategy.

    Note: The production dedup key isn't published here.
    This key uses only canonical fields that are plausible given the
    agent structure and the dashboard's AI Search index naming.
    """

    return f"{c.source}:{c.kind}:{c.subject}"


def arbitrate(candidates: Iterable[AlertCandidate]) -> List[AlertCandidate]:
    """Cheap arbitration: collapse duplicates by key, keep the latest."""

    by_key: Dict[str, Tuple[datetime, AlertCandidate]] = {}
    for c in candidates:
        k = dedup_key(c)
        prev = by_key.get(k)
        if prev is None or c.timestamp >= prev[0]:
            by_key[k] = (c.timestamp, c)

    return [item[1] for item in by_key.values()]


if __name__ == "__main__":
    now = datetime.utcnow()
    items = [
        AlertCandidate(now, "ai_search", "indexer", "atlas-candidates", "failed run"),
        AlertCandidate(now, "ai_search", "indexer", "atlas-candidates", "degraded"),
    ]
    print(arbitrate(items))

What I like about this pattern is that it’s cheap enough to always run. Arbitration shouldn’t become a second ML problem. It should be a small, predictable reducer.

Stage 4: dispatch with idempotency and exponential backoff

Teams dispatch is where the concrete implementation details live:

  • TeamsNotifier reads TEAMS_DEVOPS_WEBHOOK_URL from the environment (the teams notifier module).
  • It uses an httpx.AsyncClient(timeout=30.0) and keeps it cached (_client) so it can be reused.
  • The notifications API has a /test endpoint that returns 503 if the webhook URL is not configured (the notifications module).

Those are the bones of a defensive sender.

What the code doesn’t include:

  • any Redis client
  • any DB-backed idempotency row
  • any implemented exponential backoff routine

Instead, here’s a runnable sender wrapper that:

  • uses the same httpx.AsyncClient(timeout=30.0) pattern
  • implements exponential backoff in a generic way (algorithmic structure only)
  • includes explicit comments where the real system would add a one-shot guard and a sliding-window limiter, because those components aren’t in the repo code

(A note on client lifecycle: reusing an AsyncClient rather than creating a new client per request is an established practice to avoid resource churn and connection overhead, and the pattern used in the repo follows the client reuse guidance in httpx.)

# sender.py
from __future__ import annotations

import asyncio
from dataclasses import dataclass
from typing import Optional

import httpx


@dataclass
class SendResult:
    success: bool
    message: str


class DefensiveTeamsSender:
    """Defensive Teams sender.

    From the repo:
    - Uses httpx.AsyncClient(timeout=30.0) as in TeamsNotifier.

    Omitted here:
    - Redis/DB idempotency guard
    - sliding-window rate limiter

    Those should be inserted where marked below in a real deployment.
    """

    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
        self._client: Optional[httpx.AsyncClient] = None

    async def _get_client(self) -> httpx.AsyncClient:
        if self._client is None:
            self._client = httpx.AsyncClient(timeout=30.0)
        return self._client

    async def close(self) -> None:
        if self._client is not None:
            await self._client.aclose()
            self._client = None

    async def send_with_backoff(self, payload: dict, *, max_attempts: int = 3) -> SendResult:
        if not self.webhook_url:
            return SendResult(False, "Teams webhook URL not configured")

        # TODO:
        # - Idempotency guard (Redis key or DB row) to ensure one-shot send.
        # - Sliding-window rate limiter to prevent alert storms.

        client = await self._get_client()

        delay = 1.0
        last_err: Optional[str] = None

        for attempt in range(1, max_attempts + 1):
            try:
                resp = await client.post(self.webhook_url, json=payload)
                if 200 <= resp.status_code < 300:
                    return SendResult(True, "Sent")
                last_err = f"HTTP {resp.status_code}"
            except Exception as e:
                last_err = str(e)

            if attempt < max_attempts:
                await asyncio.sleep(delay)
                delay *= 2

        return SendResult(False, last_err or "Unknown error")


async def _demo() -> None:
    sender = DefensiveTeamsSender(webhook_url="")
    r = await sender.send_with_backoff({"text": "hello"})
    print(r)
    await sender.close()


if __name__ == "__main__":
    asyncio.run(_demo())

The subtle win here is lifecycle discipline. Caching the AsyncClient, as the repo’s TeamsNotifier does, is one of those small choices that makes a sender behave like a service instead of a script. See httpx’s client guidance for the same pattern.

Concrete example: three related events in ~30 seconds → one Teams message

The dashboard’s AI Search view tracks fields that are tailor-made for correlated symptoms:

  • failed_indexer_runs_24h
  • throttled_queries_24h
  • service_status: 'running' | 'degraded' | 'error'

(all defined in the AISearchDetailView module).

In the failure mode I designed for, you might see:

  1. an indexer run fails (incrementing failed_indexer_runs_24h)
  2. service status flips to degraded
  3. queries begin throttling (incrementing throttled_queries_24h)

Emit three messages and an operator learns nothing new after the first one.

So adjudication collapses them under one dedup key (same source=ai_search, kind=indexer, subject=<index name>), and the final notification becomes something like:

  • title: degraded indexer
  • body: latest status + relevant counters
  • operator action: trigger an indexer run

That last action comes from the UI code. The detail view includes a handler that POSTs to:

  • POST ${API_URL}/api/v1/ops/search/indexers/${indexName}/run

and reports “Indexer run triggered” on success (the AISearchDetailView module).

My agent does not call that endpoint. But the design alignment is worth pointing out: the notification should link to the same operator action the dashboard already exposes.

Nuances and tradeoffs I accepted

Normalization has a cost. You lose some source-specific richness, and I’m okay with that, because the whole point is producing a small number of messages that humans can act on.

Arbitration can hide distinct root causes if your key is too coarse. More ML won’t fix that. What fixes it is choosing keys that match operational reality (service, subsystem, subject), and making sure the collapsed message still carries enough attributes to diagnose.

Backoff introduces delay on retries. Acceptable, because the first attempt is immediate and the backoff only matters when the downstream Teams webhook is unhappy.

And the repo code shows the Teams webhook can be unconfigured: the /test endpoint returns 503. That’s a real operational state rather than an edge case, so I treat “no webhook configured” as a first-class outcome instead of an exception.

Closing

Detector count is the wrong scoreboard. The most useful notification system I’ve built is the one that turns a burst of messy telemetry into exactly one message a human will actually read, and then refuses to send the second one until there’s genuinely something new to say.