Bridged child's on_activated frames are dropped if it's activated before the parent's BusBridge starts

[simopot]

pipecat version

1.3.0

Python version

3.13.7

Operating System

macOS 15.2 (local) / Pipecat Cloud (prod)

Issue description

A bridged=() child worker activated at startup loses the frames its on_activated emits (e.g. LLMSetToolsFrame). They cross the bus to the parent's BusBridgeProcessor, which — being a mid-pipeline FrameProcessor that hasn't received its own StartFrame yet — silently drops them at the "… but StartFrame not received yet" guard. They're never replayed.

Tools are advertised to the model only from context.tools (_advertised_tool_names), set by LLMSetToolsFrame. So a dropped frame leaves the LLM with no tools, and it emits the function call as plain text instead of calling it. A non-Flows worker sets tools once, so the whole session stays tool-less.

It only bites when the child is activated before the parent's StartFrame reaches the bridge — common at startup, since the child's small pipeline is ready long before the parent's large one. (Flows is immune: it injects tools at the pipeline source and re-advertises per node. The official WebRTC example is immune: connect latency outlasts startup.)

Full root-cause walkthrough + our production workaround

• LLMWorker.on_activated queues LLMSetToolsFrame via self.queue_frame(...) on activation.
• For a bridged=() child it crosses the bus and is pushed into the parent by BusBridgeProcessor.on_bus_message.
• If the parent's StartFrame hasn't reached the bridge, it's dropped here.
• BaseWorker._activate only guarantees a worker won't run its own on_activated before it started — nothing guarantees the peer bridge it sends to has started. There's no native "wait until the bridge is live" barrier.
• Reproduces from a @worker_ready handler, or from on_client_connected on an already-connected transport (Twilio media-stream websocket) where that hook fires during startup.

Our workaround (confirmed fix in production): gate activation on the parent's on_pipeline_started (fires only once StartFrame reaches the sink, i.e. past the bridge):

def __init__(self, ...):
    self._pipeline_started = asyncio.Event()

    @self.event_handler("on_pipeline_started")
    async def _on_started(worker, frame):
        self._pipeline_started.set()

async def activate_worker(self, worker_name, *, args=None, deactivate_self=False):
    await self._pipeline_started.wait()
    await super().activate_worker(worker_name, args=args, deactivate_self=deactivate_self)

App-side only; the bridge-buffering fix below is the general one.

Reproduction steps

1. Session/main worker with a BusBridgeProcessor in its LLM slot and a non-trivial pipeline (transport in → STT → user aggregator → bridge → TTS → transport out → assistant aggregator), shared LLMContext.
2. Add a bridged=() LLMWorker child whose on_activated sets tools via build_tools() → LLMSetToolsFrame.
3. Activate the child before the main worker's StartFrame reaches the bridge — e.g. from a @worker_ready handler (child's small pipeline becomes ready before the main one finishes starting), or from on_client_connected on an already-connected transport.
4. Send a first user turn that should trigger a tool call.
5. Observe the drop log during startup and a text-only LLM response with no tool call.

Expected behavior

Frames a bridged child emits in on_activated are delivered to the parent regardless of relative startup timing. Tools set in on_activated are advertised on the first inference.

Actual behavior

The LLMSetToolsFrame reaching the bridge before its StartFrame is silently dropped and never replayed; the model runs tool-less and narrates the action as text instead of calling the tool.

Proposed fix — buffer instead of drop at the bridge. Have BusBridgeProcessor hold frames received before its own StartFrame and replay them once it arrives. General fix, no app changes, keeps the canonical example robust. (Same likely applies to _BusEdgeProcessor.)

class BusBridgeProcessor(FrameProcessor, BusSubscriber):
    def __init__(self, ...):
        ...
        self._started = False
        self._pending_inbound: list[tuple[Frame, FrameDirection]] = []

    async def process_frame(self, frame, direction):
        await super().process_frame(frame, direction)
        if isinstance(frame, StartFrame):
            self._started = True
            await self.push_frame(frame, direction)
            for f, d in self._pending_inbound:
                await self.push_frame(f, d)
            self._pending_inbound.clear()
            return
        # ... existing handling ...

    async def on_bus_message(self, message):
        # ... existing filtering ...
        if not self._started:
            self._pending_inbound.append((message.frame, message.direction))
            return
        await self.push_frame(message.frame, message.direction)

Logs

# Real call, fast/already-connected websocket transport. Timestamps are UTC wall-clock.

12:47:24.777  main worker: Starting. Waiting for StartFrame#0 to reach the end of the pipeline...
12:47:24.986  on_client_connected
12:47:25.079  child worker: StartFrame#1 reached the end of the pipeline, pipeline is now ready.   (~4 ms — tiny pipeline)
12:47:25.080  Worker 'child': activated   -> on_activated queues LLMSetToolsFrame
12:47:25.161  main::BusBridge Trying to process LLMSetToolsFrame#0 but StartFrame not received yet   <- DROPPED, never replayed
12:47:25.308  main worker: StartFrame#0 reached the end of the pipeline, pipeline is now ready.   (~531 ms — large pipeline; ~227 ms too late)

# Resulting LLM turn — text output, no tools advertised. The model emits the function call as
# plain text content, spoken aloud verbatim by TTS:
gen_ai.output.type: "text"
output: {"name":"change_language","arguments":{"language":"fi"}}   # emitted as text instead of calling the change_language tool

# on_client_connected (12:47:24.986) fired 322 ms BEFORE the bridge was ready (12:47:25.308) — so activating
# from on_client_connected (as the official example does) would drop the same frame on this transport.

[amogh-nagri-11]

I reproduced this on current main — same drop path (BusBridgeProcessor.on_bus_message → push_frame → _check_started guard). Confirmed _BusEdgeProcessor is not affected the same way since it routes inbound bus frames through worker.queue_frame (no started-gate), so only BusBridgeProcessor needs the fix.

I'd like to take this — planning to buffer pending inbound bus frames and replay them once StartFrame is processed (either via the explicit pending-list from the issue, or by routing through self.queue_frame to reuse the same ordering guarantee _BusEdgeProcessor already has), plus a regression test for the pre-StartFrame delivery case. Will open a PR shortly — flagging here first in case there's a preferred approach