From cb5634c80b479b02162536c1b5435c9e841ff388 Mon Sep 17 00:00:00 2001 From: Ryan Rouleau Date: Wed, 1 Jul 2026 16:26:21 -0700 Subject: [PATCH] feat(callbacks): allow registering multiple listener/customizer callbacks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make the additive callbacks accumulate instead of overwrite, so a wrapping package (e.g. a cloud connector) can register its own callback without displacing an application's. Follows the standard Python split by return semantics — fan-out for fire-and-forget listeners, layered merge for the value-returning customizer, single slot for the one that produces the reply. - VoiceChannel.on_inbound_call_twiml: now appends. All registered customizers run in registration order; each output is a per-call TwiML layer, later registrations winning per-field (same precedence as default_twiml_options → customizer). _build_twiml_options / _resolve_action_url take a list; action_url resolves across layers highest-priority-first. - TAC.on_interrupt, TAC.on_conversation_ended: now append. All registered listeners fire in registration order (fan-out; results are None). - TAC.on_message_ready: unchanged — stays single-slot, because it produces the reply TAC routes back to the channel and there can be only one responder (like a Flask route handler / sys.excepthook). Behavior change: registering twice now accumulates rather than replaces. Single registration (the overwhelmingly common case) is unaffected. Tests: multiple customizers layer per-field + action_url precedence across layers (voice); multiple interrupt/conversation-ended listeners all fire in order (TAC core). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/tac/channels/voice/channel.py | 55 ++++++++++++++++++++----------- src/tac/core/tac.py | 40 ++++++++++++++-------- tests/test_interrupt.py | 20 +++++++++++ tests/test_tac.py | 24 ++++++++++++++ tests/test_voice_channel.py | 51 ++++++++++++++++++++++++++++ 5 files changed, 157 insertions(+), 33 deletions(-) diff --git a/src/tac/channels/voice/channel.py b/src/tac/channels/voice/channel.py index 2615c0d..bfb72fa 100644 --- a/src/tac/channels/voice/channel.py +++ b/src/tac/channels/voice/channel.py @@ -79,7 +79,7 @@ def __init__( super().__init__(tac, memory_mode=config.memory_mode) self.config = config self.session_manager = config.session_manager - self._on_inbound_call_twiml: InboundCallTwiMLHandler | None = None + self._inbound_call_twiml_customizers: list[InboundCallTwiMLHandler] = [] self._websocket_manager = WebSocketManager() self._twilio_client: Client | None = None @@ -92,6 +92,13 @@ def on_inbound_call_twiml(self, callback: InboundCallTwiMLHandler) -> None: the callback explicitly sets override ``default_twiml_options`` and TAC defaults; unset fields fall through. + Multiple customizers may be registered: this method *appends* — it + does not replace. They run in registration order, each layered over + the previous (later registrations win per-field, same as the + ``default_twiml_options`` → customizer precedence). This lets a + wrapping package (e.g. a cloud connector) register its own customizer + without displacing an application's. + Example: ```python async def by_country(req: TwiMLRequest) -> TwiMLOptions: @@ -106,7 +113,7 @@ async def by_country(req: TwiMLRequest) -> TwiMLOptions: Outbound calls don't use this — pass per-call TwiML via ``InitiateVoiceConversationOptions.twiml_options`` directly. """ - self._on_inbound_call_twiml = callback + self._inbound_call_twiml_customizers.append(callback) def _resolve_websocket_url(self, action: str) -> str: """Resolve the public WebSocket URL from @@ -191,27 +198,32 @@ async def handle_incoming_call( """ websocket_url = self._resolve_websocket_url("handle_incoming_call") - customized: TwiMLOptions | None = None - if self._on_inbound_call_twiml is not None and twiml_request is not None: - customized = await self._on_inbound_call_twiml(twiml_request) + # Run every registered customizer in registration order; each output + # is one per-call layer, later ones winning per-field. + per_call_layers: list[TwiMLOptions] = [] + if twiml_request is not None: + for customizer in self._inbound_call_twiml_customizers: + per_call_layers.append(await customizer(twiml_request)) - merged = self._build_twiml_options(customized) + merged = self._build_twiml_options(per_call_layers) return twiml.generate_twiml(websocket_url, merged) - def _build_twiml_options(self, per_call: TwiMLOptions | None) -> TwiMLOptions: + def _build_twiml_options(self, per_call_layers: list[TwiMLOptions]) -> TwiMLOptions: """Layer TwiML options: TAC defaults → channel ``default_twiml_options`` - → ``per_call`` (customizer output for inbound, or - ``InitiateVoiceConversationOptions.twiml_options`` for outbound). + → ``per_call_layers`` in order (customizer outputs for inbound, or a + single ``InitiateVoiceConversationOptions.twiml_options`` for outbound). + + Later per-call layers win over earlier ones, per-field. """ merged = TwiMLOptions( welcome_greeting=DEFAULT_WELCOME_GREETING, conversation_configuration=self.tac.config.conversation_configuration_id, - action_url=self._resolve_action_url(per_call), + action_url=self._resolve_action_url(per_call_layers), ) if self.config.default_twiml_options is not None: self._overlay_fields(merged, self.config.default_twiml_options) - if per_call is not None: - self._overlay_fields(merged, per_call) + for layer in per_call_layers: + self._overlay_fields(merged, layer) return merged @staticmethod @@ -234,11 +246,11 @@ def _overlay_fields(target: TwiMLOptions, source: TwiMLOptions) -> None: continue setattr(target, field, getattr(source, field)) - def _resolve_action_url(self, customized: TwiMLOptions | None) -> str | None: + def _resolve_action_url(self, per_call_layers: list[TwiMLOptions]) -> str | None: """Resolve the TwiML ```` URL. Precedence (highest to lowest): - 1. customizer + 1. customizer layers (later registrations win over earlier ones) 2. channel ``default_twiml_options`` 3. Studio handoff (when ``studio_handoff_flow_sid`` is configured) 4. Channel default — derived from ``TACConfig.voice_public_domain`` @@ -256,8 +268,11 @@ def _resolve_action_url(self, customized: TwiMLOptions | None) -> str | None: from a customizer) or channel-wide. ``action_url`` left unset (not in ``model_fields_set``) falls through to the next layer. """ - if customized is not None and "action_url" in customized.model_fields_set: - return customized.action_url + # Highest-priority customizer that explicitly set action_url wins; scan + # the per-call layers from last-registered (highest) to first. + for layer in reversed(per_call_layers): + if "action_url" in layer.model_fields_set: + return layer.action_url if ( self.config.default_twiml_options is not None and "action_url" in self.config.default_twiml_options.model_fields_set @@ -526,10 +541,12 @@ async def initiate_outbound_conversation( from_number=mask_phone(from_number), ) - # Same layering as handle_incoming_call, minus the customizer + # Same layering as handle_incoming_call, minus the customizers # (customizers receive a TwiMLRequest from an inbound webhook; there - # is no equivalent for outbound). - merged = self._build_twiml_options(options.twiml_options) + # is no equivalent for outbound). Outbound has a single per-call layer: + # options.twiml_options, if provided. + per_call_layers = [options.twiml_options] if options.twiml_options is not None else [] + merged = self._build_twiml_options(per_call_layers) try: twiml_xml = twiml.generate_twiml(websocket_url, merged) diff --git a/src/tac/core/tac.py b/src/tac/core/tac.py index df51cdb..500665d 100644 --- a/src/tac/core/tac.py +++ b/src/tac/core/tac.py @@ -106,23 +106,24 @@ def __init__(self, config: TACConfig | dict[str, Any]): ) self.logger.info("Conversation Intelligence processor initialized") + # on_message_ready is single-slot on purpose: it produces the reply + # that TAC routes back to the channel, and there can only be one + # responder. on_interrupt / on_conversation_ended are fire-and-forget + # listeners, so they accumulate and all fire (see their setters). self._message_ready_callback: ( Callable[[str, ConversationSession, TACMemoryResponse | None], str | None] | Callable[[str, ConversationSession, TACMemoryResponse | None], Awaitable[str | None]] | None ) = None - self._interrupt_callback: ( + self._interrupt_callbacks: list[ Callable[[ConversationSession, Any], None] | Callable[[ConversationSession, Any], Awaitable[None]] - | None - ) = None + ] = [] - self._conversation_ended_callback: ( - Callable[[ConversationSession], None] - | Callable[[ConversationSession], Awaitable[None]] - | None - ) = None + self._conversation_ended_callbacks: list[ + Callable[[ConversationSession], None] | Callable[[ConversationSession], Awaitable[None]] + ] = [] def is_orchestrator_enabled(self) -> bool: """True if TAC is configured with Conversation Orchestrator (not relay-only mode).""" @@ -286,6 +287,11 @@ def on_interrupt( ) -> None: """Register callback invoked on user interrupt. + Multiple callbacks may be registered: this method *appends* — it does + not replace. All registered callbacks are invoked (in registration + order) on each interrupt. This lets a wrapping package register its own + listener alongside an application's. + Example: ```python def handle_interrupt(context: ConversationSession, interrupt_data: Any): @@ -299,7 +305,7 @@ def handle_interrupt(context: ConversationSession, interrupt_data: Any): Args: callback: Function to call with (context, interrupt_data). Supports sync and async. """ - self._interrupt_callback = callback + self._interrupt_callbacks.append(callback) def on_conversation_ended( self, @@ -309,6 +315,12 @@ def on_conversation_ended( ) -> None: """Register callback invoked when conversation ends. + Multiple callbacks may be registered: this method *appends* — it does + not replace. All registered callbacks are invoked (in registration + order) when a conversation ends. This lets a wrapping package clean up + its own state (e.g. a cached agent session) alongside an application's + listener. + Example: ```python def handle_conversation_ended(context: ConversationSession): @@ -322,7 +334,7 @@ def handle_conversation_ended(context: ConversationSession): Args: callback: Function to call with conversation context. Supports sync and async. """ - self._conversation_ended_callback = callback + self._conversation_ended_callbacks.append(callback) def register_partner_connector( self, @@ -406,8 +418,8 @@ def trigger_interrupt( conversation_context: Session containing conversation information. interrupt_data: Interrupt event data from voice channel. """ - if self._interrupt_callback: - result = self._interrupt_callback(conversation_context, interrupt_data) + for callback in self._interrupt_callbacks: + result = callback(conversation_context, interrupt_data) if inspect.isawaitable(result): try: asyncio.ensure_future(result) @@ -429,7 +441,7 @@ async def trigger_conversation_ended( Args: conversation_context: Session containing conversation information. """ - if self._conversation_ended_callback: - result = self._conversation_ended_callback(conversation_context) + for callback in self._conversation_ended_callbacks: + result = callback(conversation_context) if inspect.isawaitable(result): await result diff --git a/tests/test_interrupt.py b/tests/test_interrupt.py index f59fa3d..95278aa 100644 --- a/tests/test_interrupt.py +++ b/tests/test_interrupt.py @@ -308,3 +308,23 @@ async def async_handler_with_extra_arg( assert len(received) == 1 assert received[0] == "conv_wrapped" + + def test_multiple_interrupt_callbacks_all_fire(self) -> None: + """on_interrupt appends: every registered callback fires, in order.""" + tac = TAC(get_test_config()) + + calls: list[str] = [] + tac.on_interrupt(lambda ctx, data: calls.append("first")) + tac.on_interrupt(lambda ctx, data: calls.append("second")) + + context = ConversationSession(conversation_id="conv_multi", channel="voice") + interrupt = InterruptMessage( + type="interrupt", + utteranceUntilInterrupt="Hello", + durationUntilInterruptMs=1000, + ) + + tac.trigger_interrupt(context, interrupt) + + # Both fire, in registration order (fan-out — not replace). + assert calls == ["first", "second"] diff --git a/tests/test_tac.py b/tests/test_tac.py index 74cd0d3..c732977 100644 --- a/tests/test_tac.py +++ b/tests/test_tac.py @@ -214,3 +214,27 @@ def good_callback(user_message, context, memory_response): result = await tac.trigger_message_ready("test message", session, None) assert result is None + + @pytest.mark.asyncio + async def test_multiple_conversation_ended_callbacks_all_fire(self): + """on_conversation_ended appends: every registered callback fires, in order. + + Mixes a sync and an async callback to confirm both are awaited/run. + """ + from tac.models.session import ConversationSession + + tac = TAC(get_test_config()) + + calls = [] + tac.on_conversation_ended(lambda ctx: calls.append("sync")) + + async def async_cb(ctx): + calls.append("async") + + tac.on_conversation_ended(async_cb) + + session = ConversationSession(conversation_id="CH123", channel="sms") + await tac.trigger_conversation_ended(session) + + # Both fire, in registration order (fan-out — not replace). + assert calls == ["sync", "async"] diff --git a/tests/test_voice_channel.py b/tests/test_voice_channel.py index d28d563..4d3e801 100644 --- a/tests/test_voice_channel.py +++ b/tests/test_voice_channel.py @@ -1557,6 +1557,57 @@ async def test_default_options_action_url_none_suppresses(self) -> None: twiml = await channel.handle_incoming_call() assert "action=" not in twiml + @pytest.mark.asyncio + async def test_multiple_customizers_layer_in_order(self) -> None: + """on_inbound_call_twiml appends: multiple customizers all run and + layer per-field, later registrations winning. Lets a wrapping package + set one field while the app's customizer sets others.""" + from tac.channels.voice import VoiceChannelConfig + from tac.models.voice import TwiMLRequest + + async def app_customizer(req: TwiMLRequest) -> TwiMLOptions: + return TwiMLOptions(voice="en-US-Journey-D", welcome_greeting="App greeting") + + async def wrapper_customizer(req: TwiMLRequest) -> TwiMLOptions: + # A wrapping package sets only the greeting; voice falls through. + return TwiMLOptions(welcome_greeting="Wrapper greeting") + + tac = TAC(get_test_config()) + channel = VoiceChannel(tac, config=VoiceChannelConfig()) + channel.on_inbound_call_twiml(app_customizer) + channel.on_inbound_call_twiml(wrapper_customizer) + + twiml = await channel.handle_incoming_call(twiml_request=TwiMLRequest()) + + # app's voice (only it set voice) survives... + assert 'voice="en-US-Journey-D"' in twiml + # ...and the later-registered wrapper wins the field both set. + assert 'welcomeGreeting="Wrapper greeting"' in twiml + assert "App greeting" not in twiml + + @pytest.mark.asyncio + async def test_multiple_customizers_action_url_highest_priority_wins(self) -> None: + """action_url resolves across all customizer layers, later-registered + winning (exercises the reversed-scan in _resolve_action_url).""" + from tac.channels.voice import VoiceChannelConfig + from tac.models.voice import TwiMLRequest + + async def first(req: TwiMLRequest) -> TwiMLOptions: + return TwiMLOptions(action_url="https://first.example.com/end") + + async def second(req: TwiMLRequest) -> TwiMLOptions: + return TwiMLOptions(action_url="https://second.example.com/end") + + tac = TAC(get_test_config()) + channel = VoiceChannel(tac, config=VoiceChannelConfig()) + channel.on_inbound_call_twiml(first) + channel.on_inbound_call_twiml(second) + + twiml = await channel.handle_incoming_call(twiml_request=TwiMLRequest()) + + assert 'action="https://second.example.com/end"' in twiml + assert "first.example.com" not in twiml + class TestStaticTwiMLOptions: """VoiceChannelConfig.twiml_options applies to every call without a callback."""