Routing High-Severity Claims to Senior Adjusters Under Concurrency

This guide extends the parent Adjuster Assignment Algorithms component with the concurrency control, streaming deserialization, and idempotency patterns that keep severity-gated routing deterministic when catastrophe volume hits the pipeline all at once.

The base assignment function is pure and reproducible, but it assumes one claim, one in-memory adjuster snapshot, and infinite heap. A real catastrophe event breaks all three assumptions in the same five-minute window:

• Race-driven overcommitment. Hundreds of CRITICAL claims arrive concurrently and each reader sees the same “least-loaded senior adjuster” before any writer commits. Naive logic stacks twenty total-loss files onto one person and starves the rest of the senior pool.
• Duplicate owners under retry storms. When the broker redelivers a partially processed event, a second worker re-runs assignment and writes a second owner for a claim that already has one — corrupting SLA timers and the audit trail.
• Heap exhaustion from forensic payloads. A high-severity FNOL routinely carries embedded photos, drone telemetry, and engineering reports. Eagerly json.load-ing a multi-megabyte body per worker, multiplied across a surge, triggers the GC pauses that blow the assignment SLA.

The fix is to keep the severity-ceiling decision deterministic — exactly as the parent component does — while wrapping the commit in optimistic concurrency control and streaming the input so memory stays bounded. The seniority gate itself reuses the bands emitted by Automated Severity Scoring Models and the cut points governed by Dynamic Threshold Tuning; this page only adds the machinery that makes the commit safe at scale.

This pattern assumes the deterministic decision spine from the parent cluster is already in place and that the inbound event has cleared coverage checks from Coverage Validation Rules. Pin the runtime so audit replay is exact:

• Python 3.11+ for precise timezone-aware datetime and structural pattern matching.
• Pydantic v2 (pydantic>=2.6) for the strict ClaimPayload / AdjusterProfile contracts reused from the parent component.
• ijson>=3.2 for incremental, event-driven JSON parsing of large FNOL bodies.
• redis>=5.0 for WATCH/MULTI/EXEC optimistic locking on the per-adjuster workload counter — capacity state must be a versioned snapshot, never a live blind read.
• An append-only audit sink (WORM object store or compacted log topic) for the decision records described below.

Process the forensic FNOL body as a token stream and pull only the routing-relevant fields into memory. ijson yields parse events incrementally, so a 40 MB submission costs kilobytes of resident memory instead of tens of megabytes per worker. The relevant incremental-parsing patterns are documented in the Python standard library json reference, which ijson mirrors with a generator API.

Copy
import ijson
import logging
from typing import BinaryIO

logger = logging.getLogger("triage.routing.stream")

# Only these fields drive senior-adjuster routing; the rest of the FNOL
# (media, telemetry, narratives) is never materialized in the router.
_ROUTING_FIELDS = frozenset(
    {"claim_id", "jurisdiction", "coverage_type", "severity_tier", "fnol_timestamp"}
)


def extract_routing_fields(body: BinaryIO) -> dict[str, str]:
    """Stream-parse only the fields the router needs from a large FNOL body."""
    fields: dict[str, str] = {}
    parser = ijson.parse(body)
    for prefix, event, value in parser:
        if event in ("string", "number") and prefix in _ROUTING_FIELDS:
            fields[prefix] = str(value)
            if _ROUTING_FIELDS.issubset(fields):
                break  # short-circuit once routing fields are complete
    missing = _ROUTING_FIELDS - fields.keys()
    if missing:
        raise ValueError(f"FNOL missing routing fields: {sorted(missing)}")
    logger.info("Streamed routing fields for claim_id=%s", fields["claim_id"])
    return fields

Build the typed ClaimPayload, apply the same ordered filter stack as the parent engine, then narrow to the senior tier. The selection is a total-order sort (workload, then seniority rank, then adjuster id) so two replicas processing the same retry converge on the identical adjuster. Licensing remains the non-negotiable first gate, mirroring State Regulation Mapping.

Copy
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, Field

_TIER_ORDER = {"low": 1, "medium": 2, "high": 3, "critical": 4}


class SeniorAdjuster(BaseModel):
    adjuster_id: str
    licensed_states: list[str]
    specializations: list[str]
    max_severity_tier: str
    seniority_rank: int = Field(..., ge=1)  # lower = more senior
    workload_index: float = Field(..., ge=0.0, le=1.0)
    workload_version: int = Field(..., ge=0)  # optimistic-lock token


def select_senior_adjuster(
    fields: dict[str, str],
    pool: list[SeniorAdjuster],
    workload_ceiling: float = 0.95,
) -> SeniorAdjuster | None:
    """Deterministic senior-tier selection; returns None to signal escalation."""
    claim_tier = _TIER_ORDER[fields["severity_tier"]]
    qualified = [
        a for a in pool
        if fields["jurisdiction"] in a.licensed_states
        and fields["coverage_type"] in a.specializations
        and _TIER_ORDER[a.max_severity_tier] >= claim_tier
        and a.workload_index < workload_ceiling
    ]
    if not qualified:
        return None
    # Total order guarantees identical output for identical input.
    qualified.sort(
        key=lambda a: (round(a.workload_index, 3), a.seniority_rank, a.adjuster_id)
    )
    return qualified[0]

The selection is reproducible, but the commit must reject any candidate whose workload changed between read and write. Wrap the capacity update in a Redis WATCH/MULTI/EXEC transaction keyed on the adjuster’s workload_version, and gate the whole commit behind an idempotency key derived from claim_id. A losing compare-and-swap re-reads a fresh snapshot and re-selects rather than overwriting; a redelivered event is a no-op.

Copy
import hashlib
import json
import redis

logger = logging.getLogger("triage.routing.commit")


class AssignmentConflict(Exception):
    """Raised when the adjuster's capacity changed mid-decision."""


def commit_assignment(
    client: redis.Redis,
    fields: dict[str, str],
    adjuster: SeniorAdjuster,
    ruleset_version: str,
) -> dict[str, str]:
    """Idempotent, compare-and-swap commit with a tamper-evident audit record."""
    claim_id = fields["claim_id"]
    idem_key = f"assign:idem:{claim_id}"
    wl_key = f"adjuster:workload:{adjuster.adjuster_id}"

    with client.pipeline() as pipe:
        pipe.watch(idem_key, wl_key)
        if pipe.get(idem_key):  # already owned — replay is a no-op
            pipe.unwatch()
            logger.info("Replay suppressed for claim_id=%s", claim_id)
            return {"claim_id": claim_id, "status": "ALREADY_ASSIGNED"}

        current_version = int(pipe.hget(wl_key, "version") or 0)
        if current_version != adjuster.workload_version:
            pipe.unwatch()
            raise AssignmentConflict(adjuster.adjuster_id)

        record = {
            "claim_id": claim_id,
            "assigned_adjuster_id": adjuster.adjuster_id,
            "status": "ASSIGNED",
            "ruleset_version": ruleset_version,
            "severity_tier": fields["severity_tier"],
            "decided_at": datetime.now().astimezone().isoformat(),
        }
        record["record_hash"] = hashlib.sha256(
            json.dumps(record, sort_keys=True).encode("utf-8")
        ).hexdigest()

        pipe.multi()
        pipe.set(idem_key, adjuster.adjuster_id)
        pipe.hincrby(wl_key, "version", 1)
        pipe.rpush("audit:assignments", json.dumps(record))
        try:
            pipe.execute()
        except redis.WatchError as exc:  # someone committed first
            raise AssignmentConflict(adjuster.adjuster_id) from exc

    logger.info("Committed claim_id=%s -> %s", claim_id, adjuster.adjuster_id)
    return record

Wrap steps 2 and 3 in a bounded retry loop: on AssignmentConflict, refetch the senior pool with current workload_version values and re-select. When the loop exhausts its budget or select_senior_adjuster returns None, emit a FALLBACK_ESCALATION and defer to the patterns in designing fallback routes for missing adjuster data — never relax the licensing or severity gate to force a match.

Two properties must hold under test: determinism (identical input → identical adjuster) and single-owner safety (concurrent commits yield exactly one ASSIGNED record). Assert both directly.

Copy
import threading

def test_selection_is_deterministic():
    fields = {
        "claim_id": "C-1", "jurisdiction": "TX",
        "coverage_type": "property", "severity_tier": "critical",
        "fnol_timestamp": "2026-06-27T12:00:00Z",
    }
    pool = [
        SeniorAdjuster(adjuster_id="adj-b", licensed_states=["TX"],
                       specializations=["property"], max_severity_tier="critical",
                       seniority_rank=1, workload_index=0.40, workload_version=7),
        SeniorAdjuster(adjuster_id="adj-a", licensed_states=["TX"],
                       specializations=["property"], max_severity_tier="critical",
                       seniority_rank=1, workload_index=0.40, workload_version=3),
    ]
    # Equal workload + equal seniority -> adjuster_id breaks the tie, every run.
    assert select_senior_adjuster(fields, pool).adjuster_id == "adj-a"
    assert select_senior_adjuster(fields, list(reversed(pool))).adjuster_id == "adj-a"


def test_concurrent_commit_yields_one_owner(redis_client):
    results: list[str] = []
    def worker():
        try:
            r = commit_assignment(redis_client, FIELDS, ADJUSTER, "v2026.06")
            results.append(r["status"])
        except AssignmentConflict:
            results.append("CONFLICT")
    threads = [threading.Thread(target=worker) for _ in range(8)]
    for t in threads: t.start()
    for t in threads: t.join()
    assert results.count("ASSIGNED") == 1  # exactly one committed owner

Round the workload_index before sorting (as in select_senior_adjuster) so floating-point noise can never make two equal loads compare unequal across replicas. Run the concurrency test against a real Redis instance, not a mock, so WATCH semantics are exercised faithfully.

Every committed assignment writes a content-hashed, ruleset_version-stamped record to an append-only sink inside the same transaction that claims the adjuster, so the proof of who was assigned and under which rules cannot drift from the capacity mutation. This satisfies the NAIC market-conduct expectation that a licensed senior adjuster handled each high-severity claim within their authorized scope, and the hash makes post-hoc tampering detectable. Keeping these records interoperable with downstream reporting follows ACORD data standards, so a Department of Insurance examiner can replay the decision lineage for any disputed CRITICAL claim from a single claim_id lookup.

• Duplicate senior owners after a surge. The capacity update is running outside the WATCH transaction. Confirm set(idem_key), hincrby(version), and the audit push are all inside the same MULTI/EXEC, so a WatchError aborts every side effect atomically.
• Conflict-retry livelock under sustained contention. A hot adjuster is re-selected every retry. Bound the retry loop (e.g. 5 attempts), and on exhaustion escalate rather than spin — a livelocked worker holds throughput hostage during the exact window you need it.
• Heap still climbing during catastrophe load. Something downstream is materializing the full FNOL. Verify extract_routing_fields short-circuits on _ROUTING_FIELDS.issubset(fields) and that no later stage re-reads the raw body; pass only the slim fields dict forward.
• Stale workload_version causing constant conflicts. The read model lags the Redis counter. Refetch the pool after a conflict instead of reusing the stale snapshot, and record the snapshot timestamp so an unexpectedly old version triggers escalation rather than a guess.
• Senior gate admitting junior adjusters. A _TIER_ORDER comparison is inverted or a severity band arrived unmapped. Unit-test _TIER_ORDER[a.max_severity_tier] >= _TIER_ORDER[claim_tier] across every tier pair, and reject any payload whose severity_tier is not a known key.

• Adjuster Assignment Algorithms — the parent component whose deterministic decision spine this pattern extends
• Implementing priority queues for catastrophic claims — the backpressure and queue ordering that feeds this router during a surge
• Validating deductible thresholds automatically — the upstream coverage verdict each routed claim already carries
• Designing fallback routes for missing adjuster data — where an exhausted retry loop or empty senior pool escalates to
• Handling multi-state compliance in claims routing — the licensing rules behind the non-negotiable jurisdictional gate