From e546471bef2936019c4bf66d9de0276a75b62c9a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 13 May 2026 21:58:17 -0700 Subject: [PATCH] Rename _handle_task_* job dispatchers to _handle_job_* MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes a name collision where `_handle_task_cancel` was defined twice — once for `BusCancelTaskMessage` (task lifecycle) and once for `BusJobCancelMessage` (job RPC) — the second silently shadowing the first. Job-side dispatchers are now consistently named `_handle_job_*` and the internal helpers `_run_task_handler` / `_send_task_request` become `_run_job_handler` / `_send_job_request`. Task-lifecycle handlers (`_handle_task_end`, `_handle_task_cancel`, `_handle_task_activate`, `_handle_task_deactivate`, `_handle_task_error`) keep their names. --- src/pipecat/pipeline/base_task.py | 46 ++++++++++++++++--------------- 1 file changed, 24 insertions(+), 22 deletions(-) diff --git a/src/pipecat/pipeline/base_task.py b/src/pipecat/pipeline/base_task.py index 1440124c8..08a052024 100644 --- a/src/pipecat/pipeline/base_task.py +++ b/src/pipecat/pipeline/base_task.py @@ -417,14 +417,16 @@ class BaseTask(BaseObject, BusSubscriber): """ pass - async def on_job_response(self, message: BusJobResponseMessage) -> None: + async def on_job_response( + self, message: BusJobResponseMessage | BusJobResponseUrgentMessage + ) -> None: """Called when a worker sends a response. Override to process individual results as they arrive. """ pass - async def on_job_update(self, message: BusJobUpdateMessage) -> None: + async def on_job_update(self, message: BusJobUpdateMessage | BusJobUpdateUrgentMessage) -> None: """Called when a worker sends a progress update.""" pass @@ -498,21 +500,21 @@ class BaseTask(BaseObject, BusSubscriber): elif isinstance(message, BusCancelTaskMessage): await self._handle_task_cancel(message) elif isinstance(message, BusJobRequestMessage): - await self._handle_task_request(message) + await self._handle_job_request(message) elif isinstance(message, (BusJobResponseMessage, BusJobResponseUrgentMessage)): - await self._handle_task_response(message) + await self._handle_job_response(message) elif isinstance(message, (BusJobUpdateMessage, BusJobUpdateUrgentMessage)): - await self._handle_task_update(message) + await self._handle_job_update(message) elif isinstance(message, BusJobUpdateRequestMessage): - await self._handle_task_update_request(message) + await self._handle_job_update_request(message) elif isinstance(message, BusJobCancelMessage): - await self._handle_task_cancel(message) + await self._handle_job_cancel(message) elif isinstance(message, BusJobStreamStartMessage): - await self._handle_task_stream_start(message) + await self._handle_job_stream_start(message) elif isinstance(message, BusJobStreamDataMessage): - await self._handle_task_stream_data(message) + await self._handle_job_stream_data(message) elif isinstance(message, BusJobStreamEndMessage): - await self._handle_task_stream_end(message) + await self._handle_job_stream_end(message) await self._call_event_handler("on_bus_message", message) @@ -1085,11 +1087,11 @@ class BaseTask(BaseObject, BusSubscriber): group = self._create_job_group(task_names, timeout=timeout, cancel_on_error=cancel_on_error) for task_name in task_names: - await self._send_task_request(task_name, group.job_id, job_name=name, payload=payload) + await self._send_job_request(task_name, group.job_id, job_name=name, payload=payload) return group - async def _send_task_request( + async def _send_job_request( self, task_name: str, job_id: str, @@ -1179,7 +1181,7 @@ class BaseTask(BaseObject, BusSubscriber): BusCancelTaskMessage(source=self.name, target=child.name, reason=message.reason) ) - async def _handle_task_request(self, message: BusJobRequestMessage) -> None: + async def _handle_job_request(self, message: BusJobRequestMessage) -> None: """Handle an incoming job request. Dispatches to @job handlers if any match, otherwise falls back @@ -1199,14 +1201,14 @@ class BaseTask(BaseObject, BusSubscriber): lock = self._job_locks.setdefault(message.job_name, asyncio.Lock()) task = self.create_task( - self._run_task_handler(message.job_id, handler, message, lock), + self._run_job_handler(message.job_id, handler, message, lock), f"{self.name}::job_{message.job_name or 'default'}", ) self._job_handler_tasks[message.job_id] = task await self._call_event_handler("on_job_request", message) - async def _run_task_handler( + async def _run_job_handler( self, job_id: str, handler, @@ -1224,7 +1226,7 @@ class BaseTask(BaseObject, BusSubscriber): finally: self._job_handler_tasks.pop(job_id, None) - async def _handle_task_response( + async def _handle_job_response( self, message: BusJobResponseMessage | BusJobResponseUrgentMessage ) -> None: """Handle a job response and track group completion.""" @@ -1245,7 +1247,7 @@ class BaseTask(BaseObject, BusSubscriber): await self._track_job_group_response(message.job_id, message.source, message.response) - async def _handle_task_update( + async def _handle_job_update( self, message: BusJobUpdateMessage | BusJobUpdateUrgentMessage ) -> None: """Handle a job progress update.""" @@ -1255,13 +1257,13 @@ class BaseTask(BaseObject, BusSubscriber): message.job_id, JobGroupEvent(JobGroupEvent.UPDATE, message.source, message.update) ) - async def _handle_task_update_request(self, message: BusJobUpdateRequestMessage) -> None: + async def _handle_job_update_request(self, message: BusJobUpdateRequestMessage) -> None: """Handle a job update request from the requester.""" if message.job_id in self._active_jobs: await self.on_job_update_requested(message) await self._call_event_handler("on_job_update_requested", message) - async def _handle_task_cancel(self, message: BusJobCancelMessage) -> None: + async def _handle_job_cancel(self, message: BusJobCancelMessage) -> None: """Handle a job cancellation. Cancels the running handler task (if any), calls the @@ -1278,7 +1280,7 @@ class BaseTask(BaseObject, BusSubscriber): await self._call_event_handler("on_job_cancelled", message) await self.send_job_response(message.job_id, status=JobStatus.CANCELLED) - async def _handle_task_stream_start(self, message: BusJobStreamStartMessage) -> None: + async def _handle_job_stream_start(self, message: BusJobStreamStartMessage) -> None: """Handle the start of a streaming job response.""" await self.on_job_stream_start(message) await self._call_event_handler("on_job_stream_start", message) @@ -1287,7 +1289,7 @@ class BaseTask(BaseObject, BusSubscriber): JobGroupEvent(JobGroupEvent.STREAM_START, message.source, message.data), ) - async def _handle_task_stream_data(self, message: BusJobStreamDataMessage) -> None: + async def _handle_job_stream_data(self, message: BusJobStreamDataMessage) -> None: """Handle a streaming job data chunk.""" await self.on_job_stream_data(message) await self._call_event_handler("on_job_stream_data", message) @@ -1296,7 +1298,7 @@ class BaseTask(BaseObject, BusSubscriber): JobGroupEvent(JobGroupEvent.STREAM_DATA, message.source, message.data), ) - async def _handle_task_stream_end(self, message: BusJobStreamEndMessage) -> None: + async def _handle_job_stream_end(self, message: BusJobStreamEndMessage) -> None: """Handle the end of a streaming job response.""" await self.on_job_stream_end(message) await self._call_event_handler("on_job_stream_end", message)