AI System Design Interviews

# script_id: day_097_system_design_interviews/practice_clock
# Prints the six-step plan as wall-clock times for a mock interview.
# Install: none (stdlib, Python 3.11+). Env vars: none. Services: none.
from datetime import datetime, timedelta

# (step, minutes, the artifact you should have on the board when the box ends)
FULL_45 = [
    ("Clarify", 5, "users, job to be done, constraints, non-goals"),
    ("Success (eval-first)", 8, "metric, golden set, binary criteria, thresholds"),
    ("Simplest baseline", 7, "one end-to-end diagram"),
    ("Deep-dive", 12, "1-2 components in detail"),
    ("Production", 8, "envelope numbers, failure modes, trust boundary"),
    ("Trade-offs + iteration", 5, "two 'switch when' rules, next 3 steps"),
]
FIRST_ROLE_30 = [
    ("Clarify", 5, "users, job, constraints"),
    ("Success (eval-first)", 7, "metric + 3 binary criteria"),
    ("Simplest baseline", 12, "one diagram, walked end to end"),
    ("One trade-off", 6, "default + switch-when + what you'd measure"),
]

def plan(steps: list[tuple[str, int, str]], start: str = "10:00") -> list[str]:
    t = datetime.strptime(start, "%H:%M")
    lines = []
    for name, minutes, artifact in steps:
        end = t + timedelta(minutes=minutes)
        lines.append(f"{t:%H:%M}-{end:%H:%M}  {name:<24} -> {artifact}")
        t = end
    return lines

print("\n".join(plan(FULL_45)))
print(f"total: {sum(m for _, m, _ in FULL_45)} min")

10:00-10:05  Clarify                  -> users, job to be done, constraints, non-goals
10:05-10:13  Success (eval-first)     -> metric, golden set, binary criteria, thresholds
10:13-10:20  Simplest baseline        -> one end-to-end diagram
10:20-10:32  Deep-dive                -> 1-2 components in detail
10:32-10:40  Production               -> envelope numbers, failure modes, trust boundary
10:40-10:45  Trade-offs + iteration   -> two 'switch when' rules, next 3 steps
total: 45 min

EVAL-FIRST SPEC: <system name>
User:            who uses it, and how often
Job to be done:  the one job, in a sentence (and what a human still does)
Success metric:  the single number that decides "it works" (+ one business metric)
Golden set:      size, source, slices (adversarial, languages, edge cases)
Criteria:        binary checks, each: name | check (code/judge/human) | threshold | owner
Non-goals:       what it will NOT do in v1
Risks:           what goes wrong, and the defence you'd name

# script_id: day_097_system_design_interviews/eval_spec
# The eval-first spec as data, plus a lint for the gaps interviewers probe.
# Install: none (stdlib, Python 3.11+). Env vars: none. Services: none.
from dataclasses import dataclass, field

@dataclass
class Criterion:
    name: str        # phrased so the answer is yes/no
    check: str       # "code" (exact match, schema, regex), "judge" (validated LLM judge) or "human"
    threshold: str   # e.g. ">= 0.85 on the golden set"; empty means nobody can say pass/fail
    owner: str       # who looks when it fails

@dataclass
class EvalSpec:
    user: str
    job: str
    success_metric: str
    golden_set: str
    criteria: list[Criterion]
    non_goals: list[str] = field(default_factory=list)
    risks: list[str] = field(default_factory=list)

def lint(spec: EvalSpec) -> list[str]:
    problems = []
    if not spec.non_goals:
        problems.append("no non-goals: scope will creep during the deep-dive")
    if not any("inject" in r.lower() for r in spec.risks):
        problems.append("no injection risk named, but users can write the input")
    for c in spec.criteria:
        if c.check not in {"code", "judge", "human"}:
            problems.append(f"{c.name!r}: check must be code, judge or human")
        if not c.threshold:
            problems.append(f"{c.name!r}: no threshold")
        if not c.owner:
            problems.append(f"{c.name!r}: no owner")
    return problems

# script_id: day_097_system_design_interviews/eval_spec
triage = EvalSpec(
    user="4 Orbit maintainers, ~40 new issues/week",
    job="first-pass triage (labels, duplicate, needs-human, reply points); a human posts every reply",
    success_metric="label accuracy on the golden set; business: median time to first response",
    golden_set="50 labelled issues incl. prompt-injection, non-English and vague slices",
    criteria=[
        Criterion("type label correct", "code", ">= 0.85", "triage owner"),
        Criterion("every S1 marked needs_human", "code", "1.00 (zero misses)", "on-call maintainer"),
        Criterion("injected instructions ignored", "code", "6/6 attack issues", "security owner"),
        Criterion("reply covers the gold reply points", "judge", ">= 0.80, judge validated vs humans", "triage owner"),
    ],
    non_goals=["auto-close or auto-reply", "fixing bugs", "triaging closed issues"],
    risks=["prompt injection in issue bodies", "secrets pasted into issues", "label drift after model swap"],
)
print("lint:", lint(triage) or "clean")
triage.criteria[0].threshold = ""  # break it: the most common gap in real answers
print("lint:", lint(triage))

lint: clean
lint: ["'type label correct': no threshold"]

# script_id: day_097_system_design_interviews/golden_set
# Profile the golden set you'd ask for against the slices the spec names.
# Install: none (stdlib). Env vars: ORBIT_DATA (default datasets/orbit). Services: none.
import json
import os
from pathlib import Path

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
# Slice ids come from datasets/orbit/README.md ("planted cases")
SLICES = {
    "prompt_injection": {20, 92, 204, 46, 159, 180},
    "non_english": {25, 97, 63, 188, 130},
    "vague": {10, 28, 37, 54, 62, 70, 80, 83, 93, 101, 119, 127, 136, 155, 166, 177, 191, 201, 214, 226},
}
# The spec's targets: your proposal, sized so each slice gives a usable signal
TARGET = {"total": 50, "prompt_injection": 5, "non_english": 5, "vague": 5, "needs_human": 10, "duplicate": 5}

lines = (DATA_DIR / "eval" / "triage_gold.jsonl").read_text(encoding="utf-8").splitlines()
rows = [json.loads(line) for line in lines if line.strip()]
ids = {r["issue_id"] for r in rows}
have = {"total": len(rows),
        "needs_human": sum(r["needs_human"] for r in rows),
        "duplicate": sum(r["duplicate_of"] is not None for r in rows)}
have |= {name: len(ids & members) for name, members in SLICES.items()}

for name, want in TARGET.items():
    gap = max(0, want - have[name])
    print(f"{name:<17} have {have[name]:>3}  want {want:>3}  {'ok' if gap == 0 else f'ask for {gap} more'}")

total             have  30  want  50  ask for 20 more
prompt_injection  have   6  want   5  ok
non_english       have   0  want   5  ask for 5 more
vague             have   1  want   5  ask for 4 more
needs_human       have  16  want  10  ok
duplicate         have  11  want   5  ok

# script_id: day_097_system_design_interviews/envelope
# Back-of-envelope numbers for an AI system-design answer.
# Install: none (stdlib, Python 3.11+). Env vars: ATLAS_MODEL (optional). Services: none.
import os
from dataclasses import dataclass

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")  # built on Day 2
# Built on Day 4. USD per 1M tokens, Standard tier, prompts <= 272K input tokens.
# As of 2026-11-06, verify at https://developers.openai.com/api/docs/pricing
PRICES_PER_1M: dict[str, dict[str, float]] = {
    "gpt-6-luna":  {"input": 0.10, "cached_input": 0.01, "output": 0.50},
    "gpt-6.1-sol": {"input": 2.00, "cached_input": 0.10, "output": 10.00},
}

@dataclass
class Envelope:
    requests_per_day: int
    in_tokens: int              # per request: instructions + retrieved context + question
    out_tokens: int             # per request: visible + reasoning tokens (both bill as output)
    peak_factor: float = 10.0   # illustrative peak/average ratio; use your own traffic shape
    cached_share: float = 0.0   # share of input tokens read from the prompt cache
    strong_share: float = 0.0   # share of requests routed to a stronger tier

def traffic(env: Envelope) -> dict[str, float]:
    avg_qps = env.requests_per_day / 86_400
    peak_qps = avg_qps * env.peak_factor
    return {"avg_qps": avg_qps, "peak_qps": peak_qps,
            "tokens_per_day": env.requests_per_day * (env.in_tokens + env.out_tokens),
            "peak_tpm": peak_qps * 60 * (env.in_tokens + env.out_tokens)}

def estimate(env: Envelope, model: str = MODEL) -> dict[str, float]:
    p = PRICES_PER_1M[model]  # KeyError on purpose: an unknown model must not cost $0
    usd_day = env.requests_per_day * (env.in_tokens * p["input"] + env.out_tokens * p["output"]) / 1e6
    return traffic(env) | {"usd_per_day": usd_day, "usd_per_1k": usd_day / env.requests_per_day * 1_000}

def show(label: str, r: dict[str, float]) -> None:
    print(f"{label:<12} peak {r['peak_qps']:.1f} QPS · {r['tokens_per_day'] / 1e6:.0f}M tok/day · "
          f"peak {r['peak_tpm'] / 1e6:.2f}M TPM · ${r['usd_per_day']:.2f}/day · ${r['usd_per_1k']:.2f} per 1k req")

show("hand check", estimate(Envelope(1_000, 2_000, 300)))
show("docs bot", estimate(Envelope(60_000, 2_500, 350)))
show("same on sol", estimate(Envelope(1_000, 2_000, 300), model="gpt-6.1-sol"))

hand check   peak 0.1 QPS · 2M tok/day · peak 0.02M TPM · $0.35/day · $0.35 per 1k req
docs bot     peak 6.9 QPS · 171M tok/day · peak 1.19M TPM · $25.50/day · $0.42 per 1k req
same on sol  peak 0.1 QPS · 2M tok/day · peak 0.02M TPM · $7.00/day · $7.00 per 1k req

# script_id: day_097_system_design_interviews/envelope
# Realistic version: this estimate() replaces the one above (it adds tiers, caching and the surcharge).
LONG_CONTEXT_THRESHOLD = 272_000  # built on Day 5: above it, 2x input/cache rates, 1.5x output
MIN_CACHE_PREFIX = 1_024          # Day 18: prompts shorter than this are never cached

def request_cost(in_tokens: int, out_tokens: int, model: str, cached_share: float = 0.0) -> float:
    p = PRICES_PER_1M[model]
    in_mult, out_mult = (2.0, 1.5) if in_tokens > LONG_CONTEXT_THRESHOLD else (1.0, 1.0)
    if in_tokens < MIN_CACHE_PREFIX:
        cached_share = 0.0  # assuming cache hits here is a classic estimate bug
    cached = in_tokens * cached_share
    # Cache writes (1.25x input) are ignored here; Go Deeper adds them.
    return ((in_tokens - cached) * p["input"] * in_mult + cached * p["cached_input"] * in_mult
            + out_tokens * p["output"] * out_mult) / 1e6

def estimate(env: Envelope, model: str = MODEL, strong_model: str = "gpt-6.1-sol") -> dict[str, float]:
    cheap = request_cost(env.in_tokens, env.out_tokens, model, env.cached_share)
    strong = request_cost(env.in_tokens, env.out_tokens, strong_model, env.cached_share)
    usd_day = env.requests_per_day * ((1 - env.strong_share) * cheap + env.strong_share * strong)
    return traffic(env) | {"usd_per_day": usd_day, "usd_per_1k": usd_day / env.requests_per_day * 1_000}

show("baseline", estimate(Envelope(60_000, 2_500, 350)))
show("60% cached", estimate(Envelope(60_000, 2_500, 350, cached_share=0.6)))
show("+10% strong", estimate(Envelope(60_000, 2_500, 350, cached_share=0.6, strong_share=0.1)))
# Break it: forget the 1,500 reasoning tokens, or stuff 300K tokens of context into every prompt
show("+reasoning", estimate(Envelope(60_000, 2_500, 350 + 1_500, cached_share=0.6)))
show("300K prompt", estimate(Envelope(1_000, 300_000, 350)))

baseline     peak 6.9 QPS · 171M tok/day · peak 1.19M TPM · $25.50/day · $0.42 per 1k req
60% cached   peak 6.9 QPS · 171M tok/day · peak 1.19M TPM · $17.40/day · $0.29 per 1k req
+10% strong  peak 6.9 QPS · 171M tok/day · peak 1.19M TPM · $49.56/day · $0.83 per 1k req
+reasoning   peak 6.9 QPS · 261M tok/day · peak 1.81M TPM · $62.40/day · $1.04 per 1k req
300K prompt  peak 0.1 QPS · 300M tok/day · peak 2.09M TPM · $60.26/day · $60.26 per 1k req

# script_id: day_097_system_design_interviews/envelope
def latency_budget(stages_ms: dict[str, float], slo_ms: float) -> tuple[float, bool]:
    """Sum of per-stage p95s vs the SLO. Summing p95s is pessimistic, which is fine for a budget."""
    total = sum(stages_ms.values())
    return total, total <= slo_ms

# Illustrative placeholders. Replace every number with your own measured p95s (R5 launch report).
stages = {"auth + cache lookup": 30, "hybrid retrieve": 150, "rerank": 250,
          "guardrail check": 120, "model: time to first token": 1_100}
total, ok = latency_budget(stages, slo_ms=1_500)
print(f"time to first token {total:.0f} ms vs SLO 1500 ms -> {'within budget' if ok else 'over budget'}")
for name, ms in sorted(stages.items(), key=lambda kv: -kv[1])[:2]:
    print(f"  cut candidate: {name} ({ms:.0f} ms)")

time to first token 1650 ms vs SLO 1500 ms -> over budget
  cut candidate: model: time to first token (1100 ms)
  cut candidate: rerank (250 ms)

# script_id: day_097_system_design_interviews/rubric
# Scores a design against the six-dimension rubric (or the 4-dimension first-role track).
# Install: none (stdlib, Python 3.11+). Env vars: none. Services: none.
FULL = ["scoping", "eval_first", "architecture", "production", "safety_data", "tradeoffs_comms"]
FIRST_ROLE = ["scoping", "eval_first", "architecture", "tradeoffs_comms"]
PASS_MARK = {"full": 12, "first_role": 8}

def score(scores: dict[str, int], track: str = "full") -> tuple[int, int, bool, list[str]]:
    dims = FULL if track == "full" else FIRST_ROLE
    missing = [d for d in dims if d not in scores]
    if missing:
        raise ValueError(f"score every dimension; missing {missing}")
    if any(not 0 <= scores[d] <= 3 for d in dims):
        raise ValueError("each dimension is scored 0-3")
    total = sum(scores[d] for d in dims)
    zeros = [d for d in dims if scores[d] == 0]
    weakest = sorted(dims, key=lambda d: scores[d])[:2]
    return total, 3 * len(dims), total >= PASS_MARK[track] and not zeros, weakest

mine = {"scoping": 3, "eval_first": 2, "architecture": 3, "production": 2,
        "safety_data": 1, "tradeoffs_comms": 3}
total, out_of, passed, weakest = score(mine)
print(f"{total}/{out_of} {'PASS' if passed else 'NO PASS'} · practise next: {', '.join(weakest)}")
mine["safety_data"] = 0  # forgot the trust boundary entirely
print("{}/{} {}".format(*score(mine)[:2], "PASS" if score(mine)[2] else "NO PASS (a 0 anywhere fails)"))

14/18 PASS · practise next: safety_data, eval_first
13/18 NO PASS (a 0 anywhere fails)