AI Lab Coding Interview: Python Systems

After the system design capstones, the final section turns architecture knowledge into interview execution. The first test is whether you can build small, correct systems while requirements change.

AI lab coding formats vary by team. Public frontier-lab guidance emphasizes well-designed solutions, tests, live coding, debugging, standard-library fluency, experience, motivation, communication, and tradeoff analysis.^{[1]Reference 1Interview guidehttps://openai.com/interview-guide/[2]Reference 2Careershttps://www.anthropic.com/careers} Practice that bar with one base prompt followed by staged requirements: add TTLs, add concurrency, add cancellation, add rate limits, preserve deterministic output, or explain why your state model won't corrupt itself.

The final interview-prep section ends with a coding session. Build speed at practical Python: clear state, small APIs, local tests, and honest concurrency invariants.

Four-round prep map

Treat the final prep section as one packet. Each round tests a different artifact, but the artifacts should agree with each other.

Use the map to avoid uneven prep. A strong candidate doesn't only solve Python prompts; they can also explain why a design boundary exists, what failure changed their judgment, and which project artifact proves depth.

The operating model

Use this loop for every prompt:

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

Good interview code isn't the most abstract code. It's code whose invariants can be defended while requirements change.

Python tools to know cold

Know these without documentation:

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

Prompt bank

Use these as drills. Don't memorize wording. Learn the patterns.

Frontier live-practice set

Use this as the main circuit. Each linked drill carries a full prompt, clarification questions, sample tests, hidden tests, solution guide, and validated Python plus Java paths. Run each problem twice: first in Python for speed, then in Java to force explicit classes, maps, and boundary checks.

Live drill contract

For every practice run, write these artifacts before opening the guide:

1. Problem proposition in your own words: API, input shape, output shape, failure behavior.
2. Three clarification questions that can change tests.
3. Invariant sentence: state owner, allowed transition, and cheap lookup.
4. Public tests: happy path, duplicate input, missing input, boundary.
5. Hidden tests: malformed input, ordering tie-break, expiry equality, retry exhaustion, or conflict.
6. Python implementation with plain assert tests.
7. Java implementation with explicit classes and deterministic collections.
8. Follow-up answer: concurrency, persistence, scale, observability, or rollback.

That routine makes LeetLLM practice closer to a real frontier coding loop: define semantics, ship correct code, prove boundaries, then extend without rewriting the state model.

Question pattern taxonomy

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

Use this sentence when stuck:

The invariant is X; this data structure makes X cheap to maintain; these tests prove the boundary cases.

Answer rhythm

Use the same rhythm for every timed solution:

1. Clarify semantics that change tests: ordering, equality boundary, failure behavior, and concurrency expectation.
2. Name invariant and data structures before code.
3. Implement smallest single-threaded version.
4. Add sample tests while code is still small.
5. Add follow-up by changing one boundary: time, ordering, retries, or locking.
6. Explain complexity and the first production hardening step.

For a frontier-lab coding round, communication is part of correctness. Say when you're making an assumption, say which invariant you're protecting, and say what test you'll write before you write it.

Pattern drills by week

Levelled prompt ladder

Many frontier coding rounds feel like one product-shaped problem that grows across levels. Practice each ladder by shipping level 1 quickly, then preserving the same invariant as requirements change.

Use this timing target:

Coding readiness rubric

Score yourself after each drill.

When a drill scores "not ready," repeat the same prompt family with different nouns. Train transfer, not memorization.

Pattern testcase checklist

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

Use this sentence before running tests:

I expect these tests to fail if I lose the invariant. The invariant is X; the edge case most likely to break it: Y.

Debugging protocol

When a test fails, narrate the smallest useful investigation:

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

Common debug pivots:

Strong candidates aren't bug-free. They recover fast because they can name which invariant is broken.

Live coding narration bank

Practice short narration so you don't go silent while thinking. Use these sentences as scaffolding, then replace the placeholders with prompt-specific details.

Follow-up triage map

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

Good follow-up handling sounds calm because the state boundary is still visible.

Mock coding prompts

Use these as timed practice. Read the prompt, write a small solution and tests, then open the solution guide.

Prompt 1: same-host crawler

You're given a starting URL and a function get_links(url) -> list[str]. Implement a crawler that visits only URLs on the same host as the start URL.

Requirements:

• Return visited URLs in deterministic breadth-first order for the single-threaded version.
• Don't visit the same URL twice.
• Ignore malformed URLs and off-host URLs.
• Follow-up: add worker fanout without corrupting visited.
• Follow-up: add a per-host rate limit and cancellation event.

Clarifying questions to ask:

• 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?

Solution guide

Use a FIFO work queue for the base version because the prompt asks for breadth-first order. Put URL normalization and host checks before enqueueing. The invariant is: a URL enters the queue only after it has been accepted into visited.

same-host-crawler.py

1from collections import deque 2from urllib.parse import urlparse, urljoin 3 4PAGES = { 5 "https://repo.example/start": ["/tests", "/ci", "https://other.example/x"], 6 "https://repo.example/tests": ["/ci"], 7 "https://repo.example/ci": [], 8} 9 10def crawl(start_url: str, get_links) -> list[str]: 11 start = urlparse(start_url) 12 if not start.scheme or not start.netloc: 13 return [] 14 15 visited = {start_url} 16 queue = deque([start_url]) 17 ordered: list[str] = [] 18 19 while queue: 20 url = queue.popleft() 21 ordered.append(url) 22 23 for raw_link in get_links(url): 24 candidate = urljoin(url, raw_link) 25 parsed = urlparse(candidate) 26 if parsed.scheme not in {"http", "https"}: 27 continue 28 if parsed.netloc != start.netloc: 29 continue 30 if candidate in visited: 31 continue 32 visited.add(candidate) 33 queue.append(candidate) 34 35 return ordered 36 37print(crawl("https://repo.example/start", lambda url: PAGES.get(url, [])))

Output

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

Follow-up guide

For worker fanout, say explicitly that deterministic return order becomes expensive unless the prompt still requires it. Protect the claim step, not the whole fetch:

crawler-claim-step.py

1from threading import Lock 2 3visited = {"https://repo.example/start"} 4visited_lock = Lock() 5 6def claim(candidate: str) -> bool: 7 with visited_lock: 8 if candidate in visited: 9 return False 10 visited.add(candidate) 11 return True 12 13print(claim("https://repo.example/tests")) 14print(claim("https://repo.example/tests")) 15print(sorted(visited))

Output

1True 2False 3['https://repo.example/start', 'https://repo.example/tests']

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. The full answer should name all three behaviors: duplicate prevention, bounded fetch concurrency, and graceful stop.

Prompt 2: TTL key/value store

Implement an in-memory store with set(key, value, ttl=None), get(key), delete(key), and scan(prefix).

Requirements:

• ttl is measured in seconds.
• Expired keys behave as missing.
• Tests must not call sleep.
• Follow-up: add compare-and-set.
• Follow-up: add thread safety.

Clarifying questions to 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?

Solution guide

Inject now so tests can advance time directly. Store expires_at beside each value. The invariant is: every public read path either returns a non-expired value or removes the expired key.

ttl-store.py

1from dataclasses import dataclass 2from typing import Callable 3 4@dataclass 5class Entry: 6 value: str 7 expires_at: float | None 8 9class Store: 10 def __init__(self, now: Callable[[], float]) -> None: 11 self.now = now 12 self.items: dict[str, Entry] = {} 13 14 def set(self, key: str, value: str, ttl: float | None = None) -> None: 15 expires_at = None if ttl is None else self.now() + ttl 16 self.items[key] = Entry(value, expires_at) 17 18 def get(self, key: str) -> str | None: 19 entry = self.items.get(key) 20 if entry is None: 21 return None 22 if entry.expires_at is not None and entry.expires_at <= self.now(): 23 self.items.pop(key, None) 24 return None 25 return entry.value 26 27 def delete(self, key: str) -> None: 28 self.items.pop(key, None) 29 30 def scan(self, prefix: str) -> list[str]: 31 return sorted(key for key in list(self.items) if key.startswith(prefix) and self.get(key) is not None) 32 33clock = {"now": 10.0} 34store = Store(lambda: clock["now"]) 35store.set("task:1", "running", ttl=5.0) 36store.set("task:2", "ready") 37print(store.get("task:1"), store.scan("task:")) 38clock["now"] = 15.0 39print(store.get("task:1"), store.scan("task:"))

Output

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

Follow-up guide

For compare-and-set, 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

1from threading import RLock 2 3class AtomicStore: 4 def __init__(self) -> None: 5 self.items: dict[str, str] = {} 6 self.lock = RLock() 7 8 def compare_and_set(self, key: str, expected: str | None, value: str) -> bool: 9 with self.lock: 10 if self.items.get(key) != expected: 11 return False 12 self.items[key] = value 13 return True 14 15store = AtomicStore() 16print(store.compare_and_set("task:1", None, "running")) 17print(store.compare_and_set("task:1", None, "ready")) 18print(store.compare_and_set("task:1", "running", "ready"))

Output

1True 2False 3True

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. The important answer is atomicity: no other thread can change the key between compare and set.

Prompt 3: task scheduler with retries

Implement a scheduler for tasks with dependencies. A task becomes runnable when all dependencies have completed.

Requirements:

• Detect dependency cycles before running.
• Run ready tasks in priority order.
• Retry failed tasks up to max_attempts.
• Return completed tasks, permanently failed tasks, and blocked dependents separately.
• Follow-up: add worker fanout.

Clarifying questions to ask:

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

Solution guide

Represent the graph explicitly: dependents[task] points to 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.

dependency-scheduler.py

1import heapq 2from collections import defaultdict 3from dataclasses import dataclass 4 5@dataclass(frozen=True) 6class Task: 7 name: str 8 priority: int 9 deps: tuple[str, ...] = () 10 11def schedule(tasks: list[Task], run, max_attempts: int = 2) -> tuple[list[str], list[str], list[str]]: 12 if max_attempts <= 0: 13 raise ValueError("max_attempts must be positive") 14 by_name = {task.name: task for task in tasks} 15 if len(by_name) != len(tasks): 16 raise ValueError("duplicate task name") 17 dependents: dict[str, list[str]] = defaultdict(list) 18 remaining = {task.name: len(task.deps) for task in tasks} 19 20 for task in tasks: 21 for dep in task.deps: 22 if dep not in by_name: 23 raise ValueError(f"unknown dependency: {dep}") 24 dependents[dep].append(task.name) 25 26 state: dict[str, int] = {} 27 28 def visit(name: str) -> None: 29 marker = state.get(name, 0) 30 if marker == 1: 31 raise ValueError("cycle") 32 if marker == 2: 33 return 34 state[name] = 1 35 for dep in by_name[name].deps: 36 visit(dep) 37 state[name] = 2 38 39 for name in by_name: 40 visit(name) 41 42 ready = [(task.priority, task.name) for task in tasks if remaining[task.name] == 0] 43 heapq.heapify(ready) 44 attempts = defaultdict(int) 45 completed: list[str] = [] 46 failed: list[str] = [] 47 48 while ready: 49 _, name = heapq.heappop(ready) 50 attempts[name] += 1 51 if not run(name): 52 if attempts[name] < max_attempts: 53 heapq.heappush(ready, (by_name[name].priority, name)) 54 else: 55 failed.append(name) 56 continue 57 58 completed.append(name) 59 for child in dependents[name]: 60 remaining[child] -= 1 61 if remaining[child] == 0: 62 heapq.heappush(ready, (by_name[child].priority, child)) 63 64 blocked = sorted(set(by_name) - set(completed) - set(failed)) 65 return completed, failed, blocked 66 67tasks = [ 68 Task("pack", priority=2), 69 Task("ship", priority=3, deps=("pack",)), 70 Task("label", priority=1), 71] 72fail_once = {"pack"} 73 74def run_with_one_retry(name: str) -> bool: 75 if name in fail_once: 76 fail_once.remove(name) 77 return False 78 return True 79 80print(schedule(tasks, run_with_one_retry)) 81print(schedule([Task("pack", 1), Task("ship", 2, ("pack",))], lambda _: False, max_attempts=1))

Output

1(['label', 'pack', 'ship'], [], []) 2([], ['pack'], ['ship'])

Follow-up guide

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.

For 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/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."

Drill 1: token bucket with deterministic time

Rate limiters show up because they combine state, boundary conditions, and production behavior. Retry policies often need jitter to avoid synchronized retry storms, so this prompt is really about overload control as much as counters.^{[3]Reference 3Exponential Backoff And Jitterhttps://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/} A strong implementation injects time so tests don't sleep.

What to say out loud:

• The key can be a user, organization, endpoint, or model.
• now is injected for deterministic tests.
• Impossible costs fail immediately instead of returning retry advice that can never succeed.
• 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.

Drill 2: thread-safe same-host crawler shape

A crawler prompt tests graph traversal plus concurrency. The invariant is simple: each URL is claimed once before it's fetched or enqueued.

Concurrency follow-up:

• Protect visited with a lock.
• Claim a URL while holding the lock, before scheduling a worker.
• Keep output nondeterministic unless the prompt explicitly asks for deterministic order.
• Add timeouts and failed-fetch handling without retry storms.

Drill 3: ledger with idempotency

Ledger prompts test whether you can keep money-like state consistent. Use append-only events when possible; if you maintain balances, update balance and event together.

A replay with the same key and operation returns the stored result, even if the account changed later. Reusing a key for a different operation fails loudly instead of silently returning an unrelated balance.

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:

Drill 4: multiline log parser

Log parsers test string handling and explicit malformed-input policy. 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.

Concurrency answer template

When asked to make a solution concurrent, say this before writing code:

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

This sounds mechanical because it should. Concurrency interview failures usually come from vague ownership.

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.

Mastery checklist

• Implement a rate limiter with deterministic time and boundary tests.
• Implement a same-host crawler and explain where to place the visited lock.
• Implement a ledger with idempotency and insufficient-funds behavior.
• Explain how to add TTLs to a key/value store without sleeping in tests.
• Build a dependency scheduler with cycle detection and a ready queue.
• Parse multiline logs with malformed-line handling.
• State complexity and memory growth for each solution.