Execute an agent's tool plan with retries and skips

Agent buildHard40 min

Your task

An agent's planner emits a plan of tool steps, each naming the steps whose results it needs. The executor runs the plan in rounds, a few tool calls at a time, and has to cope with tools that fail. Write the executor as a deterministic simulation: run_plan(steps, outcomes, max_parallel, retry_budget).

  • steps: a list of [step_id, needs], where needs is a list of step ids. The list order is the plan order. Step ids are unique.
  • outcomes: what each attempt of a step returns. outcomes[step_id] is a list of "ok" and "fail" for its first, second, third attempt and so on. A step with no entry succeeds on every attempt. An attempt past the end of a step's list fails. Entries for ids that are not steps are ignored.
  • max_parallel: at most this many attempts per round (at least 1).
  • retry_budget: the total number of retries the whole plan may use (zero or more).

Each round

  1. A step is ready if it has not succeeded, has not failed for good, and every step it needs has succeeded.
  2. If no step is ready, execution ends. Otherwise the round runs the first max_parallel ready steps in plan order, one attempt each. Rounds are numbered from 1.
  3. A step whose attempt returns "ok" has succeeded in this round.
  4. A step whose attempt fails uses one retry from the budget if any is left, and is ready again from the next round, where it waits its turn in plan order like any other ready step. When several steps fail in the same round, they draw on the budget in plan order. A step that fails when the budget is empty has failed for good.

When execution ends, a step that never succeeded or failed is "skipped" if it needs a step that failed for good, directly or through a chain of needs; otherwise it is "blocked" (it needs an unknown id, sits in a cycle, or needs a blocked step).

Return a dict mapping every step id to [status, round]: ["ok", r] with the round it succeeded, ["failed", r] with the round of its last attempt, ["skipped", 0] or ["blocked", 0].

Examples, all with steps = [["fetch", []], ["parse", ["fetch"]], ["lint", ["fetch"]], ["report", ["parse", "lint"]]] and max_parallel 2:

  • outcomes = {"parse": ["fail", "ok"]}, retry_budget 1 returns {"fetch": ["ok", 1], "parse": ["ok", 3], "lint": ["ok", 2], "report": ["ok", 4]}. Round 2 runs parse and lint; parse fails and uses the only retry; round 3 runs parse again.
  • The same with retry_budget 0 returns {"fetch": ["ok", 1], "parse": ["failed", 2], "lint": ["ok", 2], "report": ["skipped", 0]}.
  • run_plan([["a", ["b"]], ["b", ["a"]], ["c", ["zzz"]], ["d", []]], {}, 2, 0) returns {"a": ["blocked", 0], "b": ["blocked", 0], "c": ["blocked", 0], "d": ["ok", 1]}.

Examples

run_plan(steps: list[list], outcomes: dict[str, list[str]], max_parallel: int, retry_budget: int) -> dict[str, list]
Example 1
Inputrun_plan([["fetch", []], ["parse", ["fetch"]], ["lint", ["fetch"]], ["report", ["parse", "lint"]]], {"parse": ["fail", "ok"]}, 2, 1)
Output{"fetch": ["ok", 1], "parse": ["ok", 3], "lint": ["ok", 2], "report": ["ok", 4]}
Example 2
Inputrun_plan([["fetch", []], ["parse", ["fetch"]], ["lint", ["fetch"]], ["report", ["parse", "lint"]]], {"parse": ["fail", "ok"]}, 2, 0)
Output{"fetch": ["ok", 1], "parse": ["failed", 2], "lint": ["ok", 2], "report": ["skipped", 0]}
Example 3
Inputrun_plan([["a", ["b"]], ["b", ["a"]], ["c", ["zzz"]], ["d", []]], {}, 2, 0)
Output{"a": ["blocked", 0], "b": ["blocked", 0], "c": ["blocked", 0], "d": ["ok", 1]}