{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# \ud83e\udd16 AI-Assisted & Practical Coding Interview Guide\n",
    "### Stateful Simulations, Robot Navigation, Interactive Games & Progressive System Design\n",
    "\n",
    "**Target Roles:** Machine Learning Engineer (MLE), AI Systems Engineer, Applied Scientist, Software Engineer\n",
    "\n",
    "---\n",
    "\n",
    "## \ud83d\udccc Overview: Practical Implementation & AI-Assisted Coding Rounds\n",
    "\n",
    "Across the software engineering and machine learning interview landscape, coding rounds have increasingly expanded beyond traditional algorithmic riddles. These kinds of questions get asked in companies like **Shopify**, **LinkedIn**, **Meta**, and similar tech organizations, while many other companies have also adopted comparable AI-assisted pair-programming or progressive implementation formats.\n",
    "\n",
    "In these rounds (often conducted in local IDEs like Cursor/VS Code with Copilot or in shared CoderPad environments with AI panels), candidates are given bare-bones class structures and domain rules. Rather than testing memorization of obscure competitive programming tricks, these sessions evaluate:\n",
    "\n",
    "1. **Extensible Architecture (Open-Closed Principle):** Can your class design cleanly absorb follow-up requirements (e.g., Undo, Jump, Eviction policies) without rewriting existing logic?\n",
    "2. **State Machine Integrity:** Can you manage multi-step states (headings, obstacle sets, body queues, remaining TTL intervals) without bugs?\n",
    "3. **AI Verification & Edge Cases:** AI assistants often produce code with subtle bugs (boundary off-by-ones, snake tail vacation timing, expired records in backups). Can you spot and correct them?\n",
    "4. **Clean Code & Verification:** Can you write concise, functional code and immediately prove its correctness with assertions?\n",
    "\n",
    "---\n",
    "\n",
    "## \ud83d\uddfa\ufe0f Curriculum Overview\n",
    "\n",
    "| Module | # | Problem | Core Pattern / Archetype | Concepts Tested |\n",
    "|---|---|---|---|---|\n",
    "| **1. Robot Navigation & Grid Simulation** | 1 | **Mars Rover / Grid Robot Command Engine** | Grid Navigation & Collision | Cardinal directions, ray-step obstacle collision, undo history |\n",
    "| | 2 | **Blind-Grid Robot Room Cleaner** | Sensor Traversal & Backtracking | Sensor API, virtual coordinates, blind DFS backtracking with orientation restoration |\n",
    "| | 3 | **Cleaning Robot with Cycle Detection** | Matrix Simulation & Cycle Detection | Deterministic grid reflection, state space cycle detection |\n",
    "| **2. Interactive Game Engines & State Machines** | 4 | **Snake Game Simulator with Growth & Collision** | Stateful Game Simulation | deque body, sequential food ingestion, tail vacation vs head collision order |\n",
    "| | 5 | **Connect Four / Tic-Tac-Toe with O(1) Win Checking** | Low-Level System Design | Vector counters (rows, cols, diagonals), O(1) evaluation, gravity drop |\n",
    "| | 6 | **Battleship Fleet Coordinator** | Grid Geometry & Event Handling | Multi-deck ship geometry, hit/miss tracking, sinking validation |\n",
    "| **3. Progressive Systems & Storage** | 7 | **In-Memory Key-Field DB with TTL & Backup/Restore** | Progressive 4-Level Architecture | CRUD -> Prefix Scan -> TTL interval -> Remaining TTL restore |\n",
    "| | 8 | **Cloud Storage Manager with Quota & Auto-Eviction** | Resource Quotas & Auto-Eviction | User quota management, largest-first tie-broken auto-eviction, file compression |\n",
    "| | 9 | **Text Editor with Undo/Redo & Find/Replace** | Buffer Manipulation & Command History | Buffer slicing, pattern replacement count, transactional Command Undo/Redo |\n",
    "| **4. Enterprise Infrastructure & Scheduling** | 10 | **Multi-Account Financial Ledger with Rollbacks** | Double-Entry & Atomic Transactions | Double-entry bookkeeping, point-in-time binary search balance, atomic rollback |\n",
    "| | 11 | **Sliding Window Log & Token Bucket Rate Limiter** | Traffic Shaping & Concurrency | Timestamp sliding window log, continuous refill token bucket |\n",
    "| | 12 | **DAG Task Engine with Dependency Resolution** | Graph Pipelines & Orchestration | Directed acyclic graph, Kahn's topological sort, cycle detection, execution pipeline |\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 1. Mars Rover / Grid Robot Command Engine\n",
    "\n",
    "### Problem Statement\n",
    "Design a controller for a robot operating on an unbounded 2D Cartesian grid.\n",
    "- The robot starts at coordinate `(x, y)` facing one of four cardinal directions: `'N'`, `'E'`, `'S'`, or `'W'`.\n",
    "- It executes a sequence of commands passed as a space-separated string:\n",
    "  - `'F'`: Move forward 1 unit in current direction.\n",
    "  - `'B'`: Move backward 1 unit in current direction (maintaining heading).\n",
    "  - `'L'`: Turn 90\u00b0 counter-clockwise (e.g., `'N'` $\\to$ `'W'`).\n",
    "  - `'R'`: Turn 90\u00b0 clockwise (e.g., `'N'` $\\to$ `'E'`).\n",
    "  - `'J<k>'`: Jump forward $k$ units in current direction (e.g., `'J3'`).\n",
    "- **Obstacle Collision:** If moving forward/backward/jumping would cross or land on a known obstacle `(ox, oy)`, the robot **halts immediately before the obstacle** and completes no further steps of that movement command.\n",
    "- **Undo Capability:** `undo(steps)` reverts the robot's state to what it was `steps` commands prior.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class GridRobot:\n",
    "    def __init__(self, x: int = 0, y: int = 0, direction: str = 'N', obstacles: set[tuple[int, int]] = None):\n",
    "        pass\n",
    "\n",
    "    def execute(self, commands: str) -> tuple[int, int, str]:\n",
    "        \"\"\"Executes commands; returns (x, y, direction).\"\"\"\n",
    "        pass\n",
    "\n",
    "    def undo(self, steps: int = 1) -> tuple[int, int, str]:\n",
    "        \"\"\"Reverts the last `steps` actions; returns (x, y, direction).\"\"\"\n",
    "        pass\n",
    "```\n",
    "\n",
    "### Key Traps & AI Failure Modes\n",
    "- **Obstacle Jumping:** AI assistants often only check if the final destination has an obstacle, forgetting that the robot cannot teleport through an obstacle between the start and destination! Ray-step checking is required.\n",
    "- **Direction Modulo:** Modulo arithmetic with negative numbers in some languages can produce negative offsets; in Python `(idx - 1) % 4` is safe.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from typing import List, Tuple, Set, Optional\n",
    "\n",
    "class GridRobot:\n",
    "    \"\"\"Stateful 2D grid robot simulator with cardinal navigation, obstacle avoidance, and undo history.\"\"\"\n",
    "    DIRS = ['N', 'E', 'S', 'W']\n",
    "    OFFSETS = {'N': (0, 1), 'E': (1, 0), 'S': (0, -1), 'W': (-1, 0)}\n",
    "\n",
    "    def __init__(self, x: int = 0, y: int = 0, direction: str = 'N', obstacles: Optional[Set[Tuple[int, int]]] = None):\n",
    "        self.x = x\n",
    "        self.y = y\n",
    "        self.direction = direction\n",
    "        self.obstacles = set(obstacles) if obstacles else set()\n",
    "        self.history: List[Tuple[int, int, str]] = []\n",
    "\n",
    "    def execute(self, commands: str) -> Tuple[int, int, str]:\n",
    "        self.history.append((self.x, self.y, self.direction))\n",
    "        for token in commands.split():\n",
    "            if token == 'L':\n",
    "                idx = self.DIRS.index(self.direction)\n",
    "                self.direction = self.DIRS[(idx - 1) % 4]\n",
    "            elif token == 'R':\n",
    "                idx = self.DIRS.index(self.direction)\n",
    "                self.direction = self.DIRS[(idx + 1) % 4]\n",
    "            elif token == 'F':\n",
    "                self._step(1)\n",
    "            elif token == 'B':\n",
    "                self._step(-1)\n",
    "            elif token.startswith('J'):\n",
    "                dist = int(token[1:]) if len(token) > 1 else 2\n",
    "                self._step(dist)\n",
    "        return (self.x, self.y, self.direction)\n",
    "\n",
    "    def _step(self, steps: int) -> None:\n",
    "        \"\"\"Step incrementally along line of sight to catch intermediate obstacles.\"\"\"\n",
    "        dx, dy = self.OFFSETS[self.direction]\n",
    "        step_dir = 1 if steps > 0 else -1\n",
    "        for _ in range(abs(steps)):\n",
    "            nx = self.x + dx * step_dir\n",
    "            ny = self.y + dy * step_dir\n",
    "            if (nx, ny) in self.obstacles:\n",
    "                break\n",
    "            self.x, self.y = nx, ny\n",
    "\n",
    "    def undo(self, steps: int = 1) -> Tuple[int, int, str]:\n",
    "        for _ in range(min(steps, len(self.history))):\n",
    "            self.x, self.y, self.direction = self.history.pop()\n",
    "        return (self.x, self.y, self.direction)\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "robot = GridRobot(x=0, y=0, direction='N', obstacles={(0, 2), (5, 5)})\n",
    "assert robot.execute(\"F F\") == (0, 1, 'N'), \"Robot should stop at (0, 1) before obstacle at (0, 2)\"\n",
    "assert robot.execute(\"R F F\") == (2, 1, 'E'), \"Turn East and advance 2 units\"\n",
    "assert robot.undo(1) == (0, 1, 'N'), \"Undo should restore state prior to 'R F F'\"\n",
    "assert robot.execute(\"L J3\") == (-3, 1, 'W'), \"Turn West and jump 3 units\"\n",
    "print(\"\u2713 Problem 1: GridRobot passed all tests.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 2. Blind-Grid Robot Room Cleaner\n",
    "\n",
    "### Problem Statement\n",
    "You control a cleaning robot placed inside an unknown room represented by an enclosed grid of cells.\n",
    "- You are **not** given room dimensions or starting coordinates.\n",
    "- You can interact with the room strictly via four sensory API calls on `RobotAPI`:\n",
    "  - `move() -> bool`: Moves forward 1 cell. Returns `True` on success, or `False` if blocked by a wall.\n",
    "  - `turn_left() -> None`: Rotates 90\u00b0 counter-clockwise.\n",
    "  - `turn_right() -> None`: Rotates 90\u00b0 clockwise.\n",
    "  - `clean() -> None`: Cleans the current cell.\n",
    "- **Goal:** Design an algorithm that cleans every reachable space in the room and returns the total number of unique cells cleaned.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class RobotAPI:\n",
    "    def move(self) -> bool: pass\n",
    "    def turn_left(self) -> None: pass\n",
    "    def turn_right(self) -> None: pass\n",
    "    def clean(self) -> None: pass\n",
    "\n",
    "class RobotRoomCleaner:\n",
    "    def clean_room(self, robot: RobotAPI) -> int:\n",
    "        \"\"\"Cleans all reachable spaces and returns count of unique cleaned cells.\"\"\"\n",
    "        pass\n",
    "```\n",
    "\n",
    "### Key Traps & AI Failure Modes\n",
    "- **Backtracking State Restoration:** When backtracking in recursion, returning to the previous cell is not enough\u2014you must also **restore the robot's physical heading** so higher-level call frames preserve their directional invariant!\n",
    "- **Virtual Coordinates:** Start at virtual `(0, 0)` with heading `0` (Up). Maintain visited cells in a hash set relative to `(0, 0)`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from typing import Set, Tuple, List\n",
    "\n",
    "class MockRobot:\n",
    "    \"\"\"Mock environment representing an enclosed room with walls.\"\"\"\n",
    "    DIRS = [(-1, 0), (0, 1), (1, 0), (0, -1)]  # Up, Right, Down, Left\n",
    "\n",
    "    def __init__(self, room: List[List[int]], start_r: int, start_c: int):\n",
    "        self.room = room\n",
    "        self.r = start_r\n",
    "        self.c = start_c\n",
    "        self.d = 0  # 0: Up, 1: Right, 2: Down, 3: Left\n",
    "        self.cleaned: Set[Tuple[int, int]] = set()\n",
    "\n",
    "    def move(self) -> bool:\n",
    "        dr, dc = self.DIRS[self.d]\n",
    "        nr, nc = self.r + dr, self.c + dc\n",
    "        if 0 <= nr < len(self.room) and 0 <= nc < len(self.room[0]) and self.room[nr][nc] == 1:\n",
    "            self.r, self.c = nr, nc\n",
    "            return True\n",
    "        return False\n",
    "\n",
    "    def turn_left(self) -> None:\n",
    "        self.d = (self.d - 1) % 4\n",
    "\n",
    "    def turn_right(self) -> None:\n",
    "        self.d = (self.d + 1) % 4\n",
    "\n",
    "    def clean(self) -> None:\n",
    "        self.cleaned.add((self.r, self.c))\n",
    "\n",
    "class RobotRoomCleaner:\n",
    "    DIRS = [(-1, 0), (0, 1), (1, 0), (0, -1)]  # Relative: Up, Right, Down, Left\n",
    "\n",
    "    def clean_room(self, robot: MockRobot) -> int:\n",
    "        visited: Set[Tuple[int, int]] = set()\n",
    "\n",
    "        def go_back():\n",
    "            \"\"\"Physically move 1 step backward and restore original orientation.\"\"\"\n",
    "            robot.turn_right()\n",
    "            robot.turn_right()\n",
    "            robot.move()\n",
    "            robot.turn_right()\n",
    "            robot.turn_right()\n",
    "\n",
    "        def backtrack(r: int, c: int, d: int):\n",
    "            visited.add((r, c))\n",
    "            robot.clean()\n",
    "\n",
    "            # Explore 4 directions clockwise starting from current heading\n",
    "            for i in range(4):\n",
    "                new_d = (d + i) % 4\n",
    "                dr, dc = self.DIRS[new_d]\n",
    "                nr, nc = r + dr, c + dc\n",
    "\n",
    "                if (nr, nc) not in visited:\n",
    "                    if robot.move():\n",
    "                        backtrack(nr, nc, new_d)\n",
    "                        go_back()\n",
    "                robot.turn_right()\n",
    "\n",
    "        backtrack(0, 0, 0)\n",
    "        return len(visited)\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "room_map = [\n",
    "    [1, 1, 1, 0],\n",
    "    [1, 0, 1, 1],\n",
    "    [1, 1, 1, 1]\n",
    "]\n",
    "mock_robot = MockRobot(room_map, start_r=0, start_c=0)\n",
    "cleaner = RobotRoomCleaner()\n",
    "\n",
    "cleaned_count = cleaner.clean_room(mock_robot)\n",
    "expected_reachable = sum(row.count(1) for row in room_map)\n",
    "\n",
    "assert cleaned_count == expected_reachable, f\"Expected {expected_reachable}, got {cleaned_count}\"\n",
    "assert len(mock_robot.cleaned) == expected_reachable\n",
    "print(f\"\u2713 Problem 2: RobotRoomCleaner cleaned all {cleaned_count} reachable cells.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 3. Autonomous Cleaning Robot with Cycle Detection\n",
    "\n",
    "### Problem Statement\n",
    "An autonomous cleaning robot is placed in an $R \\times C$ room where `0` represents cleanable empty space and `1` represents an obstacle.\n",
    "- The robot starts at cell `(0, 0)` facing **East** (Right).\n",
    "- **Movement Rules:**\n",
    "  1. It marks the current space as cleaned.\n",
    "  2. It attempts to move straight forward in its current direction.\n",
    "  3. If it hits an obstacle or room boundary, it **turns 90\u00b0 clockwise** without moving and tries again.\n",
    "- **Termination:** Because movement is deterministic, the robot terminates when it enters an infinite loop (it reaches an `(r, c, direction)` state it has already visited).\n",
    "- **Goal:** Return the total number of **unique** spaces cleaned.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class CleaningSimulator:\n",
    "    def number_of_clean_spaces(self, room: list[list[int]]) -> int:\n",
    "        \"\"\"Returns total unique spaces cleaned before looping.\"\"\"\n",
    "        pass\n",
    "```\n",
    "\n",
    "### Key Traps & AI Failure Modes\n",
    "- **Cleaned Cells vs State History:** AI assistants often confuse unique cleaned cells with visited loop states. You need two separate sets: `cleaned` tracks `(r, c)`, while `visited_states` tracks `(r, c, direction)`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from typing import List, Set, Tuple\n",
    "\n",
    "class CleaningSimulator:\n",
    "    \"\"\"Simulates deterministic cleaning robot with cycle detection on an obstacle matrix.\"\"\"\n",
    "    DIRS = [(0, 1), (1, 0), (0, -1), (-1, 0)]  # East (0), South (1), West (2), North (3)\n",
    "\n",
    "    def number_of_clean_spaces(self, room: List[List[int]]) -> int:\n",
    "        R, C = len(room), len(room[0])\n",
    "        r, c, d = 0, 0, 0\n",
    "        \n",
    "        cleaned: Set[Tuple[int, int]] = set()\n",
    "        visited_states: Set[Tuple[int, int, int]] = set()\n",
    "\n",
    "        while True:\n",
    "            cleaned.add((r, c))\n",
    "            state = (r, c, d)\n",
    "            \n",
    "            if state in visited_states:\n",
    "                break  # Cycle detected: deterministic loop reached\n",
    "            visited_states.add(state)\n",
    "\n",
    "            dr, dc = self.DIRS[d]\n",
    "            nr, nc = r + dr, c + dc\n",
    "            \n",
    "            if 0 <= nr < R and 0 <= nc < C and room[nr][nc] == 0:\n",
    "                r, c = nr, nc\n",
    "            else:\n",
    "                d = (d + 1) % 4  # Turn 90\u00b0 clockwise\n",
    "\n",
    "        return len(cleaned)\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "sim = CleaningSimulator()\n",
    "test_room = [\n",
    "    [0, 0, 0],\n",
    "    [1, 1, 0],\n",
    "    [0, 0, 0]\n",
    "]\n",
    "cleaned_cells = sim.number_of_clean_spaces(test_room)\n",
    "assert cleaned_cells == 7, f\"Expected 7 cleaned cells, got {cleaned_cells}\"\n",
    "print(f\"\u2713 Problem 3: CleaningSimulator successfully detected cycle with {cleaned_cells} unique spaces cleaned.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 4. Snake Game Simulator with Growth & Collision\n",
    "\n",
    "### Problem Statement\n",
    "Design a stateful Snake Game running on a grid of size `width` $\\times$ `height`.\n",
    "- The snake starts at coordinate `(0, 0)` with initial length 1.\n",
    "- Food coordinates appear sequentially from a predefined list `food = [[r1, c1], [r2, c2], ...]`.\n",
    "- `move(direction: str) -> int`:\n",
    "  - Direction is `'U'` (Up), `'D'` (Down), `'L'` (Left), or `'R'` (Right).\n",
    "  - Returns the **current score** (equal to number of food items consumed so far), or `-1` if the snake collides with a boundary or its own body.\n",
    "  - If the new head coordinate matches the current food item:\n",
    "    - The snake eats the food, grows by 1 (its tail does **not** move), and advances to the next food item.\n",
    "  - If no food is eaten:\n",
    "    - The tail moves forward (vacates its previous cell) as the head advances.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class SnakeGame:\n",
    "    def __init__(self, width: int, height: int, food: list[list[int]]):\n",
    "        pass\n",
    "\n",
    "    def move(self, direction: str) -> int:\n",
    "        \"\"\"Advances snake by 1 step. Returns current score or -1 on collision.\"\"\"\n",
    "        pass\n",
    "```\n",
    "\n",
    "### Key Traps & AI Failure Modes\n",
    "- **The Tail Vacation Rule:** If the snake's head moves into the cell currently occupied by its tail on a normal non-eating step, **this is NOT a collision** because the tail moves forward simultaneously!\n",
    "- **Data Structures:** Using a pure list for the body causes $\\mathcal{O}(L)$ membership checks. Combining `collections.deque` with a `set` yields $\\mathcal{O}(1)$ moves.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from collections import deque\n",
    "from typing import List, Tuple, Set\n",
    "\n",
    "class SnakeGame:\n",
    "    \"\"\"Snake game engine with O(1) step execution via double-ended queue + hash set.\"\"\"\n",
    "    DIR_OFFSETS = {'U': (-1, 0), 'D': (1, 0), 'L': (0, -1), 'R': (0, 1)}\n",
    "\n",
    "    def __init__(self, width: int, height: int, food: List[List[int]]):\n",
    "        self.width = width\n",
    "        self.height = height\n",
    "        self.food = food\n",
    "        self.food_idx = 0\n",
    "        self.score = 0\n",
    "        \n",
    "        self.body: deque[Tuple[int, int]] = deque([(0, 0)])\n",
    "        self.body_set: Set[Tuple[int, int]] = {(0, 0)}\n",
    "\n",
    "    def move(self, direction: str) -> int:\n",
    "        dr, dc = self.DIR_OFFSETS[direction]\n",
    "        head_r, head_c = self.body[0]\n",
    "        nr, nc = head_r + dr, head_c + dc\n",
    "\n",
    "        # 1. Boundary check\n",
    "        if not (0 <= nr < self.height and 0 <= nc < self.width):\n",
    "            return -1\n",
    "\n",
    "        # 2. Check food consumption\n",
    "        is_food = (self.food_idx < len(self.food) and [nr, nc] == self.food[self.food_idx])\n",
    "\n",
    "        # 3. Vacate tail before collision check if no food eaten\n",
    "        tail = self.body[-1]\n",
    "        if not is_food:\n",
    "            self.body.pop()\n",
    "            self.body_set.remove(tail)\n",
    "\n",
    "        # 4. Check body collision\n",
    "        if (nr, nc) in self.body_set:\n",
    "            return -1\n",
    "\n",
    "        # 5. Place new head\n",
    "        self.body.appendleft((nr, nc))\n",
    "        self.body_set.add((nr, nc))\n",
    "\n",
    "        # 6. Update food and score\n",
    "        if is_food:\n",
    "            self.food_idx += 1\n",
    "            self.score += 1\n",
    "\n",
    "        return self.score\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "game = SnakeGame(width=3, height=2, food=[[1, 2], [0, 1]])\n",
    "\n",
    "assert game.move('R') == 0, \"Move to (0, 1) - no food\"\n",
    "assert game.move('D') == 0, \"Move to (1, 1) - no food\"\n",
    "assert game.move('R') == 1, \"Move to (1, 2) - consumed food 1, score = 1\"\n",
    "assert game.move('U') == 1, \"Move to (0, 2) - no food\"\n",
    "assert game.move('L') == 2, \"Move to (0, 1) - consumed food 2, score = 2\"\n",
    "assert game.move('U') == -1, \"Out of bounds collision -> Game Over (-1)\"\n",
    "\n",
    "print(\"\u2713 Problem 4: SnakeGame simulation verified.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 5. Connect Four / Tic-Tac-Toe Game Engine with $\\mathcal{O}(1)$ Win Checking\n",
    "\n",
    "### Problem Statement\n",
    "Design an $N \\times N$ board game engine for 2 players.\n",
    "- `make_move(row: int, col: int, player: int) -> int`:\n",
    "  - Records a move by `player` (1 or 2) at coordinate `(row, col)`.\n",
    "  - Determines if the current move results in a win **in strictly $\\mathcal{O}(1)$ time** without scanning the $N \\times N$ board.\n",
    "  - Returns:\n",
    "    - `0` if no player has won yet.\n",
    "    - `1` if Player 1 wins.\n",
    "    - `2` if Player 2 wins.\n",
    "- A player wins when they successfully place $N$ tokens in any horizontal row, vertical column, main diagonal, or anti-diagonal.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class ConnectEngine:\n",
    "    def __init__(self, n: int = 3):\n",
    "        pass\n",
    "\n",
    "    def make_move(self, row: int, col: int, player: int) -> int:\n",
    "        \"\"\"Places token; returns winner (1 or 2) or 0 if game continues.\"\"\"\n",
    "        pass\n",
    "```\n",
    "\n",
    "### Key Traps & AI Failure Modes\n",
    "- **Full Board Scan Trap:** Scanning the board after each move takes $\\mathcal{O}(N)$ or $\\mathcal{O}(N^2)$ time. In production and interviews, interviewers expect **$\\mathcal{O}(1)$ amortized tracking** using count vectors.\n",
    "- **Anti-Diagonal Math:** The anti-diagonal condition is strictly `row + col == n - 1`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from collections import defaultdict\n",
    "\n",
    "class ConnectEngine:\n",
    "    \"\"\"O(1) win-checking game engine using linear accumulator arrays per player.\"\"\"\n",
    "    def __init__(self, n: int = 3):\n",
    "        self.n = n\n",
    "        self.rows = defaultdict(lambda: defaultdict(int))\n",
    "        self.cols = defaultdict(lambda: defaultdict(int))\n",
    "        self.diag = defaultdict(int)\n",
    "        self.anti_diag = defaultdict(int)\n",
    "        self.winner = 0\n",
    "\n",
    "    def make_move(self, row: int, col: int, player: int) -> int:\n",
    "        if self.winner != 0:\n",
    "            return self.winner\n",
    "\n",
    "        self.rows[player][row] += 1\n",
    "        self.cols[player][col] += 1\n",
    "        \n",
    "        if row == col:\n",
    "            self.diag[player] += 1\n",
    "        if row + col == self.n - 1:\n",
    "            self.anti_diag[player] += 1\n",
    "\n",
    "        if (self.rows[player][row] == self.n or\n",
    "            self.cols[player][col] == self.n or\n",
    "            self.diag[player] == self.n or\n",
    "            self.anti_diag[player] == self.n):\n",
    "            self.winner = player\n",
    "            return player\n",
    "\n",
    "        return 0\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "game = ConnectEngine(n=3)\n",
    "\n",
    "assert game.make_move(0, 0, player=1) == 0\n",
    "assert game.make_move(0, 2, player=2) == 0\n",
    "assert game.make_move(1, 1, player=1) == 0\n",
    "assert game.make_move(1, 0, player=2) == 0\n",
    "assert game.make_move(2, 2, player=1) == 1, \"Player 1 should win on main diagonal in O(1)\"\n",
    "\n",
    "print(\"\u2713 Problem 5: ConnectEngine O(1) win verification passed.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 6. Battleship Fleet Coordinator\n",
    "\n",
    "### Problem Statement\n",
    "Build the coordinator for a two-player naval combat Battleship game on a grid of size `size \\times size`.\n",
    "- `place_ship(player: int, ship_name: str, coords: list[tuple[int, int]]) -> bool`:\n",
    "  - Places a named ship for `player` (1 or 2).\n",
    "  - Rejects if any coordinate is out of bounds or overlaps with an existing ship belonging to that player.\n",
    "- `fire(player: int, target: tuple[int, int]) -> str`:\n",
    "  - `player` fires at the opponent's coordinates.\n",
    "  - Returns:\n",
    "    - `\"Already Targeted\"` if this player previously fired at this cell.\n",
    "    - `\"Miss\"` if the target contains no ship.\n",
    "    - `\"Hit\"` if a ship coordinate was struck.\n",
    "    - `\"Sunk <ship_name>\"` if all cells of the hit ship are now struck.\n",
    "    - `\"Victory\"` if the last remaining opponent ship has been sunk.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class BattleshipGame:\n",
    "    def __init__(self, size: int = 10):\n",
    "        pass\n",
    "\n",
    "    def place_ship(self, player: int, ship_name: str, coords: list[tuple[int, int]]) -> bool:\n",
    "        pass\n",
    "\n",
    "    def fire(self, player: int, target: tuple[int, int]) -> str:\n",
    "        \"\"\"Fires shot; returns 'Miss', 'Hit', 'Sunk <name>', or 'Victory'.\"\"\"\n",
    "        pass\n",
    "```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from typing import Dict, Set, Tuple, List\n",
    "\n",
    "class BattleshipGame:\n",
    "    \"\"\"Two-player Battleship coordinator managing fleet geometry and damage tracking.\"\"\"\n",
    "    def __init__(self, size: int = 10):\n",
    "        self.size = size\n",
    "        self.fleet: Dict[int, Dict[str, Set[Tuple[int, int]]]] = {1: {}, 2: {}}\n",
    "        self.fired_shots: Dict[int, Set[Tuple[int, int]]] = {1: set(), 2: set()}\n",
    "\n",
    "    def place_ship(self, player: int, ship_name: str, coords: List[Tuple[int, int]]) -> bool:\n",
    "        c_set = set(coords)\n",
    "        for r, c in c_set:\n",
    "            if not (0 <= r < self.size and 0 <= c < self.size):\n",
    "                return False\n",
    "            for existing_ship in self.fleet[player].values():\n",
    "                if (r, c) in existing_ship:\n",
    "                    return False\n",
    "        self.fleet[player][ship_name] = c_set\n",
    "        return True\n",
    "\n",
    "    def fire(self, player: int, target: Tuple[int, int]) -> str:\n",
    "        opponent = 2 if player == 1 else 1\n",
    "        if target in self.fired_shots[player]:\n",
    "            return \"Already Targeted\"\n",
    "        self.fired_shots[player].add(target)\n",
    "\n",
    "        for name, remaining_coords in self.fleet[opponent].items():\n",
    "            if target in remaining_coords:\n",
    "                remaining_coords.remove(target)\n",
    "                if not remaining_coords:\n",
    "                    del self.fleet[opponent][name]\n",
    "                    if not self.fleet[opponent]:\n",
    "                        return \"Victory\"\n",
    "                    return f\"Sunk {name}\"\n",
    "                return \"Hit\"\n",
    "                \n",
    "        return \"Miss\"\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "game = BattleshipGame(size=5)\n",
    "\n",
    "assert game.place_ship(player=1, ship_name=\"Destroyer\", coords=[(0, 0), (0, 1)]) is True\n",
    "assert game.place_ship(player=2, ship_name=\"Patrol\", coords=[(2, 2)]) is True\n",
    "\n",
    "assert game.fire(player=1, target=(4, 4)) == \"Miss\"\n",
    "assert game.fire(player=1, target=(4, 4)) == \"Already Targeted\"\n",
    "assert game.fire(player=1, target=(2, 2)) == \"Victory\", \"Sinking opponent's last ship grants Victory\"\n",
    "\n",
    "print(\"\u2713 Problem 6: BattleshipGame fleet coordination passed.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 7. In-Memory Key-Field Database with TTL & Backup/Restore\n",
    "\n",
    "### Problem Statement\n",
    "Build an in-memory database storing `(key, field) -> value` entries with 4 evolving levels:\n",
    "- **Level 1 (Basic CRUD):**\n",
    "  - `set(key, field, value)`: Inserts or updates a field value.\n",
    "  - `get(key, field) -> str | None`: Returns value or `None` if absent.\n",
    "  - `delete(key, field) -> bool`: Deletes field; returns `True` if deleted, `False` if absent.\n",
    "- **Level 2 (Scanning & Formatted Output):**\n",
    "  - `scan(key) -> list[str]`: Returns lexicographically sorted `[\"field(value)\"]` strings.\n",
    "  - `scan_by_prefix(key, prefix) -> list[str]`: Returns matching fields filtered by prefix.\n",
    "- **Level 3 (Timestamps & Time-to-Live):**\n",
    "  - `set_at_with_ttl(key, field, value, timestamp, ttl)`: Expiration occurs at `timestamp + ttl` (half-open: valid on `[set_time, set_time + ttl)`).\n",
    "  - `get_at(key, field, timestamp) -> str | None`: Visible only if not expired at query time.\n",
    "- **Level 4 (Backup & Restore with Recomputed Expiration):**\n",
    "  - `backup(timestamp) -> int`: Snapshots all non-expired entries at `timestamp`. For each entry, records its remaining TTL: $rem\\_ttl = (set\\_time + ttl) - timestamp$. Returns backup ID.\n",
    "  - `restore(timestamp, backup_id) -> bool`: Restores DB from backup. New expiration becomes $timestamp + rem\\_ttl$. Already-expired entries are omitted.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class InMemoryDB:\n",
    "    def set(self, key: str, field: str, value: str) -> None: pass\n",
    "    def get(self, key: str, field: str) -> str | None: pass\n",
    "    def delete(self, key: str, field: str) -> bool: pass\n",
    "    def scan(self, key: str) -> list[str]: pass\n",
    "    def scan_by_prefix(self, key: str, prefix: str) -> list[str]: pass\n",
    "    def set_at_with_ttl(self, key: str, field: str, value: str, timestamp: int, ttl: int) -> None: pass\n",
    "    def get_at(self, key: str, field: str, timestamp: int) -> str | None: pass\n",
    "    def backup(self, timestamp: int) -> int: pass\n",
    "    def restore(self, timestamp: int, backup_id: int) -> bool: pass\n",
    "```\n",
    "\n",
    "### Key Traps & AI Failure Modes\n",
    "- **Backup TTL Recomputation:** When restoring at time $T_{restore}$, resetting TTL to original $ttl$ is wrong. The remaining TTL must be recomputed: $T_{new\\_expiry} = T_{restore} + rem\\_ttl$.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from collections import defaultdict\n",
    "from typing import Dict, List, Optional, Tuple\n",
    "\n",
    "class InMemoryDB:\n",
    "    \"\"\"4-Level In-Memory Database with CRUD, Prefix Scan, TTL, and Recomputed Backup/Restore.\"\"\"\n",
    "    def __init__(self):\n",
    "        self.db: Dict[str, Dict[str, Tuple[str, int, Optional[int]]]] = defaultdict(dict)\n",
    "        self.backups: Dict[int, Dict[str, Dict[str, Tuple[str, Optional[int]]]]] = {}\n",
    "        self.backup_counter = 0\n",
    "\n",
    "    # Level 1: Basic CRUD\n",
    "    def set(self, key: str, field: str, value: str) -> None:\n",
    "        self.set_at_with_ttl(key, field, value, timestamp=0, ttl=None)\n",
    "\n",
    "    def get(self, key: str, field: str) -> Optional[str]:\n",
    "        return self.get_at(key, field, timestamp=0)\n",
    "\n",
    "    def delete(self, key: str, field: str) -> bool:\n",
    "        if key in self.db and field in self.db[key]:\n",
    "            del self.db[key][field]\n",
    "            return True\n",
    "        return False\n",
    "\n",
    "    # Level 2: Scanning & Output Formatting\n",
    "    def scan(self, key: str) -> List[str]:\n",
    "        return self.scan_at(key, timestamp=0)\n",
    "\n",
    "    def scan_by_prefix(self, key: str, prefix: str) -> List[str]:\n",
    "        return [entry for entry in self.scan(key) if entry.startswith(prefix)]\n",
    "\n",
    "    # Level 3: Timestamps & Time-To-Live\n",
    "    def set_at_with_ttl(self, key: str, field: str, value: str, timestamp: int, ttl: Optional[int]) -> None:\n",
    "        self.db[key][field] = (value, timestamp, ttl)\n",
    "\n",
    "    def get_at(self, key: str, field: str, timestamp: int) -> Optional[str]:\n",
    "        if key in self.db and field in self.db[key]:\n",
    "            val, set_t, ttl = self.db[key][field]\n",
    "            if ttl is None or timestamp < set_t + ttl:\n",
    "                return val\n",
    "        return None\n",
    "\n",
    "    def scan_at(self, key: str, timestamp: int) -> List[str]:\n",
    "        if key not in self.db:\n",
    "            return []\n",
    "        res = []\n",
    "        for f, (v, set_t, ttl) in self.db[key].items():\n",
    "            if ttl is None or timestamp < set_t + ttl:\n",
    "                res.append(f\"{f}({v})\")\n",
    "        res.sort()\n",
    "        return res\n",
    "\n",
    "    # Level 4: Backup & Restore with Remaining TTL Recomputation\n",
    "    def backup(self, timestamp: int) -> int:\n",
    "        self.backup_counter += 1\n",
    "        snapshot: Dict[str, Dict[str, Tuple[str, Optional[int]]]] = {}\n",
    "        for k, fields in self.db.items():\n",
    "            valid_fields = {}\n",
    "            for f, (v, set_t, ttl) in fields.items():\n",
    "                if ttl is None:\n",
    "                    valid_fields[f] = (v, None)\n",
    "                elif timestamp < set_t + ttl:\n",
    "                    remaining_ttl = (set_t + ttl) - timestamp\n",
    "                    valid_fields[f] = (v, remaining_ttl)\n",
    "            if valid_fields:\n",
    "                snapshot[k] = valid_fields\n",
    "        self.backups[self.backup_counter] = snapshot\n",
    "        return self.backup_counter\n",
    "\n",
    "    def restore(self, timestamp: int, backup_id: int) -> bool:\n",
    "        if backup_id not in self.backups:\n",
    "            return False\n",
    "        snapshot = self.backups[backup_id]\n",
    "        self.db = defaultdict(dict)\n",
    "        for k, fields in snapshot.items():\n",
    "            for f, (v, rem_ttl) in fields.items():\n",
    "                self.db[k][f] = (v, timestamp, rem_ttl)\n",
    "        return True\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "db = InMemoryDB()\n",
    "\n",
    "# Level 1 & 2: CRUD and Scan\n",
    "db.set(\"user1\", \"name\", \"Alice\")\n",
    "db.set(\"user1\", \"age\", \"30\")\n",
    "assert db.scan(\"user1\") == [\"age(30)\", \"name(Alice)\"]\n",
    "\n",
    "# Level 3: TTL expiration at set_time + ttl\n",
    "db.set_at_with_ttl(\"user1\", \"session\", \"tok123\", timestamp=10, ttl=5)\n",
    "assert db.get_at(\"user1\", \"session\", timestamp=12) == \"tok123\"\n",
    "assert db.get_at(\"user1\", \"session\", timestamp=15) is None, \"Expired at 10+5=15\"\n",
    "\n",
    "# Level 4: Backup at t=12 (remaining TTL of session = 3)\n",
    "bk = db.backup(timestamp=12)\n",
    "db.delete(\"user1\", \"name\")\n",
    "assert db.get(\"user1\", \"name\") is None\n",
    "\n",
    "# Restore at t=20 -> session should expire at 20 + 3 = 23\n",
    "assert db.restore(timestamp=20, backup_id=bk) is True\n",
    "assert db.get_at(\"user1\", \"name\", timestamp=20) == \"Alice\"\n",
    "assert db.get_at(\"user1\", \"session\", timestamp=22) == \"tok123\", \"Valid before 20+3\"\n",
    "assert db.get_at(\"user1\", \"session\", timestamp=24) is None, \"Expired after 20+3\"\n",
    "\n",
    "print(\"\u2713 Problem 7: InMemoryDB 4-Level assessment passed.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 8. Cloud Storage Manager with User Quotas & Auto-Eviction\n",
    "\n",
    "### Problem Statement\n",
    "Design a multi-tenant cloud storage manager:\n",
    "- **Level 1 (Files):** `add_file(name, size)`, `get_file_size(name)`, `copy_file(src, dest)`.\n",
    "- **Level 2 (Search):** `find_by_prefix(prefix)`, `find_by_suffix(suffix)` returning lexicographically sorted filenames.\n",
    "- **Level 3 (User Quotas & Eviction):** `add_user(user_id, capacity)`, `add_file_by_user(user_id, name, size)`.\n",
    "  - If adding the file exceeds user capacity, the system **auto-evicts the user's largest files first** until enough space is available.\n",
    "  - **Tie-breaker:** If multiple files have the same size, evict the **lexicographically larger name** first.\n",
    "  - If the new file size alone exceeds the user's total capacity, reject the operation (`return False`).\n",
    "- **Level 4 (Compression):** `compress_file(name)`:\n",
    "  - Halves the file size (`size // 2`) and updates the owning user's consumed quota.\n",
    "  - Returns `False` if file is already compressed.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class CloudStorage:\n",
    "    def add_file(self, name: str, size: int) -> bool: pass\n",
    "    def get_file_size(self, name: str) -> int | None: pass\n",
    "    def copy_file(self, src: str, dest: str) -> bool: pass\n",
    "    def find_by_prefix(self, prefix: str) -> list[str]: pass\n",
    "    def find_by_suffix(self, suffix: str) -> list[str]: pass\n",
    "    def add_user(self, user_id: str, capacity: int) -> bool: pass\n",
    "    def add_file_by_user(self, user_id: str, name: str, size: int) -> bool: pass\n",
    "    def compress_file(self, name: str) -> bool: pass\n",
    "```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from typing import Dict, List, Optional, Set, Tuple\n",
    "\n",
    "class CloudStorage:\n",
    "    \"\"\"Multi-tenant cloud storage with prefix/suffix search, auto-eviction, and compression.\"\"\"\n",
    "    def __init__(self):\n",
    "        self.files: Dict[str, Tuple[int, Optional[str], bool]] = {}\n",
    "        self.users: Dict[str, Dict[str, any]] = {}\n",
    "\n",
    "    def add_file(self, name: str, size: int) -> bool:\n",
    "        if name in self.files:\n",
    "            return False\n",
    "        self.files[name] = (size, None, False)\n",
    "        return True\n",
    "\n",
    "    def get_file_size(self, name: str) -> Optional[int]:\n",
    "        return self.files[name][0] if name in self.files else None\n",
    "\n",
    "    def copy_file(self, src: str, dest: str) -> bool:\n",
    "        if src not in self.files or dest in self.files:\n",
    "            return False\n",
    "        size, user_id, comp = self.files[src]\n",
    "        if user_id:\n",
    "            return self.add_file_by_user(user_id, dest, size)\n",
    "        self.files[dest] = (size, None, comp)\n",
    "        return True\n",
    "\n",
    "    def find_by_prefix(self, prefix: str) -> List[str]:\n",
    "        return sorted([k for k in self.files if k.startswith(prefix)])\n",
    "\n",
    "    def find_by_suffix(self, suffix: str) -> List[str]:\n",
    "        return sorted([k for k in self.files if k.endswith(suffix)])\n",
    "\n",
    "    def add_user(self, user_id: str, capacity: int) -> bool:\n",
    "        if user_id in self.users:\n",
    "            return False\n",
    "        self.users[user_id] = {'capacity': capacity, 'used': 0, 'files': set()}\n",
    "        return True\n",
    "\n",
    "    def add_file_by_user(self, user_id: str, name: str, size: int) -> bool:\n",
    "        if user_id not in self.users or name in self.files:\n",
    "            return False\n",
    "        user = self.users[user_id]\n",
    "        if size > user['capacity']:\n",
    "            return False\n",
    "\n",
    "        # Evict largest files first, tie breaker: lexicographically larger name\n",
    "        while user['used'] + size > user['capacity']:\n",
    "            candidate = sorted(\n",
    "                user['files'],\n",
    "                key=lambda fname: (self.files[fname][0], fname),\n",
    "                reverse=True\n",
    "            )[0]\n",
    "            cand_size = self.files[candidate][0]\n",
    "            user['used'] -= cand_size\n",
    "            user['files'].remove(candidate)\n",
    "            del self.files[candidate]\n",
    "\n",
    "        self.files[name] = (size, user_id, False)\n",
    "        user['files'].add(name)\n",
    "        user['used'] += size\n",
    "        return True\n",
    "\n",
    "    def compress_file(self, name: str) -> bool:\n",
    "        if name not in self.files:\n",
    "            return False\n",
    "        size, user_id, comp = self.files[name]\n",
    "        if comp:\n",
    "            return False\n",
    "        new_size = size // 2\n",
    "        saved = size - new_size\n",
    "        self.files[name] = (new_size, user_id, True)\n",
    "        if user_id and user_id in self.users:\n",
    "            self.users[user_id]['used'] -= saved\n",
    "        return True\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "cs = CloudStorage()\n",
    "cs.add_user(\"alice\", capacity=100)\n",
    "\n",
    "assert cs.add_file_by_user(\"alice\", \"doc.pdf\", 40) is True\n",
    "assert cs.add_file_by_user(\"alice\", \"video.mp4\", 50) is True\n",
    "assert cs.users[\"alice\"][\"used\"] == 90\n",
    "\n",
    "# Adding 30 exceeds 100: should evict largest file (\"video.mp4\", 50)\n",
    "assert cs.add_file_by_user(\"alice\", \"photo.png\", 30) is True\n",
    "assert \"video.mp4\" not in cs.files, \"video.mp4 should have been auto-evicted\"\n",
    "assert cs.users[\"alice\"][\"used\"] == 70\n",
    "\n",
    "# Compression halves photo.png (30 -> 15)\n",
    "assert cs.compress_file(\"photo.png\") is True\n",
    "assert cs.get_file_size(\"photo.png\") == 15\n",
    "assert cs.users[\"alice\"][\"used\"] == 55\n",
    "\n",
    "print(\"\u2713 Problem 8: CloudStorage auto-eviction and compression verified.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 9. Text Editor with Undo/Redo & Find/Replace\n",
    "\n",
    "### Problem Statement\n",
    "Implement a stateful text editor:\n",
    "- **Level 1 (Typing & Cursor):**\n",
    "  - `insert(text: str)`: Inserts string at the current cursor position and advances cursor.\n",
    "  - `delete(k: int) -> int`: Deletes up to $k$ characters directly before the cursor.\n",
    "  - `get_text() -> str`: Returns current text buffer.\n",
    "- **Level 2 (Find & Replace):**\n",
    "  - `find(pattern: str) -> list[int]`: Returns starting indices of all non-overlapping matches.\n",
    "  - `replace_all(pattern: str, replacement: str) -> int`: Replaces all occurrences and returns the count of substitutions made.\n",
    "- **Level 3 (Undo/Redo):**\n",
    "  - `undo() -> bool` & `redo() -> bool`: Transactional command stack reverting/restoring buffer state. Any new modification clears the redo stack.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class TextEditor:\n",
    "    def insert(self, text: str) -> None: pass\n",
    "    def delete(self, k: int) -> int: pass\n",
    "    def get_text(self) -> str: pass\n",
    "    def find(self, pattern: str) -> list[int]: pass\n",
    "    def replace_all(self, pattern: str, replacement: str) -> int: pass\n",
    "    def undo(self) -> bool: pass\n",
    "    def redo(self) -> bool: pass\n",
    "```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from typing import List, Tuple\n",
    "\n",
    "class TextEditor:\n",
    "    \"\"\"Text editor with cursor manipulation, pattern replacement, and dual-stack undo/redo.\"\"\"\n",
    "    def __init__(self):\n",
    "        self.buffer: List[str] = []\n",
    "        self.cursor: int = 0\n",
    "        self.undo_stack: List[Tuple[List[str], int]] = []\n",
    "        self.redo_stack: List[Tuple[List[str], int]] = []\n",
    "\n",
    "    def _save_snapshot(self) -> None:\n",
    "        self.undo_stack.append((list(self.buffer), self.cursor))\n",
    "        self.redo_stack.clear()\n",
    "\n",
    "    def insert(self, text: str) -> None:\n",
    "        self._save_snapshot()\n",
    "        self.buffer[self.cursor:self.cursor] = list(text)\n",
    "        self.cursor += len(text)\n",
    "\n",
    "    def delete(self, k: int) -> int:\n",
    "        self._save_snapshot()\n",
    "        actual_k = min(k, self.cursor)\n",
    "        start = self.cursor - actual_k\n",
    "        del self.buffer[start:self.cursor]\n",
    "        self.cursor = start\n",
    "        return actual_k\n",
    "\n",
    "    def get_text(self) -> str:\n",
    "        return \"\".join(self.buffer)\n",
    "\n",
    "    def find(self, pattern: str) -> List[int]:\n",
    "        text = self.get_text()\n",
    "        indices = []\n",
    "        idx = text.find(pattern)\n",
    "        while idx != -1:\n",
    "            indices.append(idx)\n",
    "            idx = text.find(pattern, idx + len(pattern))\n",
    "        return indices\n",
    "\n",
    "    def replace_all(self, pattern: str, replacement: str) -> int:\n",
    "        self._save_snapshot()\n",
    "        text = self.get_text()\n",
    "        count = text.count(pattern)\n",
    "        if count > 0:\n",
    "            new_text = text.replace(pattern, replacement)\n",
    "            self.buffer = list(new_text)\n",
    "            self.cursor = min(self.cursor, len(self.buffer))\n",
    "        return count\n",
    "\n",
    "    def undo(self) -> bool:\n",
    "        if not self.undo_stack:\n",
    "            return False\n",
    "        self.redo_stack.append((list(self.buffer), self.cursor))\n",
    "        self.buffer, self.cursor = self.undo_stack.pop()\n",
    "        return True\n",
    "\n",
    "    def redo(self) -> bool:\n",
    "        if not self.redo_stack:\n",
    "            return False\n",
    "        self.undo_stack.append((list(self.buffer), self.cursor))\n",
    "        self.buffer, self.cursor = self.redo_stack.pop()\n",
    "        return True\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "editor = TextEditor()\n",
    "editor.insert(\"hello world\")\n",
    "assert editor.get_text() == \"hello world\"\n",
    "assert editor.find(\"world\") == [6]\n",
    "\n",
    "# Replace and Undo\n",
    "assert editor.replace_all(\"world\", \"python\") == 1\n",
    "assert editor.get_text() == \"hello python\"\n",
    "\n",
    "assert editor.undo() is True\n",
    "assert editor.get_text() == \"hello world\", \"Undo should restore previous text\"\n",
    "\n",
    "assert editor.redo() is True\n",
    "assert editor.get_text() == \"hello python\", \"Redo should restore replaced text\"\n",
    "\n",
    "print(\"\u2713 Problem 9: TextEditor undo/redo verification passed.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 10. Multi-Account Financial Ledger with Atomic Rollbacks\n",
    "\n",
    "### Problem Statement\n",
    "Implement a double-entry financial ledger:\n",
    "- `create_account(account_id: str, initial_balance: int) -> bool`\n",
    "- `transfer(src: str, dest: str, amount: int, timestamp: int) -> str | None`:\n",
    "  - Debits `src` and credits `dest` atomically.\n",
    "  - Returns a unique transaction ID on success, or `None` if invalid/insufficient funds.\n",
    "- `rollback(tx_id: str, timestamp: int) -> bool`:\n",
    "  - Reverses the transaction atomically if `dest` currently has sufficient balance to refund the transfer.\n",
    "- `get_balance_at(account_id: str, timestamp: int) -> int | None`:\n",
    "  - Point-in-time balance query using historical logging and binary search ($\\mathcal{O}(\\log T)$).\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class TransactionLedger:\n",
    "    def create_account(self, account_id: str, initial_balance: int) -> bool: pass\n",
    "    def transfer(self, src: str, dest: str, amount: int, timestamp: int) -> str | None: pass\n",
    "    def rollback(self, transaction_id: str, timestamp: int) -> bool: pass\n",
    "    def get_balance_at(self, account_id: str, timestamp: int) -> int | None: pass\n",
    "```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "import bisect\n",
    "from collections import defaultdict\n",
    "from typing import Dict, List, Optional, Tuple\n",
    "\n",
    "class TransactionLedger:\n",
    "    \"\"\"Double-entry transaction ledger with point-in-time binary search balance queries and atomic rollbacks.\"\"\"\n",
    "    def __init__(self):\n",
    "        self.accounts: Dict[str, int] = {}\n",
    "        self.balance_history: Dict[str, List[Tuple[int, int]]] = defaultdict(list)\n",
    "        self.transactions: Dict[str, Dict[str, any]] = {}\n",
    "        self.tx_counter = 0\n",
    "\n",
    "    def create_account(self, account_id: str, initial_balance: int) -> bool:\n",
    "        if account_id in self.accounts or initial_balance < 0:\n",
    "            return False\n",
    "        self.accounts[account_id] = initial_balance\n",
    "        self.balance_history[account_id].append((0, initial_balance))\n",
    "        return True\n",
    "\n",
    "    def transfer(self, src: str, dest: str, amount: int, timestamp: int) -> Optional[str]:\n",
    "        if src not in self.accounts or dest not in self.accounts:\n",
    "            return None\n",
    "        if amount <= 0 or self.accounts[src] < amount:\n",
    "            return None\n",
    "\n",
    "        self.accounts[src] -= amount\n",
    "        self.accounts[dest] += amount\n",
    "        self.balance_history[src].append((timestamp, self.accounts[src]))\n",
    "        self.balance_history[dest].append((timestamp, self.accounts[dest]))\n",
    "\n",
    "        self.tx_counter += 1\n",
    "        tx_id = f\"tx_{self.tx_counter}\"\n",
    "        self.transactions[tx_id] = {\n",
    "            'src': src, 'dest': dest, 'amount': amount, 'status': 'COMMITTED'\n",
    "        }\n",
    "        return tx_id\n",
    "\n",
    "    def rollback(self, tx_id: str, timestamp: int) -> bool:\n",
    "        if tx_id not in self.transactions:\n",
    "            return False\n",
    "        tx = self.transactions[tx_id]\n",
    "        if tx['status'] != 'COMMITTED':\n",
    "            return False\n",
    "\n",
    "        src, dest, amt = tx['src'], tx['dest'], tx['amount']\n",
    "        if self.accounts[dest] < amt:\n",
    "            return False\n",
    "\n",
    "        self.accounts[dest] -= amt\n",
    "        self.accounts[src] += amt\n",
    "        self.balance_history[dest].append((timestamp, self.accounts[dest]))\n",
    "        self.balance_history[src].append((timestamp, self.accounts[src]))\n",
    "        tx['status'] = 'ROLLED_BACK'\n",
    "        return True\n",
    "\n",
    "    def get_balance_at(self, account_id: str, timestamp: int) -> Optional[int]:\n",
    "        if account_id not in self.accounts:\n",
    "            return None\n",
    "        history = self.balance_history[account_id]\n",
    "        idx = bisect.bisect_right(history, (timestamp, float('inf'))) - 1\n",
    "        return history[idx][1] if idx >= 0 else 0\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "ledger = TransactionLedger()\n",
    "ledger.create_account(\"acc1\", 100)\n",
    "ledger.create_account(\"acc2\", 50)\n",
    "\n",
    "tx = ledger.transfer(\"acc1\", \"acc2\", amount=30, timestamp=10)\n",
    "assert tx is not None\n",
    "\n",
    "assert ledger.get_balance_at(\"acc1\", timestamp=5) == 100\n",
    "assert ledger.get_balance_at(\"acc1\", timestamp=10) == 70\n",
    "assert ledger.get_balance_at(\"acc2\", timestamp=10) == 80\n",
    "\n",
    "assert ledger.rollback(tx, timestamp=15) is True\n",
    "assert ledger.get_balance_at(\"acc1\", timestamp=15) == 100\n",
    "assert ledger.get_balance_at(\"acc2\", timestamp=15) == 50\n",
    "\n",
    "print(\"\u2713 Problem 10: TransactionLedger atomic rollback and history verified.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 11. Sliding Window Log & Token Bucket Rate Limiter\n",
    "\n",
    "### Problem Statement\n",
    "Implement two foundational rate-limiting algorithms:\n",
    "1. **Sliding Window Log:**\n",
    "   - Tracks exact request timestamps per client.\n",
    "   - For an incoming request at `timestamp`, prunes timestamps older than `timestamp - window_seconds`.\n",
    "   - Allows request if the active count $< max\\_requests$.\n",
    "2. **Token Bucket:**\n",
    "   - Maintains a bucket with `capacity` and `refill_rate` (tokens/sec).\n",
    "   - Dynamically computes refill: $tokens = \\min(capacity, tokens + elapsed \\times refill\\_rate)$.\n",
    "   - Allows request if $tokens \\ge cost$.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class SlidingWindowRateLimiter:\n",
    "    def __init__(self, max_requests: int, window_seconds: float): pass\n",
    "    def allow_request(self, client_id: str, timestamp: float) -> bool: pass\n",
    "\n",
    "class TokenBucketRateLimiter:\n",
    "    def __init__(self, capacity: float, refill_rate: float): pass\n",
    "    def allow_request(self, client_id: str, timestamp: float, cost: float = 1.0) -> bool: pass\n",
    "```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from collections import defaultdict, deque\n",
    "from typing import Dict, Tuple\n",
    "\n",
    "class SlidingWindowRateLimiter:\n",
    "    \"\"\"Sliding window log rate limiter with microsecond timestamp pruning.\"\"\"\n",
    "    def __init__(self, max_requests: int, window_seconds: float):\n",
    "        self.max_requests = max_requests\n",
    "        self.window = window_seconds\n",
    "        self.logs: Dict[str, deque[float]] = defaultdict(deque)\n",
    "\n",
    "    def allow_request(self, client_id: str, timestamp: float) -> bool:\n",
    "        q = self.logs[client_id]\n",
    "        cutoff = timestamp - self.window\n",
    "        while q and q[0] <= cutoff:\n",
    "            q.popleft()\n",
    "            \n",
    "        if len(q) < self.max_requests:\n",
    "            q.append(timestamp)\n",
    "            return True\n",
    "        return False\n",
    "\n",
    "class TokenBucketRateLimiter:\n",
    "    \"\"\"Token bucket rate limiter supporting bursts with continuous fractional refill.\"\"\"\n",
    "    def __init__(self, capacity: float, refill_rate: float):\n",
    "        self.capacity = capacity\n",
    "        self.refill_rate = refill_rate\n",
    "        self.buckets: Dict[str, Tuple[float, float]] = {}\n",
    "\n",
    "    def allow_request(self, client_id: str, timestamp: float, cost: float = 1.0) -> bool:\n",
    "        if client_id not in self.buckets:\n",
    "            self.buckets[client_id] = (self.capacity, timestamp)\n",
    "\n",
    "        tokens, last_time = self.buckets[client_id]\n",
    "        elapsed = max(0.0, timestamp - last_time)\n",
    "        tokens = min(self.capacity, tokens + elapsed * self.refill_rate)\n",
    "\n",
    "        if tokens >= cost:\n",
    "            self.buckets[client_id] = (tokens - cost, timestamp)\n",
    "            return True\n",
    "        else:\n",
    "            self.buckets[client_id] = (tokens, timestamp)\n",
    "            return False\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "sw = SlidingWindowRateLimiter(max_requests=2, window_seconds=5.0)\n",
    "assert sw.allow_request(\"c1\", 1.0) is True\n",
    "assert sw.allow_request(\"c1\", 2.0) is True\n",
    "assert sw.allow_request(\"c1\", 3.0) is False, \"Exceeded 2 requests in 5s window\"\n",
    "assert sw.allow_request(\"c1\", 6.5) is True, \"Timestamp 1.0 slid out of window [1.5, 6.5]\"\n",
    "\n",
    "tb = TokenBucketRateLimiter(capacity=3.0, refill_rate=1.0)\n",
    "assert tb.allow_request(\"c1\", 0.0) is True\n",
    "assert tb.allow_request(\"c1\", 0.0) is True\n",
    "assert tb.allow_request(\"c1\", 0.0) is True\n",
    "assert tb.allow_request(\"c1\", 0.0) is False\n",
    "assert tb.allow_request(\"c1\", 2.0) is True\n",
    "\n",
    "print(\"\u2713 Problem 11: RateLimiters passed all window and token assertions.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 12. DAG Task Execution Engine with Dependency Resolution\n",
    "\n",
    "### Problem Statement\n",
    "Design an orchestration engine for Directed Acyclic Graph (DAG) task pipelines:\n",
    "- `register_task(task_id: str, fn: callable, dependencies: list[str] = None)`:\n",
    "  - Registers a task node and its parent dependency task IDs.\n",
    "- `get_execution_order() -> list[str]`:\n",
    "  - Computes a valid topological execution order using **Kahn's Algorithm**.\n",
    "  - Raises `ValueError(\"Cycle detected in DAG task graph\")` if a dependency cycle exists.\n",
    "- `run() -> dict[str, any]`:\n",
    "  - Executes the registered tasks in valid topological order.\n",
    "  - Automatically feeds parent task outputs into child functions as a dictionary `{parent_id: parent_result}`.\n",
    "  - Returns `{task_id: result}` dictionary.\n",
    "\n",
    "### Starter Interface\n",
    "```python\n",
    "class DAGTaskEngine:\n",
    "    def register_task(self, task_id: str, fn: callable, dependencies: list[str] = None) -> None: pass\n",
    "    def get_execution_order(self) -> list[str]: pass\n",
    "    def run(self) -> dict[str, any]: pass\n",
    "```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from collections import defaultdict, deque\n",
    "from typing import Callable, Dict, List, Optional, Set, Any\n",
    "\n",
    "class DAGTaskEngine:\n",
    "    \"\"\"Topological DAG task executor with circular dependency detection and output piping.\"\"\"\n",
    "    def __init__(self):\n",
    "        self.tasks: Dict[str, Callable[[Dict[str, Any]], Any]] = {}\n",
    "        self.deps: Dict[str, Set[str]] = {}\n",
    "        self.dependents: Dict[str, Set[str]] = defaultdict(set)\n",
    "\n",
    "    def register_task(self, task_id: str, fn: Callable[[Dict[str, Any]], Any], dependencies: Optional[List[str]] = None) -> None:\n",
    "        self.tasks[task_id] = fn\n",
    "        self.deps[task_id] = set(dependencies) if dependencies else set()\n",
    "        for dep in self.deps[task_id]:\n",
    "            self.dependents[dep].add(task_id)\n",
    "\n",
    "    def get_execution_order(self) -> List[str]:\n",
    "        in_degree = {t: len(self.deps[t]) for t in self.tasks}\n",
    "        queue = deque([t for t, deg in in_degree.items() if deg == 0])\n",
    "        order = []\n",
    "\n",
    "        while queue:\n",
    "            curr = queue.popleft()\n",
    "            order.append(curr)\n",
    "            for child in self.dependents[curr]:\n",
    "                in_degree[child] -= 1\n",
    "                if in_degree[child] == 0:\n",
    "                    queue.append(child)\n",
    "\n",
    "        if len(order) != len(self.tasks):\n",
    "            raise ValueError(\"Cycle detected in DAG task graph\")\n",
    "        return order\n",
    "\n",
    "    def run(self) -> Dict[str, Any]:\n",
    "        order = self.get_execution_order()\n",
    "        results: Dict[str, Any] = {}\n",
    "        for task_id in order:\n",
    "            parent_outputs = {p: results[p] for p in self.deps[task_id]}\n",
    "            results[task_id] = self.tasks[task_id](parent_outputs)\n",
    "        return results\n",
    "\n",
    "# ---------------- Verification Tests ----------------\n",
    "dag = DAGTaskEngine()\n",
    "dag.register_task(\"fetch_features\", lambda _: [1.0, 2.0, 3.0])\n",
    "dag.register_task(\"fetch_weights\", lambda _: [0.5, 0.5, 0.5])\n",
    "dag.register_task(\n",
    "    \"compute_prediction\",\n",
    "    lambda parents: sum(f * w for f, w in zip(parents[\"fetch_features\"], parents[\"fetch_weights\"])),\n",
    "    dependencies=[\"fetch_features\", \"fetch_weights\"]\n",
    ")\n",
    "dag.register_task(\n",
    "    \"format_response\",\n",
    "    lambda parents: f\"Prediction: {parents['compute_prediction']:.2f}\",\n",
    "    dependencies=[\"compute_prediction\"]\n",
    ")\n",
    "\n",
    "order = dag.get_execution_order()\n",
    "assert order.index(\"fetch_features\") < order.index(\"compute_prediction\")\n",
    "assert order.index(\"fetch_weights\") < order.index(\"compute_prediction\")\n",
    "assert order.index(\"compute_prediction\") < order.index(\"format_response\")\n",
    "\n",
    "outputs = dag.run()\n",
    "assert outputs[\"compute_prediction\"] == 3.0\n",
    "assert outputs[\"format_response\"] == \"Prediction: 3.00\"\n",
    "\n",
    "cyclic_dag = DAGTaskEngine()\n",
    "cyclic_dag.register_task(\"X\", lambda _: 1, dependencies=[\"Y\"])\n",
    "cyclic_dag.register_task(\"Y\", lambda _: 2, dependencies=[\"X\"])\n",
    "try:\n",
    "    cyclic_dag.get_execution_order()\n",
    "    assert False, \"Should have raised ValueError on cycle\"\n",
    "except ValueError as e:\n",
    "    assert \"Cycle detected\" in str(e)\n",
    "\n",
    "print(\"\u2713 Problem 12: DAGTaskEngine Kahn's topological sort and cycle detection verified.\")\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python",
   "version": "3.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}