Practical Interviews

# script_id: day_098_practical_interviews/kata_extract
# Kata 1: turn one Orbit issue into a typed record, then score 10 gold cases per field.
# Install: uv add "openai>=3.46,<4" pydantic
# Env: OPENAI_API_KEY; ORBIT_DATA (default datasets/orbit). Services: none.
import json
import os
from pathlib import Path
from typing import Literal

from openai import OpenAI
from pydantic import BaseModel

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
COMPONENTS = ("core scheduler flows broker-redis broker-postgres broker-sqs cli "
              "dashboard retries results observability auth docs")
client = OpenAI()


class QuickTriage(BaseModel):  # interview-sized; Atlas's full IssueRecord is Day 15's
    type: Literal["bug", "feature", "question", "docs"]
    components: list[str]
    severity: Literal["S1", "S2", "S3", "S4"] | None  # nullable so "not a bug" is expressible
    summary: str


def extract(issue: dict) -> QuickTriage | None:
    resp = client.responses.parse(
        model=MODEL,
        instructions=f"Triage this Orbit issue. components: one or more of: {COMPONENTS}. "
                     "severity only for bugs (S1 is worst), else null. summary: one English sentence.",
        input=f"# {issue['title']}\n\n{issue['body'][:6000]}",  # one issue is ~8,600 words
        text_format=QuickTriage,
    )
    if resp.status != "completed" or resp.output_parsed is None:
        return None  # refusal or truncation: validate, don't trust
    return resp.output_parsed


with open(DATA_DIR / "issues/issues.jsonl", encoding="utf-8") as f:
    ISSUES = {(row := json.loads(line))["id"]: row for line in f}
record = extract(ISSUES[4])
print(record.model_dump_json(indent=2) if record else "no record")

# script_id: day_098_practical_interviews/kata_extract
FIELDS = ("type", "components", "severity")


def field_accuracy(cases: list[dict], predict) -> dict[str, float]:
    """Share of cases where each field matches gold exactly (components compared as sets)."""
    hits = dict.fromkeys(FIELDS, 0)
    for case in cases:
        pred = predict(case)
        if pred is None:  # still in the denominator: refusals must cost you
            continue
        for field in FIELDS:
            want, got = case[field], getattr(pred, field)
            if field == "components":
                want, got = set(want), set(got)
            hits[field] += want == got
    return {f: round(hits[f] / len(cases), 2) for f in FIELDS}


with open(DATA_DIR / "eval/extraction_gold.jsonl", encoding="utf-8") as f:
    gold = [json.loads(line) for line in f][:10]
print(field_accuracy(gold, lambda case: extract(ISSUES[case["issue_id"]])))

{
  "type": "bug",
  "components": ["core"],
  "severity": "S2",
  "summary": "A long-running statement job runs on two workers, so customers get two emails."
}
{'type': 0.9, 'components': 0.7, 'severity': 0.6}

# script_id: day_098_practical_interviews/kata_tool_loop
# Kata 2: a strict tool, a dispatcher and a bounded loop over the Responses API.
# Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY; ORBIT_DATA. Services: none.
import json
import os
from dataclasses import dataclass
from pathlib import Path

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
client = OpenAI()
with open(DATA_DIR / "issues/issues.jsonl", encoding="utf-8") as f:
    ISSUES = {(row := json.loads(line))["id"]: row for line in f}


def get_issue(issue_id: int) -> dict:
    issue = ISSUES.get(issue_id)
    if issue is None:
        return {"error": f"no issue {issue_id}"}  # an error is data the model can act on
    return {k: issue[k] for k in ("id", "title", "state", "labels", "duplicate_of")}


TOOLS = [{"type": "function", "name": "get_issue", "strict": True,
          "description": "Fetch one Orbit issue by its numeric id.",
          "parameters": {"type": "object", "properties": {"issue_id": {"type": "integer"}},
                         "required": ["issue_id"], "additionalProperties": False}}]
REGISTRY = {"get_issue": get_issue}


def dispatch(name: str, arguments: str) -> str:
    try:
        result = REGISTRY[name](**json.loads(arguments))  # arguments is a JSON string
    except Exception as e:  # unknown tool, bad args: report, don't crash the loop
        result = {"error": f"{type(e).__name__}: {e}"}
    return json.dumps(result)

# script_id: day_098_practical_interviews/kata_tool_loop
@dataclass
class LoopResult:
    text: str
    rounds: int
    stop_reason: str  # "done" | "max_rounds"


def run_tool_loop(question: str, *, max_rounds: int = 5) -> LoopResult:
    items: list = [{"role": "user", "content": question}]
    for rnd in range(1, max_rounds + 1):
        resp = client.responses.create(model=MODEL, input=items, tools=TOOLS)
        calls = [item for item in resp.output if item.type == "function_call"]
        if not calls:
            return LoopResult(resp.output_text, rnd, "done")
        items += resp.output  # keep the call items (and any reasoning) in the history
        for call in calls:    # answer EVERY call; parallel calls are on by default
            items.append({"type": "function_call_output", "call_id": call.call_id,
                          "output": dispatch(call.name, call.arguments)})
    return LoopResult("", max_rounds, "max_rounds")


result = run_tool_loop("Is issue 145 a duplicate? If so, give the original's title and state.")
print(result.text)
print(f"rounds={result.rounds} stop={result.stop_reason}")

Yes. #145 is a duplicate of #143, "3.1.0: reconnect storm after Redis failover, thousands of
connections and an ORB-1001 flood take Redis down again", which is closed.
rounds=3 stop=done

# script_id: day_098_practical_interviews/kata_tool_loop_broken
# fragment: the one-line bug; drop it into run_tool_loop above to see the failure
        for call in calls[:1]:  # BUG: a parallel second call never gets an output
            items.append({"type": "function_call_output", "call_id": call.call_id,
                          "output": dispatch(call.name, call.arguments)})

# script_id: day_098_practical_interviews/kata_rag
# Kata 3: embed 45 pages, cosine top-3, typed answer, reject citations you never retrieved.
# Install: uv add "openai>=3.46,<4" pydantic numpy · Env: OPENAI_API_KEY; ORBIT_DATA.
import os
from pathlib import Path

import numpy as np
from openai import OpenAI
from pydantic import BaseModel

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
EMBED_MODEL = os.environ.get("ATLAS_EMBED_MODEL", "text-embedding-3-small")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
client = OpenAI()
DOCS = {p.stem: p.read_text(encoding="utf-8") for p in sorted((DATA_DIR / "docs").glob("*.md"))}
IDS = list(DOCS)


class Citation(BaseModel):
    doc_id: str
    quote: str


class Answer(BaseModel):
    text: str
    citations: list[Citation]


def embed(texts: list[str]) -> np.ndarray:
    resp = client.embeddings.create(model=EMBED_MODEL, input=texts)
    vecs = np.array([d.embedding for d in resp.data], dtype=np.float32)
    return vecs / np.linalg.norm(vecs, axis=1, keepdims=True)  # unit length: dot = cosine


DOC_VECS = embed([DOCS[i][:6000] for i in IDS])  # whole pages, well under the 8,192-token cap

# script_id: day_098_practical_interviews/kata_rag


def search_docs(query: str, k: int = 3) -> list[str]:
    return [IDS[i] for i in np.argsort(-(DOC_VECS @ embed([query])[0]))[:k]]


def ask(question: str) -> Answer:
    ids = search_docs(question)
    context = "\n\n".join(f"<doc id='{i}'>\n{DOCS[i][:4000]}\n</doc>" for i in ids)
    resp = client.responses.parse(model=MODEL, text_format=Answer, input=f"{context}\n\nQ: {question}",
                                  instructions="Answer only from the docs and cite their ids. "
                                               "If the docs don't say, say so.")
    if resp.output_parsed is None:
        raise RuntimeError(f"no parsed answer (status={resp.status})")
    answer = resp.output_parsed
    real = [c for c in answer.citations if c.doc_id in ids]
    print(f"rejected {len(answer.citations) - len(real)} fake citations")
    return answer.model_copy(update={"citations": real})


ans = ask("How long does Orbit wait between retry attempts by default?")
print(ans.text, "\ncitations:", [c.doc_id for c in ans.citations])

rejected 0 fake citations
By default Orbit uses exponential backoff between attempts (3 attempts in total), with any single
delay capped at 600 seconds.
citations: ['retries-and-backoff']

# script_id: day_098_practical_interviews/mock_takehome
# atlas-dupes: suggest up to 3 EARLIER Orbit issues a new issue may duplicate, plus an eval.
# Install: uv add numpy pydantic  (OpenAI backend or --verify: uv add "openai>=3.46,<4")
# Env: ORBIT_DATA; DUPES_BACKEND=keyword (default: offline, $0) | openai (needs OPENAI_API_KEY);
#      DUPES_MIN_SCORE (default 0.25, tuned on the keyword backend; re-tune for any other).
# Run: python mock_takehome.py   |   python mock_takehome.py --title "..." --body "..."
import argparse
import json
import os
import re
import time
import zlib
from collections import Counter
from functools import cache
from pathlib import Path

import numpy as np
from pydantic import BaseModel

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
BACKEND = os.environ.get("DUPES_BACKEND", "keyword")
MIN_SCORE = float(os.environ.get("DUPES_MIN_SCORE", "0.25"))
MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
EMBED_MODEL = os.environ.get("ATLAS_EMBED_MODEL", "text-embedding-3-small")
MAX_CHARS = 6000  # issue 227 is ~8,600 words: over the 8,192-token embedding cap untruncated
USAGE: Counter = Counter()  # tokens by kind; only the OpenAI paths add to it


def load_issues() -> list[dict]:
    with open(DATA_DIR / "issues/issues.jsonl", encoding="utf-8") as f:
        return [json.loads(line) for line in f]


def issue_text(issue: dict) -> str:
    # The title is the densest signal, so it counts twice; long log tails mostly add noise.
    return f"{issue['title']}\n{issue['title']}\n{issue.get('body', '')}"[:MAX_CHARS]


ISSUES = load_issues()
BY_ID = {i["id"]: i for i in ISSUES}

# script_id: day_098_practical_interviews/mock_takehome
TOKEN = re.compile(r"[a-z0-9][a-z0-9_.\-]+")  # keeps ORB-1101, orbit.toml, lease_seconds whole
DF = Counter(t for i in ISSUES for t in set(TOKEN.findall(issue_text(i).lower())))


def keyword_embed(texts: list[str], dim: int = 4096) -> np.ndarray:
    """Offline fallback: hashed TF-IDF vectors, so the take-home runs without a key."""
    out = np.zeros((len(texts), dim), dtype=np.float32)
    for row, text in enumerate(texts):
        for tok, n in Counter(TOKEN.findall(text.lower())).items():
            idf = max(np.log(len(ISSUES) / (1 + DF[tok])), 0.0)  # rare terms weigh most
            out[row, zlib.crc32(tok.encode()) % dim] += (1 + np.log(n)) * idf
    return out / np.maximum(np.linalg.norm(out, axis=1, keepdims=True), 1e-9)


@cache
def llm():
    from openai import OpenAI  # lazy: the keyword path needs neither the package nor a key
    return OpenAI()


def openai_embed(texts: list[str]) -> np.ndarray:
    vecs = []
    for start in range(0, len(texts), 64):  # small batches keep each request modest
        resp = llm().embeddings.create(model=EMBED_MODEL, input=texts[start:start + 64])
        USAGE["embed"] += resp.usage.prompt_tokens
        vecs += [d.embedding for d in resp.data]
    v = np.array(vecs, dtype=np.float32)
    return v / np.linalg.norm(v, axis=1, keepdims=True)


EMBED = {"keyword": keyword_embed, "openai": openai_embed}[BACKEND]

# script_id: day_098_practical_interviews/mock_takehome
class Index:
    """In-memory vector index: the interview-sized stand-in for pgvector."""

    def __init__(self) -> None:
        self.ids: list[str] = []
        self.meta: list[dict] = []
        self.rows: list[np.ndarray] = []
        self.pos: dict[str, int] = {}

    def upsert(self, ids: list[str], vectors, meta: list[dict]) -> None:
        for id_, vec, m in zip(ids, vectors, meta):
            if id_ in self.pos:  # re-ingesting an issue replaces it, so reruns are idempotent
                self.rows[self.pos[id_]], self.meta[self.pos[id_]] = vec, m
                continue
            self.pos[id_] = len(self.ids)
            self.ids.append(id_)
            self.meta.append(m)
            self.rows.append(vec)

    def search(self, vector: np.ndarray, k: int, where=None) -> list[tuple[str, float]]:
        keep = [j for j, m in enumerate(self.meta) if where is None or where(m)]
        if not keep:
            return []
        scores = np.stack([self.rows[j] for j in keep]) @ vector
        return [(self.ids[keep[t]], float(scores[t])) for t in np.argsort(-scores)[:k]]


VECS = dict(zip((i["id"] for i in ISSUES), EMBED([issue_text(i) for i in ISSUES])))
INDEX = Index()
INDEX.upsert([str(i) for i in VECS], list(VECS.values()),
             [{"created_at": BY_ID[i]["created_at"]} for i in VECS])

# script_id: day_098_practical_interviews/mock_takehome
class DupVerdict(BaseModel):
    is_duplicate: bool
    reason: str


VERDICTS: dict[tuple, DupVerdict | None] = {}


def verify_pair(new: dict, old: dict) -> DupVerdict | None:
    key = (new.get("id"), old["id"])
    if key[0] is not None and key in VERDICTS:
        return VERDICTS[key]
    resp = llm().responses.parse(
        model=MODEL, text_format=DupVerdict,
        instructions="Is NEW the same underlying problem or request as OLD? Same symptom and "
                     "cause = duplicate. Same component but a different bug = not a duplicate.",
        input=f"OLD #{old['id']}:\n{issue_text(old)[:1500]}\n\nNEW:\n{issue_text(new)[:1500]}")
    USAGE["in"] += resp.usage.input_tokens
    USAGE["out"] += resp.usage.output_tokens
    verdict = resp.output_parsed if resp.status == "completed" else None  # None = unconfirmed
    VERDICTS[key] = verdict
    return verdict

# script_id: day_098_practical_interviews/mock_takehome
def suggest_duplicates(issue: dict, k: int = 3, verify: bool = False,
                       min_score: float = MIN_SCORE) -> list[tuple[int, float, str]]:
    vec = VECS.get(issue.get("id"))
    if vec is None:  # a brand-new issue: embed it now
        vec = EMBED([issue_text(issue)])[0]
    cutoff = issue.get("created_at", "9999")  # strictly earlier: excludes itself and the future
    out = []
    for sid, score in INDEX.search(vec, k, where=lambda m: m["created_at"] < cutoff):
        old = BY_ID[int(sid)]
        if score < min_score:
            continue
        reason = f"similar wording to #{old['id']}: {old['title'][:60]}"
        if verify:
            verdict = verify_pair(issue, old)
            if verdict is None or not verdict.is_duplicate:
                continue
            reason = verdict.reason
        out.append((old["id"], round(score, 3), reason))
    return out


def gold_for(issue: dict) -> set[int]:
    """Earlier members of the issue's duplicate cluster: any of them is a correct suggestion."""
    root = issue["duplicate_of"]
    cluster = {root} | {i["id"] for i in ISSUES if i["duplicate_of"] == root}
    return {c for c in cluster if BY_ID[c]["created_at"] < issue["created_at"]}

# script_id: day_098_practical_interviews/mock_takehome
def hit_at_k(k: int = 3, verify: bool = False, misses: list | None = None,
             min_score: float = 0.0) -> dict[str, float]:
    hits1 = hitsk = 0
    ms: list[float] = []
    queries = [i for i in ISSUES if i["duplicate_of"]]
    for q in queries:
        t0 = time.perf_counter()
        got = [sid for sid, _, _ in suggest_duplicates(q, k, verify, min_score)]
        ms.append((time.perf_counter() - t0) * 1000)
        gold = gold_for(q)
        hits1 += bool(got) and got[0] in gold
        hitsk += bool(set(got) & gold)
        if misses is not None and not set(got) & gold:
            misses.append(q["id"])
    n = len(queries)
    return {"n": n, "hit@1": round(hits1 / n, 2), f"hit@{k}": round(hitsk / n, 2),
            "p95_ms": round(float(np.percentile(ms, 95)), 1)}


def false_alarm_rate(min_score: float = MIN_SCORE, verify: bool = False) -> dict[str, float]:
    queries = [i for i in ISSUES if not i["duplicate_of"]]
    shown = sum(bool(suggest_duplicates(q, 3, verify, min_score)) for q in queries)
    return {"n": len(queries), "false_alarms": round(shown / len(queries), 2)}

# script_id: day_098_practical_interviews/mock_takehome
PRICE_PER_1M = {"embed": 0.02, "in": 0.10, "out": 0.50}  # USD; as of 2026-11, verify at pricing


def cost_usd() -> float:
    return sum(USAGE[kind] * price for kind, price in PRICE_PER_1M.items()) / 1e6


def main(argv: list[str] | None = None) -> None:
    ap = argparse.ArgumentParser(description="Suggest earlier Orbit issues a new one may duplicate.")
    ap.add_argument("--verify", action="store_true", help="LLM check per suggestion (needs a key)")
    ap.add_argument("--issue", type=int, help="suggest for an existing issue id")
    ap.add_argument("--title", default="", help="suggest for a new issue")
    ap.add_argument("--body", default="")
    args = ap.parse_args(argv)
    if args.issue or args.title:  # suggestion mode
        issue = BY_ID[args.issue] if args.issue else {"title": args.title, "body": args.body}
        for sid, score, reason in suggest_duplicates(issue, verify=args.verify):
            print(f"#{sid}  score={score:.3f}  {reason}")
    else:  # eval mode (the default, so a grader's first run shows the numbers)
        misses: list[int] = []
        ranking = hit_at_k(3, args.verify, misses=misses)
        shown = hit_at_k(3, args.verify, min_score=MIN_SCORE)
        alarms = false_alarm_rate(MIN_SCORE, args.verify)
        print(f"ranking: {ranking} misses={misses}")
        print(f"at threshold {MIN_SCORE}: hit@3 {shown['hit@3']} · false alarms {alarms}")
    print(f"backend={BACKEND} verify={args.verify} cost=${cost_usd():.4f}")


if __name__ == "__main__":
    main()

ranking: {'n': 37, 'hit@1': 0.97, 'hit@3': 1.0, 'p95_ms': 0.2} misses=[]
at threshold 0.25: hit@3 0.68 · false alarms {'n': 203, 'false_alarms': 0.29}
backend=keyword verify=False cost=$0.0000

# script_id: day_098_practical_interviews/test_takehome
# pytest for atlas-dupes, offline on the keyword backend. Save next to mock_takehome.py.
# Install: uv add pytest numpy pydantic · Run: pytest -q test_takehome.py
import os

os.environ.setdefault("DUPES_BACKEND", "keyword")  # tests must not spend money or need a key

from mock_takehome import BY_ID, ISSUES, hit_at_k, suggest_duplicates  # noqa: E402


def test_never_suggests_itself_or_the_future():
    for issue in ISSUES:
        for sid, _, _ in suggest_duplicates(issue, min_score=0.0):
            assert sid != issue["id"]
            assert BY_ID[sid]["created_at"] < issue["created_at"]


def test_new_issue_without_an_id():
    got = suggest_duplicates({"title": "Reconnect storm after Redis Sentinel failover",
                              "body": "ORB-1001 flood, Redis CPU at 100%"}, min_score=0.0)
    assert [sid for sid, _, _ in got][:3] and 143 in {sid for sid, _, _ in got}


def test_ranking_does_not_regress():
    assert hit_at_k(3)["hit@3"] >= 0.9  # set below today's baseline: n is only 37


if __name__ == "__main__":
    import pytest
    raise SystemExit(pytest.main([__file__, "-q"]))

# atlas-dupes
Suggests up to 3 earlier Orbit issues a new issue may duplicate, with a reason each.

## Run
uv sync && python mock_takehome.py                  # eval (keyword backend, offline, $0)
DUPES_BACKEND=openai python mock_takehome.py --verify   # needs OPENAI_API_KEY

## Results (n=37 duplicate queries, 203 non-duplicates; small sample)
| Config | hit@1 | hit@3 | false alarms | $/run | p95 |
| keyword, no threshold | … | … | … | $0 | … ms |

## Decisions
- Search only earlier issues (leakage guard in the filter, tested).
- Truncate to 6,000 chars: one issue exceeds the embedding input cap.

## Known failures
- …three concrete ones, with issue ids…

## With more time
- …the next three improvements and the metric each would move…

# script_id: day_098_practical_interviews/dsa_clusters
# Two DSA patterns inside AI code: BFS over a graph (duplicate clusters) and a heap (top-k).
# Install: none (stdlib, Python 3.11+). Env: ORBIT_DATA. Services: none.
import heapq
import json
import os
from collections import defaultdict, deque
from pathlib import Path

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
with open(DATA_DIR / "issues/issues.jsonl", encoding="utf-8") as f:
    issues = [json.loads(line) for line in f]

graph: dict[int, set[int]] = defaultdict(set)  # undirected: duplicate <-> original
for issue in issues:
    if issue["duplicate_of"]:
        graph[issue["id"]].add(issue["duplicate_of"])
        graph[issue["duplicate_of"]].add(issue["id"])

seen: set[int] = set()
clusters: list[list[int]] = []
for start in sorted(graph):  # BFS from each unvisited node: O(V + E) overall
    if start in seen:
        continue
    queue, members = deque([start]), []
    seen.add(start)
    while queue:
        node = queue.popleft()
        members.append(node)
        for nxt in graph[node] - seen:
            seen.add(nxt)
            queue.append(nxt)
    clusters.append(sorted(members))

print(f"{len(clusters)} clusters, {sum(map(len, clusters))} issues, first: {clusters[0]}")
scores = {143: 0.41, 145: 0.38, 61: 0.12, 209: 0.09, 161: 0.35}  # toy similarity scores
print("top 3 via heap:", heapq.nlargest(3, scores, key=scores.get))  # O(n log k), no full sort

15 clusters, 52 issues, first: [4, 22, 41, 67]
top 3 via heap: [143, 145, 161]