7. Barge-In Interrupt Priority Queue

Problem

A voice agent is mid-sentence when the caller starts talking over it. The moment that barge-in is detected, whatever the agent had queued up to say next should be scrapped — there's no point finishing a sentence the user has already interrupted.

The agent's response to the interruption should play as soon as it's ready, ahead of anything else. Given the full event stream of queued segments and detected interrupts, in arrival order, determine the actual final playback order.

Source: src/7_barge_in_priority_queue.py

def schedule_playback(events: list[tuple[str, str, int]]) -> list[str]:
    ...

Examples:
>>> schedule_playback([("play", "n1", 5), ("interrupt", "i1", 3), ("play", "n2", 5)])
["i1", "n2"]

>>> schedule_playback([("play", "n1", 1), ("play", "n2", 1)])
["n1", "n2"]

Step-by-Step Approach

1. Maintain two separate FIFO structures: a normal_queue for regular queued segments, and an interrupt_queue for detected barge-ins.
2. Process events in arrival order. A "play" event simply appends its segment id to the back of normal_queue.
3. An "interrupt" event does two things: it clears normal_queue entirely (everything queued but not yet played is scrapped), and it appends its own segment id to the back of interrupt_queue.
4. Interrupts never clear each other — only normal segments get dropped by a barge-in, so interrupt_queue only ever grows, never gets cleared mid-stream.
5. After processing every event, the final playback order is deterministic: every interrupt, in the order they were detected, followed by whatever normal segments survived (necessarily all queued after the very last interrupt, since any earlier ones were wiped).
6. Handle the edge cases directly: no interrupts at all just returns normal_queue unchanged; interrupts with nothing queued between them simply accumulate back-to-back with no normal segments in between.

The key insight is that you don't need to simulate playback time or clock ticks at all — because an interrupt always wins immediately and unconditionally, the whole schedule collapses to "all interrupts, then whatever's left," which two queues and one clear operation are enough to compute.

Reference solution

def schedule_playback(events: list[tuple[str, str, int]]) -> list[str]:
    normal_queue: list[str] = []
    interrupt_queue: list[str] = []
    for kind, segment_id, _duration in events:
        if kind == "interrupt":
            # a barge-in wipes out anything queued but not yet played
            normal_queue.clear()
            interrupt_queue.append(segment_id)
        else:
            normal_queue.append(segment_id)
    # interrupts always win, in the order they were detected; survivors play after
    return interrupt_queue + normal_queue

Key Functions & Tricks

• normal_queue.clear() — the entire preemption mechanism: one call, triggered only by an interrupt event.
• Two independent FIFO lists instead of one priority queue — since priority is binary (interrupt beats normal, always), a heap is unnecessary overhead; plain lists in arrival order suffice.
• interrupt_queue + normal_queue — the final merge only needs to happen once, at the end, because within each queue relative order never changes after appends.
• Unpacking _duration with a throwaway name — the field is part of the event's realistic shape but unused by this scheduling policy.

How to Recognize This Pattern