Skip to content

Pipecat-native alignment and horizon agents

Status: Draft | Last reviewed: 2026-09-01

This doc compares the orchestrator's homegrown multi-agent machinery — subagents, text-turn coordination, consult messaging, transfer — against the native multi-worker features that ship in the pipecat release the repo already depends on, and proposes what to adopt, what to keep custom, and in what order. It closes with a layered architecture for horizon agents: journey-scale AI orchestration (an insurance claim that starts on a voice call and runs for days or weeks through workflow steps, human waits, and outbound calls) spanning both ai-orchestrator and async-agent. Read it before building new multi-agent, transfer, or durable-workflow features.

Non-goals: this doc proposes no code changes by itself — every item lands in the backlog with effort and risk. The A2UI protocol comparison against the a2ui.org specification lives in the companion doc, A2UI spec convergence.

Current state: custom machinery vs pipecat-native equivalents

ai-orchestrator depends on pipecat-ai>=1.4.0,<2.0.0 (pyproject.toml#L12) as a plain pip dependency. lib/pipecat/ is a first-party extension layer (~268 files) that imports from the upstream package; nothing is vendored or forked. The 1.4 line already ships a complete multi-worker surface — pipecat/bus (a WorkerBus abstraction with in-process AsyncQueueBus and network RedisBus/PgmqBus backends, typed job-coordination messages, worker lifecycle activate/deactivate), pipecat/registry (WorkerRegistry tracking local and remote workers with watch/notify), pipecat/workers (LLMWorker/LLMContextWorker with an @tool decorator, WebSocketProxyClient/Server proxy workers, and the UIWorker), plus pipeline/llm_switcher.py. Almost none of it is used today:

TrueServe mechanism Where Pipecat-native concept Adoption status
One PipelineWorker + WorkerRunner per session grpc_provider.py pipecat/workers runner Adopted (single worker, no sharing)
Subagents: per-subagent SubagentWorker + bus jobs (legacy executor fallback) worker_adapter.py BaseWorker + @job + AsyncQueueBus Adopted 2026-07-31 (below)
TextTurnCoordinator/TextTurnGate floor arbitration text_turn_serializer.py LLMSwitcher + worker activation No direct native equivalent — keep
AgentBus consult pairing over Redis agent_bus.py pipecat/bus RedisBus + WorkerRegistry Name collision; unrelated implementation
Session switch (S2S/Realtime instruction swap) session_switch_manager.py — No native analog — keep
A2A consume/expose a2a_manager.py, a2a_server.py — (pipecat has no A2A) Custom — keep
A2UI generative UI a2ui_tool.py + a2ui-react UIWorker (different problem — below) Custom — keep
— — Bus, registry, jobs, LLMSwitcher, proxy workers, UIWorker Unused upstream surface

The rest of this doc walks each row that has a decision attached.

UIWorker: complementary, not a replacement

Pipecat's UIWorker is a split-LLM screen agent: the client streams accessibility snapshots and user events up (ui-snapshot, ui-event over RTVI), a dedicated worker LLM reasons over page state, and the worker drives the screen back down (ui-command, ui-job-group) with primitives like click(ref), set_input_value(ref, value), scroll_to(ref). It solves perceiving and manipulating an existing application UI.

TrueServe's A2UI stack solves the opposite problem: the agent generates UI (surfaces, components, data models) that a2ui-react renders from scratch. There is no page to snapshot and no pre-existing controls to click — the agent owns the surface. Migrating A2UI onto UIWorker would discard the catalog system, the surface/data-model protocol, and the multimodal permission flow while gaining nothing, because UIWorker has no component-rendering vocabulary at all.

Verdict: do not migrate A2UI to UIWorker. One candidate future use fits the screen-driving model:

  1. Copilot driving CRM screens — the copilot path already listens to live calls; a UIWorker could act on the human agent's Salesforce/Genesys screen (highlight the relevant case, prefill a form) instead of only suggesting text. It depends on an RTVI-capable client channel that does not exist in the current transports. Effort L.

Subagents: adopt pipecat workers and jobs, in phases

Note: Phases 0–2 SHIPPED 2026-07-31. This section records the before-state and the delivered shape; the runtime documentation lives in universal-pipeline-builder.md.

Before the change, subagents were tool calls running a stateless, text-only LLM request via direct SDK calls — no pipecat pipeline, no worker, one hand-rolled adapter (OpenAI-only, llm_adapters/). workers mode was a shim that detected LLMContextWorker but never registered it (pipecat workers require a dedicated LLMService per worker, and subagents bypassed LLMService entirely).

Phase 0 — record reality (SHIPPED)

  1. context_strategy was dead config — plumbed end to end (UI, config schema, execute() parameter) but no call site passed conversation context; every subagent ran task_only. Now wired: tool callbacks hand the parent's live LLMContext to the adapter and context_strategies.py slices it (task_only / last_n_turns / full; system/tool messages and non-text content always stripped).
  2. Nested subagent pipelines are not built — the executor's _pipeline_to_inline_config extracts only the first LLM stage. Docstrings now say so and the build warns when non-LLM stages are ignored. Real recursive builds remain future work (backlog P0b).

Phase 1 — real workers per subagent (SHIPPED)

llm_service_factory.py constructs a lightweight per-subagent pipecat LLMService (OpenAI, OpenAI Responses, Google; Anthropic/Bedrock when their SDKs ship in the image) with the same credential resolution as the parent stage (get_env_credential, integration row → env). subagent_worker.py is a bus-only BaseWorker — not LLMContextWorker: subagents are request/response with a fresh context per call, so a persistent worker-owned context and frame pipeline would be dead weight — serving a @job(name="run") handler that runs one-shot run_inference. Workers register as children of the session PipelineWorker (a root bus-only worker never finishes on its own and would block the runner's auto-end, wedging session teardown).

Phase 2 — jobs over the in-process bus (SHIPPED)

Dispatch flows through pipecat job coordination: the session's PipelineWorker is the requester, one job(worker, name="run") per invocation over the default AsyncQueueBus (readiness-wait, per-job asyncio tasks, cancellation propagation for free). Two findings sharpened the original plan:

  • The old parallel mode was broken upstream. Its SubagentToolHandler frame interceptor imported FunctionCallsFinishedFrame, which does not exist in pipecat 1.4 — the import crashed the build for every parallel-mode agent. It was also unnecessary: pipecat's LLMService already runs a tool-call batch concurrently (run_in_parallel=True default, grouped so the LLM re-triggers once per batch). The handler is deleted; all non-session-switch modes share one per-tool-callback path through the adapter.
  • The legacy executor stays as fallback. ref subagents (config resolves from the DB per call), unsupported providers, and factory failures dispatch through SubagentExecutor with identical result contracts.

The payoff is not the in-process refactor itself; it is that job semantics are bus-portable. The same code runs over RedisBus later, which is what makes cross-session messaging and horizon-agent seams coherent.

Explicit non-adoption: session switch

session_switch (S2S/Realtime) is not delegation — it rewrites the live provider session (session.update swapping instructions and tools, a [SYSTEM HANDOFF] primer item, history retained by the provider). Pipecat has no analog; worker activation would tear down and rebuild the provider socket. Keep it custom.

TextTurnCoordinator vs LLMSwitcher: keep custom

Pipecat's LLMSwitcher selects one active LLM at a time — a routing primitive. The multi-agent text-turn system solves a different shape: N branches run concurrently, one holds the floor and streams live, the others buffer whole Start→Text→End runs and replay in order, agents abstain via the [[SILENT]] sentinel, and sequential_shared mode cross-injects completed peer turns into other branches' contexts (see text turn-taking). None of that maps onto a switcher, and forcing it would lose concurrent generation — the feature's point.

Verdict: keep the coordinator. One convergence note, contingent on subagent Phases 1–2: if text branches ever become bus-registered workers, floor grant/release could ride worker activate/deactivate messages instead of the bespoke coordinator channel. Revisit after Phase 2; not a proposal today.

RedisBus vs the homegrown AgentBus

agent_bus.py is the cross-session control channel for warm transfer/consult: a Redis pair registry (agentbus:pair:{consult_id} mapping consult id → origin and consult conversation ids) plus message delivery by publishing into the peer session's existing inject:{conversation_id} channel. It is unrelated to pipecat's pipecat/bus despite the name — a collision that will misdirect every reader who knows upstream pipecat.

Proposal Detail Effort Risk
Resolve the name collision Either rename AgentBus → e.g. ConsultChannel, or (better) fold it into the migration below S Low
Migrate the consult control channel to pipecat RedisBus + WorkerRegistry Each session's runner joins a RedisBus scoped to the consult; origin and consult agents discover each other via registry watch/notify instead of the hand-rolled pair registry; structured events (merge_requested etc.) become typed bus messages instead of [event:…] strings on the inject channel M M — Redis semantics change under live consults; run both paths behind a flag for a compat window

Explicit non-adoption: distributed pipeline workers. Pipecat's distributed mode can span pipeline stages across processes. Do not adopt that for media sessions — a session is a long-lived gRPC stream pinned to one pod, HPA scales on the active-session gauge, and the SIP/media leg (ari-controller → media-proxy) must stay untouched. The bus carries control messages only; media never crosses it.

Horizon agents: a three-layer architecture

A horizon agent is a real end-to-end journey orchestrated by AI: an insurance claim opened on a voice call, then a workflow filing forms, waiting on adjusters (human or AI), placing outbound status calls, and resuming on each trigger until the claim is filed, adjudicated, and paid — days to weeks of wall-clock time. No single runtime should own that span. The pieces already exist in the stack; what is missing is the seams.

Layer Timescale Runtime Machinery
In-session ms–minutes ai-orchestrator Pipecat pipeline; subagents/jobs over AsyncQueueBus (post Phase 2)
Cross-session minutes ai-orchestrator + Redis RedisBus control channel (consult, injection, session-to-session events)
Journey days–weeks async-agent LangGraph durable graphs: Postgres checkpoints, RabbitMQ/SQS consumers, cron poller, Redis locks

Note: the layering invariant — pipecat job coordination never escapes the session, and async-agent never holds a realtime media session. Pipecat's bus is a live-process construct with no durability; LangGraph checkpoints are durable but have no media path. Each layer delegates across the seam rather than stretching itself.

async-agent is already the journey layer: a LangGraph workflow executor with llm_agent/tool/code/foreach/subworkflow nodes, durable Postgres checkpoints, queue consumers, and a cron poller. It shares the tool_configs table and prompt-token conventions with the orchestrator but no pipecat code — correctly so.

Seam inventory

Five seams make the layers compose. Each row is a proposal:

# Seam Exists today Gap Effort
H1 Session → journey: a live call spawns a workflow Partially — workflow triggers fire on conversation lifecycle A first-class handoff artifact: conversation id, context snapshot, tenant, tool refs, resumption token — a typed contract, not ad-hoc trigger payloads M
H2 Journey → session: a workflow node initiates an outbound call or chat No An initiate-session node: originate via ari-controller for voice / conversation-api session for chat, then suspend on a LangGraph checkpoint until the session posts its result. The biggest gap and the highest-value seam L
H3 Await external event: workflow waits on a human or another agent Mechanics exist (LangGraph interrupt + checkpoint + cron resume) A standard wake-signal contract — which queue message or entries:{id} event resumes which checkpoint, with timeout/escalation semantics M
H4 A2A at journey timescale In-session only — a2a_manager.py collapses streaming tasks synchronously with a ~30 s cap (required for Realtime) Long-running A2A tasks belong in async-agent nodes that poll/subscribe to the remote Task, never in an in-session tool. Add an a2a_task node type M
H5 Journey status → humans: claim state visible in conversations No Workflow progress events published into entries:{conversation_id} so surface-app and the Activity Log show journey state alongside the transcript S–M

The claim journey, end to end

sequenceDiagram
    participant C as Customer
    participant O as ai-orchestrator (voice session)
    participant W as async-agent (claim workflow)
    participant H as Human adjuster

    C->>O: Inbound call — starts a claim
    O->>W: H1 handoff artifact (context snapshot, tenant, refs)
    Note over O: Session ends; nothing durable remains here
    W->>W: File claim, run checks (LLM + tool nodes)
    W->>H: Assign adjudication task
    Note over W: H3 — checkpoint, await wake signal
    H-->>W: Decision posted (queue message resumes checkpoint)
    W->>O: H2 — originate outbound status call
    O->>C: "Your claim was approved…"
    O-->>W: Session result; workflow resumes
    W->>W: Trigger payout, close claim
    W-->>C: H5 — status events visible in conversation history

Diagram source: pipecat-native-alignment.mmd.

Explicit non-adoptions at this layer: do not build a second durable engine inside ai-orchestrator (async-agent is that engine), and do not stretch pipecat's bus to journey timescales (it has no persistence; a pod restart loses everything in flight).

Consolidated backlog

Numbered for reference; rejections are recorded as decisions so they are not re-litigated.

# Proposal Decision Effort Risk Prerequisite
P0 Wire context_strategy + honest nested-pipeline warnings Done 2026-07-31 S Low —
P0b Recursive nested-subagent pipeline builds (currently first-LLM-stage-only) Backlog L M P2
P1 Per-subagent pipecat LLMService + registered SubagentWorker Done 2026-07-31 (bus-only BaseWorker, not LLMContextWorker) M M —
P2 Subagent dispatch → bus jobs on AsyncQueueBus; broken SubagentToolHandler deleted Done 2026-07-31 M M —
P3 Consult control channel → pipecat RedisBus + registry; retire/rename homegrown AgentBus Adopt M M P2 recommended
P4 UIWorker as A2UI replacement Reject — complementary problems — — —
P5 UIWorker for copilot screen-driving / embedded views Backlog (no commitment) L M RTVI client channel
P6 LLMSwitcher for text-turn coordination Reject — switcher ≠ concurrent floor arbitration — — Revisit floor-over-bus after P2
P7 Distributed pipecat workers across pods for media sessions Reject — session-pinned pod + HPA stays — — —
P8 Replace session_switch with worker activation Reject — no native analog for live provider-session rewrite — — —
H1 Session→journey handoff artifact contract Adopt M Low —
H2 Journey→session initiate-session node (outbound voice/chat + checkpoint suspend) Adopt L M H1
H3 Await-external-event wake-signal contract Adopt M Low —
H4 a2a_task node in async-agent for long-running A2A Adopt M Low —
H5 Journey status events into conversation entries Adopt S–M Low —
P9 Journey-scale pipecat bus / durable engine in orchestrator Reject — async-agent is the durable layer — — —

Suggested order (P0/P1/P2 done): H1/H3/H5 (independent, unlock the journey demo) → P3 → H2 → H4.

See also