Skill-Based Technician Dispatch: Competency Matching and Scoring

Skill-based dispatch is the technician-selection stage of the Technician Routing & SLA Enforcement pipeline — the decision that picks who works a job by matching the trades and certifications a work order demands against the people who actually hold them.

Sending an unqualified technician to a job is expensive in ways that compound: a wasted truck roll, a second dispatch to redo the work, a certification-compliance gap that surfaces in an audit, and an SLA clock that keeps running the whole time. This stage closes that gap by treating technician selection as a deterministic scoring problem rather than a first-available grab. A competency matrix records which technician can perform which trade at what proficiency; a certification registry records which regulated qualifications are current; a scoring function combines those signals with priority and availability into a single ranked assignment. For facilities managers and reliability engineers, deterministic selection means every dispatch is defensible and repeatable; for Python automation and CMMS integration teams, it means an auditable function whose output depends only on its inputs. This guide implements that stage end to end — prerequisites, the input/output data contract, a step-by-step build, a configuration reference, validation checks, and the failure modes you will hit in production.

Prerequisites

This component runs as a synchronous selection step invoked by the routing worker once a work order has been validated and its required skills resolved. Before you deploy it, confirm the following are in place.

• Python 3.11+ with pydantic>=2.6 for modelling and validating the work order, technician, and match records; python-dateutil>=2.9 for certification-expiry arithmetic across timezones; and no scheduler of its own — the selector is called inline by the routing consumer.
• A skills registry (the competency matrix) that maps every technician to the trade codes they can perform and a proficiency level per trade. This is authoritative reference data, not free text: trade codes must be drawn from the same controlled taxonomy the work order uses, or matching silently fails. Model it as a table keyed by technician_id and skill_code.
• A certification registry exposing, per technician, each held certification with an issue date, expiry date, and issuing authority. Regulated trades (electrical, refrigerant handling, confined-space, hot-work) require a current certification, so expiry is load-bearing, not decorative.
• CMMS REST API v1 exposing a read-only technician roster (GET /api/v1/technicians) that returns each technician’s skills, certifications, home zone, and current shift state. The selector reads this; it never writes assignments — committing the assignment is a downstream step.
• Environment variables: CMMS_BASE_URL, CMMS_API_TOKEN (carrying the technicians:read scope), and DISPATCH_MIN_SCORE (default 0.5). A token missing technicians:read must fail closed at the roster fetch rather than returning an empty candidate list that looks like “no one is qualified.” Scope boundaries are defined in security & access boundaries.

Architecture and Data Contract

The selector sits between a validated work order and the assignment committer. It never mutates roster state and it never starts or stops an SLA clock — it reads required skills, filters the roster to genuinely eligible technicians, scores each one, and emits a ranked selection. Four boundaries keep the decision honest and stop selection logic from leaking into scheduling or compliance layers.

• Requirement boundary: the work order’s required_skills and required_certifications enter only after they have been resolved against the controlled taxonomy; an unresolved trade code is rejected before it silently excludes every technician.
• Eligibility boundary: a technician is a candidate only if they hold every required skill and every required certification is currently valid. Eligibility is a hard gate — it is never traded off against a high score elsewhere.
• Scoring boundary: a deterministic function consumes the eligible candidates and emits a score in [0, 1] per technician. Same roster, same work order, same ranking, every time.
• Selection boundary: the top-scored technician is returned with the full ranked list and the reason codes behind the score, so the caller can commit the assignment, break a tie, or escalate without re-querying.

Selection is not a single number — ties are common and must resolve deterministically. When two technicians score equally on competency and availability, the tie breaks by proximity, delegating the distance calculation to geographic zone routing so drive-time is respected without duplicating that logic here.

The contract across the selection boundary is explicit: the input is a DispatchRequest derived from the canonical work order, and the output is a DispatchDecision carrying the ranked matches. Encoding both with pydantic means an unresolved trade code or a malformed roster record is rejected at the boundary instead of producing a plausible-but-wrong assignment. The certification-validity rules that gate eligibility are covered in depth in matching technician certifications to work order trades.

from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional
from pydantic import BaseModel, Field


class Priority(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    STANDARD = "standard"
    PLANNED = "planned"


class SkillProficiency(BaseModel):
    """One trade a technician can perform, at a graded proficiency."""
    skill_code: str = Field(..., min_length=2, max_length=32)
    level: int = Field(..., ge=1, le=5, description="1=novice, 5=master")


class Certification(BaseModel):
    """A held qualification with an expiry the selector must honour."""
    cert_code: str = Field(..., min_length=2, max_length=32)
    issued_on: datetime
    expires_on: datetime
    authority: str


class Technician(BaseModel):
    """A roster member the selector may consider for assignment."""
    technician_id: str = Field(..., min_length=4)
    skills: List[SkillProficiency] = Field(default_factory=list)
    certifications: List[Certification] = Field(default_factory=list)
    home_zone: str
    on_shift: bool = True


class DispatchRequest(BaseModel):
    """Input contract: what the job needs and how urgent it is."""
    work_order_id: str = Field(..., min_length=8, max_length=64)
    asset_id: str = Field(..., min_length=6)
    location_id: str
    required_skills: Dict[str, int] = Field(
        ..., description="skill_code -> minimum proficiency level"
    )
    required_certifications: List[str] = Field(default_factory=list)
    priority: Priority = Priority.STANDARD
    escalation_tier: int = Field(0, ge=0, le=3)


class SkillMatch(BaseModel):
    """One scored candidate plus the reasons behind the score."""
    technician_id: str
    score: float = Field(..., ge=0.0, le=1.0)
    reasons: List[str] = Field(default_factory=list)


class DispatchDecision(BaseModel):
    """Output contract: the chosen technician and the full ranking."""
    selected: Optional[str]
    ranked: List[SkillMatch] = Field(default_factory=list)
    unmatched_reason: Optional[str] = None

Step-by-Step Implementation

1. Define the canonical work order payload

The selector is invoked against a work order, so it imports the same WorkOrderPayload used everywhere on this site rather than redefining its own shape. The SLA fields — priority, requested_completion, and escalation_tier — are what the routing engine reads when it decides how hard to push for a scarce skill, and they are the same fields the SLA timer & escalation stage consumes downstream. The selector narrows the payload to just the requirement fields it needs; the required-skills map is resolved upstream from the asset’s maintenance procedure.

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional


@dataclass
class WorkOrderPayload:
    """Canonical CMMS work order — SLA fields are mandatory site-wide."""
    work_order_id: str
    asset_id: str
    part_skus: List[str]
    required_quantities: Dict[str, int]
    location_id: str
    priority: Priority = Priority.STANDARD
    requested_completion: Optional[datetime] = None
    escalation_tier: int = 0
    status: str = "open"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


def to_dispatch_request(
    wo: WorkOrderPayload,
    required_skills: Dict[str, int],
    required_certifications: List[str],
) -> DispatchRequest:
    """Narrow the work order down to exactly what the selector needs."""
    return DispatchRequest(
        work_order_id=wo.work_order_id,
        asset_id=wo.asset_id,
        location_id=wo.location_id,
        required_skills=required_skills,
        required_certifications=required_certifications,
        priority=wo.priority,
        escalation_tier=wo.escalation_tier,
    )

2. Look up the competency matrix

The competency matrix is the authoritative map from a technician to the trades they can perform and at what level. Represent it as an index keyed by technician_id so a proficiency lookup is a constant-time dictionary access rather than a linear scan of every skill on every candidate. Build the index once per selection from the roster and reuse it across all candidates.

from typing import Dict, List


def build_competency_index(
    technicians: List[Technician],
) -> Dict[str, Dict[str, int]]:
    """technician_id -> {skill_code: proficiency_level}."""
    index: Dict[str, Dict[str, int]] = {}
    for tech in technicians:
        index[tech.technician_id] = {
            skill.skill_code: skill.level for skill in tech.skills
        }
    return index


def meets_skill_requirements(
    required: Dict[str, int],
    held: Dict[str, int],
) -> bool:
    """True only if every required skill is held at or above its level."""
    return all(held.get(code, 0) >= level for code, level in required.items())

3. Validate certification currency

A technician who once held a certification is not the same as one who holds it today. Compare each required certification’s expires_on against the moment of dispatch in UTC, and treat a missing or expired certification as a hard disqualification. Comparing timezone-aware datetimes avoids the classic off-by-one where a naive local timestamp crosses midnight and a still-valid card reads as expired.

from datetime import datetime, timezone
from typing import List


def has_valid_certifications(
    technician: Technician,
    required_certs: List[str],
    now: datetime | None = None,
) -> bool:
    """Every required certification must be present and unexpired."""
    now = now or datetime.now(timezone.utc)
    held = {
        cert.cert_code: cert.expires_on
        for cert in technician.certifications
    }
    for code in required_certs:
        expiry = held.get(code)
        if expiry is None or expiry <= now:
            return False
    return True

4. Score eligible candidates with a weighted function

Only technicians who pass both hard gates — every skill at level, every certification current — reach the scorer. The score blends three normalised signals: how far each held proficiency exceeds the requirement (headroom rewards depth), certification currency (a card expiring next week is worth less than one valid for a year), and a priority weighting that biases toward higher-proficiency technicians for critical work. Weights sum to one so the output lands in [0, 1] and stays comparable across jobs. The math is pure, which is what makes the ranking testable in isolation.

from datetime import datetime, timezone
from typing import Dict, List

# Weights sum to 1.0 so the score is normalised into [0, 1].
W_PROFICIENCY = 0.55
W_CERT_CURRENCY = 0.25
W_PRIORITY_FIT = 0.20
CERT_HORIZON_DAYS = 365.0


def score_candidate(
    request: DispatchRequest,
    technician: Technician,
    held_skills: Dict[str, int],
    now: datetime | None = None,
) -> SkillMatch:
    """Blend proficiency headroom, cert currency, and priority fit."""
    now = now or datetime.now(timezone.utc)
    reasons: List[str] = []

    # Proficiency: average headroom above the required level, capped at 1.0.
    headrooms = []
    for code, required_level in request.required_skills.items():
        held = held_skills.get(code, 0)
        headrooms.append(min((held - required_level) / 4.0, 1.0))
    proficiency = sum(headrooms) / len(headrooms) if headrooms else 0.0
    reasons.append(f"proficiency={proficiency:.2f}")

    # Certification currency: fraction of the horizon still remaining.
    if request.required_certifications:
        remaining = []
        by_code = {c.cert_code: c for c in technician.certifications}
        for code in request.required_certifications:
            days_left = (by_code[code].expires_on - now).days
            remaining.append(max(0.0, min(days_left / CERT_HORIZON_DAYS, 1.0)))
        currency = sum(remaining) / len(remaining)
    else:
        currency = 1.0
    reasons.append(f"cert_currency={currency:.2f}")

    # Priority fit: critical work rewards deeper proficiency.
    priority_fit = proficiency if request.priority == Priority.CRITICAL else 0.6
    reasons.append(f"priority_fit={priority_fit:.2f}")

    score = (
        W_PROFICIENCY * proficiency
        + W_CERT_CURRENCY * currency
        + W_PRIORITY_FIT * priority_fit
    )
    return SkillMatch(
        technician_id=technician.technician_id,
        score=round(score, 4),
        reasons=reasons,
    )

5. Select the top match and break ties by proximity

Rank the scored candidates and return the highest. When two technicians tie on score — common when both comfortably exceed the requirement — resolve deterministically on proximity so the shorter drive wins, delegating the distance lookup to the zone router rather than reimplementing it. If no candidate clears DISPATCH_MIN_SCORE, or the eligible pool is empty, return an explicit unmatched reason so the caller can escalate rather than silently dropping the work order.

import logging
from typing import Callable, Dict, List

logger = logging.getLogger(__name__)


def select_technician(
    request: DispatchRequest,
    technicians: List[Technician],
    proximity_rank: Callable[[str, str], float],
    min_score: float = 0.5,
    now=None,
) -> DispatchDecision:
    """Filter, score, rank, and break ties by proximity."""
    competency = build_competency_index(technicians)
    ranked: List[SkillMatch] = []

    for tech in technicians:
        if not tech.on_shift:
            continue
        held = competency[tech.technician_id]
        if not meets_skill_requirements(request.required_skills, held):
            continue
        if not has_valid_certifications(tech, request.required_certifications, now):
            continue
        ranked.append(score_candidate(request, tech, held, now))

    if not ranked:
        logger.info("no eligible technician wo:%s", request.work_order_id)
        return DispatchDecision(
            selected=None, ranked=[], unmatched_reason="no_eligible_technician"
        )

    # Sort by score desc, then by proximity to the job location (asc), then id.
    ranked.sort(
        key=lambda m: (
            -m.score,
            proximity_rank(m.technician_id, request.location_id),
            m.technician_id,
        )
    )
    top = ranked[0]
    if top.score < min_score:
        return DispatchDecision(
            selected=None, ranked=ranked, unmatched_reason="below_min_score"
        )
    logger.info(
        "dispatched wo:%s tech:%s score:%.2f candidates:%d",
        request.work_order_id, top.technician_id, top.score, len(ranked),
    )
    return DispatchDecision(selected=top.technician_id, ranked=ranked)

Configuration Reference

Keep every tunable in a version-controlled configuration registry, not in the worker source. The defaults below are conservative starting points for a general-purpose multi-trade maintenance team.

Validation and Testing

Selection is pure once the roster is fixed, so the highest-value test asserts that a given work order and roster always yield the same technician. A single deterministic assertion catches accidental nondeterminism — an unstable sort, dict ordering, or a clock-dependent certification check — before it reaches production.

from datetime import datetime, timedelta, timezone


def test_expired_certification_is_never_dispatched():
    now = datetime(2026, 7, 16, tzinfo=timezone.utc)
    request = DispatchRequest(
        work_order_id="WO-2026-0512",
        asset_id="CHLR-007",
        location_id="ZONE-N",
        required_skills={"HVAC-REFRIG": 3},
        required_certifications=["EPA-608"],
        priority=Priority.CRITICAL,
    )
    qualified = Technician(
        technician_id="TECH-100",
        skills=[SkillProficiency(skill_code="HVAC-REFRIG", level=4)],
        certifications=[Certification(
            cert_code="EPA-608", issued_on=now - timedelta(days=200),
            expires_on=now + timedelta(days=300), authority="EPA")],
        home_zone="ZONE-N",
    )
    expired = Technician(
        technician_id="TECH-200",
        skills=[SkillProficiency(skill_code="HVAC-REFRIG", level=5)],
        certifications=[Certification(
            cert_code="EPA-608", issued_on=now - timedelta(days=800),
            expires_on=now - timedelta(days=5), authority="EPA")],
        home_zone="ZONE-N",
    )

    def proximity(tech_id: str, location_id: str) -> float:
        return 0.0  # co-located for this test

    decision = select_technician(
        request, [qualified, expired], proximity, now=now
    )
    # The higher-skilled tech loses because the certification is expired.
    assert decision.selected == "TECH-100"
    assert all(m.technician_id != "TECH-200" for m in decision.ranked)

On a successful selection the component emits a single structured line per work order — dispatched wo:WO-2026-0512 tech:TECH-100 score:0.88 candidates:1 — which is the canonical signal that a defensible assignment was made. When no candidate qualifies it instead emits no eligible technician wo:..., and the returned unmatched_reason distinguishes an empty eligible pool from a pool that scored below threshold. Assert against both the log line and the unmatched_reason in integration tests to verify the full requirement-to-assignment path.

Failure Modes and Troubleshooting

Expand each scenario for the root cause, the diagnostic log excerpt, and the fix. The checklist items render as interactive checkboxes — work through them in order.

A technician was dispatched on an expired certification

• Confirm has_valid_certifications compares timezone-aware datetimes — a naive expires_on compared against a UTC now can read a lapsed card as still valid across a midnight boundary.
• Check that the roster feed actually carries expires_on per certification; a null expiry that defaults to “far future” is the usual root cause of a lapsed dispatch.
• Verify the certification code in required_certifications matches the roster’s cert_code exactly — a mismatch means the required cert is never found, and an empty required set trivially passes the gate. The currency rules are detailed in matching technician certifications to work order trades.

No qualified technician is available for a critical work order

• Look for no eligible technician wo:<id> and inspect unmatched_reason — no_eligible_technician means the hard gates emptied the pool; below_min_score means someone qualified but scored too low.
• Confirm require_on_shift is not filtering the whole roster during a coverage gap; cross-check the roster against shift-calendar integration before concluding no one is skilled.
• Escalate rather than lowering DISPATCH_MIN_SCORE on the fly — a weak match on a critical job trades an SLA miss for a rework truck roll and should route to the SLA timer & escalation path.

Skill taxonomy mismatch silently excludes every candidate

• Diagnostic: a full roster of skilled technicians yields no_eligible_technician for a routine trade — a sign the work order’s required_skills codes do not exist in the competency matrix.
• Confirm both sides resolve against the same controlled taxonomy; a free-text "HVAC" will never match a coded "HVAC-REFRIG", and the mismatch never raises — it just excludes.
• Add a startup assertion that every skill_code in an incoming request exists in the competency index, so an unresolved code fails loudly at ingestion instead of quietly at selection.

Ties resolve inconsistently between runs

• Reproduce with two equally scored candidates and confirm the same technician is chosen every time; a differing result means the sort key is unstable.
• Verify the sort includes the technician_id final tiebreaker after proximity, so genuinely identical candidates still resolve deterministically.
• Confirm proximity_rank is a pure function of technician and location; a call that hits a live distance API with variable results will reorder ties non-deterministically — cache it per selection via geographic zone routing.

Frequently Asked Questions

Why is certification validity a hard gate rather than a scored factor?

Because a lapsed regulated certification is not a weaker qualification — it is a compliance violation. Trading it off against a high proficiency score would let the scorer dispatch an out-of-compliance technician whenever their skill was strong enough, which is exactly the outcome the gate exists to prevent. Currency still influences the score among valid candidates, but presence-and-validity is binary.

How does tie-breaking by proximity avoid duplicating the zone router?

The selector never computes distance itself. It accepts a proximity_rank callable and uses it only as the second sort key, so all geographic logic — zones, drive-time estimates, home-base offsets — stays in geographic zone routing. The selector’s contract is “given a way to rank proximity, break ties deterministically,” which keeps the two stages independently testable.

What should happen when no technician qualifies?

The selector returns selected=None with an explicit unmatched_reason and logs it, rather than forcing a weak assignment. The caller decides the escalation path — widen the eligible pool by relaxing a non-regulated skill floor, wait for shift coverage, or route to manual dispatch. Silently assigning the closest warm body is the failure mode this design refuses.

Can the scoring weights be tuned per site or trade?

Yes, as long as W_PROFICIENCY, W_CERT_CURRENCY, and W_PRIORITY_FIT still sum to 1.0. Keeping them normalised means scores stay comparable across work orders and sites, so a 0.8 on one job means the same thing as a 0.8 on another. Store the weights in the configuration registry, not in code, so a change is reviewable and reversible.

Related

Gate eligibility on live regulated credentials with matching technician certifications to work order trades, break scoring ties on drive-time via geographic zone routing, confirm the candidate is actually on-shift through shift-calendar integration, hand escalations to SLA timer & escalation, and scope the roster read with security & access boundaries.

Part of: Technician Routing & SLA Enforcement.