1. Fixed-Capacity Ring Buffer for Streaming Audio Chunks

Problem

A real-time voice pipeline can't let its audio buffer grow without bound while audio keeps streaming in — memory has to stay flat no matter how long the call runs. The standard fix is a fixed-capacity ring (circular) buffer: push new chunks in, and once it's full, pushing a new chunk silently evicts the oldest one still waiting.

Implement the ring buffer's core push/pop mechanics using a fixed-size array and wraparound indices, not a Python list that resizes. Every operation must run in O(1).

Source: src/1_ring_buffer_audio_chunks.py

def run_ring_buffer(capacity: int, ops: list[tuple[str, int | None]]) -> list[int | None]:
    ...

Examples:
>>> run_ring_buffer(2, [("push", 1), ("push", 2), ("push", 3), ("pop", None)])
[None, None, 1, 2]

>>> run_ring_buffer(1, [("push", 5), ("pop", None), ("pop", None)])
[None, 5, None]

Step-by-Step Approach

1. Back the buffer with a fixed-size array of length capacity, plus a head index (oldest live element) and a count of how many slots are currently occupied — never resize the array.
2. On push: if count < capacity, write into the next free slot at (head + count) % capacity and increment count; the push doesn't evict anything, so return None.
3. On push when count == capacity: the buffer is full, so overwrite the slot at head, advance head by one (mod capacity), and return the value that was evicted.
4. On pop: if count == 0 return None; otherwise read the value at head, advance head by one (mod capacity), decrement count, and return that value.
5. The modulo arithmetic is what gives wraparound — once head or the write index reaches the end of the array it should cycle back to index 0 rather than growing the array.
6. Process every op in order, collecting one result per op, and return the full results list.

The key insight is that head and count alone are enough state to treat a fixed array as a circular queue — you never shift elements or resize, so every push and pop is O(1) regardless of how long the stream runs.

Reference solution

def run_ring_buffer(capacity: int, ops: list[tuple[str, int | None]]) -> list[int | None]:
    # fixed-size backing array; head is the index of the oldest live element
    buf: list[int | None] = [None] * capacity
    head = 0
    count = 0
    results: list[int | None] = []
    for kind, value in ops:
        if kind == "push":
            if count == capacity:
                # full: overwrite the oldest slot and return what it held
                evicted = buf[head]
                buf[head] = value
                head = (head + 1) % capacity
                results.append(evicted)
            else:
                # not full: write into the next free slot after the tail
                tail = (head + count) % capacity
                buf[tail] = value
                count += 1
                results.append(None)
        else:  # "pop"
            if count == 0:
                results.append(None)
            else:
                results.append(buf[head])
                head = (head + 1) % capacity
                count -= 1
    return results

Key Functions & Tricks

• (head + count) % capacity — computes the next free write slot without a separate tail pointer.
• head = (head + 1) % capacity — the wraparound step that makes a fixed array behave as a circular queue.
• head + count instead of head + tail — tracking a count alongside head disambiguates "empty" from "full" (both would otherwise look like head == tail).
• count == capacity check — the single branch that decides whether a push is a plain write or an evict-and-overwrite.
• Fixed-size list preallocation ([None] * capacity) — guarantees O(1) writes with no resizing/copying ever occurring.

How to Recognize This Pattern