An LLM produced a plan of tool calls. Execute it safely:
{"id", "tool", "args": {...}, "after": [step ids]}.$ references an earlier result: "$s1" is step
s1's whole output, "$s1.total" is a key of s1's output (dict). Resolve references recursively inside
lists and dicts.after steps have all finished
successfully, in plan order. Steps are run one at a time within a wave."skipped". A step referencing an unknown step or a
missing key fails with "bad_reference". An unknown tool fails with "unknown_tool".ToolError(retryable=True); retry those once. Other exceptions fail the step with
"tool_error".budget. When the budget is exhausted,
remaining runnable steps fail with "budget_exhausted"."blocked".Return {"status": {id: "ok" | <reason>}, "outputs": {id: output for ok steps}, "waves": [[ids]], "calls": n}.
Implement execute_plan. Input: plan: list[dict], budget: int. Returns: dict.
This code runs before yours (fake clients, clocks and the test harness). You can use its names; don't edit them.
class ToolError(Exception):
def __init__(self, message, retryable=False):
super().__init__(message)
self.retryable = retryable
_flaky = {}
def _search(query):
return [f"{query}-doc{i}" for i in range(1, 3)]
def _fetch(doc):
return {"doc": doc, "words": len(doc) * 10}
def _sum(values):
return {"total": sum(values)}
def _flaky_tool(key):
_flaky[key] = _flaky.get(key, 0) + 1
if _flaky[key] == 1:
raise ToolError("temporarily unavailable", retryable=True)
return f"ok:{key}"
def _broken(**kwargs):
raise ValueError("bad input")
def _always_down(**kwargs):
raise ToolError("down", retryable=True)
TOOLS = {"search": _search, "fetch": _fetch, "sum": _sum, "flaky": _flaky_tool, "broken": _broken, "down": _always_down}
def __harness__(fn, plan, budget):
_flaky.clear()
return fn(plan, budget)
def execute_plan(plan: list[dict], budget: int) -> dict:
outputs = {}
for step in plan:
outputs[step["id"]] = TOOLS[step["tool"]](**step["args"])
return {"status": {s["id"]: "ok" for s in plan}, "outputs": outputs, "waves": [[s["id"] for s in plan]], "calls": len(plan)}
Run executes the public tests. Submit grades against public and hidden tests.
An LLM produced a plan of tool calls. Execute it safely:
{"id", "tool", "args": {...}, "after": [step ids]}.$ references an earlier result: "$s1" is step
s1's whole output, "$s1.total" is a key of s1's output (dict). Resolve references recursively inside
lists and dicts.after steps have all finished
successfully, in plan order. Steps are run one at a time within a wave."skipped". A step referencing an unknown step or a
missing key fails with "bad_reference". An unknown tool fails with "unknown_tool".ToolError(retryable=True); retry those once. Other exceptions fail the step with
"tool_error".budget. When the budget is exhausted,
remaining runnable steps fail with "budget_exhausted"."blocked".Return {"status": {id: "ok" | <reason>}, "outputs": {id: output for ok steps}, "waves": [[ids]], "calls": n}.
Implement execute_plan. Input: plan: list[dict], budget: int. Returns: dict.
This code runs before yours (fake clients, clocks and the test harness). You can use its names; don't edit them.
class ToolError(Exception):
def __init__(self, message, retryable=False):
super().__init__(message)
self.retryable = retryable
_flaky = {}
def _search(query):
return [f"{query}-doc{i}" for i in range(1, 3)]
def _fetch(doc):
return {"doc": doc, "words": len(doc) * 10}
def _sum(values):
return {"total": sum(values)}
def _flaky_tool(key):
_flaky[key] = _flaky.get(key, 0) + 1
if _flaky[key] == 1:
raise ToolError("temporarily unavailable", retryable=True)
return f"ok:{key}"
def _broken(**kwargs):
raise ValueError("bad input")
def _always_down(**kwargs):
raise ToolError("down", retryable=True)
TOOLS = {"search": _search, "fetch": _fetch, "sum": _sum, "flaky": _flaky_tool, "broken": _broken, "down": _always_down}
def __harness__(fn, plan, budget):
_flaky.clear()
return fn(plan, budget)
def execute_plan(plan: list[dict], budget: int) -> dict:
outputs = {}
for step in plan:
outputs[step["id"]] = TOOLS[step["tool"]](**step["args"])
return {"status": {s["id"]: "ok" for s in plan}, "outputs": outputs, "waves": [[s["id"] for s in plan]], "calls": len(plan)}
Run executes the public tests. Submit grades against public and hidden tests.