Alert Routing Workflows: Deterministic SOC Case Dispatch

A correct verdict delivered to the wrong responder is a missed detection with extra steps. Routing is the stage that takes a scored, correlated case and decides who — which team, which on-call responder, which escalation tier — and how fast it must be worked, and it has to make that decision the same way every time so an incident post-mortem can explain why a case went where it did. This page is part of the broader SOAR playbook & response automation discipline, and it sits between the decision gate and the ticketing gate: once the engine has decided a case needs a human (or a human sign-off on an action), routing resolves the destination. Get routing wrong and a critical domain-controller alert lands in a low-priority queue at 3 a.m. while the on-call for that asset never gets paged.

Problem Framing

Consider a concrete failure scenario. A SOC covers three business units, each with its own on-call rotation and its own crown-jewel systems. A case fires: severity 92, technique T1486 (Data Encrypted for Impact) on a host that a stale spreadsheet says belongs to “IT-General.” The routing logic, written as a nested if/elif that grew organically, checks severity first and pages the global tier-1 queue — but this host is actually a finance database whose owner rotation should have been paged directly, and the case waits forty minutes in a shared queue before anyone notices the business context. Meanwhile a flood of severity-40 informational cases from a noisy scanner are routed to the same tier-1 queue by a rule nobody remembers adding, burying the ransomware case under noise.

The job of a routing engine is to make dispatch deterministic and explainable: resolve the asset’s true owner from an authoritative source, evaluate priority-ordered rules where the first match wins and the match is recorded on the case, and attach an escalation timer so an unacknowledged high-severity case climbs the ladder automatically. That requires three things working together: an authoritative asset-and-owner lookup, a priority-ordered rule set with a recorded decision, and an escalation policy with acknowledgement timers. The rest of this page builds exactly that.

Prerequisites & Environment

The reference implementation targets Python 3.11+. The only third-party dependency is pydantic; the rest is standard library. A production deployment adds a directory/CMDB client and an on-call schedule client, modeled here as injectable async lookups.

python3 -m venv .venv
source .venv/bin/activate
pip install "pydantic>=2.6,<3.0"

Infrastructure assumptions:

• An authoritative asset inventory (CMDB or asset-management API) mapping host.id/user.name to an owning team and a criticality tier. Routing quality is bounded by inventory freshness — a stale owner map is the most common cause of mis-routing.
• An on-call schedule source (the paging provider’s API) resolving a team to the human currently on call, so routing targets a person, not just a queue.
• A dead-letter queue for cases that match no rule or whose owner cannot be resolved, so an un-routable case is visible and manually triaged rather than silently dropped into a default bucket.

Architecture Overview

Routing is a single logical stage with three internal phases — resolve, match, dispatch — feeding the ticketing and notification gates. A case that resolves to no owner or matches no rule fails closed to the DLQ rather than defaulting silently to a catch-all queue.

The pivotal design decision is that the matched rule is recorded on the case, and no rule is a routing failure — not a default. A first-match-wins rule set is only explainable if the winning rule’s id travels with the case; and a case that matches nothing must surface for triage rather than silently landing in a catch-all, because a silent catch-all is where high-severity cases go to be forgotten.

Step-by-Step Implementation

Step 1 — Model the case, the destination, and a routing rule

A rule is a typed predicate plus a destination and a priority. Rules are data, not code branches, so the set is reviewable and testable.

from __future__ import annotations

from dataclasses import dataclass
from typing import Callable, Optional

from pydantic import BaseModel, Field


class RoutableCase(BaseModel):
    case_id: str = Field(min_length=1)
    entity_id: str = Field(min_length=1)
    severity: int = Field(ge=0, le=100)
    category: str = Field(min_length=1)        # e.g. "ransomware", "recon"
    technique_id: Optional[str] = None
    owner_team: Optional[str] = None           # filled by resolution
    criticality: int = Field(default=0, ge=0, le=3)


@dataclass(frozen=True)
class Route:
    dest: str            # queue / pager target
    tier: int            # escalation start tier
    ack_seconds: int     # acknowledgement SLA before escalation


@dataclass(frozen=True)
class Rule:
    rule_id: str
    priority: int                                   # lower = evaluated first
    match: Callable[[RoutableCase], bool]
    route: Route

Step 2 — Resolve the authoritative owner and criticality

Resolution replaces the case’s owner and criticality from the inventory of record. It fails closed: an entity with no owner is routed to the DLQ, never guessed.

from typing import Awaitable, Protocol


class Inventory(Protocol):
    async def owner_of(self, entity_id: str) -> Optional[tuple[str, int]]: ...


async def resolve_owner(case: RoutableCase, inv: Inventory) -> Optional[RoutableCase]:
    """Return the case with owner/criticality set, or None if unresolvable."""
    found = await inv.owner_of(case.entity_id)
    if found is None:
        return None                                 # -> ERR_ROUTE_001
    team, crit = found
    return case.model_copy(update={"owner_team": team, "criticality": crit})

Step 3 — Evaluate priority-ordered rules, first match wins

Rules are sorted once by priority. The first matching rule wins and its id is returned with the route, so the decision is auditable. Ordering rules by specificity (crown-jewel asset before generic severity) is what prevents the “everything to tier-1” failure.

def compile_ruleset(rules: list[Rule]) -> list[Rule]:
    """Sort by priority once; specific rules must precede broad ones."""
    return sorted(rules, key=lambda r: r.priority)


def match_route(case: RoutableCase, rules: list[Rule]) -> Optional[tuple[str, Route]]:
    for rule in rules:                              # already priority-sorted
        if rule.match(case):
            return rule.rule_id, rule.route
    return None                                     # -> ERR_ROUTE_011


# Example ruleset: crown-jewel criticality beats raw severity.
RULESET = compile_ruleset([
    Rule("crown-jewel-critical", 10,
         lambda c: c.criticality >= 3 and c.severity >= 70,
         Route(dest="ir-crown-jewel", tier=2, ack_seconds=300)),
    Rule("high-sev-generic", 50,
         lambda c: c.severity >= 80,
         Route(dest="ir-tier1", tier=1, ack_seconds=600)),
    Rule("recon-low", 90,
         lambda c: c.category == "recon" and c.severity < 50,
         Route(dest="hunt-backlog", tier=0, ack_seconds=86400)),
])

Step 4 — Dispatch and arm the escalation timer

Dispatch resolves the on-call human for the destination and schedules an escalation coroutine that climbs the tier ladder if no acknowledgement arrives in time.

import asyncio
import logging

logger = logging.getLogger("soc.soar.router")


class OnCall(Protocol):
    async def responder_for(self, dest: str) -> Optional[str]: ...


async def dispatch(
    case: RoutableCase, rule_id: str, route: Route,
    oncall: OnCall, ack_events: "asyncio.Queue[str]",
) -> None:
    responder = await oncall.responder_for(route.dest)
    logger.info(
        '{"msg":"routed","case":"%s","rule":"%s","dest":"%s","responder":"%s","tier":%d}',
        case.case_id, rule_id, route.dest, responder or "queue-only", route.tier,
    )
    # Arm escalation: if no ack within ack_seconds, climb one tier.
    try:
        await asyncio.wait_for(_await_ack(case.case_id, ack_events), route.ack_seconds)
    except asyncio.TimeoutError:
        logger.warning("ERR_ROUTE_021 no-ack case=%s escalating to tier=%d",
                       case.case_id, route.tier + 1)


async def _await_ack(case_id: str, ack_events: "asyncio.Queue[str]") -> None:
    while True:
        acked = await ack_events.get()
        if acked == case_id:
            return

Schema & Validation Integration

Routing keys on fields the upstream pipeline guarantees: entity_id (ECS host.id/user.name), severity, and category. Those come normalized from JSON event normalization and scored by dynamic severity scoring, so a routing rule can trust that severity means the same thing across sources. The RoutableCase model is the contract boundary — a case missing category or entity_id is rejected before routing (ERR_ROUTE_031) rather than mis-matched. The owner and criticality the router resolves become part of the case that flows into automated ticket creation, so the ticket lands in the right project with the right priority without a second lookup.

Error Handling & DLQ Routing

Every failure produces a stable, queryable code. Codes follow the ERR_CATEGORY_NNN convention.

The cardinal rule is no silent default. Both an unresolved owner and an unmatched rule are failures that surface for a human, because a default queue is exactly where a mis-routed critical case becomes invisible. Escalation timers (ERR_ROUTE_021) are the safety net that guarantees an unacknowledged case does not sit forever, even if the initial route was imperfect.

Performance Tuning

Routing is lightweight per case; the cost is in the two external lookups. Tune in this order:

• Cache the inventory and on-call lookups with a short TTL (30–120 s). Owner and rotation data change on the order of minutes, so caching removes almost all per-case latency while staying fresh enough.
• Compile the ruleset once, not per case. Sort by priority at load and re-sort only on a config change.
• Keep rule predicates pure and cheap — no I/O inside a match function. All external data (owner, criticality) must be resolved before matching, so evaluation is a fast in-memory pass.
• Bound escalation timers as tasks under a supervising TaskGroup so tens of thousands of armed timers do not leak; cancel a timer the moment an ack arrives.
• Batch dispatch during storms by coalescing acknowledgement waits, and lean on the dedupe stage from playbook orchestration patterns so a flood of identical cases produces one route, not thousands.

Verification & Observability

Confirm correct operation with a structured log per routing decision (carrying the winning rule id), counters per error code, and tests that assert first-match ordering.

def test_crown_jewel_beats_generic_severity() -> None:
    case = RoutableCase(case_id="C-9", entity_id="FIN-DB-1", severity=85,
                        category="malware", criticality=3)
    result = match_route(case, RULESET)
    assert result is not None
    rule_id, route = result
    assert rule_id == "crown-jewel-critical"   # specificity wins over high-sev-generic
    assert route.dest == "ir-crown-jewel"


def test_unmatched_case_is_a_failure_not_a_default() -> None:
    case = RoutableCase(case_id="C-10", entity_id="H-2", severity=55,
                        category="informational", criticality=1)
    assert match_route(case, RULESET) is None   # -> ERR_ROUTE_011, surfaced for triage

Operationally, emit routed_cases_total{rule_id,dest}, route_dlq_total{code}, escalations_total{from_tier}, and route_latency_seconds. A rising route_dlq_total{code="ERR_ROUTE_001"} points at CMDB staleness; a rising escalations_total points at under-staffed rotations or an ack SLA set too tight, and feeds directly into threshold tuning strategies.

Troubleshooting

• Critical cases land in a generic queue. Root cause: a broad high-severity rule has a lower priority number than the specific crown-jewel rule, so it wins first. Fix: order specific rules before broad ones and assert the ordering in a test.
• Cases route to the wrong team. Root cause: stale inventory maps the asset to a former owner. Fix: shorten the owner-cache TTL and reconcile the CMDB; treat owner resolution as authoritative, not advisory.
• Nobody gets paged though the case routed. Root cause: the on-call source was unreachable (ERR_ROUTE_041) and the case went queue-only. Fix: alert on degraded routing and fail over to a secondary rotation source.
• Escalation never fires. Root cause: the ack event stream is not wired to the timer, so _await_ack blocks forever without a timeout path. Fix: verify asyncio.wait_for wraps the ack wait and that acks publish the exact case_id.
• A storm floods one queue. Root cause: routing runs before dedupe, so identical cases each route. Fix: dedupe upstream in the orchestration engine so one case per entity+technique reaches the router.

FAQ

Related

• SOAR Playbook & Response Automation
• Playbook Orchestration Patterns
• Automated Ticket Creation
• Case Enrichment Pipelines
• Dynamic Severity Scoring