refactor(trigger): share automation turn delivery

nanobot/agent/automation_turns.py

@@ -0,0 +1,145 @@
+"""Shared coordination for session-bound automation turns."""
+
+from __future__ import annotations
+
+import asyncio
+import dataclasses
+from collections.abc import Awaitable, Callable, Iterable
+
+from nanobot.bus.events import InboundMessage, OutboundMessage
+
+
+class AutomationTurnError(RuntimeError):
+    """Raised when an automation turn reaches the agent and finishes with an error."""
+
+
+async def publish_next_deferred_turn(
+    *,
+    deferred_queues: dict[str, list[InboundMessage]],
+    publish_inbound: Callable[[InboundMessage], Awaitable[None]],
+    session_key: str,
+) -> bool:
+    """Publish the next deferred automation turn for a session."""
+    queue = deferred_queues.get(session_key)
+    if not queue:
+        return False
+    msg = queue.pop(0)
+    if not queue:
+        deferred_queues.pop(session_key, None)
+    await publish_inbound(msg)
+    return True
+
+
+class AutomationTurnCoordinator:
+    """Manage automation turns without mixing them into live injections."""
+
+    def __init__(
+        self,
+        *,
+        publish_inbound: Callable[[InboundMessage], Awaitable[None]],
+        dispatch: Callable[[InboundMessage], Awaitable[object]],
+        is_running: Callable[[], bool],
+        turn_id: Callable[[InboundMessage], str | None],
+        pending_id: Callable[[InboundMessage], str | None],
+        should_defer_turn: Callable[[InboundMessage, str, Iterable[str]], bool],
+        missing_id_error: str,
+        duplicate_id_error: Callable[[str], str],
+        deferred_queues: dict[str, list[InboundMessage]] | None = None,
+    ) -> None:
+        self._publish_inbound = publish_inbound
+        self._dispatch = dispatch
+        self._is_running = is_running
+        self._turn_id = turn_id
+        self._pending_id = pending_id
+        self._should_defer_turn = should_defer_turn
+        self._missing_id_error = missing_id_error
+        self._duplicate_id_error = duplicate_id_error
+        self.deferred_queues = deferred_queues if deferred_queues is not None else {}
+        self._waiters: dict[str, asyncio.Future[OutboundMessage | None]] = {}
+        self._pending_messages_by_turn_id: dict[str, InboundMessage] = {}
+
+    async def submit(self, msg: InboundMessage) -> OutboundMessage | None:
+        """Submit an automation turn and wait for its session response."""
+        turn_id = self._turn_id(msg)
+        if not turn_id:
+            raise ValueError(self._missing_id_error)
+        if turn_id in self._waiters:
+            raise RuntimeError(self._duplicate_id_error(turn_id))
+
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[OutboundMessage | None] = loop.create_future()
+        self._waiters[turn_id] = future
+        self._pending_messages_by_turn_id[turn_id] = msg
+        try:
+            if self._is_running():
+                await self._publish_inbound(msg)
+            else:
+                await self._dispatch(msg)
+            try:
+                return await future
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                raise AutomationTurnError(str(exc) or exc.__class__.__name__) from exc
+        finally:
+            self._waiters.pop(turn_id, None)
+            self._pending_messages_by_turn_id.pop(turn_id, None)
+
+    def defer_if_active(
+        self,
+        msg: InboundMessage,
+        *,
+        session_key: str,
+        active_session_keys: Iterable[str],
+    ) -> bool:
+        """Defer an automation turn when its target session is already active."""
+        if not self._should_defer_turn(msg, session_key, active_session_keys):
+            return False
+        pending_msg = msg
+        if session_key != msg.session_key:
+            pending_msg = dataclasses.replace(
+                msg,
+                session_key_override=session_key,
+            )
+        self.deferred_queues.setdefault(session_key, []).append(pending_msg)
+        return True
+
+    def complete(
+        self,
+        msg: InboundMessage,
+        *,
+        response: OutboundMessage | None = None,
+        error: BaseException | None = None,
+    ) -> None:
+        turn_id = self._turn_id(msg)
+        if not turn_id:
+            return
+        future = self._waiters.get(turn_id)
+        if future is None or future.done():
+            return
+        if error is not None:
+            future.set_exception(error)
+        else:
+            future.set_result(response)
+
+    def pending_ids_for_session(self, session_key: str) -> set[str]:
+        """Return automation IDs that are waiting for or running in *session_key*."""
+        pending_ids: set[str] = set()
+        for msg in self.deferred_queues.get(session_key, []):
+            pending_id = self._pending_id(msg)
+            if pending_id:
+                pending_ids.add(pending_id)
+        for msg in self._pending_messages_by_turn_id.values():
+            if msg.session_key != session_key:
+                continue
+            pending_id = self._pending_id(msg)
+            if pending_id:
+                pending_ids.add(pending_id)
+        return pending_ids
+
+    async def publish_next_deferred(self, session_key: str) -> bool:
+        return await publish_next_deferred_turn(
+            deferred_queues=self.deferred_queues,
+            publish_inbound=self._publish_inbound,
+            session_key=session_key,
+        )

nanobot/agent/cron_turns.py

@@ -2,11 +2,10 @@
 
 from __future__ import annotations
 
-import asyncio
-import dataclasses
 from collections.abc import Awaitable, Callable, Iterable
 
-from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.agent.automation_turns import AutomationTurnCoordinator
+from nanobot.bus.events import InboundMessage
 from nanobot.cron.session_turns import (
     cron_run_id,
     cron_trigger,
@@ -14,7 +13,7 @@ from nanobot.cron.session_turns import (
 )
 
 
-class CronTurnCoordinator:
+class CronTurnCoordinator(AutomationTurnCoordinator):
     """Manage scheduled cron turns without mixing them into live injections."""
 
     def __init__(
@@ -23,116 +22,31 @@ class CronTurnCoordinator:
         publish_inbound: Callable[[InboundMessage], Awaitable[None]],
         dispatch: Callable[[InboundMessage], Awaitable[object]],
         is_running: Callable[[], bool],
+        deferred_queues: dict[str, list[InboundMessage]] | None = None,
     ) -> None:
-        self._publish_inbound = publish_inbound
-        self._dispatch = dispatch
-        self._is_running = is_running
-        self.deferred_queues: dict[str, list[InboundMessage]] = {}
-        self._waiters: dict[str, asyncio.Future[OutboundMessage | None]] = {}
-        self._pending_messages_by_run_id: dict[str, InboundMessage] = {}
-
-    async def submit(self, msg: InboundMessage) -> OutboundMessage | None:
-        """Submit a scheduled cron turn and wait for its session response."""
-        run_id = cron_run_id(msg.metadata)
-        if not run_id:
-            raise ValueError("cron turn metadata must include a run_id")
-        if run_id in self._waiters:
-            raise RuntimeError(f"cron run {run_id!r} is already pending")
-
-        loop = asyncio.get_running_loop()
-        future: asyncio.Future[OutboundMessage | None] = loop.create_future()
-        self._waiters[run_id] = future
-        self._pending_messages_by_run_id[run_id] = msg
-        try:
-            if self._is_running():
-                await self._publish_inbound(msg)
-            else:
-                await self._dispatch(msg)
-            return await future
-        finally:
-            self._waiters.pop(run_id, None)
-            self._pending_messages_by_run_id.pop(run_id, None)
-
-    def should_defer(
-        self,
-        msg: InboundMessage,
-        *,
-        session_key: str,
-        active_session_keys: Iterable[str],
-    ) -> bool:
-        return (
-            defer_cron_until_session_idle(msg.metadata)
-            and session_key in active_session_keys
+        super().__init__(
+            publish_inbound=publish_inbound,
+            dispatch=dispatch,
+            is_running=is_running,
+            turn_id=lambda msg: cron_run_id(msg.metadata),
+            pending_id=_cron_job_id,
+            should_defer_turn=_should_defer_cron_turn,
+            missing_id_error="cron turn metadata must include a run_id",
+            duplicate_id_error=lambda run_id: f"cron run {run_id!r} is already pending",
+            deferred_queues=deferred_queues,
         )
 
-    def defer_if_active(
-        self,
-        msg: InboundMessage,
-        *,
-        session_key: str,
-        active_session_keys: Iterable[str],
-    ) -> bool:
-        """Defer a cron turn when its target session is already active."""
-        if not self.should_defer(
-            msg,
-            session_key=session_key,
-            active_session_keys=active_session_keys,
-        ):
-            return False
-        pending_msg = msg
-        if session_key != msg.session_key:
-            pending_msg = dataclasses.replace(
-                msg,
-                session_key_override=session_key,
-            )
-        self.defer(session_key, pending_msg)
-        return True
-
-    def complete(
-        self,
-        msg: InboundMessage,
-        *,
-        response: OutboundMessage | None = None,
-        error: BaseException | None = None,
-    ) -> None:
-        run_id = cron_run_id(msg.metadata)
-        if not run_id:
-            return
-        future = self._waiters.get(run_id)
-        if future is None or future.done():
-            return
-        if error is not None:
-            future.set_exception(error)
-        else:
-            future.set_result(response)
-
-    def defer(self, session_key: str, msg: InboundMessage) -> None:
-        self.deferred_queues.setdefault(session_key, []).append(msg)
-
     def pending_job_ids_for_session(self, session_key: str) -> set[str]:
         """Return cron jobs that are waiting for or running in *session_key*."""
-        job_ids: set[str] = set()
-        for msg in self.deferred_queues.get(session_key, []):
-            job_id = _cron_job_id(msg)
-            if job_id:
-                job_ids.add(job_id)
-        for msg in self._pending_messages_by_run_id.values():
-            if msg.session_key != session_key:
-                continue
-            job_id = _cron_job_id(msg)
-            if job_id:
-                job_ids.add(job_id)
-        return job_ids
+        return self.pending_ids_for_session(session_key)
 
-    async def publish_next_deferred(self, session_key: str) -> bool:
-        queue = self.deferred_queues.get(session_key)
-        if not queue:
-            return False
-        msg = queue.pop(0)
-        if not queue:
-            self.deferred_queues.pop(session_key, None)
-        await self._publish_inbound(msg)
-        return True
+
+def _should_defer_cron_turn(
+    msg: InboundMessage,
+    session_key: str,
+    active_session_keys: Iterable[str],
+) -> bool:
+    return defer_cron_until_session_idle(msg.metadata) and session_key in active_session_keys
 
 
 def _cron_job_id(msg: InboundMessage) -> str | None:

nanobot/agent/loop.py

@@ -18,6 +18,7 @@ from loguru import logger
 from nanobot.agent import context as agent_context
 from nanobot.agent import model_presets as preset_helpers
 from nanobot.agent.autocompact import AutoCompact
+from nanobot.agent.automation_turns import publish_next_deferred_turn
 from nanobot.agent.context import ContextBuilder
 from nanobot.agent.cron_turns import CronTurnCoordinator
 from nanobot.agent.hook import AgentHook, CompositeHook
@@ -225,6 +226,7 @@ class AgentLoop:
         runtime_events: RuntimeEventBus | None = None,
         runtime_model_publisher: Callable[[str, str | None], None] | None = None,
         restart_mode: str = "auto",
+        local_trigger_store: Any | None = None,
     ):
         from nanobot.config.schema import ToolsConfig
 
@@ -272,6 +274,7 @@ class AgentLoop:
         ):
             self._image_generation_provider_configs["openrouter"] = image_generation_provider_config
         self.cron_service = cron_service
+        self.local_trigger_store = local_trigger_store
         self.restrict_to_workspace = restrict_to_workspace
         self.workspace_scopes = WorkspaceScopeResolver(
             default_workspace=workspace,
@@ -316,15 +319,22 @@ class AgentLoop:
         # When a session has an active task, new messages for that session
         # are routed here instead of creating a new task.
         self._pending_queues: dict[str, asyncio.Queue] = {}
+        self._deferred_automation_turns: dict[str, list[InboundMessage]] = {}
         self._cron_turns = CronTurnCoordinator(
             publish_inbound=self.bus.publish_inbound,
             dispatch=self._dispatch,
             is_running=lambda: self._running,
+            deferred_queues=self._deferred_automation_turns,
         )
         self._local_trigger_turns = LocalTriggerTurnCoordinator(
             publish_inbound=self.bus.publish_inbound,
             dispatch=self._dispatch,
             is_running=lambda: self._running,
+            deferred_queues=self._deferred_automation_turns,
+        )
+        self._automation_turn_coordinators = (
+            ("cron", self._cron_turns),
+            ("local trigger", self._local_trigger_turns),
         )
         # NANOBOT_MAX_CONCURRENT_REQUESTS: <=0 means unlimited; default 3.
         _max = int(os.environ.get("NANOBOT_MAX_CONCURRENT_REQUESTS", "3"))
@@ -605,9 +615,11 @@ class AgentLoop:
         return self._local_trigger_turns.pending_trigger_ids_for_session(session_key)
 
     async def _publish_next_deferred_automation_turn(self, session_key: str) -> None:
-        if await self._cron_turns.publish_next_deferred(session_key):
-            return
-        await self._local_trigger_turns.publish_next_deferred(session_key)
+        await publish_next_deferred_turn(
+            deferred_queues=self._deferred_automation_turns,
+            publish_inbound=self.bus.publish_inbound,
+            session_key=session_key,
+        )
 
     def _persist_user_message_early(
         self,
@@ -938,25 +950,21 @@ class AgentLoop:
                         self.commands.dispatch_priority,
                     )
                     continue
-                if self._cron_turns.defer_if_active(
-                    msg,
-                    session_key=effective_key,
-                    active_session_keys=self._pending_queues.keys(),
-                ):
-                    logger.info(
-                        "Deferred cron turn for active session {}",
-                        effective_key,
-                    )
-                    continue
-                if self._local_trigger_turns.defer_if_active(
-                    msg,
-                    session_key=effective_key,
-                    active_session_keys=self._pending_queues.keys(),
-                ):
-                    logger.info(
-                        "Deferred local trigger turn for active session {}",
-                        effective_key,
-                    )
+                deferred = False
+                for label, coordinator in self._automation_turn_coordinators:
+                    if coordinator.defer_if_active(
+                        msg,
+                        session_key=effective_key,
+                        active_session_keys=self._pending_queues.keys(),
+                    ):
+                        logger.info(
+                            "Deferred {} turn for active session {}",
+                            label,
+                            effective_key,
+                        )
+                        deferred = True
+                        break
+                if deferred:
                     continue
                 # If this session already has an active pending queue (i.e. a task
                 # is processing this session), route the message there for mid-turn
@@ -1079,17 +1087,11 @@ class AgentLoop:
                             session_key=session_key,
                             metadata=msg.metadata,
                         )
-                    self._cron_turns.complete(msg, response=response)
-                    self._local_trigger_turns.complete(msg, response=response)
+                    for _, coordinator in self._automation_turn_coordinators:
+                        coordinator.complete(msg, response=response)
                 except asyncio.CancelledError:
-                    self._cron_turns.complete(
-                        msg,
-                        error=asyncio.CancelledError(),
-                    )
-                    self._local_trigger_turns.complete(
-                        msg,
-                        error=asyncio.CancelledError(),
-                    )
+                    for _, coordinator in self._automation_turn_coordinators:
+                        coordinator.complete(msg, error=asyncio.CancelledError())
                     logger.info("Task cancelled for session {}", session_key)
                     # Preserve partial context from the interrupted turn so
                     # the user does not lose tool results and assistant
@@ -1128,8 +1130,8 @@ class AgentLoop:
                             session_key=session_key,
                             metadata=msg.metadata,
                         )
-                    self._cron_turns.complete(msg, error=exc)
-                    self._local_trigger_turns.complete(msg, error=exc)
+                    for _, coordinator in self._automation_turn_coordinators:
+                        coordinator.complete(msg, error=exc)
                 finally:
                     # Drain any messages still in the pending queue and re-publish
                     # them to the bus so they are processed as fresh inbound messages
@@ -1481,7 +1483,7 @@ class AgentLoop:
             # message.  Mark messages with _command so get_history can filter
             # them out of LLM context.  /new is excluded because it
             # intentionally clears the session.
-            if raw.lower() != "/new":
+            if cmd_ctx.raw.lower() != "/new":
                 ctx.user_persisted_early = self._persist_user_message_early(
                     ctx.msg, ctx.session, _command=True
                 )

nanobot/cli/commands.py

@@ -948,14 +948,13 @@ def _run_gateway(
         runtime_events=runtime_events,
         provider_signature=provider_snapshot.signature,
         hooks=[TokenUsageHook(timezone_name=config.agents.defaults.timezone)],
+        local_trigger_store=trigger_store,
     )
     WebuiTurnCoordinator(
         bus=bus,
         sessions=session_manager,
         schedule_background=lambda coro: agent._schedule_background(coro),
     ).subscribe(runtime_events)
-    agent.local_trigger_store = trigger_store
-
     from nanobot.bus.events import OutboundMessage
     from nanobot.session.keys import session_key_for_channel
 
@@ -1302,7 +1301,6 @@ def _run_gateway(
                 asyncio.create_task(
                     run_local_trigger_queue(
                         store=trigger_store,
-                        bus=bus,
                         submit_turn=getattr(agent, "submit_local_trigger_turn", None),
                     ),
                     name="nanobot-local-triggers",

nanobot/command/builtin.py

@@ -753,11 +753,18 @@ async def cmd_trigger(ctx: CommandContext) -> OutboundMessage:
     if store is None:
         store = LocalTriggerStore(workspace)
 
+    from nanobot.session.keys import UNIFIED_SESSION_KEY
+
+    session_key = (
+        ctx.msg.session_key
+        if ctx.key == UNIFIED_SESSION_KEY
+        else ctx.key
+    )
     trigger = store.create(
         name=name,
         channel=ctx.msg.channel,
         chat_id=ctx.msg.chat_id,
-        session_key=ctx.key,
+        session_key=session_key,
         sender_id="trigger",
         origin_metadata=dict(ctx.msg.metadata or {}),
     )

nanobot/triggers/local_runner.py

@@ -9,8 +9,8 @@ from typing import Any
 
 from loguru import logger
 
+from nanobot.agent.automation_turns import AutomationTurnError
 from nanobot.bus.events import InboundMessage, OutboundMessage
-from nanobot.bus.queue import MessageBus
 from nanobot.triggers.local_session_turns import LOCAL_TRIGGER_META
 from nanobot.triggers.local_store import LocalTriggerStore
 from nanobot.triggers.local_types import LocalTrigger, TriggerDelivery
@@ -20,14 +20,13 @@ from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN
 async def run_local_trigger_queue(
     *,
     store: LocalTriggerStore,
-    bus: MessageBus | None = None,
     submit_turn: Callable[[InboundMessage], Awaitable[OutboundMessage | None]] | None = None,
     poll_interval_s: float = 0.5,
     batch_size: int = 20,
 ) -> None:
-    """Poll local trigger deliveries and publish them as normal inbound messages."""
-    if bus is None and submit_turn is None:
-        raise ValueError("run_local_trigger_queue requires bus or submit_turn")
+    """Poll local trigger deliveries and submit them as session turns."""
+    if submit_turn is None:
+        raise ValueError("run_local_trigger_queue requires submit_turn")
     logger.info("Local trigger queue started")
     recovered = store.recover_processing_deliveries()
     if recovered:
@@ -46,7 +45,6 @@ async def run_local_trigger_queue(
                 await _deliver_delivery(
                     store,
                     delivery,
-                    bus=bus,
                     submit_turn=submit_turn,
                 )
                 store.complete_delivery(delivery)
@@ -67,6 +65,21 @@ async def run_local_trigger_queue(
                     delivery.trigger_id,
                     exc,
                 )
+            except AutomationTurnError as exc:
+                error = str(exc) or exc.__class__.__name__
+                store.record_delivery(
+                    delivery.trigger_id,
+                    status="error",
+                    error=error,
+                    run_at_ms=delivery.created_at_ms,
+                )
+                store.complete_delivery(delivery)
+                logger.warning(
+                    "Trigger: delivery {} for {} reached the agent but failed: {}",
+                    delivery.id,
+                    delivery.trigger_id,
+                    error,
+                )
             except Exception as exc:
                 error = str(exc) or exc.__class__.__name__
                 retried = store.retry_delivery(delivery, error)
@@ -92,8 +105,7 @@ async def _deliver_delivery(
     store: LocalTriggerStore,
     delivery: TriggerDelivery,
     *,
-    bus: MessageBus | None,
-    submit_turn: Callable[[InboundMessage], Awaitable[OutboundMessage | None]] | None,
+    submit_turn: Callable[[InboundMessage], Awaitable[OutboundMessage | None]],
 ) -> None:
     trigger = store.get(delivery.trigger_id)
     if trigger is None:
@@ -109,12 +121,7 @@ async def _deliver_delivery(
         metadata=_delivery_metadata(trigger, delivery),
         session_key_override=trigger.session_key,
     )
-    if submit_turn is not None:
-        await submit_turn(msg)
-    else:
-        if bus is None:
-            raise RuntimeError("bus unavailable for local trigger delivery")
-        await bus.publish_inbound(msg)
+    await submit_turn(msg)
     store.record_delivery(
         trigger.id,
         status="ok",
@@ -129,6 +136,7 @@ def _delivery_metadata(trigger: LocalTrigger, delivery: TriggerDelivery) -> dict
         "trigger_name": trigger.name,
         "delivery_id": delivery.id,
         "created_at_ms": delivery.created_at_ms,
+        "persist_content": _history_content(trigger, delivery),
     }
     if trigger.channel == "websocket":
         metadata.pop(WEBUI_TURN_METADATA_KEY, None)
@@ -138,3 +146,8 @@ def _delivery_metadata(trigger: LocalTrigger, delivery: TriggerDelivery) -> dict
             source["label"] = trigger.name
         metadata[WEBUI_MESSAGE_SOURCE_METADATA_KEY] = source
     return metadata
+
+
+def _history_content(trigger: LocalTrigger, delivery: TriggerDelivery) -> str:
+    label = trigger.name.strip() if trigger.name else trigger.id
+    return f"Local trigger received: {label}\n\n{delivery.content}"

nanobot/triggers/local_session_turns.py

@@ -14,6 +14,9 @@ LOCAL_TRIGGER_META = "_local_trigger"
 
 
 def _local_trigger_history_text(trigger: Mapping[str, Any]) -> str:
+    persist_content = trigger.get("persist_content")
+    if isinstance(persist_content, str) and persist_content.strip():
+        return persist_content
     name = trigger.get("trigger_name")
     trigger_id = trigger.get("trigger_id")
     label = name if isinstance(name, str) and name.strip() else trigger_id

nanobot/triggers/local_turns.py

@@ -2,15 +2,14 @@
 
 from __future__ import annotations
 
-import asyncio
-import dataclasses
 from collections.abc import Awaitable, Callable, Iterable
 
-from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.agent.automation_turns import AutomationTurnCoordinator
+from nanobot.bus.events import InboundMessage
 from nanobot.triggers.local_session_turns import local_trigger, local_trigger_delivery_id
 
 
-class LocalTriggerTurnCoordinator:
+class LocalTriggerTurnCoordinator(AutomationTurnCoordinator):
     """Manage local trigger turns without mixing them into live injections."""
 
     def __init__(
@@ -19,113 +18,33 @@ class LocalTriggerTurnCoordinator:
         publish_inbound: Callable[[InboundMessage], Awaitable[None]],
         dispatch: Callable[[InboundMessage], Awaitable[object]],
         is_running: Callable[[], bool],
+        deferred_queues: dict[str, list[InboundMessage]] | None = None,
     ) -> None:
-        self._publish_inbound = publish_inbound
-        self._dispatch = dispatch
-        self._is_running = is_running
-        self.deferred_queues: dict[str, list[InboundMessage]] = {}
-        self._waiters: dict[str, asyncio.Future[OutboundMessage | None]] = {}
-        self._pending_messages_by_delivery_id: dict[str, InboundMessage] = {}
-
-    async def submit(self, msg: InboundMessage) -> OutboundMessage | None:
-        """Submit a local trigger turn and wait for its session response."""
-        delivery_id = local_trigger_delivery_id(msg.metadata)
-        if not delivery_id:
-            raise ValueError("local trigger turn metadata must include a delivery_id")
-        if delivery_id in self._waiters:
-            raise RuntimeError(f"local trigger delivery {delivery_id!r} is already pending")
-
-        loop = asyncio.get_running_loop()
-        future: asyncio.Future[OutboundMessage | None] = loop.create_future()
-        self._waiters[delivery_id] = future
-        self._pending_messages_by_delivery_id[delivery_id] = msg
-        try:
-            if self._is_running():
-                await self._publish_inbound(msg)
-            else:
-                await self._dispatch(msg)
-            return await future
-        finally:
-            self._waiters.pop(delivery_id, None)
-            self._pending_messages_by_delivery_id.pop(delivery_id, None)
-
-    def should_defer(
-        self,
-        msg: InboundMessage,
-        *,
-        session_key: str,
-        active_session_keys: Iterable[str],
-    ) -> bool:
-        return local_trigger(msg.metadata) is not None and session_key in active_session_keys
-
-    def defer_if_active(
-        self,
-        msg: InboundMessage,
-        *,
-        session_key: str,
-        active_session_keys: Iterable[str],
-    ) -> bool:
-        """Defer a local trigger turn when its target session is already active."""
-        if not self.should_defer(
-            msg,
-            session_key=session_key,
-            active_session_keys=active_session_keys,
-        ):
-            return False
-        pending_msg = msg
-        if session_key != msg.session_key:
-            pending_msg = dataclasses.replace(
-                msg,
-                session_key_override=session_key,
-            )
-        self.defer(session_key, pending_msg)
-        return True
-
-    def complete(
-        self,
-        msg: InboundMessage,
-        *,
-        response: OutboundMessage | None = None,
-        error: BaseException | None = None,
-    ) -> None:
-        delivery_id = local_trigger_delivery_id(msg.metadata)
-        if not delivery_id:
-            return
-        future = self._waiters.get(delivery_id)
-        if future is None or future.done():
-            return
-        if error is not None:
-            future.set_exception(error)
-        else:
-            future.set_result(response)
-
-    def defer(self, session_key: str, msg: InboundMessage) -> None:
-        self.deferred_queues.setdefault(session_key, []).append(msg)
+        super().__init__(
+            publish_inbound=publish_inbound,
+            dispatch=dispatch,
+            is_running=is_running,
+            turn_id=lambda msg: local_trigger_delivery_id(msg.metadata),
+            pending_id=_local_trigger_id,
+            should_defer_turn=_should_defer_local_trigger_turn,
+            missing_id_error="local trigger turn metadata must include a delivery_id",
+            duplicate_id_error=lambda delivery_id: (
+                f"local trigger delivery {delivery_id!r} is already pending"
+            ),
+            deferred_queues=deferred_queues,
+        )
 
     def pending_trigger_ids_for_session(self, session_key: str) -> set[str]:
         """Return local triggers waiting for or running in *session_key*."""
-        trigger_ids: set[str] = set()
-        for msg in self.deferred_queues.get(session_key, []):
-            trigger_id = _local_trigger_id(msg)
-            if trigger_id:
-                trigger_ids.add(trigger_id)
-        for msg in self._pending_messages_by_delivery_id.values():
-            if msg.session_key != session_key:
-                continue
-            trigger_id = _local_trigger_id(msg)
-            if trigger_id:
-                trigger_ids.add(trigger_id)
-        return trigger_ids
+        return self.pending_ids_for_session(session_key)
 
-    async def publish_next_deferred(self, session_key: str) -> bool:
-        queue = self.deferred_queues.get(session_key)
-        if not queue:
-            return False
-        msg = queue.pop(0)
-        if not queue:
-            self.deferred_queues.pop(session_key, None)
-        await self._publish_inbound(msg)
-        return True
+
+def _should_defer_local_trigger_turn(
+    msg: InboundMessage,
+    session_key: str,
+    active_session_keys: Iterable[str],
+) -> bool:
+    return local_trigger(msg.metadata) is not None and session_key in active_session_keys
 
 
 def _local_trigger_id(msg: InboundMessage) -> str | None: