No public way to settle outstanding function calls before `on_pipeline_finished` runs

Author: mannyb223Created Sep 16, 2026Updated Sep 16, 2026

Question

pipecat 1.10.0. The pipeline-termination docs name on_pipeline_finished as "the single write point for end-of-call work like saving a transcript or recording". A function call that is still running when the call ends is not settled by the time that handler runs, and LLMService exposes nothing public to ask whether calls are outstanding, to wait for them, or to cancel them.

Is there a documented way to settle outstanding function calls from on_pipeline_finished, or should the write point move after processor cleanup for applications whose record depends on tool results?

Mechanism, with source lines at v1.10.0

The worker emits on_pipeline_finished on the terminal-frame path: pipeline/worker.py:1555 for an EndFrame, :1558 for a StopFrame, :1228 on the cancel path.
Only afterwards does it clean the processors: pipeline/worker.py:1328-1342, which awaits the worker's own event handlers (await self.cleanup()) and then await self._pipeline.cleanup().
Outstanding parallel function-call tasks are cancelled inside LLMService.cleanup() (services/llm_service.py:555-566, _cancel_all_function_call_tasks at :1552). Neither stop() (:533-543) nor cancel() (:544-554) touches them.

So a tool handler that is still running when the call ends is alive while the documented write point builds the record, and can complete its side effect after that snapshot has been taken. The persisted transcript, and any post-call analysis built from it, can therefore miss an action the application really took.

What is public on LLMService: start / stop / cancel / cleanup (:523, :533, :544, :555), run_function_calls (:1475), has_function (:1448), the register and unregister pairs (:917, :1024, :1394, :1415), run_inference (:444), append_system_instruction (:594), and the events on_function_calls_started, on_function_calls_cancelled, on_completion_timeout (:411-413). Every route that reports or settles outstanding calls is private: _cancel_all_function_call_tasks (:1552), _cancel_function_call_tasks (:1942), _cancel_function_calls_by_tool_call_id (:2059), _cancel_function_call (:2069), and the _function_call_tasks dict itself (:403).

The one public method that does cancel them is cleanup(), but it also runs every registered tool's cleanup callable (_run_tool_cleanups, :567) and is the method pipecat itself calls from Pipeline.cleanup() moments later, so calling it from the handler would run those cleanups twice and pre-empt pipecat's own teardown order. It is not documented for that use. LLMAssistantAggregator.has_function_calls_in_progress (processors/aggregators/llm_response_universal.py:1602) reports what the aggregator has seen, which is not a settle, and a late result frame may not reach it once the pipeline is tearing down.

The docs' only word on background work at termination is "Create background tasks through the pipeline task manager so they are tracked and cancelled on shutdown", which does not cover a function-call task pipecat created itself.

Reproduction

Stock only: real WorkerRunner, PipelineWorker, Pipeline, LLMService.register_function and the public LLMService.run_function_calls. ObservableLLMService is a no-provider subclass that wraps cleanup() to log entry and exit before delegating unchanged, and overrides no function-call or cancellation method. One registered tool sleeps 3 s. The termination paths tested are the public PipelineWorker.queue_frame(EndFrame) and PipelineWorker.cancel, each with cancel_on_interruption true and false. No InterruptionFrame is constructed or injected.

function_finish_order.py

python

#!/tmp/pc110/venv/bin/python
"""Offline Pipecat 1.10.0 reproduction of function-call/finalization order."""

from __future__ import annotations

import argparse
import asyncio
import json
import time
from dataclasses import dataclass, field
from typing import Any

from pipecat.frames.frames import EndFrame, FunctionCallFromLLM
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.services.llm_service import FunctionCallParams, LLMService
from pipecat.services.settings import LLMSettings
from pipecat.workers.runner import WorkerRunner


@dataclass
class Probe:
    case: str
    started_at: float = field(default_factory=time.monotonic)
    events: list[dict[str, Any]] = field(default_factory=list)
    handler_active: bool = False
    handler_completed: bool = False
    handler_cancelled: bool = False
    snapshot_active: bool | None = None
    snapshot_completed: bool | None = None

    def log(self, event: str, **details: Any) -> None:
        item = {
            "case": self.case,
            "t": round(time.monotonic() - self.started_at, 6),
            "event": event,
            **details,
        }
        self.events.append(item)
        print(json.dumps(item, sort_keys=True), flush=True)


class ObservableLLMService(LLMService):
    """Adds observation only; production cleanup remains the base implementation."""

    def __init__(self, probe: Probe):
        self.probe = probe
        # Base LLMService is sufficient because this probe invokes no inference.
        # Store-mode settings must nevertheless initialize every inherited field.
        super().__init__(
            settings=LLMSettings(
                model=None,
                system_instruction=None,
                temperature=None,
                max_tokens=None,
                top_p=None,
                top_k=None,
                frequency_penalty=None,
                presence_penalty=None,
                seed=None,
                filter_incomplete_user_turns=None,
                user_turn_completion_config=None,
            )
        )

    async def cleanup(self) -> None:
        self.probe.log(
            "cleanup_enter",
            handler_active=self.probe.handler_active,
            handler_completed=self.probe.handler_completed,
        )
        await super().cleanup()
        self.probe.log(
            "cleanup_exit",
            handler_active=self.probe.handler_active,
            handler_completed=self.probe.handler_completed,
            handler_cancelled=self.probe.handler_cancelled,
        )

    async def process_frame(self, frame: Any, direction: Any) -> None:
        """Pass frames through; inference is intentionally absent from this probe."""
        await super().process_frame(frame, direction)
        await self.push_frame(frame, direction)


async def run_case(mode: str, cancel_on_interruption: bool) -> dict[str, Any]:
    case = f"{mode}-cancel_on_interruption-{str(cancel_on_interruption).lower()}"
    probe = Probe(case=case)
    probe.log("case_start", interruption_frame_injected=False)

    service = ObservableLLMService(probe)

    async def slow_tool(params: FunctionCallParams) -> None:
        probe.handler_active = True
        probe.log("handler_enter", tool_call_id=params.tool_call_id)
        try:
            await asyncio.sleep(3.0)
            probe.handler_completed = True
            probe.log("handler_completion")
            await params.result_callback({"ok": True})
            probe.log("result_callback_returned")
        except asyncio.CancelledError:
            probe.handler_cancelled = True
            probe.log("handler_cancelled")
            raise
        finally:
            probe.handler_active = False
            probe.log("handler_exit")

    service.register_function(
        "slow_tool",
        slow_tool,
        cancel_on_interruption=cancel_on_interruption,
    )

    worker = PipelineWorker(
        Pipeline([service]),
        enable_rtvi=False,
        enable_tracing=False,
        enable_turn_tracking=False,
        idle_timeout_secs=None,
        check_dangling_tasks=True,
    )
    pipeline_started = asyncio.Event()

    @worker.event_handler("on_pipeline_started")
    async def on_pipeline_started(_worker: PipelineWorker, frame: Any) -> None:
        probe.log("pipeline_started", frame=type(frame).__name__)
        pipeline_started.set()

    @worker.event_handler("on_pipeline_finished")
    async def on_pipeline_finished(_worker: PipelineWorker, frame: Any) -> None:
        probe.snapshot_active = probe.handler_active
        probe.snapshot_completed = probe.handler_completed
        probe.log(
            "on_pipeline_finished_enter_snapshot",
            frame=type(frame).__name__,
            handler_active=probe.handler_active,
            handler_completed=probe.handler_completed,
            handler_cancelled=probe.handler_cancelled,
        )
        await asyncio.sleep(3.25)
        probe.log(
            "on_pipeline_finished_after_wait",
            handler_active=probe.handler_active,
            handler_completed=probe.handler_completed,
            handler_cancelled=probe.handler_cancelled,
        )
        probe.log("on_pipeline_finished_exit")

    runner = WorkerRunner(
        handle_sigint=False,
        handle_sigterm=False,
        check_dangling_tasks=True,
    )
    await runner.add_workers(worker)
    worker_run = asyncio.create_task(runner.run(), name=f"runner-run-{case}")
    await asyncio.wait_for(pipeline_started.wait(), timeout=5.0)

    probe.log("run_function_calls_request")
    await service.run_function_calls(
        [
            FunctionCallFromLLM(
                function_name="slow_tool",
                tool_call_id=f"tool-{case}",
                arguments={},
                context=LLMContext(),
            )
        ]
    )
    probe.log("run_function_calls_returned")

    await asyncio.sleep(1.0)
    if mode == "end":
        probe.log("termination_request", api="worker.queue_frame(EndFrame)")
        await worker.queue_frame(EndFrame(reason="stock ordering probe"))
    else:
        probe.log("termination_request", api="worker.cancel")
        await worker.cancel(reason="stock ordering probe")
    probe.log("termination_request_returned", api=mode)

    await asyncio.wait_for(worker_run, timeout=10.0)
    probe.log("worker_run_returned")

    names = [item["event"] for item in probe.events]
    expected = {
        "snapshot_saw_outstanding_handler": (
            probe.snapshot_active is True and probe.snapshot_completed is False
        ),
        "handler_completed_during_finished_handler": (
            names.index("on_pipeline_finished_enter_snapshot")
            < names.index("handler_completion")
            < names.index("on_pipeline_finished_exit")
        ),
        "cleanup_began_after_finished_handler": (
            names.index("on_pipeline_finished_exit") < names.index("cleanup_enter")
        ),
        "handler_not_cancelled_before_snapshot": not probe.handler_cancelled,
    }
    passed = all(expected.values())
    result = {"case": case, "passed": passed, "checks": expected}
    print("CASE_RESULT " + json.dumps(result, sort_keys=True), flush=True)
    if not passed:
        raise AssertionError(result)
    return result


async def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--mode", choices=("end", "cancel", "all"), default="all")
    args = parser.parse_args()
    modes = ("end", "cancel") if args.mode == "all" else (args.mode,)
    results = []
    for mode in modes:
        for cancel_on_interruption in (True, False):
            results.append(await run_case(mode, cancel_on_interruption))
    print(
        "RUN_RESULT "
        + json.dumps(
            {
                "passed": all(result["passed"] for result in results),
                "case_count": len(results),
            },
            sort_keys=True,
        ),
        flush=True,
    )


if __name__ == "__main__":
    asyncio.run(main())

Observed

All four cases behave the same way. At on_pipeline_finished entry the handler is active and incomplete; it completes while the event handler is still running (the handler deliberately waits so that completion is visible); LLMService.cleanup() is entered only after the event handler has exited; and the handler is not cancelled before the snapshot.

{"case": "end-cancel_on_interruption-true", "event": "case_start", "interruption_frame_injected": false, "t": 1e-06}
{"case": "end-cancel_on_interruption-true", "event": "pipeline_started", "frame": "StartFrame", "t": 0.694014}
{"case": "end-cancel_on_interruption-true", "event": "run_function_calls_request", "t": 0.694107}
{"case": "end-cancel_on_interruption-true", "event": "run_function_calls_returned", "t": 0.694209}
{"case": "end-cancel_on_interruption-true", "event": "handler_enter", "t": 0.694314, "tool_call_id": "tool-end-cancel_on_interruption-true"}
{"api": "worker.queue_frame(EndFrame)", "case": "end-cancel_on_interruption-true", "event": "termination_request", "t": 1.696477}
{"api": "end", "case": "end-cancel_on_interruption-true", "event": "termination_request_returned", "t": 1.696743}
{"case": "end-cancel_on_interruption-true", "event": "on_pipeline_finished_enter_snapshot", "frame": "EndFrame", "handler_active": true, "handler_cancelled": false, "handler_completed": false, "t": 1.697621}
{"case": "end-cancel_on_interruption-true", "event": "handler_completion", "t": 3.696494}
{"case": "end-cancel_on_interruption-true", "event": "result_callback_returned", "t": 3.697462}
{"case": "end-cancel_on_interruption-true", "event": "handler_exit", "t": 3.697633}
{"case": "end-cancel_on_interruption-true", "event": "on_pipeline_finished_after_wait", "handler_active": false, "handler_cancelled": false, "handler_completed": true, "t": 4.949755}
{"case": "end-cancel_on_interruption-true", "event": "on_pipeline_finished_exit", "t": 4.950124}
{"case": "end-cancel_on_interruption-true", "event": "cleanup_enter", "handler_active": false, "handler_completed": true, "t": 4.951008}
{"case": "end-cancel_on_interruption-true", "event": "cleanup_exit", "handler_active": false, "handler_cancelled": false, "handler_completed": true, "t": 4.951979}
{"case": "end-cancel_on_interruption-true", "event": "worker_run_returned", "t": 4.955302}

All four cases:

CASE_RESULT {"case": "end-cancel_on_interruption-true", "checks": {"cleanup_began_after_finished_handler": true, "handler_completed_during_finished_handler": true, "handler_not_cancelled_before_snapshot": true, "snapshot_saw_outstanding_handler": true}, "passed": true}
CASE_RESULT {"case": "end-cancel_on_interruption-false", "checks": {"cleanup_began_after_finished_handler": true, "handler_completed_during_finished_handler": true, "handler_not_cancelled_before_snapshot": true, "snapshot_saw_outstanding_handler": true}, "passed": true}
CASE_RESULT {"case": "cancel-cancel_on_interruption-true", "checks": {"cleanup_began_after_finished_handler": true, "handler_completed_during_finished_handler": true, "handler_not_cancelled_before_snapshot": true, "snapshot_saw_outstanding_handler": true}, "passed": true}
CASE_RESULT {"case": "cancel-cancel_on_interruption-false", "checks": {"cleanup_began_after_finished_handler": true, "handler_completed_during_finished_handler": true, "handler_not_cancelled_before_snapshot": true, "snapshot_saw_outstanding_handler": true}, "passed": true}
RUN_RESULT {"case_count": 4, "passed": true}

Expected

Some documented way for the application to know that a function call is outstanding at the documented write point, and either to wait for it or to settle it, without reaching into private state or calling cleanup() out of turn. A documented statement that the write point is expected to run before function calls are settled, with the recommended alternative, would answer it equally well.

#4948 (closed for inactivity) reports in-flight _run_function_call tasks being flagged as dangling on cancel. That describes the later cleanup barrier, not settlement before on_pipeline_finished, so this question is separate.

Environment

pipecat-ai==1.10.0, Python 3.13.11, macOS. No model, provider, transport, database or network is involved in the reproduction.

Source: pipecat-ai/pipecat

View original on GitHub View discussion on GitHub