{"record":{"id":"6fe811da5cfcc97b","repo":"microsoft/semantic-kernel","slug":"unexpected-failure-broadcasting-to-channel-type","errorCode":null,"errorMessage":"Unexpected failure broadcasting to channel: {type(channel_ref.channel)}, failure: {failure}","messagePattern":"Unexpected failure broadcasting to channel: (.+?), failure: (.+?)","errorType":"exception","errorClass":"Exception","httpStatus":null,"severity":"error","filePath":"python/semantic_kernel/agents/group_chat/broadcast_queue.py","lineNumber":94,"sourceCode":"    async def ensure_synchronized(self, channel_ref: ChannelReference) -> None:\n        \"\"\"Blocks until a channel-queue is not in a receive state to ensure that channel history is complete.\n\n        Args:\n            channel_ref: The channel reference.\n        \"\"\"\n        if channel_ref.hash not in self.queues:\n            return\n\n        queue_ref = self.queues[channel_ref.hash]\n\n        while True:\n            async with queue_ref.queue_lock:\n                is_empty = queue_ref.is_empty\n\n                if queue_ref.receive_failure is not None:\n                    failure = queue_ref.receive_failure\n                    queue_ref.receive_failure = None\n                    raise Exception(\n                        f\"Unexpected failure broadcasting to channel: {type(channel_ref.channel)}, failure: {failure}\"\n                    ) from failure\n\n                if not is_empty and (not queue_ref.receive_task or queue_ref.receive_task.done()):\n                    queue_ref.receive_task = asyncio.create_task(self.receive(channel_ref, queue_ref))\n\n            if is_empty:\n                break\n\n            await asyncio.sleep(self.block_duration)\n\n    async def receive(self, channel_ref: ChannelReference, queue_ref: QueueReference) -> None:\n        \"\"\"Processes the specified queue with the provided channel, until the queue is empty.\n\n        Args:\n            channel_ref: The channel reference.\n            queue_ref: The queue reference.\n        \"\"\"","sourceCodeStart":76,"sourceCodeEnd":112,"githubUrl":"https://github.com/microsoft/semantic-kernel/blob/c028a0c7dc4f0814cdcbaba9d998f187a41197bf/python/semantic_kernel/agents/group_chat/broadcast_queue.py#L76-L112","documentation":"The broadcast loop drains each channel's QueueReference. If a prior receive() coroutine recorded a failure in queue_ref.receive_failure, that failure is surfaced on the next iteration by raising a new Exception chained with `from failure`. The actual error occurred asynchronously inside AgentChannel.receive; this raise is how the broadcasting caller learns about it.","triggerScenarios":"AgentChannel.receive() raised while draining a broadcast queue. On the next broadcast iteration the stored receive_failure is non-None, so the loop raises this wrapped exception. Triggers include a buggy custom channel, a serialization/deserialization error in channel.receive, or a downstream transport failure.","commonSituations":"Custom AgentChannel subclass whose receive() throws on malformed input; a channel backed by a remote service that drops the connection mid-receive; incompatible message types reaching the channel; partial state after an earlier crash that left receive_failure set.","solutions":["Read exc.__cause__ (the original `failure`) to identify the real channel error rather than the broadcast wrapper.","Inspect and fix the AgentChannel implementation whose receive() raised — add input validation and proper error handling inside receive.","Ensure messages placed on the queue are types the channel can deserialize; verify channel compatibility with the broadcasting agent.","Reproduce with logging inside receive() to capture the exact failing payload before it surfaces here."],"exampleFix":"// before\nclass MyChannel(AgentChannel):\n    async def receive(self, *args, **kwargs):\n        return json.loads(self._raw)  # raises if _raw invalid -> surfaces as broadcast failure\n\n// after\nclass MyChannel(AgentChannel):\n    async def receive(self, *args, **kwargs):\n        try:\n            return json.loads(self._raw)\n        except json.JSONDecodeError:\n            logger.warning(\"dropping malformed channel payload\")\n            return None","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try:\n    await broadcaster.broadcast(message)\nexcept Exception as e:\n    cause = e.__cause__  # the original channel.receive failure\n    logger.error(\"channel receive failed: %s\", cause)","preventionTips":["Make AgentChannel.receive() defensive: validate inputs and never throw on normal payloads.","Add logging inside receive() to capture failing payloads at the source.","Unit-test channel.receive() with the exact message types it will receive."],"tags":["broadcast","channel","async","agentchat"],"backgroundTag":null,"analyzedSha":"c028a0c7dc4f0814cdcbaba9d998f187a41197bf","analyzedAt":"2026-08-13T13:48:05.040Z","schemaVersion":2},"datasetVersion":"2026-08-13T14:17:21.547Z"}