Anthropic Software Engineer Phone Screen Questions

## Problem Build a simple agent orchestration framework. Each agent runs a task (a callable) and may spawn sub-tasks. Tasks have dependencies — a task cannot start until all its dependencies complete. The system should run independent tasks in parallel (simulated with threads or asyncio) and return results in topological order. ```python class AgentOrchestrator: def add_task(self, task_id: str, fn: callable, deps: list[str]) -> None: ... def run_all(self) -> dict[str, any]: ... ``` ## Example ``` orchestrator = AgentOrchestrator() orchestrator.add_task("fetch", lambda: fetch_data(), deps=[]) orchestrator.add_task("parse", lambda: parse(results["fetch"]), deps=["fetch"]) orchestrator.add_task("store", lambda: store(results["parse"]), deps=["parse"]) orchestrator.add_task("notify", lambda: send_alert(), deps=["fetch"]) results = orchestrator.run_all() # fetch runs first; parse and notify run after fetch; store runs after parse ``` ## Follow-ups 1. How do you detect and report circular dependencies before execution? 2. What happens if a task raises an exception — do dependent tasks still run? 3. How would you add timeout and retry logic per task? 4. How would this design change if tasks are distributed across machines?

## Problem Extend or implement a caching system, possibly adding eviction policies (LRU/LFU) or cache-miss handling. ## Likely LeetCode equivalent Similar to LRU Cache (LC 146) or LFU Cache (LC 460). ## Tags hash_table, design, cache, anthropic

## Problem Find duplicate files in a directory tree, likely by comparing file content hashes or sizes. ## Likely LeetCode equivalent Similar to Find Duplicate File in System (LC 609). ## Tags hash_table, strings, anthropic, filesystem

## Problem Compute both the mode (most frequent element) and median of a data stream or array. ## Likely LeetCode equivalent Similar to Find Median from Data Stream (LC 295) combined with frequency counting. ## Tags heap, hash_table, math, anthropic

## Problem Design an image processing pipeline that applies a sequence of filters to a 2D pixel grid. Each filter is a function `(grid) -> grid`. Filters include: `grayscale`, `blur(radius)`, `threshold(value)`, and `crop(x1,y1,x2,y2)`. The pipeline must be lazy — filters are composed but not applied until `execute()` is called. ```python class ImagePipeline: def __init__(self, image: list[list[int]]): ... def add_filter(self, name: str, **kwargs) -> 'ImagePipeline': ... def execute(self) -> list[list[int]]: ... ``` ## Example ``` Input image (grayscale): 4x4 grid of pixel values 0-255 pipeline = ImagePipeline(image) result = ( pipeline .add_filter("blur", radius=1) .add_filter("threshold", value=128) .add_filter("crop", x1=1, y1=1, x2=3, y2=3) .execute() ) # result is a 2x2 grid with filters applied in order ``` ## Follow-ups 1. How would you parallelize the blur filter across rows or tiles? 2. If the image doesn't fit in memory, how do you apply filters in a streaming fashion? 3. How would you optimize a pipeline where two consecutive filters can be fused? 4. What interface changes are needed to support color (RGB) images?

## Problem You receive a stream of events describing an object's state over time. Events are either: - `SNAPSHOT t obj_state` — full state at time `t` - `DIFF t patch` — a JSON-patch-style diff applied at time `t` Given a query time `q`, reconstruct the object's state at time `q` using the nearest preceding snapshot plus any diffs between the snapshot and `q`. ```python def reconstruct_state( events: list[dict], # [{"type": "SNAPSHOT"|"DIFF", "t": int, "data": dict}] query_t: int ) -> dict: pass ``` ## Example ``` events = [ {"type": "SNAPSHOT", "t": 0, "data": {"x": 1, "y": 2}}, {"type": "DIFF", "t": 5, "data": {"op": "set", "key": "x", "val": 10}}, {"type": "SNAPSHOT", "t": 10, "data": {"x": 10, "y": 5}}, {"type": "DIFF", "t": 15, "data": {"op": "del", "key": "y"}}, ] reconstruct_state(events, query_t=12) # -> {"x": 10, "y": 5} (snapshot at t=10, no diffs between 10 and 12) reconstruct_state(events, query_t=17) # -> {"x": 10} (snapshot at t=10 + diff at t=15 removes y) ``` ## Follow-ups 1. How do you index events to find the nearest preceding snapshot in O(log n)? 2. What if diffs can arrive out of order? 3. How would you design a compaction strategy to prune old snapshots and diffs?

## Problem You are given a string containing a simple markup language with tags `[b]`, `[i]`, `[u]`, `[link=url]`, and `[/tag]`. Implement a parser that strips all markup and returns clean plain text. Additionally, implement a renderer that converts markup to HTML equivalents. ```python class TextHandler: def __init__(self, markup: str): ... def to_plain(self) -> str: ... def to_html(self) -> str: ... def extract_links(self) -> list[str]: ... ``` ## Example ``` Input: "Hello [b]world[/b]! Visit [link=https://example.com]our site[/link]." to_plain() -> "Hello world! Visit our site." to_html() -> 'Hello <b>world</b>! Visit <a href="https://example.com">our site</a>.' extract_links() -> ["https://example.com"] ``` ## Follow-ups 1. How do you handle malformed or unclosed tags gracefully? 2. What if tags can be nested (`[b][i]text[/i][/b]`) — how does your parser handle overlapping? 3. How would you support custom user-defined tags? 4. What are the XSS risks if this output is rendered in a browser and how do you mitigate them?