[bugf][planner-worker-swarm][run accepts img and never passes it on]

PlannerWorkerSwarm.run takes img, documents it, and lists it in the trace
decorator's input_params, but the parameter appears nowhere else in the class.
The planner, every worker and the judge are all called with task only, so a
vision request runs as a text-only workflow and returns a confident answer
about an image no agent ever saw.

img now reaches the planner (and its sub-planners), the worker pool, and the
judge. WorkerPool carries it because workers claim tasks from a queue rather
than being called directly.

Author: ayaangazali

swarms/structs/planner_worker_swarm.py

@@ -361,10 +361,12 @@ def __init__(
         max_workers: Optional[int] = None,
         poll_interval: float = 0.1,
         task_timeout: Optional[float] = None,
+        img: Optional[str] = None,
     ):
         self.agents = agents
         self.task_queue = task_queue
         self.conversation = conversation
+        self.img = img
         self.max_workers = max_workers or min(
             len(agents), os.cpu_count() or 4
         )
@@ -462,7 +464,7 @@ def _worker_loop(
                         max_workers=1
                     ) as task_executor:
                         future = task_executor.submit(
-                            agent.run, task=context
+                            agent.run, task=context, img=self.img
                         )
                         try:
                             result = future.result(
@@ -473,7 +475,7 @@ def _worker_loop(
                                 f"Task execution exceeded {self.task_timeout}s timeout"
                             )
                 else:
-                    result = agent.run(task=context)
+                    result = agent.run(task=context, img=self.img)
 
                 current = self.task_queue.get_task(task.id)
                 if current and self.task_queue.complete(
@@ -691,6 +693,7 @@ def _run_planner(
         task: str,
         depth: int = 0,
         parent_task_id: Optional[str] = None,
+        img: Optional[str] = None,
     ) -> List[PlannerTask]:
         """Run a planner and add produced tasks to the queue.
 
@@ -711,7 +714,7 @@ def _run_planner(
             f"[PlannerWorkerSwarm] Running {planner_name} (depth={depth})"
         )
 
-        raw_output = planner.run(task=task)
+        raw_output = planner.run(task=task, img=img)
 
         spec = self._parse_structured_output(
             raw_output, PlannerTaskSpec
@@ -759,12 +762,13 @@ def _run_planner(
                         task=f"Decompose this task into smaller subtasks:\n\n{ptask.description}",
                         depth=depth + 1,
                         parent_task_id=ptask.id,
+                        img=img,
                     )
                     added_tasks.extend(sub_tasks)
 
         return added_tasks
 
-    def _run_judge(self) -> CycleVerdict:
+    def _run_judge(self, img: Optional[str] = None) -> CycleVerdict:
         """Run the judge agent to evaluate cycle results."""
         schema = BaseTool().base_model_to_dict(CycleVerdict)
 
@@ -795,7 +799,7 @@ def _run_judge(self) -> CycleVerdict:
             "If not, identify specific gaps and provide instructions for the next planning cycle."
         )
 
-        raw_output = judge.run(task=eval_task)
+        raw_output = judge.run(task=eval_task, img=img)
 
         try:
             verdict = self._parse_structured_output(
@@ -889,7 +893,7 @@ def run(
                     "Create new tasks to address these gaps."
                 )
 
-            self._run_planner(planner_task)
+            self._run_planner(planner_task, img=img)
 
             # Phase 2: Worker execution
             worker_pool = WorkerPool(
@@ -898,6 +902,7 @@ def run(
                 conversation=self.conversation,
                 max_workers=self.max_workers,
                 task_timeout=self.task_timeout,
+                img=img,
             )
             worker_pool.run(timeout=self.worker_timeout)
 
@@ -910,7 +915,7 @@ def run(
             )
 
             # Phase 3: Judge evaluation
-            verdict = self._run_judge()
+            verdict = self._run_judge(img=img)
 
             logger.info(
                 f"[PlannerWorkerSwarm] Cycle {cycle + 1} done. "

tests/structs/test_planner_worker_swarm.py

@@ -492,6 +492,38 @@ def test_drains_queue(self):
         ).run(timeout=10)
         assert q.is_all_done() and a.run.call_count == 3
 
+    def test_img_reaches_every_worker(self):
+        q = TaskQueue()
+        q.add_tasks(
+            [
+                PlannerTask(title=f"T{i}", description=f"D{i}")
+                for i in range(2)
+            ]
+        )
+        a = _mock_agent("W1")
+        WorkerPool(
+            agents=[a],
+            task_queue=q,
+            conversation=Conversation(time_enabled=False),
+            max_workers=1,
+            img="chart.png",
+        ).run(timeout=10)
+        assert a.run.call_count == 2
+        for call in a.run.call_args_list:
+            assert call.kwargs["img"] == "chart.png"
+
+    def test_no_img_still_passes_none(self):
+        q = TaskQueue()
+        q.add_task(PlannerTask(title="T", description="D"))
+        a = _mock_agent("W1")
+        WorkerPool(
+            agents=[a],
+            task_queue=q,
+            conversation=Conversation(time_enabled=False),
+            max_workers=1,
+        ).run(timeout=10)
+        assert a.run.call_args.kwargs["img"] is None
+
     def test_worker_prompt_injected(self):
         q = TaskQueue()
         q.add_task(PlannerTask(title="MyTask", description="Do it"))