Parallelizing Adjuster Assignment with Asyncio Task Graphs

This guide sits under Adjuster Assignment Algorithms: it shows how to fan out the independent I/O-bound sub-steps of an assignment decision — skill lookup, workload query, licensing check, conflict check — with an asyncio task graph, while keeping the final assignment perfectly deterministic under bounded concurrency, timeouts, and cancellation.

An adjuster assignment is a join over several backing services. To pick an owner for a claim you need the pool’s skill tags from a competency service, each candidate’s current open-claim count from a workload store, licensing status per state from a compliance service, and any conflict-of-interest flags from a separate system of record. Done sequentially — await skills(); await workload(); await licensing(); await conflicts() — the request pays the sum of four network round-trips. At a few hundred milliseconds each that is a second per claim before any decision logic runs, and during a catastrophe surge that latency is the difference between claims flowing and claims backing up.

The naive fix is to fire the calls concurrently, but three failures follow immediately. First, one slow or hung dependency stalls the entire assignment because nothing bounds how long a gather waits. Second, an unbounded fan-out across thousands of concurrent claims opens thousands of simultaneous connections to each backing service and knocks it over — the assignment stage becomes its own denial-of-service. Third, and most damaging, teams let the concurrency leak into the result: they pick the first adjuster whose checks happen to return first, so the same claim assigned twice can produce two different owners depending on network timing. That non-determinism is exactly what a regulator’s “why this adjuster?” question cannot survive. This page fans the I/O out with asyncio.TaskGroup, bounds it with a semaphore and per-call timeouts, and then reduces the gathered facts through a pure, order-independent decision so the assignment is fast and reproducible.

This targets Python 3.11+ specifically, because it relies on asyncio.TaskGroup and asyncio.timeout, both added in 3.11. Testing uses pytest with pytest-asyncio. Structured audit lines go through structlog:

Copy
python -m venv .venv && source .venv/bin/activate   # Python 3.11+
pip install "structlog==24.*" "pytest==8.*" "pytest-asyncio==0.23.*"

The claim entering this stage must already be scored and coverage-validated — the severity tier from the parent Adjuster Assignment Algorithms hierarchy and the eligibility verdict from Coverage Validation Rules are inputs here, not outputs. The four backing services must each expose an async client; if a dependency is synchronous only, wrap it with asyncio.to_thread at its own boundary rather than blocking the event loop inside the task graph.

Define an immutable record for the facts the sub-steps return, and give every I/O call the same two guards: a shared semaphore that caps how many calls are in flight across all claims, and a per-call timeout that cancels a hung dependency. Bounding concurrency here is what stops a surge from turning the assignment stage into a load test against your own services.

Copy
from __future__ import annotations

import asyncio
from dataclasses import dataclass

import structlog

log = structlog.get_logger("adjuster.assign")

_GATE = asyncio.Semaphore(64)      # global cap on concurrent backing-service calls
CALL_TIMEOUT_S = 2.0


@dataclass(frozen=True, slots=True)
class AssignmentFacts:
    skills: dict[str, frozenset[str]]      # adjuster_id -> competency tags
    workload: dict[str, int]               # adjuster_id -> open claim count
    licensed: frozenset[str]               # adjuster_ids licensed in loss state
    conflicted: frozenset[str]             # adjuster_ids with a COI on this claim


async def _guarded(coro_factory, *, name: str):
    async with _GATE:
        try:
            async with asyncio.timeout(CALL_TIMEOUT_S):
                return await coro_factory()
        except TimeoutError:
            log.error("adjuster.substep.timeout", substep=name)
            raise

asyncio.TaskGroup runs the four calls concurrently and gives structured-concurrency guarantees: if any task raises, the group cancels the siblings and re-raises, so the assignment fails atomically instead of proceeding on partial facts. Collect the results only after the group exits, when every task is known to have succeeded.

Copy
async def gather_facts(
    claim_id: str,
    loss_state: str,
    pool: tuple[str, ...],
    svc,                                  # object exposing the four async clients
) -> AssignmentFacts:
    async with asyncio.TaskGroup() as tg:
        t_skills = tg.create_task(
            _guarded(lambda: svc.skills(pool), name="skills"))
        t_load = tg.create_task(
            _guarded(lambda: svc.workload(pool), name="workload"))
        t_lic = tg.create_task(
            _guarded(lambda: svc.licensing(pool, loss_state), name="licensing"))
        t_coi = tg.create_task(
            _guarded(lambda: svc.conflicts(claim_id, pool), name="conflicts"))
    # reached only if all four succeeded
    facts = AssignmentFacts(
        skills=t_skills.result(),
        workload=t_load.result(),
        licensed=t_lic.result(),
        conflicted=t_coi.result(),
    )
    log.info("adjuster.facts.gathered", claim_id=claim_id, pool_size=len(pool))
    return facts

The selection is a pure function of the gathered facts, and it must be independent of the order the sub-steps finished. Filter the pool to adjusters who are both licensed and conflict-free, then rank the survivors by a total-ordering key — fewest open claims first, best skill match next, and a stable tiebreaker on adjuster id so ties never resolve by chance. The same facts always yield the same owner.

Copy
@dataclass(frozen=True, slots=True)
class Assignment:
    claim_id: str
    adjuster_id: str
    rationale: str


class NoEligibleAdjuster(Exception):
    """Pool is empty after licensing and conflict filtering — divert to fallback."""


def select_adjuster(
    claim_id: str,
    required_skills: frozenset[str],
    facts: AssignmentFacts,
) -> Assignment:
    eligible = [
        a for a in facts.workload
        if a in facts.licensed and a not in facts.conflicted
    ]
    if not eligible:
        raise NoEligibleAdjuster(claim_id)

    def rank_key(a: str) -> tuple[int, int, str]:
        missing = len(required_skills - facts.skills.get(a, frozenset()))
        return (facts.workload[a], missing, a)   # id makes the order total

    chosen = min(eligible, key=rank_key)
    rationale = (f"workload={facts.workload[chosen]} "
                 f"skill_gap={len(required_skills - facts.skills.get(chosen, frozenset()))}")
    log.info("adjuster.assigned", claim_id=claim_id, adjuster_id=chosen)
    return Assignment(claim_id, chosen, rationale)

Two properties matter: the reduction is deterministic regardless of which sub-step returned first, and a hung dependency is cancelled rather than allowed to stall the assignment. Use pytest.mark.asyncio, inject fake async clients with controllable delays, and assert the same claim always resolves to the same adjuster.

Copy
import asyncio
import pytest


class FakeSvc:
    def __init__(self, delays):     # delays: dict substep -> seconds
        self.delays = delays

    async def skills(self, pool):
        await asyncio.sleep(self.delays.get("skills", 0))
        return {"a-1": frozenset({"auto"}), "a-2": frozenset({"auto", "cat"})}

    async def workload(self, pool):
        await asyncio.sleep(self.delays.get("workload", 0))
        return {"a-1": 12, "a-2": 12}          # deliberate tie

    async def licensing(self, pool, state):
        await asyncio.sleep(self.delays.get("licensing", 0))
        return frozenset({"a-1", "a-2"})

    async def conflicts(self, claim_id, pool):
        await asyncio.sleep(self.delays.get("conflicts", 0))
        return frozenset()


@pytest.mark.asyncio
async def test_assignment_is_order_independent():
    # two runs, opposite substep timings, must yield the same adjuster
    facts_a = await gather_facts("C1", "TX", ("a-1", "a-2"),
                                 FakeSvc({"skills": 0.05}))
    facts_b = await gather_facts("C1", "TX", ("a-1", "a-2"),
                                 FakeSvc({"conflicts": 0.05}))
    req = frozenset({"auto"})
    assert select_adjuster("C1", req, facts_a).adjuster_id == \
           select_adjuster("C1", req, facts_b).adjuster_id == "a-1"  # id tiebreak


@pytest.mark.asyncio
async def test_hung_dependency_times_out():
    svc = FakeSvc({"licensing": 10.0})          # exceeds CALL_TIMEOUT_S
    with pytest.raises(TimeoutError):
        await gather_facts("C2", "TX", ("a-1", "a-2"), svc)

Run with python -m pytest -q. The load-bearing assertion is that two runs with opposite sub-step timings pick the identical adjuster: the workload tie is broken by adjuster id, not by which coroutine finished first, which is what makes the decision replayable in an audit.

Concurrency is an implementation detail that must never leak into the assignment outcome. Because select_adjuster is a pure function over an immutable AssignmentFacts snapshot and its ranking key ends in a stable tiebreaker, the same claim against the same facts always produces the same adjuster — the reproducibility a market-conduct examiner requires when asking why a particular claim went to a particular person. Persist the facts snapshot, the rationale string, and the chosen adjuster to the append-only assignment ledger before the routing action is published. The licensing filter is itself a compliance control: an adjuster not in the licensed set for the loss state is removed before ranking, so an unlawful assignment cannot be written even under timing pressure. When the pool empties after filtering, NoEligibleAdjuster fails closed to a fallback route rather than assigning an ineligible owner.

• Same claim assigned to different adjusters on retry. Symptom: a re-run picks a different owner. Cause: the selection resolved a tie by which sub-step returned first. Fix: make the ranking key a total order ending in a stable field like adjuster id, and keep selection a pure function of the facts snapshot.
• One slow service stalls every assignment. Symptom: p99 assignment latency tracks a single dependency’s worst case. Cause: no per-call timeout, so a gather waits indefinitely. Fix: wrap each call in asyncio.timeout; a cancelled call raises and the TaskGroup fails the assignment atomically to the fallback route.
• Backing service falls over during a surge. Symptom: connection errors spike as claim volume climbs. Cause: unbounded fan-out opens one connection per claim per service. Fix: gate every call with a shared asyncio.Semaphore so total in-flight calls are capped regardless of claim volume.
• Partial facts produce a bad assignment. Symptom: an adjuster is chosen even though the licensing call failed. Cause: results were read from tasks individually with exceptions swallowed. Fix: use TaskGroup and read .result() only after the group exits, so any failure cancels siblings and aborts the whole assignment.
• Event loop stalls under load. Symptom: latency climbs even though services are fast. Cause: a synchronous client is being awaited inside the graph and blocks the loop. Fix: wrap any blocking client in asyncio.to_thread at its boundary so the event loop stays free.

• Adjuster Assignment Algorithms — the parent guide whose decision hierarchy this task graph parallelizes
• Routing High-Severity Claims to Senior Adjusters — the tier rule the deterministic selection feeds into
• Coverage Validation Rules — the eligibility gate that produces the validated claim entering this stage
• Orchestrating Batch Severity Scoring with Celery — the complementary concurrency model for bulk scoring
• Claims Triage & Routing Engines — the control plane this assignment stage belongs to