23. Coalesce Outbound Calls by Destination

Problem

A voice agent sometimes needs to notify third-party systems mid-conversation (a CRM update, a webhook) and those calls are cheaper in bulk. Given a single multiplexed stream of outbound call events — each timestamped and tagged with a destination — group each destination's own calls into batches independently.

A batch for a destination closes when it reaches max_size calls, or when the next call to that same destination would arrive more than max_wait seconds after that batch's first call, whichever happens first. Different destinations' batches are entirely independent of each other.

Source: src/23_coalesce_calls_by_destination.py

def coalesce_by_destination(
    events: list[tuple[float, str]], max_size: int, max_wait: float
) -> dict[str, list[list[float]]]: ...

>>> coalesce_by_destination([(0, "crm"), (1, "webhook"), (2, "crm"), (3, "crm")], max_size=2, max_wait=5)
{'crm': [[0, 2], [3]], 'webhook': [[1]]}

Step-by-Step Approach

1. Since events arrive as one interleaved but globally time-sorted stream, sweep through them once, dispatching each event to per-destination bookkeeping rather than sorting or grouping up front.
2. Keep a dict of "currently open batch" per destination, each holding the batch's first timestamp and its accumulated items so far.
3. For each incoming (t, dest) event, check whether that destination's open batch (if any) must close first: either it's already at max_size, or admitting this event would push it more than max_wait past the batch's first timestamp.
4. If it must close, move it into the destination's list of finished batches and clear the open slot; either way, if there's no open batch for this destination now, start a new one anchored at the current event's timestamp.
5. Otherwise, append the event's timestamp directly onto the currently open batch.
6. Once the stream ends, flush every destination's still-open batch into its finished list — a batch doesn't need to hit its size or wait limit to be a valid final batch.

The key insight is that per-destination coalescing is just the single-stream greedy-batching sweep, fanned out over a hashmap keyed by destination instead of assumed to be one global stream — each destination's batch-closing decision is still anchored to that batch's own first timestamp, so it stays a single O(n) pass with no sorting or per-destination re-scanning.

Reference solution

from collections import defaultdict


def coalesce_by_destination(
    events: list[tuple[float, str]], max_size: int, max_wait: float
) -> dict[str, list[list[float]]]:
    # One "open batch" per destination at a time, keyed in a dict -- this is
    # the coalescing sweep from a single-stream batcher, just fanned out by
    # key instead of assuming one global stream. Still O(n) overall since
    # each event is looked at once.
    open_batches: dict[str, dict] = {}
    closed: dict[str, list[list[float]]] = defaultdict(list)

    def close(dest: str) -> None:
        batch = open_batches.pop(dest)
        closed[dest].append(batch["items"])

    for t, dest in events:
        batch = open_batches.get(dest)
        if batch is not None and (
            len(batch["items"]) >= max_size or t - batch["first"] > max_wait
        ):
            close(dest)
            batch = None
        if batch is None:
            open_batches[dest] = {"first": t, "items": [t]}
        else:
            batch["items"].append(t)

    # flush whatever's still open once the stream ends
    for dest in list(open_batches.keys()):
        close(dest)

    return dict(closed)

Key Functions & Tricks

• open_batches dict keyed by destination — isolates each destination's own greedy batching state so they never interfere with each other.
• defaultdict(list) for closed — appends a finished batch to a destination's list without an existence check.
• Anchoring max_wait to the batch's first timestamp, not the previous event — a fixed origin per batch that lets a batch close definitively without revisiting it.
• Final flush loop — a batch still open when the stream ends is a valid batch too, not a discarded partial one.

How to Recognize This Pattern