AI Lab Coding Interview: Python Systems

Personalize this lesson

Adapt explanations and teaching visuals to your background and preferred voice.

You just designed a reasoning agent that budgets test-time compute and recovers from failed tool calls. A lab coding round asks a smaller question: can you implement one slice of that system while the requirements keep moving?

You're in a shared editor with a starting URL and a get_links callback. Twelve minutes later they want worker fanout. Then a per-host rate limit. Then cancellation. People who stall didn't forget BFS. They lost the visited invariant the moment the second requirement arrived.

OpenAI's engineering interviews look for well-designed solutions, high-quality code, performance, and test coverage, plus communication you can follow while you work.^{[1]Reference 1Interview guidehttps://openai.com/interview-guide/} Anthropic's technical process uses live coding tools such as Colab and CodeSignal; they expect you to be fluent in syntax and the standard library so lookup time doesn't eat the round, and they'll ask about experience and motivation.^{[2]Reference 2Careershttps://www.anthropic.com/careers} Formats vary by team. Practice that bar with one base prompt and staged follow-ups: TTLs, concurrency, cancellation, rate limits, deterministic output, and a state model that still makes sense when the next constraint lands.

Keep one invariant visible

Use this loop for every prompt:

Restate input, output, and failure behavior.
Ship version 1 with the smallest correct state model.
Add table-driven tests before adding stage 2.
Isolate shared mutable state before adding threads.
End by naming complexity, race risks, and production hardening.

The code that survives a staged round is the version whose visited set, TTL record, or ledger key you can still defend after the next requirement lands.

Time	Target
0-5 min	clarify ordering, failure behavior, and mutable state
5-20 min	level 1 complete with tests
20-32 min	level 2 or 3 complete without rewriting
32-38 min	edge tests and complexity
final minutes	name the next follow-up, the race risk, and what you'd harden in production

Python tools to know cold

Know these without documentation:

Need	Python building block
FIFO work queue	collections.deque, queue.Queue, asyncio.Queue
Counts and top errors	collections.Counter
LRU cache	collections.OrderedDict
Deadlines and TTL	time.monotonic, injected `now` function
Priority scheduling	heapq, queue.PriorityQueue
Thread safety	threading.Lock, threading.RLock, threading.Condition, threading.Event
Worker fanout	concurrent.futures.ThreadPoolExecutor, as_completed
Parsing	splitlines, re, explicit state machines

Don't wait for a test framework. Write a small run_tests() and use plain assert.

Same-host crawler

The running example is a CI-docs crawler. You're given https://repo.example/start and get_links(url) -> list[str]. Visit only URLs on the same host. Return them in deterministic breadth-first order for the single-threaded version. Don't visit the same URL twice. Ignore malformed URLs and off-host URLs.

Clarifying questions to ask before you type:

Should URLs be normalized for fragments, trailing slashes, query strings, and redirects?
Should failed fetches be retried, skipped, or returned separately?
Does deterministic order still matter after worker fanout, or only for the single-threaded version?

The invariant is: a URL enters the queue only after it has been accepted into visited. Put host checks and that claim before fetch. Relative links such as /tests need urljoin so they become https://repo.example/tests.

Diagram showing candidate URL, claim into visited, fetch outside lock, and already claimed. — candidate URL, claim into visited, fetch outside lock, and already claimed.

same-host-crawler.py

from collections import deque
from urllib.parse import urlparse, urljoin

PAGES = {
    "https://repo.example/start": ["/tests", "/ci", "https://other.example/x"],
    "https://repo.example/tests": ["/ci"],
    "https://repo.example/ci": [],
}

def crawl(start_url: str, get_links) -> list[str]:
    start = urlparse(start_url)
    if not start.scheme or not start.netloc:
        return []

    visited = {start_url}
    queue = deque([start_url])
    ordered: list[str] = []

    while queue:
        url = queue.popleft()
        ordered.append(url)

        for raw_link in get_links(url):
            candidate = urljoin(url, raw_link)
            parsed = urlparse(candidate)
            if parsed.scheme not in {"http", "https"}:
                continue
            if parsed.netloc != start.netloc:
                continue
            if candidate in visited:
                continue
            visited.add(candidate)
            queue.append(candidate)

    return ordered

visited_order = crawl("https://repo.example/start", lambda url: PAGES.get(url, []))
print(visited_order)
assert visited_order == [
    "https://repo.example/start",
    "https://repo.example/tests",
    "https://repo.example/ci",
]

Output

['https://repo.example/start', 'https://repo.example/tests', 'https://repo.example/ci']

/tests is enqueued before /ci, so BFS returns start, tests, then ci. The off-host https://other.example/x never enters visited. When tests later points at /ci, the claim already happened, so you don't fetch it twice.

That single-threaded version still has a hole once two workers can call get_links at the same time. If you fetch first and mark visited later, both workers can pick /ci.

Two workers race for https://repo.example/ci. The left path fetches before marking visited, so both workers call get_links. The right path adds /ci to visited under a lock, then fetches outside the lock, so worker B skips. — Two workers race for `/ci`. Fetch-then-mark lets both call `get_links`. Claim-then-fetch adds `/ci` to `visited` under the lock first, so only one worker fetches.

Protect the claim step, not the whole fetch. Deterministic return order gets expensive once workers run in parallel, unless the prompt still requires it. Say that out loud before you add a global lock just to keep BFS order.

crawler-claim-step.py

from threading import Lock

visited = {"https://repo.example/start"}
visited_lock = Lock()

def claim(candidate: str) -> bool:
    with visited_lock:
        if candidate in visited:
            return False
        visited.add(candidate)
        return True

print(claim("https://repo.example/ci"))
print(claim("https://repo.example/ci"))
print(sorted(visited))

Output

True
False
['https://repo.example/ci', 'https://repo.example/start']

For rate limiting, add a per-host token bucket or next-allowed timestamp before fetch. For cancellation, check a threading.Event before scheduling new work, before fetching, and after each fetch before enqueueing children. Name all three behaviors: duplicate prevention, bounded fetch concurrency, and a cooperative stop.

Claim-before-fetch still leaves failed fetches stuck unless you model more than one terminal state:

State	Meaning	Retry?
`in_flight`	Claimed, fetch not finished	No second worker
`visited` / success	Content processed	No
`failed` / retry set	Hard failure with policy remaining	Yes, under backoff

When the interviewer asks about failed fetches, say: the claim still happens under the lock, but success and permanent failure are different outcomes. A production sketch keeps visited for terminal success, in_flight for active work, and a retry queue with attempt counts so a timeout doesn't become permanent silence.

The crawler taught claim-before-mutate. The next prompt asks the same ownership question on a store: every public read has to observe expiry, not just get.

TTL key/value store

Implement an in-memory store with set(key, value, ttl=None), get(key), delete(key), and scan(prefix). ttl is in seconds. Expired keys behave as missing. Tests must not call sleep.

Ask:

Should scan(prefix) return keys, values, or key/value pairs?
Is a key expired when expires_at == now, or only when expires_at < now?
Should compare-and-set treat an expired key as missing?

Inject now so tests can jump the clock. Store expires_at beside each value. The invariant is: every public read path either returns a non-expired value or removes the expired key. This store treats equality as expired (expires_at <= now), so a key set at time 10.0 with ttl=5.0 is gone at 15.0.

ttl-store.py

from dataclasses import dataclass
from typing import Callable

@dataclass
class Entry:
    value: str
    expires_at: float | None

class Store:
    def __init__(self, now: Callable[[], float]) -> None:
        self.now = now
        self.items: dict[str, Entry] = {}

    def set(self, key: str, value: str, ttl: float | None = None) -> None:
        expires_at = None if ttl is None else self.now() + ttl
        self.items[key] = Entry(value, expires_at)

    def get(self, key: str) -> str | None:
        entry = self.items.get(key)
        if entry is None:
            return None
        if entry.expires_at is not None and entry.expires_at <= self.now():
            self.items.pop(key, None)
            return None
        return entry.value

    def delete(self, key: str) -> None:
        self.items.pop(key, None)

    def scan(self, prefix: str) -> list[str]:
        return sorted(
            key
            for key in list(self.items)
            if key.startswith(prefix) and self.get(key) is not None
        )

clock = {"now": 10.0}
store = Store(lambda: clock["now"])
store.set("task:1", "running", ttl=5.0)
store.set("task:2", "ready")
print(store.get("task:1"), store.scan("task:"))
clock["now"] = 15.0
print(store.get("task:1"), store.scan("task:"))

Output

running ['task:1', 'task:2']
None ['task:2']

scan calls get, so expiry cleanup isn't a get-only side path. Snapshot list(self.items) first; get may pop keys while you iterate.

Compare-and-set and locking

Compare-and-set must clean any expired value and compare while holding the same lock. The compact example below omits TTL so the atomic boundary is easy to see:

compare-and-set.py

from threading import RLock

class AtomicStore:
    def __init__(self) -> None:
        self.items: dict[str, str] = {}
        self.lock = RLock()

    def compare_and_set(self, key: str, expected: str | None, value: str) -> bool:
        with self.lock:
            if self.items.get(key) != expected:
                return False
            self.items[key] = value
            return True

store = AtomicStore()
print(store.compare_and_set("task:1", None, "running"))
print(store.compare_and_set("task:1", None, "ready"))
print(store.compare_and_set("task:1", "running", "ready"))

Output

True
False
True

For the TTL store, wrap every public method with the same lock. If compare_and_set() calls get() and set() while holding that lock, use an RLock or split the internal helpers so the lock is acquired once. What you have to defend is atomicity: no other thread can change the key between compare and set.

Expiry is a time boundary on one record. A scheduler adds a graph: some tasks are ready, some retry, and some never run.

Task scheduler with retries

Schedule CI tasks with dependencies. A task becomes runnable when all dependencies have completed. Detect dependency cycles before running anything. Run ready tasks in priority order (lower number first). Retry failed tasks up to max_attempts. Return completed tasks, permanently failed tasks, and blocked dependents separately.

Ask:

Does a lower priority number run first, or does a higher number win?
If a dependency permanently fails, should dependents be marked failed, skipped, or blocked?
Are retries immediate, delayed, or scheduled with backoff?

Represent the graph explicitly: dependents[task] lists tasks released by this task, and remaining[task] counts unmet dependencies. Use heapq for priority. The invariant is: a task enters the heap only when remaining[task] == 0. Three-state DFS (0 unseen, 1 active, 2 done) rejects a back edge to an active node in O(V + E) before run is ever called.

dependency-scheduler.py

import heapq
from collections import defaultdict
from dataclasses import dataclass

@dataclass(frozen=True)
class Task:
    name: str
    priority: int
    deps: tuple[str, ...] = ()

def schedule(tasks: list[Task], run, max_attempts: int = 2) -> tuple[list[str], list[str], list[str]]:
    if max_attempts <= 0:
        raise ValueError("max_attempts must be positive")
    by_name = {task.name: task for task in tasks}
    if len(by_name) != len(tasks):
        raise ValueError("duplicate task name")
    dependents: dict[str, list[str]] = defaultdict(list)
    remaining = {task.name: len(task.deps) for task in tasks}

    for task in tasks:
        for dep in task.deps:
            if dep not in by_name:
                raise ValueError(f"unknown dependency: {dep}")
            dependents[dep].append(task.name)

    state: dict[str, int] = {}

    def visit(name: str) -> None:
        marker = state.get(name, 0)
        if marker == 1:
            raise ValueError("cycle")
        if marker == 2:
            return
        state[name] = 1
        for dep in by_name[name].deps:
            visit(dep)
        state[name] = 2

    for name in by_name:
        visit(name)

    ready = [(task.priority, task.name) for task in tasks if remaining[task.name] == 0]
    heapq.heapify(ready)
    attempts = defaultdict(int)
    completed: list[str] = []
    failed: list[str] = []

    while ready:
        _, name = heapq.heappop(ready)
        attempts[name] += 1
        if not run(name):
            if attempts[name] < max_attempts:
                heapq.heappush(ready, (by_name[name].priority, name))
            else:
                failed.append(name)
            continue

        completed.append(name)
        for child in dependents[name]:
            remaining[child] -= 1
            if remaining[child] == 0:
                heapq.heappush(ready, (by_name[child].priority, child))

    blocked = sorted(set(by_name) - set(completed) - set(failed))
    return completed, failed, blocked

tasks = [
    Task("docs", priority=1),
    Task("test", priority=2),
    Task("deploy", priority=3, deps=("test",)),
]
fail_once = {"test"}

def run_with_one_retry(name: str) -> bool:
    if name in fail_once:
        fail_once.remove(name)
        return False
    return True

print(schedule(tasks, run_with_one_retry))
print(schedule([Task("test", 1), Task("deploy", 2, ("test",))], lambda _: False, max_attempts=1))
try:
    schedule(
        [Task("lint", 1, ("test",)), Task("test", 1, ("lint",))],
        lambda _: True,
    )
except ValueError as error:
    print(error)

Output

(['docs', 'test', 'deploy'], [], [])
([], ['test'], ['deploy'])
cycle

docs and test start ready. Priority 1 runs first, so docs completes. test fails once, retries, then succeeds and releases deploy. In the second call, test is permanently failed after one attempt, so deploy is blocked: it never ran. The cycle lint ↔ test is rejected before any run call, including for unrelated ready work.

Keep permanently failed tasks separate from blocked dependents. A blocked task never ran; callers may skip it, surface the failed prerequisite, or retry after repair.

Worker fanout

Keep graph construction and cycle detection single-threaded. Then protect only shared scheduler state: ready heap, attempts, completed, failed, and remaining dependency counts. Worker threads can run tasks outside the lock, then reacquire the lock to publish success or failure and release dependents.

If asked about retries, say whether retries preserve priority or use backoff. A concise full answer: ready tasks are claimed under a condition variable, execution happens outside the lock, and completion updates notify workers when new tasks become ready.

The crawler still needs a quota. That's the next drill: refill math with an injected clock, no sleep.

Token bucket with deterministic time

Rate limiters show up because they combine state, boundary conditions, and overload behavior. Retry storms get worse when every client wakes at the same backoff; jitter spreads those retries.^{[3]Reference 3Exponential Backoff And Jitterhttps://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/} Inject now so the refill test can jump to 0.5s without sleeping.

The bucket below starts full. allow refills from elapsed time, then either consumes cost or returns how long the caller should wait. Impossible costs fail immediately instead of returning retry advice that can never succeed.

token-bucket.py

import math
from dataclasses import dataclass

@dataclass
class Bucket:
    capacity: float
    refill_per_second: float
    tokens: float
    updated_at: float

class TokenBucketLimiter:
    def __init__(self, capacity: int, refill_per_second: float) -> None:
        if capacity <= 0:
            raise ValueError("capacity must be positive")
        if not math.isfinite(refill_per_second) or refill_per_second <= 0:
            raise ValueError("refill_per_second must be positive and finite")
        self.capacity = float(capacity)
        self.refill_per_second = float(refill_per_second)
        self._buckets: dict[str, Bucket] = {}

    def allow(self, key: str, now: float, cost: float = 1.0) -> tuple[bool, float]:
        if not math.isfinite(now):
            raise ValueError("now must be finite")
        if not math.isfinite(cost) or not 0 < cost <= self.capacity:
            raise ValueError("cost must be greater than zero and no larger than capacity")
        bucket = self._buckets.get(key)
        if bucket is None:
            bucket = Bucket(self.capacity, self.refill_per_second, self.capacity, now)
            self._buckets[key] = bucket
        if now < bucket.updated_at:
            raise ValueError("now must not move backwards")

        elapsed = max(0.0, now - bucket.updated_at)
        bucket.tokens = min(bucket.capacity, bucket.tokens + elapsed * bucket.refill_per_second)
        bucket.updated_at = now

        if bucket.tokens >= cost:
            bucket.tokens -= cost
            return True, 0.0

        missing = cost - bucket.tokens
        retry_after = missing / bucket.refill_per_second
        return False, retry_after

limiter = TokenBucketLimiter(capacity=3, refill_per_second=1.0)
print([limiter.allow("org-a", now=0.0)[0] for _ in range(4)])
print(limiter.allow("org-a", now=0.5))
print(limiter.allow("org-a", now=1.0))
try:
    limiter.allow("org-a", now=2.0, cost=4.0)
except ValueError as error:
    print(error)

Output

[True, True, True, False]
(False, 0.5)
(True, 0.0)
cost must be greater than zero and no larger than capacity

Token counts after refill for a capacity-3 bucket that refills at 1 token per second. Three cost-1 allows at t=0 leave 0 tokens. The t=0 fourth request and the t=0.5 request stay below cost 1, so both deny. At t=1.0 the bar reaches 1.0 and the request is allowed. — After three allows at t=0 the bucket is empty, so t=0.5 still denies: only 0.5 tokens have refilled and retry-after is 0.5s. At t=1.0 the bar hits cost 1 and the request goes through.

The key can be a user, organization, endpoint, or model. Cleanup for idle buckets is a production memory concern, not a correctness requirement for the base prompt. A thread-safe version needs a lock around _buckets and bucket mutation.

Denied quota calls get retried. If the retry is a debit against an org's remaining credits, "run it again" is the wrong model. You need an idempotency key.

Credit ledger with idempotency

Keep credit-like state consistent: org token budgets, GPU-second reservations, or eval-run spend. Use append-only events when you can. If you also keep a balance, update the balance and the event together.

ledger-idempotency.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Event:
    idempotency_key: str
    account: str
    delta: int
    balance_after: int

class Ledger:
    def __init__(self) -> None:
        self.balance: dict[str, int] = {}
        self.events: list[Event] = []
        self.results: dict[str, Event] = {}

    def apply(self, key: str, account: str, delta: int) -> int:
        if key in self.results:
            prior = self.results[key]
            if (prior.account, prior.delta) != (account, delta):
                raise ValueError("idempotency key reused with different operation")
            return prior.balance_after
        new_balance = self.balance.get(account, 0) + delta
        if new_balance < 0:
            raise ValueError("insufficient credits")
        self.balance[account] = new_balance
        event = Event(key, account, delta, new_balance)
        self.events.append(event)
        self.results[key] = event
        return new_balance

ledger = Ledger()
print(ledger.apply("deposit-1", "org-a", 100))
print(ledger.apply("withdraw-1", "org-a", -30))
print(ledger.apply("deposit-1", "org-a", 100))
print(ledger.balance["org-a"], len(ledger.events))
try:
    ledger.apply("deposit-1", "org-a", 200)
except ValueError as error:
    print(error)

Output

100
70
100
70 2
idempotency key reused with different operation

A replay with the same key and operation returns the stored result, even if the org's balance changed later. Reusing a key for a different operation fails loudly instead of silently returning an unrelated balance. That's idempotency: the request key names the original effect, not "try the debit again."

Transfer follow-up:

Use one idempotency key for the whole transfer.
Lock account IDs in sorted order to avoid deadlock.
Record both debit and credit events together.
Define whether external side effects happen before or after durable commit.

A reusable lock-order helper makes deadlock prevention concrete:

ordered-account-locks.py

from contextlib import ExitStack
from threading import Lock

locks = {"org-a": Lock(), "org-b": Lock()}

def lock_accounts(*account_ids: str) -> ExitStack:
    stack = ExitStack()
    for account_id in sorted(set(account_ids)):
        stack.enter_context(locks[account_id])
    return stack

with lock_accounts("org-b", "org-a"):
    print("locked:", sorted({"org-b", "org-a"}))

Output

locked: ['org-a', 'org-b']

Crawlers, stores, schedulers, and ledgers are all mutable state plus an explicit claim. Parsers are the other common shape: an explicit current record and a policy for junk lines.

Multiline log parser

Group indented continuation lines under the previous valid event. Ignore orphan continuations and malformed records in this base version; a production parser should count or collect rejects.

multiline-log-parser.py

import json

def parse_events(text: str) -> list[dict[str, str]]:
    events: list[dict[str, str]] = []
    current: dict[str, str] | None = None

    for line in text.splitlines():
        if line.startswith(" "):
            if current is not None:
                current["message"] += f" {line.strip()}"
            continue

        parts = line.split("|", maxsplit=2)
        if len(parts) != 3:
            current = None
            continue

        timestamp, level, message = parts
        current = {"timestamp": timestamp, "level": level, "message": message}
        events.append(current)

    return events

LOGS = """  orphan continuation
2026-06-02T12:00:00Z|ERROR|request failed
  timeout while calling model
malformed line
2026-06-02T12:01:00Z|INFO|retry queued"""

print(json.dumps(parse_events(LOGS), indent=2))

Output

[
  {
    "timestamp": "2026-06-02T12:00:00Z",
    "level": "ERROR",
    "message": "request failed timeout while calling model"
  },
  {
    "timestamp": "2026-06-02T12:01:00Z",
    "level": "INFO",
    "message": "retry queued"
  }
]

The leading indented line is dropped because there's no current event. The malformed non-indented line isn't an event and clears current, so a later indented line wouldn't attach to the ERROR record. Ask whether tabs count as continuation; this version only treats a leading space as a continuation.

Once the state model is right, interviewers often say "make it faster with threads." That follow-up is a concurrency-model question, not a request to sprinkle ThreadPoolExecutor on a CPU loop.

When threads help

Before you reach for threads or async, name what kind of work you're speeding up. That one sentence is a strong signal because it shows you understand why a concurrency model helps or doesn't.

On a standard GIL-enabled CPython build, only one thread executes Python bytecode at a time. Optional free-threaded CPython builds, available since 3.13, change that constraint, so state which runtime you're discussing.^{[4]Reference 4Python support for free threadinghttps://docs.python.org/3/howto/free-threading-python.html} Free-threading isn't data-race freedom: shared mutable state still needs locks, atomics, or other synchronization. The official howto recommends threading.Lock instead of relying on the internal locks of built-in dicts and sets. Dropping locks because "there is no GIL" is still wrong.

Work type	Example	GIL effect	Right tool
I/O-bound	fetching URLs, reading disk, calling a model API	blocking I/O generally releases the GIL, so threads can overlap waits	threads, `ThreadPoolExecutor`, or `asyncio`
CPU-bound Python	heavy parsing or numeric loops in Python code	threads take turns under a GIL-enabled build	`multiprocessing` or `ProcessPoolExecutor`

So the honest answer to "should we add threads?" is: only if the bottleneck is waiting, not computing. A crawler that spends its time on network round-trips is a good fit for threads. A function that spends its time crunching numbers in pure Python won't get faster with threads; it needs separate processes, or a library that drops into C and releases the GIL.

⚠️ Common mistake: Reaching for asyncio and then calling a blocking function inside a coroutine. Synchronous calls like requests.get(...) or time.sleep(...) don't yield to the event loop, so a single blocking line freezes every other task. The whole point of async is that each await hands control back so other work can run.

Each worker yields at await asyncio.sleep(0), so the event loop interleaves them:

asyncio-yield.py

import asyncio

async def worker(name: str, log: list[str]) -> None:
    log.append(f"{name} start")
    await asyncio.sleep(0)  # yields control back to the event loop
    log.append(f"{name} resume")

async def main() -> list[str]:
    log: list[str] = []
    await asyncio.gather(worker("a", log), worker("b", log))
    return log

print(asyncio.run(main()))

Output

['a start', 'b start', 'a resume', 'b resume']

Swap await asyncio.sleep(0) for a blocking time.sleep(0) and the output becomes ['a start', 'a resume', 'b start', 'b resume']: worker a runs start-to-finish before b ever begins, because a blocking call never returns control to the loop.

Before adding workers, say these five sentences:

Shared state is X.
The lock protects X.
A work item is claimed at this point.
Worker shutdown happens through this sentinel, event, or executor lifecycle.
Failed work records an error and doesn't corrupt shared state.

Debugging when a test fails

Narrate the smallest useful investigation:

Read the assertion and say expected versus actual.
Reproduce with one smaller fixture.
Print or inspect the state that owns the invariant.
Fix the state transition, not the symptom.
Add one regression test before moving to the next follow-up.

Symptom	First thing to inspect
Duplicate output	claim point or `visited` insertion timing
Missing output	enqueue condition, prefix filter, or failure policy
Wrong order	queue/heap tie-break and where sorting happens
Expired key returned	read path skipped cleanup
Retry ran too many times	attempt counter increment point
Blocked task marked failed	dependency failure policy mixed with task execution failure
Threaded result flakes	shared state changed outside the lock

You don't have to be bug-free. Recover fast by naming which invariant broke: visited claimed too late, a TTL read skipped cleanup, or a lock dropped between compare and set.

When the interviewer adds a follow-up, classify it before changing code.

Follow-up type	First move	Common trap
Time	inject `now`, store deadline, clean on read	sleeping in tests
Ordering	choose deque, heap, sort, or insertion order	mixing policy into unrelated state
Capacity	evict by explicit rule	hidden off-by-one at capacity 0 or 1
Retry	store attempt count and final state	retrying external writes without idempotency
Snapshot	version records or copy-on-write	mutating data a snapshot should freeze
Compare-and-set	lock compare and write together	calling `get` then `set` as two operations
Thread safety	protect claim point and shared maps	holding lock during slow fetch or execution
Cancellation	durable flag plus cooperative checks	stopping new work while workers keep publishing children

Pattern map

Most prompts are one of these shapes. Classify the prompt before writing code.

Pattern	Recognition signal	First state model	High-value tests
Traversal	URLs, graph nodes, dependencies, neighbors	`visited` plus queue or stack	cycle, duplicate edge, malformed node, deterministic order
Mutable store	`set`, `get`, `delete`, `scan`, cache records	dict from key to record	missing key, overwrite, delete, scan ordering, expired key
Time boundary	TTL, deadline, rate limit, retry-after	injected `now`, deadline fields	equality boundary, refill edge, expired-on-read, no `sleep`
Ordering policy	priority, deadline, LRU, LFU, ready queue	heap, deque, `OrderedDict`, frequency buckets	tie-break, stale heap entry, capacity 0, promotion
Idempotent write	request ID, retry, ledger, external action	request key to stored result	duplicate success, duplicate failure, conflicting retry
Parser	logs, events, chunks, sections, streaming lines	explicit current record or buffer	malformed line, continuation, empty input, final flush
Concurrency	workers, thread-safe, fanout, cancellation	lock-protected claim point plus queue	duplicate claim, shutdown, exception, partial result
Validation	schema, payload, permissions, filters	recursive validator or rule table	nested failure, path reporting, unknown field, missing required

Many rounds feel like one product-shaped problem that grows across levels. Ship level 1 quickly, then preserve the same invariant.

Prompt family	Level 1	Level 2	Level 3	Level 4
GPU quota reservation	add, remove, lookup	reserve/release capacity	expirations, over-allocation prevention	concurrency or audit log
Key/value store	set, get, delete	prefix scan and ordering	TTL and compare-and-set	transactions or snapshots
Crawler	same-host BFS	URL normalization	retries and rate limits	worker fanout and cancellation
Scheduler	dependencies and ready queue	cycle detection	retries and deadlines	workers, cancellation, fairness
Cache	get/put capacity eviction	update semantics	TTL or LFU	metrics and thread safety
Chat or event router	register handlers	route messages	priorities or filters	replay, idempotency, backpressure
Log processor	parse records	multiline events	top errors and windows	malformed input and streaming
Permission filter	include/exclude resources	groups and inheritance	deny precedence	audit why each item passed
Stream assembler	append chunks	sequence ordering	duplicate/missing chunks	timeout and memory cap
Experiment splitter	assign users	stable hashing	ramp percentages	sticky overrides and rollback

State the tests before coding the follow-up. That keeps the interviewer aligned and prevents late rewrites.

Pattern	Minimum public tests	Private-edge tests to rehearse
Crawler or graph traversal	start node, duplicate edge, off-host or invalid neighbor	cycles, relative links, malformed URL, empty graph, deterministic order
TTL store or rate limiter	set/get, expiry, scan, refill	equality boundary, zero TTL, time moving backward, cleanup during read
LRU or LFU cache	capacity eviction, update existing key, get promotion	capacity 0, tie-break by recency, expired entry, stale heap record
Scheduler	independent tasks, dependency release, retry	cycle, missing dependency, permanent failure, blocked dependents
Ledger	deposit, withdraw, transfer	duplicate idempotency key, conflicting retry, insufficient credits, lock order
Parser	one record, multiline continuation, malformed line	orphan continuation, empty input, final flush, very large record
Validator	required field, unknown field, nested list	error path, default value, type coercion, repeated failure
Stream assembler	in-order chunks, out-of-order chunks	duplicate chunk, missing end marker, timeout, memory cap

Practice circuit

Don't memorize wording. Learn the patterns. Each linked drill carries a full prompt, clarification questions, sample tests, hidden tests, a solution guide, and validated Python plus Java paths. Use Python first. The Java path is useful when you want maps, classes, and null checks to be explicit, not because labs require Java.

Arena	Practice drills	Interview signal
Repository and codebase systems	Repository Hash Tree, Repository Diff, Duplicate File Groups, Repository Ignore Filter, File Patch Applier, Patch Conflict Detector, Codebase Symbol Index, In-memory Filesystem, Stack Trace Reconstructor	Code review agents, workspace snapshots, deterministic patches, symbol lookup, upload filtering, incremental state
Stateful services	TTL Key/Value Store, Transactional KV Store, Idempotent Ledger, LRU TTL Cache, LFU TTL Cache, Token Budget Ledger	Durable APIs, atomic updates, idempotency, eviction policy, TTL boundaries
Scheduling and workers	Dependency Scheduler With Retries, Workflow State Machine, Multi-tenant Job Scheduler, Bounded Worker Queue, Worker Lease Registry, Priority Deadline Scheduler, Batch Inference Scheduler	Run orchestration, backpressure, retries, leases, fairness, cancellation follow-ups
Streaming and parsers	Streaming Markdown Parser, SSE Event Parser, Streaming Token Assembler, Log Error Parser, Longest-match Tokenizer, Prompt Section Extractor, Secret Redactor	Chunked state, final flush, malformed records, overlapping matches, transcript safety
Agents and webhooks	Tool Call Schema Validator, Webhook Idempotency Receiver, PR Readiness Gate, Model Fallback Router, Retry Backoff Planner, Circuit Breaker State	Tool execution, dedupe, provider fallback, proof artifacts, retry policy, outage behavior
Retrieval and context	RAG Chunk Selector, Context Window Packer, Document Permission Filter	Ranking, budget packing, access control, explainable inclusion
Evals and telemetry	Eval Failure Aggregator, Canary Metric Judge, Experiment Traffic Splitter, Agent Event Timeline, Command Log Classifier, Sliding Window Error Counter	Rollout judgment, failure clustering, reviewable proof, redacted logs, stable assignment, moving windows
Systems basics under frontier wording	Same-host Crawler, Token Bucket Rate Limiter, Notification Rate Limiter	Graph traversal, duplicate suppression, host policy, quota math, noisy automation control

These drills are Python systems work: concurrency, state machines, caches, and ledgers. If a later panel also covers ML coding, reuse earlier lessons on tensor shapes, numerically stable softmax, attention, and training-loop invariants. The same claim-before-mutate discipline applies to buffer writes and gradient steps.

Common pitfalls

Solving version 1, then adding TTL, transactions, or threads without restating the invariant.
Using wall-clock sleeps in tests instead of injected time.
Making crawler output order part of correctness after adding concurrent workers.
Treating idempotency as "retry the operation" instead of recording the request key and result semantics.
Adding a global lock around everything without explaining throughput, deadlock, and fairness tradeoffs.
Stating a free-threaded runtime and then dropping locks on shared mutable state.

AI Lab Coding Interview: Python Systems

Keep one invariant visible

What is the common failure mode in staged coding rounds?

Python tools to know cold

Why inject `now` instead of calling `time.sleep()` in a TTL or rate-limit interview test?

Same-host crawler

What invariant should you say before adding threads to a crawler, scheduler, or cache?

TTL key/value store

Task scheduler with retries

Token bucket with deterministic time

Credit ledger with idempotency

Multiline log parser

When threads help

An interviewer asks you to make a log parser "faster with threads." The parser is pure-Python string processing over one large file. What do you say?

Debugging when a test fails

Pattern map

Practice circuit

Common pitfalls

Mastery Check

Discussion

What is the common failure mode in staged coding rounds?

What is the common failure mode in staged coding rounds?

Why inject now instead of calling time.sleep() in a TTL or rate-limit interview test?

Why inject now instead of calling time.sleep() in a TTL or rate-limit interview test?

What invariant should you say before adding threads to a crawler, scheduler, or cache?

What invariant should you say before adding threads to a crawler, scheduler, or cache?

An interviewer asks you to make a log parser "faster with threads." The parser is pure-Python string processing over one large file. What do you say?

An interviewer asks you to make a log parser "faster with threads." The parser is pure-Python string processing over one large file. What do you say?

Mastery Check

Discussion

Why inject `now` instead of calling `time.sleep()` in a TTL or rate-limit interview test?