0

Agent to-do list with task dependencies

by Frontier Interview· 1 day ago· ⑂ 0 remixes

Question: Agent to-do list with task dependencies

Companies: Perplexity

Difficulty: Medium

Problem statement

Build the task tracker an AI agent uses to plan its work. The question comes in three parts. Each part builds on the one before, so keep earlier behavior working.

Approach

The round is staged. Part 1 is a plain task store with a forward-only
status machine. Part 2 adds dependencies, so status changes have to
ripple to other tasks. Part 3 renders the whole list for an LLM. The
evaluation is mostly about whether the part 1 design survives parts 2
and 3 without a rewrite, so pick the data structures for the last part
up front.

Part 1: store and status rules. Keep tasks in a dict keyed by an
auto-incrementing integer ID. A Python dict keeps insertion order, so
iterating it also gives ID order for free. Model the statuses as an
IntEnum in lifecycle order (BLOCKED=1 < READY_TO_EXECUTE=2 < IN_PROGRESS=3 < SUCCEEDED=4 < FAILED=5). Then "forward only" is one
comparison, new_status > current, and a change is refused when the ID
is unknown, the task is already terminal, or the new status is not
strictly higher.

Part 2, brute force. After every status change, loop over all tasks.
Each BLOCKED task checks its dependencies: all succeeded means ready,
any failed means failed. Repeat until nothing changes, because a failure
can travel down a chain of several tasks. That is O(n) per pass and up
to O(n) passes, which is too slow and is exactly what the prompt warns
against.

Part 2, event driven. Store the graph in both directions and keep a
counter per task:

  • dependents[d] lists the tasks that wait on d (the reverse edges).
  • pending[t] counts how many of t's dependencies have not succeeded.

When a task succeeds, visit only its direct dependents and decrement
their counters. A BLOCKED dependent whose counter hits zero becomes
READY_TO_EXECUTE. This is Kahn's topological-sort in-degree trick,
driven by events.

When a task fails, run a BFS over the reverse edges. Every BLOCKED
task reached becomes FAILED and is pushed so its own dependents are
visited. A task that is already FAILED is not pushed again, so each
task fails at most once and the whole cascade costs O(tasks + edges it
touches). Only BLOCKED tasks can be reached at all: a task is ready
only when all its dependencies succeeded, and a succeeded task can never
fail.

Dependencies always name earlier tasks, so the graph is a DAG and the
BFS cannot loop. When a task is added with a dependency that already
failed, it can never run, so it starts FAILED instead of sitting
BLOCKED forever. Confirm this choice with the interviewer.

Part 3: rendering. Prefer a format that is easy for a model to parse
over prose: a one-line summary of counts, then a tab-separated table
with one row per task (id, status, depends_on, description).
Write the description as a JSON string so newlines or tabs inside it
cannot break the table.

Invariants:

  • Status never moves backward, and a terminal task never changes.
  • pending[t] equals the number of t's dependencies whose status is
    not SUCCEEDED.
  • After every call, no BLOCKED task has pending == 0 and no
    BLOCKED task depends, even indirectly, on a FAILED task.

Code

from __future__ import annotations

import json
from collections import deque
from dataclasses import dataclass, field
from enum import IntEnum


class TaskStatus(IntEnum):
    BLOCKED = 1
    READY_TO_EXECUTE = 2
    IN_PROGRESS = 3
    SUCCEEDED = 4
    FAILED = 5


TERMINAL = {TaskStatus.SUCCEEDED, TaskStatus.FAILED}


@dataclass
class Task:
    task_id: int
    description: str
    status: TaskStatus
    dependency_ids: list[int] = field(default_factory=list)


class TodoList:
    def __init__(self) -> None:
        self._tasks: dict[int, Task] = {}          # insertion order = id order
        self._dependents: dict[int, list[int]] = {}  # dep id -> tasks waiting on it
        self._pending: dict[int, int] = {}         # task id -> deps not yet SUCCEEDED
        self._next_id = 1

    def add_task(self, description: str, dependency_ids: list[int] | None = None) -> Task:
        deps = list(dict.fromkeys(dependency_ids or []))  # drop duplicates, keep order
        task_id = self._next_id
        self._next_id += 1

        statuses = [self._tasks[d].status for d in deps]
        if TaskStatus.FAILED in statuses:
            status = TaskStatus.FAILED            # can never run; fail fast
        elif all(s == TaskStatus.SUCCEEDED for s in statuses):
            status = TaskStatus.READY_TO_EXECUTE
        else:
            status = TaskStatus.BLOCKED

        task = Task(task_id, description, status, deps)
        self._tasks[task_id] = task
        self._dependents[task_id] = []
        self._pending[task_id] = sum(s != TaskStatus.SUCCEEDED for s in statuses)
        for d in deps:
            self._dependents[d].append(task_id)
        return task

    def get_task(self, task_id: int) -> Task | None:
        return self._tasks.get(task_id)

    def change_task_status(self, task_id: int, new_status: TaskStatus) -> bool:
        task = self._tasks.get(task_id)
        if task is None or task.status in TERMINAL or new_status <= task.status:
            return False
        task.status = new_status
        if new_status == TaskStatus.SUCCEEDED:
            self._on_succeeded(task_id)
        elif new_status == TaskStatus.FAILED:
            self._on_failed(task_id)
        return True

    def _on_succeeded(self, task_id: int) -> None:
        for child_id in self._dependents[task_id]:
            self._pending[child_id] -= 1
            child = self._tasks[child_id]
            if self._pending[child_id] == 0 and child.status == TaskStatus.BLOCKED:
                child.status = TaskStatus.READY_TO_EXECUTE

    def _on_failed(self, task_id: int) -> None:
        queue = deque([task_id])
        while queue:
            for child_id in self._dependents[queue.popleft()]:
                child = self._tasks[child_id]
                if child.status == TaskStatus.BLOCKED:   # each task fails at most once
                    child.status = TaskStatus.FAILED
                    queue.append(child_id)

    def render_for_llm(self) -> str:
        counts = {s.name: 0 for s in TaskStatus}
        lines = []
        for t in self._tasks.values():
            counts[t.status.name] += 1
            deps = ",".join(map(str, t.dependency_ids)) or "-"
            lines.append(f"{t.task_id}\t{t.status.name}\t{deps}\t{json.dumps(t.description)}")
        summary = " ".join(f"{k}={v}" for k, v in counts.items() if v)
        header = f"# tasks={len(self._tasks)} {summary}".rstrip()
        return "\n".join([header, "id\tstatus\tdepends_on\tdescription", *lines])

Complexity

Let n be the number of tasks and k the dependency count of one task.

  • add_task: O(k) time; O(k) extra space for the edges.
  • get_task: O(1).
  • change_task_status to IN_PROGRESS or another non-terminal status: O(1).
  • change_task_status to SUCCEEDED: O(direct dependents).
  • change_task_status to FAILED: O(tasks and edges reached by the
    cascade). Each task fails once, so across the whole lifetime of the
    list all cascades together cost O(n + total edges).
  • render_for_llm: O(n + total edges) time and output size.
  • Total space: O(n + total edges) for tasks, reverse edges and counters.

Test cases

  • add_task("fetch") on an empty list returns task 1, READY_TO_EXECUTE.
  • get_task(99) on a list without that ID returns None.
  • Task 1 is IN_PROGRESS: change_task_status(1, READY_TO_EXECUTE) is
    False (backward), and change_task_status(1, IN_PROGRESS) is False
    (not strictly higher).
  • change_task_status(42, FAILED) for an unknown ID returns False.
  • Add 2 depending on [1], 3 on [1, 2], 4 on [3]: all three start
    BLOCKED. Succeed 1 and task 2 becomes ready, but task 3 stays
    blocked (it still waits on 2). Succeed 2 and task 3 becomes ready,
    while task 4 stays blocked.
  • After task 1 succeeds, change_task_status(1, FAILED) returns False
    (terminal).
  • A new task whose dependencies have all succeeded starts
    READY_TO_EXECUTE.
  • Cascade: 2 depends on 1, 3 on 2, 4 on [3, 1], and 5 is independent.
    Failing 1 makes 2, 3 and 4 FAILED and leaves 5 ready.
  • Adding a task that depends on a failed task returns it FAILED.
  • Duplicate dependency IDs ([1, 1]) count once, so succeeding 1 makes
    the task ready.
  • render_for_llm() on the first scenario gives: 
    # tasks=5 BLOCKED=1 READY_TO_EXECUTE=2 SUCCEEDED=2
id	status	depends_on	description
1	SUCCEEDED	-	"fetch"
2	SUCCEEDED	1	"parse"
3	READY_TO_EXECUTE	1,2	"index"
4	BLOCKED	3	"rank"
5	READY_TO_EXECUTE	1,2	"ready now"

Follow-ups

  • Retry a failed task. This breaks "forward only", and failures must
    be undone downstream. Give FAILED a retry path back to
    READY_TO_EXECUTE. Cascaded failures then need to know their cause:
    store failed_because on each task, and on retry walk the same
    reverse edges and return the tasks that failed only because of this
    one to BLOCKED, recomputing their pending counts.
  • Add dependencies after creation (cycles possible). Before adding
    edge t -> d, check that t is not reachable from d with a DFS over
    the forward edges. If it is, reject the edge and return the cycle path,
    which a DFS can name and Kahn's algorithm cannot. Then update
    pending[t], and move t back to BLOCKED if it was ready.
  • Give an agent the next task to run. Keep a min-heap (or a sorted
    set) of READY_TO_EXECUTE IDs, updated on every transition, so
    next_ready() is O(log n) instead of a scan. Add priorities by
    keying the heap on (priority, id).
  • Subtasks or a parent that finishes when its children do. This is
    the same counter idea pointed the other way: a parent with k open
    children succeeds when the count reaches 0 and fails when any child
    fails.
  • Large lists in the LLM context. Render only the parts the model
    needs: open tasks in full, finished tasks collapsed to a count, and
    long failure chains shortened to their root cause.

Interviewer expectations

Mid-level (L4)

  • Finishes part 1 cleanly: the enum in lifecycle order, one comparison
    for forward-only, and correct False cases.
  • Gets dependency creation right and handles the success transition,
    even if failure propagation starts as a rescan.
  • Writes a few of their own checks and runs them.

Senior (L5)

  • Uses reverse edges plus pending counters from the start, and knows
    why a rescan is O(n) per change.
  • Propagates failure through the whole chain with BFS or DFS, and argues
    why each task fails at most once.
  • Raises edge cases unprompted: duplicate dependency IDs, a dependency
    that already failed, a manual change on a blocked task.
  • Picks a parse-friendly render format and makes it robust to messy
    descriptions.

Staff (L6+)

  • Designs part 1 so parts 2 and 3 add code without rewriting it, and
    states the invariants the code keeps.
  • Talks through what breaks with retries, edges added later (cycle
    detection and how to report the cycle), concurrent agents updating
    the same list, and persistence.
  • Thinks about the LLM consumer: token budget, stable IDs the model can
    refer back to, and what context the agent needs in order to decide
    its next action.

Sign in to comment, vote or remix.