25. Minimum Workers to Run All Evaluation Jobs

Problem

An eval harness receives a batch of evaluation jobs, each with a [start, end) time window on a shared pool of identical workers. A worker can only run one job at a time, but can immediately pick up a new job the instant the previous one ends.

Given the jobs' start and end times, return the minimum number of workers needed to run every job without ever double-booking a worker.

Source: src/25_heap_min_workers_scheduling.py

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

Examples:
>>> min_workers_needed([(0, 30), (5, 10), (15, 20)])
2

>>> min_workers_needed([(7, 10), (2, 4)])
1

Step-by-Step Approach

1. Recognize "minimum resources to handle all overlapping intervals" as the signature use case for a min-heap of end times, processed in start-time order — this is the "meeting rooms II" pattern.
2. Sort the jobs by start time, so you always consider the next job to begin in chronological order.
3. Maintain a min-heap holding the end times of jobs currently assigned to a worker; the heap's root is always the worker that frees up soonest.
4. For each job in order: if the heap's smallest end time is <= the job's start time, that worker is already free by the time this job needs to start — reuse it (pop the old end time, push the new one).
5. Otherwise, every currently-occupied worker is still busy when this job needs to start, so allocate a brand-new worker (push its end time onto the heap without popping anything).
6. After processing every job, the heap's size is the peak number of workers that were simultaneously in use — and that peak is exactly the minimum number of workers required overall.

The key insight is that sorting by start time plus always checking against the soonest-to-free worker (the heap's root) is enough to guarantee an optimal reuse decision at every step — you never need to look at more than one candidate worker to know whether reuse is possible.

Reference solution

import heapq


def min_workers_needed(jobs: list[tuple[int, int]]) -> int:
    if not jobs:
        return 0
    jobs_sorted = sorted(jobs, key=lambda job: job[0])
    heap: list[int] = []  # min-heap of end times for workers currently occupied
    for start, end in jobs_sorted:
        if heap and heap[0] <= start:
            # the earliest-freeing worker is free by `start`: reuse it instead of
            # allocating a new one
            heapq.heapreplace(heap, end)
        else:
            heapq.heappush(heap, end)  # no free worker: allocate a new one
    return len(heap)

Key Functions & Tricks

• Sort by job[0] (start time) — processing jobs chronologically is what makes the greedy reuse decision correct
• heap[0] — O(1) peek at the worker that frees up soonest, without popping it
• heapq.heapreplace(heap, end) — atomic pop-then-push: reuse a freed worker in one O(log n) operation
• heapq.heappush(heap, end) — allocate a new worker when no existing one is free yet
• len(heap) at the end — the heap's final size is exactly the peak concurrent worker count

How to Recognize This Pattern