📞 Machine Coding: Call Center Dispatch System

📝 Overview

A multi-tiered call center dispatch system that routes incoming customer calls to the lowest available employee level. It handles dynamic escalation of calls through a hierarchy of operators, supervisors, and directors, safely queuing requests when staff are unavailable.

Why This Challenge?

• Resource Pool Management: Evaluates your skills in managing pools of shared resources (employees) and handling pending task queues.
• Hierarchical Escalation: Tests your ability to implement chain-of-responsibility-like logic for escalating tasks when conditions aren't met.
• Object-Oriented Design: A strong test of encapsulation, inheritance, and decoupling business rules from entity state.

---

🏭 The Scenario & Requirements

😡 The Problem (The Villain)

"The Infinite Hold Loop." In a chaotic, poorly designed call center, calls are routed randomly. Customers needing a simple password reset are sent to a busy Director, while Tier-1 Operators sit idle. If a call is too complex for an Operator, they have no systematic way to hand it off, resulting in dropped calls, lost context, and frustrated customers.

🦸 The System (The Hero)

"The Tiered Dispatcher." An automated routing engine that groups employees into distinct ranks. It always attempts to assign calls to the lowest tier (Operator) first to optimize payroll efficiency. If unresolved, it cleanly elevates the call's required rank and pushes it back to the dispatcher for escalation. If the entire center is at capacity, calls are safely held in a FIFO queue.

📜 Requirements & Constraints

1. (Functional): Must handle 3 distinct levels of employees: Operator, Supervisor, Director.
2. (Functional): Calls must be dispatched to the lowest available rank that can handle them, but a higher rank can take a lower-rank call if necessary.
3. (Functional): Employees must be able to escalate calls to the next tier if they cannot resolve them.
4. (Technical): If no suitable employee is available, calls must be queued and automatically dispatched when an employee finishes their current call.

---

🏗️ Design & Architecture

🧠 Thinking Process

To decouple the workers from the routing logic, we model three primary entities:
1. Call: The fundamental unit of work. It maintains its own state (READY, IN_PROGRESS) and tracks the minimum Rank required to handle it.
2. Employee: An abstract base class providing core methods like take_call and complete_call. Concrete subclasses (Operator, Supervisor) define what happens during an escalate_call.
3. CallCenter: The central orchestrator that maintains the lists of employees and the queued_calls deque. It acts as the single source of truth for dispatching.

🧩 Class Diagram

(The Object-Oriented Blueprint. Who owns what?)

classDiagram
    direction TB
    class CallCenter {
        -List~Operator~ operators
        -List~Supervisor~ supervisors
        -List~Director~ directors
        -Deque~Call~ queued_calls
        +dispatch_call(call)
        +notify_call_escalated(call)
    }
    class Employee {
        <<abstract>>
        +Rank rank
        +Call call
        +take_call(call)
        +complete_call()
        +escalate_call()*
    }
    class Call {
        +Rank rank
        +CallState state
    }

    CallCenter o-- Employee : manages
    CallCenter o-- Call : queues
    Employee --> Call : handles
    Employee <|-- Operator
    Employee <|-- Supervisor
    Employee <|-- Director

⚙️ Design Patterns Applied

• Chain of Responsibility (Conceptual): Calls logically escalate up the hierarchy (Operator \(\rightarrow\) Supervisor \(\rightarrow\) Director) if a lower rank cannot handle the request.
• State Pattern (Simplified): The Call object transitions through specific lifecycle states (READY, IN_PROGRESS, COMPLETE).
• Dependency Inversion: The CallCenter routing mechanism depends on the abstract Employee interface to assign work, rather than hardcoding assignments to specific roles.

---

💻 Solution Implementation

The Code

from abc import ABC, abstractmethod
from collections import deque
from enum import Enum
from typing import Optional, Deque, Sequence


class Rank(Enum):
    """Represents the rank hierarchy of an employee in the call center."""
    OPERATOR = 0
    SUPERVISOR = 1
    DIRECTOR = 2


class CallState(Enum):
    """Represents the lifecycle state of a customer call."""
    READY = 0
    IN_PROGRESS = 1
    COMPLETE = 2


class Call:
    """Encapsulates a customer call requiring a minimum employee rank."""

    def __init__(self, rank: Rank) -> None:
        """
        Initializes a Call.

        Args:
            rank (Rank): The minimum rank required to handle this call.
        """
        self.state: CallState = CallState.READY
        self.rank: Rank = rank
        self.employee: Optional['Employee'] = None


class Employee(ABC):
    """Abstract base class representing a generic call center employee."""

    def __init__(self, employee_id: str, name: str, rank: Rank, call_center: 'CallCenter') -> None:
        self.employee_id: str = employee_id
        self.name: str = name
        self.rank: Rank = rank
        self.call: Optional[Call] = None
        self.call_center: 'CallCenter' = call_center

    def take_call(self, call: Call) -> None:
        """
        Assigns a call to the employee and updates its state.

        Args:
            call (Call): The call to be handled.
        """
        self.call = call
        self.call.employee = self
        self.call.state = CallState.IN_PROGRESS

    def complete_call(self) -> None:
        """Completes the current call and notifies the call center to free the employee."""
        if self.call:
            self.call.state = CallState.COMPLETE
            self.call_center.notify_call_completed(self.call)
            self.call = None

    @abstractmethod
    def escalate_call(self) -> None:
        """Escalates the current call to a higher-ranking employee tier."""
        pass

    def _escalate_call(self) -> None:
        """Internal helper to reset the call state and trigger the center's escalation protocol."""
        if self.call:
            self.call.state = CallState.READY
            call_to_escalate = self.call
            self.call = None
            self.call_center.notify_call_escalated(call_to_escalate)


class Operator(Employee):
    """Represents an operator (Tier 1) employee."""

    def __init__(self, employee_id: str, name: str, call_center: 'CallCenter') -> None:
        super().__init__(employee_id, name, Rank.OPERATOR, call_center)

    def escalate_call(self) -> None:
        if self.call:
            self.call.rank = Rank.SUPERVISOR
            self._escalate_call()


class Supervisor(Employee):
    """Represents a supervisor (Tier 2) employee."""

    def __init__(self, employee_id: str, name: str, call_center: 'CallCenter') -> None:
        super().__init__(employee_id, name, Rank.SUPERVISOR, call_center)

    def escalate_call(self) -> None:
        if self.call:
            self.call.rank = Rank.DIRECTOR
            self._escalate_call()


class Director(Employee):
    """Represents a director (Tier 3) employee."""

    def __init__(self, employee_id: str, name: str, call_center: 'CallCenter') -> None:
        super().__init__(employee_id, name, Rank.DIRECTOR, call_center)

    def escalate_call(self) -> None:
        raise NotImplementedError("Directors are the highest tier and must handle the call.")


class CallCenter:
    """Orchestrator class that manages employee pools and dispatches incoming calls."""

    def __init__(self, operators: Sequence[Operator], supervisors: Sequence[Supervisor], directors: Sequence[Director]) -> None:
        self.operators: Sequence[Operator] = operators
        self.supervisors: Sequence[Supervisor] = supervisors
        self.directors: Sequence[Director] = directors
        self.queued_calls: Deque[Call] = deque()

    def dispatch_call(self, call: Call) -> None:
        """
        Dispatches a call to an available employee matching or exceeding the required rank.
        Queues the call if no suitable employee is available.

        Args:
            call (Call): The call to be dispatched.
        """
        employee: Optional[Employee] = None

        if call.rank == Rank.OPERATOR:
            employee = self._dispatch_to_group(call, self.operators)

        if employee is None and call.rank.value <= Rank.SUPERVISOR.value:
            employee = self._dispatch_to_group(call, self.supervisors)

        if employee is None and call.rank.value <= Rank.DIRECTOR.value:
            employee = self._dispatch_to_group(call, self.directors)

        if employee is None:
            self.queued_calls.append(call)

    def _dispatch_to_group(self, call: Call, employees: Sequence[Employee]) -> Optional[Employee]:
        """Iterates through an employee group to find an available handler."""
        for employee in employees:
            if employee.call is None:
                employee.take_call(call)
                return employee
        return None

    def notify_call_escalated(self, call: Call) -> None:
        """Re-enters an escalated call into the dispatch routing logic."""
        self.dispatch_call(call)

    def notify_call_completed(self, call: Call) -> None:
        """Triggered when an employee finishes a call; pulls the next priority call from the queue."""
        if self.queued_calls:
            next_call = self.queued_calls.popleft()
            self.dispatch_call(next_call)

🔬 Why This Works (Evaluation)

The system excels because of its Decoupled Escalation Protocol. When an Operator escalates a call, they do not directly search for a Supervisor. Instead, they increment the call's required Rank, transition the state back to READY, and trigger CallCenter.notify_call_escalated(). This pushes the routing responsibility back to the CallCenter, keeping the Employee classes strictly focused on their own state and preventing spaghetti dependencies.

---

⚖️ Trade-offs & Limitations

Decision	Pros	Cons / Limitations
Centralized Dispatcher (CallCenter)	Simple, predictable logic with a single source of truth for routing.	Becomes a locking bottleneck in a highly concurrent, multi-threaded environment.
Linear Search for Free Employee	Extremely easy to implement and debug.	\(O(N)\) time complexity per dispatch. Can be slow if there are thousands of employees.
Simple FIFO Queue	Guarantees fairness for waiting customers.	Ignores call priority; a VIP customer or emergency call cannot skip the line.

---

🎤 Interview Toolkit

• Concurrency Probe: How would you handle 500 calls arriving concurrently across multiple threads? (Use a thread-safe queue.Queue for pending calls and implement a threading.Lock around the _dispatch_to_group method to prevent double-assigning an employee).
• Extensibility: How easily can we add a "Specialist" rank? (Very easily: Add it to the Rank enum, create a Specialist(Employee) subclass, and add one routing block in CallCenter.dispatch_call()).
• Optimization: If you had 10,000 operators, how would you speed up finding a free one? (Instead of iterating through a list, maintain a Set or Queue of specifically "Idle" employees, achieving \(O(1)\) assignment).

🔗 Related Challenges

• Elevator Management System — For another resource allocation challenge involving state machines and dispatching.
• High-Concurrency Parking Lot — For managing tiered resources where specific requests require specific slot types.