D1 — Resumable Iterator with Serializable State

45 minutes. Four gates. Narrate out loud, recorded.

Open one gate at a time. Do not scroll past a gate until its tests pass. The entire diagnostic value of this problem is that you cannot design for requirements you have not seen yet.

Table of Contents

• Setup
• The Prompt, As An Interviewer Would Say It
• Gate 1: Basic Resume
• Gate 2: Batching
• Gate 3: Map and Filter
• Gate 4: Flat Map and Mid-Group Checkpoints
• Timing Log
• Narration Self-Score

Setup

cp starter.py attempt.py
python3 test_diagnostic.py attempt.py           # runs all gates, reports per gate
python3 test_diagnostic.py attempt.py --gate 1  # run one gate only

Start the timer. Start the recorder.

The Prompt, As An Interviewer Would Say It

"I'd like you to build a resumable iterator. The idea is: you're processing a big stream of records, and the process can die at any point. When it comes back, you want to pick up where you left off — without re-processing anything you already emitted, and without holding the whole stream in memory.

So the object wraps a source, you iterate it normally, and at any point you can ask it for a checkpoint. That checkpoint needs to be something you could write to disk — JSON, basically. Later, you hand the checkpoint plus the same source back, and you get an iterator that produces exactly the items you hadn't gotten to yet.

Assume the source is replayable — you can call the factory again and get the same sequence. Start there and we'll build on it."

What is not said, and what you should clarify out loud:

• Is the source finite? (Assume it may be very large; do not materialize it.)
• Is the source deterministic across replays? (Yes — that is the premise.)
• What if the checkpoint is from a different source? (Out of scope; a version field is enough.)
• Does state() before any iteration mean "start from the beginning"? (Yes.)

Asking these earns real points. Assuming them silently does not.

Gate 1: Basic Resume

Implement ResumableIterator:

it = ResumableIterator(lambda: range(10))
first = [next(it) for _ in range(4)]          # [0, 1, 2, 3]
ckpt = it.state()                              # JSON-serializable dict
resumed = ResumableIterator.resume(lambda: range(10), ckpt)
rest = list(resumed)                           # [4, 5, 6, 7, 8, 9]

Requirements:

• ResumableIterator(source_factory) where source_factory is a zero-argument callable returning a fresh iterable.
• Implements the iterator protocol: __iter__ returns self, __next__ yields items, raises StopIteration at the end.
• state() returns a dict that survives json.dumps / json.loads round-tripping.
• ResumableIterator.resume(source_factory, state) is a classmethod returning a new iterator positioned after the last item that was emitted.
• The source must not be materialized into a list. The test asserts this by passing a generator-backed factory and counting how many items get pulled.
• state() on a fresh, un-iterated instance must round-trip to an iterator producing the full sequence.

Run: python3 test_diagnostic.py attempt.py --gate 1

Stop. Record the time. Only then read gate 2.

Gate 2: Batching

Add batch(n):

it = ResumableIterator(lambda: range(10))
batches = it.batch(3)
assert next(batches) == [0, 1, 2]
assert next(batches) == [3, 4, 5]
ckpt = it.state()
resumed = ResumableIterator.resume(lambda: range(10), ckpt)
assert list(resumed.batch(3)) == [[6, 7, 8], [9]]

Requirements:

• batch(n) returns an iterator of lists, each of length n except possibly the last.
• n <= 0 raises ValueError.
• A checkpoint taken after a yielded batch resumes exactly at the next unemitted item.
• batch must be lazy. The test uses a source that raises if pulled more than one batch ahead.

Run: python3 test_diagnostic.py attempt.py --gate 2

Stop. Record the time. Only then read gate 3.

Gate 3: Map and Filter

Add a transform pipeline. Transforms are declared at construction and are part of the iterator's identity, not part of the state.

it = ResumableIterator(lambda: range(10)).map(lambda x: x * 10).filter(lambda x: x % 20 == 0)
first = [next(it), next(it)]                   # [0, 20]
ckpt = it.state()
resumed = ResumableIterator.resume(
    lambda: range(10), ckpt
).map(lambda x: x * 10).filter(lambda x: x % 20 == 0)
assert list(resumed) == [40, 60, 80]

Requirements:

• .map(fn) and .filter(pred) are chainable and may be interleaved in any order.
• Transforms apply in declaration order.
• Resume must be exact. The concatenation of what you emitted before the checkpoint and what the resumed iterator emits must equal the un-checkpointed output, for a checkpoint taken at every position.
• The obvious implementation — "count emitted outputs, skip that many on resume" — is wrong here, and the tests will catch it. Think about what the checkpoint has to describe: a position in the source, or a position in the output?

Run: python3 test_diagnostic.py attempt.py --gate 3

Stop. Record the time. Only then read gate 4.

Gate 4: Flat Map and Mid-Group Checkpoints

Add .flat_map(fn), where fn maps one item to zero or more items.

it = ResumableIterator(lambda: range(4)).flat_map(lambda x: [x] * x)
# source 0,1,2,3 expands to groups [], [1], [2,2], [3,3,3]
# full output: [1, 2, 2, 3, 3, 3]
first = [next(it) for _ in range(4)]           # [1, 2, 2, 3]  <- checkpoint lands INSIDE the x=3 group
ckpt = it.state()
resumed = ResumableIterator.resume(lambda: range(4), ckpt).flat_map(lambda x: [x] * x)
assert list(resumed) == [3, 3]                 # not [3, 3, 3] — one was already emitted

Requirements:

• .flat_map(fn) chains with .map and .filter in any order and any depth.
• A checkpoint may land in the middle of an expanded group. Resuming must not re-emit the sub-items already emitted from that group, and must not skip the ones that remain. This is the whole gate. A source-index-only checkpoint is insufficient.
• The state dict stays JSON-serializable and stays O(1) in size — it must not grow with the number of items consumed. The test asserts a size bound.
• Still no materialization of the source.

Run: python3 test_diagnostic.py attempt.py --gate 4

Timing Log

Fill this in as you go. Copy it into ../scores/ afterwards.

Time-to-first-passing-gate: ______ minutes

This number matters more than the total. In a gated format, an unopened gate scores zero regardless of how good your unwritten design was — so the compounding metric is how fast you get something correct and extensible. See inference I3 in ../../research/findings.md.

Narration Self-Score

Listen back to the recording. Score 0–5, one point each:

Score: ___ / 5

Going silent while stuck is the most common and most costly narration failure. The interviewer cannot give you a hint they do not know you need, and silence reads as being lost even when you are thinking productively.