42. Min Agents for Overlapping Call Schedules

Problem

A voice-agent server accepts inbound calls that stay live for a known (start, end) window on the day's timeline. Each live call needs one dedicated worker process for its duration, and a worker can't handle two overlapping calls at once.

Given every call's (start, end) window, compute the minimum number of workers that must run concurrently to handle every call without ever double-booking one.

Source: src/42_min_agents_overlapping_calls.py

def min_workers_needed(call_intervals: list[tuple[int, int]]) -> int:
    ...

Examples:
>>> min_workers_needed([(0, 45), (10, 20), (25, 40)])
2

>>> min_workers_needed([(1, 5), (5, 10)])
1

Step-by-Step Approach

1. Recognize this as the “Meeting Rooms II” pattern: the answer is the maximum number of intervals ever simultaneously active, not anything about the intervals' order in the input.
2. Split each interval into its start and end, and sort the starts and the ends separately — the pairing between a specific start and end no longer matters for counting overlap.
3. Walk both sorted lists with two pointers, one for starts and one for ends, keeping a running count of workers currently busy.
4. At each step, compare the next unprocessed start to the next unprocessed end: if the start comes strictly first, a new call begins before any existing one has finished, so increment the worker count and advance the start pointer.
5. Otherwise, the earliest still-open call has ended by (or exactly when) the next one starts, so decrement the worker count and advance the end pointer instead — a call ending exactly when another starts frees the worker for reuse.
6. Track the running maximum of the worker count after every increment; that peak is the answer. An empty input needs zero workers.

The key insight is that once starts and ends are sorted independently, the answer reduces to a single sweep tracking “how many intervals have started but not yet ended” — you never need to know which specific call's end matches which start, only the running balance.

Reference solution

def min_workers_needed(call_intervals: list[tuple[int, int]]) -> int:
    # sort starts and ends separately, two-pointer sweep, O(n log n) time, O(n) space
    if not call_intervals:
        return 0
    starts = sorted(start for start, _ in call_intervals)
    ends = sorted(end for _, end in call_intervals)
    workers = 0
    max_workers = 0
    s_ptr = e_ptr = 0
    n = len(call_intervals)
    while s_ptr < n:
        if starts[s_ptr] < ends[e_ptr]:
            # a new call starts before the earliest still-open call ends: needs its own worker
            workers += 1
            max_workers = max(max_workers, workers)
            s_ptr += 1
        else:
            # the earliest open call has ended by the time the next one starts: free a worker
            workers -= 1
            e_ptr += 1
    return max_workers


TEST_CASES = [
    {"input": {"call_intervals": [(0, 45), (10, 20), (25, 40)]}, "expected": 2},
    {"input": {"call_intervals": []}, "expected": 0},
    {"input": {"call_intervals": [(2, 4), (6, 8), (10, 12)]}, "expected": 1},
    {"input": {"call_intervals": [(0, 20), (0, 20), (0, 20)]}, "expected": 3},
    {"input": {"call_intervals": [(1, 5), (5, 10)]}, "expected": 1},
]


def main():
    for i, case in enumerate(TEST_CASES):
        call_intervals = case["input"]["call_intervals"]
        expected = case["expected"]
        print(f"Test {i}: min_workers_needed(call_intervals={call_intervals})")
        if call_intervals:
            starts = sorted(start for start, _ in call_intervals)
            ends = sorted(end for _, end in call_intervals)
            print(f"  starts={starts}, ends={ends}")
            workers = 0
            max_workers = 0
            s_ptr = e_ptr = 0
            n = len(call_intervals)
            while s_ptr < n:
                if starts[s_ptr] < ends[e_ptr]:
                    workers += 1
                    max_workers = max(max_workers, workers)
                    print(f"  call starts at {starts[s_ptr]} -> workers={workers} (max so far {max_workers})")
                    s_ptr += 1
                else:
                    workers -= 1
                    print(f"  call ends at {ends[e_ptr]} -> workers={workers}")
                    e_ptr += 1
        result = min_workers_needed(call_intervals=call_intervals)
        assert result == expected
        print(f"PASSED: {result}")
    print(f"All {len(TEST_CASES)} test cases passed.")


if __name__ == "__main__":
    main()

Key Functions & Tricks

• starts = sorted(...) — all start times, sorted independently of which call they belong to
• ends = sorted(...) — all end times, sorted independently as well
• starts[s_ptr] < ends[e_ptr] — the comparison deciding whether to open a new worker or free one
• max_workers = max(max_workers, workers) — records the peak concurrency, which is the final answer

How to Recognize This Pattern