Design a customer-support agent dispatcher

01The design, in the six-step framework

1. Requirements and use cases

Customers submit support requests that need a skill. Agents with skills become available, get assigned one request at a time, resolve it, and sometimes take a break in the middle of a request. A request abandoned by an agent on break goes back to the queue without losing its place. A resolve event that arrives after the request was already re-queued, because the agent clicked late, must be ignored. The assignment rule (least loaded, longest idle, best skill match) must be swappable without touching the entities.

Out of scope: the chat transport, persistence, and multiple requests per agent. The reported round was explicit that policy must be separated from entities and that tests must be deterministic through injected timestamps.

Constraints: fairness by original arrival time across re-queues; every state change is a checked transition; a single-threaded core, with the lock added at the boundary if several threads feed events.

2. Core entities and relationships

• Agent: id, skills, an AgentState, the request currently held, and a handled count used by the policy.
• Request: id, required skill, original arrival time, priority, a RequestState, who holds it and an attempt counter.
• Dispatcher: owns the agent and request registries and the queue; every event enters here and ends in _drain, which is the one place assignments happen.
• QueueEntry: a heap record keyed on priority then original arrival, so a re-queued request keeps its place.
• AssignmentPolicy: a protocol with one method; LeastLoadedPolicy is the default.

The dispatcher aggregates agents and requests (they have ids and lives of their own) and composes queue entries. The policy is used through a protocol so that no entity knows how it was chosen (strategy).

3. Interfaces and key methods

class AgentState(Enum):   AVAILABLE = auto(); BUSY = auto(); ON_BREAK = auto(); OFFLINE = auto()
class RequestState(Enum): QUEUED = auto(); ASSIGNED = auto(); RESOLVED = auto()

class AssignmentPolicy(Protocol):
    def pick(self, agents: Iterable["Agent"], request: "Request") -> "Agent | None": ...

class Dispatcher:
    def __init__(self, policy: AssignmentPolicy, clock: Callable[[], float]) -> None: ...
    def submit(self, request: "Request") -> None:
        """Queue or assign immediately."""
    def agent_available(self, agent_id: str) -> None: ...
    def agent_break(self, agent_id: str) -> None:
        """Agent steps away; an in-flight request is re-queued at its original time."""
    def resolve(self, request_id: str, agent_id: str) -> None:
        """Ignored if the request was re-queued since (stale event)."""

Invariants: an agent is BUSY if and only if current is set; a request is ASSIGNED if and only if some agent's current is it; a QUEUED request has exactly one live heap entry; arrived_at never changes.

4. Design patterns and the choices behind them

Two state machines, checked at every transition

Agents and requests each have a small state machine. agent_available refuses to flip a BUSY agent, forcing the caller to resolve or break first, which is how a lost request is made impossible rather than merely unlikely.

Stale events handled by identity, not time

A late resolve is detected by checking that the request is still ASSIGNED to that agent. Comparing timestamps was rejected: clocks on agent devices are not trustworthy, and the state check needs no clock at all. The attempt counter exists for the harder version where the same agent could be assigned the same request twice.

Fairness through the heap key

Re-queueing pushes a new entry with the original arrived_at, so a customer whose agent took a break does not go to the back. Priority sits in front of arrival in the key, which is the one deliberate unfairness; say it (see priority queues in object design).

Policy as a strategy, drain as the single assignment path

Every event funnels into _drain, so there is one loop to reason about. Entries the policy cannot place are held aside and re-pushed, which prevents the loop from spinning on an unplaceable head while later requests could be served.

5. Core flow and code

1. Request r1 (billing) arrives at t=100 and r2 (billing) at t=101; one billing agent A is available. submit(r1) drains: the policy picks A, r1 becomes ASSIGNED, A BUSY. r2 waits.
2. A calls agent_break: r1 goes back to QUEUED with its original arrival of 100 and a new heap entry; A is ON_BREAK; the drain finds no able agent and holds both entries.
3. Agent B (billing) calls agent_available: the drain pops r1 first because 100 beats 101, assigns it to B, and r2 keeps waiting.
4. A's late resolve(r1, A) arrives: r1 is assigned to B, so the event is dropped.
5. B resolves r1: B's handled count rises, B becomes AVAILABLE, the drain assigns r2 to B.
6. Every timestamp came in with the events, so a test replays this exact sequence with no sleeping.

import heapq, time
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Callable, Iterable, Protocol

class AgentState(Enum):
    AVAILABLE = auto(); BUSY = auto(); ON_BREAK = auto(); OFFLINE = auto()

class RequestState(Enum):
    QUEUED = auto(); ASSIGNED = auto(); RESOLVED = auto()

@dataclass
class Agent:
    agent_id: str
    skills: frozenset[str]
    state: AgentState = AgentState.AVAILABLE
    current: "Request | None" = None
    handled: int = 0

@dataclass
class Request:
    request_id: str
    skill: str
    arrived_at: float
    priority: int = 0
    state: RequestState = RequestState.QUEUED
    assigned_to: str | None = None
    attempt: int = 0                      # incremented on every assignment

@dataclass(order=True)
class QueueEntry:
    neg_priority: int
    arrived_at: float                      # fairness: original arrival, never reset
    request: Request = field(compare=False)

class AssignmentPolicy(Protocol):
    def pick(self, agents: Iterable[Agent], request: Request) -> Agent | None: ...

class LeastLoadedPolicy:
    def pick(self, agents: Iterable[Agent], request: Request) -> Agent | None:
        able = [a for a in agents if a.state is AgentState.AVAILABLE and request.skill in a.skills]
        return min(able, key=lambda a: a.handled, default=None)

class Dispatcher:
    def __init__(self, policy: AssignmentPolicy, clock: Callable[[], float] = time.monotonic) -> None:
        self._policy, self._clock = policy, clock
        self._agents: dict[str, Agent] = {}
        self._requests: dict[str, Request] = {}
        self._queue: list[QueueEntry] = []

    def add_agent(self, agent: Agent) -> None:
        self._agents[agent.agent_id] = agent

    def submit(self, request: Request) -> None:
        self._requests[request.request_id] = request
        heapq.heappush(self._queue, QueueEntry(-request.priority, request.arrived_at, request))
        self._drain()

    def _drain(self) -> None:
        """Assign while a queued request has an able, available agent."""
        held: list[QueueEntry] = []
        while self._queue:
            entry = heapq.heappop(self._queue)
            req = entry.request
            if req.state is not RequestState.QUEUED:
                continue                       # resolved or re-queued under a newer entry
            agent = self._policy.pick(self._agents.values(), req)
            if agent is None:
                held.append(entry); continue
            self._assign(agent, req)
        for e in held:
            heapq.heappush(self._queue, e)

    def _assign(self, agent: Agent, req: Request) -> None:
        agent.state, agent.current = AgentState.BUSY, req
        req.state, req.assigned_to, req.attempt = RequestState.ASSIGNED, agent.agent_id, req.attempt + 1

    def agent_available(self, agent_id: str) -> None:
        agent = self._agents[agent_id]
        if agent.state is AgentState.BUSY:
            raise ValueError("resolve or break first")
        agent.state = AgentState.AVAILABLE
        self._drain()

    def agent_break(self, agent_id: str) -> None:
        agent = self._agents[agent_id]
        req, agent.current = agent.current, None
        agent.state = AgentState.ON_BREAK
        if req is not None and req.state is RequestState.ASSIGNED:
            req.state, req.assigned_to = RequestState.QUEUED, None
            heapq.heappush(self._queue, QueueEntry(-req.priority, req.arrived_at, req))  # keeps its place
            self._drain()

    def resolve(self, request_id: str, agent_id: str) -> None:
        req = self._requests[request_id]
        if req.state is not RequestState.ASSIGNED or req.assigned_to != agent_id:
            return                               # stale: the request moved on after a break
        req.state = RequestState.RESOLVED
        agent = self._agents[agent_id]
        agent.handled += 1
        agent.state, agent.current = AgentState.AVAILABLE, None
        self._drain()

    def queue_depth(self) -> int:
        return sum(1 for e in self._queue if e.request.state is RequestState.QUEUED)

6. Extensions, concurrency and what interviewers probe

• Skill fallback: a request that no agent can serve for a long time could be offered to a generalist; that is a policy change (a second protocol method or a wrapper policy), not an entity change.
• Break with hand-back delay: let the agent finish within a grace period before re-queueing; add an ON_BREAK_PENDING state and a deadline on the agent, checked in _drain with the injected clock.
• Where a lock is needed: if web handlers call the dispatcher from several threads, one lock around each public method is enough because every method is short and ends in _drain; finer locks buy nothing here.
• Domain events: emitting Assigned, Requeued and Resolved records lets a notifier and a metrics sink react without the dispatcher knowing them (see domain events).
• Test first: the six-step scenario above as one test with fixed times; then a resolve from the wrong agent; then two agents and three requests checking least-loaded selection.
• Common mistake: resetting a request's arrival time on re-queue, which punishes the customer for the agent's break, and letting a late resolve flip a re-assigned request to RESOLVED.

02Where this question was reported

• PracHub · Google onsite: agent/request state machines, fair re-queue, stale events, policy separated from entities, deterministic tests via injected timestamps