Route subtasks to specialised workers by capability, respect task dependencies, isolate worker failures, and enforce a global step budget.
Full problem description visible. Upgrade to unlock the editor, test cases, and solution.
Implement run_supervisor(goal, workers, tasks, max_steps) that manages a multi-worker task execution workflow.
workers is a dict mapping worker names to callable functions.
tasks is a list of dicts, each with:
- id (str): unique task identifier.
- worker (str): name of the worker to dispatch to.
- depends_on (list[str], optional): task IDs that must complete first.
- params (any): arguments to pass to the worker.
Requirements:
- Each iteration, identify ready tasks (all dependencies in completed).
- Dispatch each ready task to its assigned worker via workers[name](params).
- Collect results in a completed dict by task ID.
- On worker failure, mark the task as failed with None result but do not abort other tasks.
- If a worker name is not in workers, mark the task as failed with an error message.
- Stop when all tasks are done or max_steps steps have been used.
- Return {"completed": dict, "trace": list, "steps_used": int, "all_completed": bool}.
- The trace should contain one entry per task: {"task": str, "worker": str, "status": str, "result": any}.
₹999/monthLimited period launch pricing.
Get the full problem statement, test cases, interactive editor, solution explanation, and visual diagram.
def run_supervisor(goal, workers, tasks, max_steps):
completed = {}
trace = []
steps = 0Find ready tasks and dispatch to workers
return {"completed": completed, "trace": trace,
"steps_used": steps,
"all_completed": len(completed) == len(tasks)}Already have Pro access? Sign in