fix(trigger): defer local triggers until session idle

docs/chat-commands.md

@@ -79,13 +79,16 @@ Replace `"Review PR #4502"` with the message you want nanobot to receive. The
 trigger is bound to the session where it was created, so the message goes back
 to that same chat. Keep `nanobot gateway` running so trigger messages can be
 delivered. The trigger message starts an automation turn; it is not shown in
-the chat as a normal user message.
+the chat as a normal user message. If that session is already running a turn,
+the trigger waits until the session is idle instead of being injected into the
+active turn.
 
-Trigger deliveries are stored in the workspace until the gateway consumes them.
-If the gateway exits after claiming a delivery but before marking it complete,
-the next gateway start requeues that delivery. This is an at-least-once local
-queue: a delivery may run more than once if the process exits at the wrong time,
-so external scripts should make repeated trigger messages safe.
+Trigger deliveries are stored in the workspace until their linked agent turn
+finishes successfully. If the gateway exits after claiming a delivery but before
+the turn completes, the next gateway start requeues that delivery. This is an
+at-least-once local queue: a delivery may run more than once if the process
+exits at the wrong time, so external scripts should make repeated trigger
+messages safe.
 
 For longer or generated content, omit the message argument and pipe stdin:
 

docs/cli-reference.md

@@ -136,11 +136,13 @@ Keep `nanobot gateway` running so the message can be delivered to the linked
 chat/session.
 
 The command writes to a workspace-local durable queue. If `nanobot gateway` is
-not running yet, the message waits in that workspace. If the gateway exits after
-claiming a delivery but before completing it, the next gateway start requeues
-that delivery. The queue is at-least-once, not exactly-once, so the same message
-can be delivered again after an interrupted process. Run one gateway consumer
-per workspace; this local queue is not a distributed multi-consumer queue.
+not running yet, the message waits in that workspace. If the target session is
+already running a turn, the trigger waits for that session to become idle. If the
+gateway exits after claiming a delivery but before the linked turn completes,
+the next gateway start requeues that delivery. The queue is at-least-once, not
+exactly-once, so the same message can be delivered again after an interrupted
+process. Run one gateway consumer per workspace; this local queue is not a
+distributed multi-consumer queue.
 
 Use stdin when another local process generates the message:
 

docs/concepts.md

@@ -148,8 +148,9 @@ schedule. Create one from the target chat with `/trigger <name>`, then call
 `nanobot trigger <id> "<message>"` when a local script or external service wants
 nanobot to respond in that session. Webhook servers, third-party auth, and
 event-to-message formatting stay outside nanobot. Trigger deliveries are stored
-in the workspace until the gateway consumes them and are requeued on gateway
-restart if processing was interrupted. Delivery is at-least-once, so external
+in the workspace until the linked agent turn finishes successfully. If the
+target session is busy, the trigger waits until that session is idle instead of
+being injected into the active turn. Delivery is at-least-once, so external
 systems should tolerate repeated trigger messages.
 
 ## Where to Go Next

docs/webui.md

@@ -133,9 +133,11 @@ that webhook/service outside nanobot and have it call the trigger command with
 the final message.
 
 Trigger deliveries use the same workspace as the gateway. They survive gateway
-restarts and are requeued if the process exits before marking a delivery
-complete. This is an at-least-once local queue, so repeated delivery is possible
-after an interrupted process.
+restarts and are requeued if the process exits before the linked turn completes.
+If the linked session is already running a turn, the local trigger waits until
+that session is idle instead of being injected into the active turn. This is an
+at-least-once local queue, so repeated delivery is possible after an interrupted
+process.
 
 For recurring background checks that should stay quiet unless there is something
 useful to report, use the protected heartbeat job by editing `HEARTBEAT.md`

nanobot/agent/cron_turns.py

@@ -124,14 +124,15 @@ class CronTurnCoordinator:
                 job_ids.add(job_id)
         return job_ids
 
-    async def publish_next_deferred(self, session_key: str) -> None:
+    async def publish_next_deferred(self, session_key: str) -> bool:
         queue = self.deferred_queues.get(session_key)
         if not queue:
-            return
+            return False
         msg = queue.pop(0)
         if not queue:
             self.deferred_queues.pop(session_key, None)
         await self._publish_inbound(msg)
+        return True
 
 
 def _cron_job_id(msg: InboundMessage) -> str | None:

nanobot/agent/loop.py

@@ -67,6 +67,7 @@ from nanobot.session.manager import (
     SessionManager,
     replay_max_messages_for_context,
 )
+from nanobot.triggers.local_turns import LocalTriggerTurnCoordinator
 from nanobot.utils.document import extract_documents, reference_non_image_attachments
 from nanobot.utils.helpers import image_placeholder_text
 from nanobot.utils.helpers import truncate_text as truncate_text_fn
@@ -320,6 +321,11 @@ class AgentLoop:
             dispatch=self._dispatch,
             is_running=lambda: self._running,
         )
+        self._local_trigger_turns = LocalTriggerTurnCoordinator(
+            publish_inbound=self.bus.publish_inbound,
+            dispatch=self._dispatch,
+            is_running=lambda: self._running,
+        )
         # NANOBOT_MAX_CONCURRENT_REQUESTS: <=0 means unlimited; default 3.
         _max = int(os.environ.get("NANOBOT_MAX_CONCURRENT_REQUESTS", "3"))
         self._concurrency_gate: asyncio.Semaphore | None = (
@@ -589,9 +595,20 @@ class AgentLoop:
     async def submit_cron_turn(self, msg: InboundMessage) -> OutboundMessage | None:
         return await self._cron_turns.submit(msg)
 
+    async def submit_local_trigger_turn(self, msg: InboundMessage) -> OutboundMessage | None:
+        return await self._local_trigger_turns.submit(msg)
+
     def pending_cron_job_ids_for_session(self, session_key: str) -> set[str]:
         return self._cron_turns.pending_job_ids_for_session(session_key)
 
+    def pending_local_trigger_ids_for_session(self, session_key: str) -> set[str]:
+        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)
+
     def _persist_user_message_early(
         self,
         msg: InboundMessage,
@@ -931,6 +948,16 @@ class AgentLoop:
                         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,
+                    )
+                    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
                 # injection instead of creating a competing task.
@@ -1053,11 +1080,16 @@ class AgentLoop:
                             metadata=msg.metadata,
                         )
                     self._cron_turns.complete(msg, response=response)
+                    self._local_trigger_turns.complete(msg, response=response)
                 except asyncio.CancelledError:
                     self._cron_turns.complete(
                         msg,
                         error=asyncio.CancelledError(),
                     )
+                    self._local_trigger_turns.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
@@ -1097,6 +1129,7 @@ class AgentLoop:
                             metadata=msg.metadata,
                         )
                     self._cron_turns.complete(msg, error=exc)
+                    self._local_trigger_turns.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
@@ -1127,14 +1160,14 @@ class AgentLoop:
                             msg, session_key, "idle"
                         )
                         self._runtime_events().clear_turn(session_key)
-                    await self._cron_turns.publish_next_deferred(session_key)
+                    await self._publish_next_deferred_automation_turn(session_key)
         finally:
             if pending is None:
                 await self._runtime_events().run_status_changed(
                     msg, session_key, "idle"
                 )
                 self._runtime_events().clear_turn(session_key)
-                await self._cron_turns.publish_next_deferred(session_key)
+                await self._publish_next_deferred_automation_turn(session_key)
 
     async def close_mcp(self) -> None:
         """Drain pending background archives, then close MCP connections."""

nanobot/cli/commands.py

@@ -1295,7 +1295,11 @@ def _run_gateway(
                 asyncio.create_task(agent.run(), name="nanobot-agent-loop"),
                 asyncio.create_task(channels.start_all(), name="nanobot-channels"),
                 asyncio.create_task(
-                    run_local_trigger_queue(store=trigger_store, bus=bus),
+                    run_local_trigger_queue(
+                        store=trigger_store,
+                        bus=bus,
+                        submit_turn=getattr(agent, "submit_local_trigger_turn", None),
+                    ),
                     name="nanobot-local-triggers",
                 ),
             ]

nanobot/triggers/local_runner.py

@@ -4,11 +4,12 @@ from __future__ import annotations
 
 import asyncio
 import uuid
+from collections.abc import Awaitable, Callable
 from typing import Any
 
 from loguru import logger
 
-from nanobot.bus.events import InboundMessage
+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
@@ -19,11 +20,14 @@ from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN
 async def run_local_trigger_queue(
     *,
     store: LocalTriggerStore,
-    bus: MessageBus,
+    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")
     logger.info("Local trigger queue started")
     recovered = store.recover_processing_deliveries()
     if recovered:
@@ -39,7 +43,12 @@ async def run_local_trigger_queue(
 
         for delivery in deliveries:
             try:
-                await _publish_delivery(store, bus, delivery)
+                await _deliver_delivery(
+                    store,
+                    delivery,
+                    bus=bus,
+                    submit_turn=submit_turn,
+                )
                 store.complete_delivery(delivery)
             except asyncio.CancelledError as exc:
                 store.retry_delivery(delivery, str(exc) or exc.__class__.__name__)
@@ -79,10 +88,12 @@ class _TerminalDeliveryError(RuntimeError):
     pass
 
 
-async def _publish_delivery(
+async def _deliver_delivery(
     store: LocalTriggerStore,
-    bus: MessageBus,
     delivery: TriggerDelivery,
+    *,
+    bus: MessageBus | None,
+    submit_turn: Callable[[InboundMessage], Awaitable[OutboundMessage | None]] | None,
 ) -> None:
     trigger = store.get(delivery.trigger_id)
     if trigger is None:
@@ -90,8 +101,7 @@ async def _publish_delivery(
     if not trigger.enabled:
         raise _TerminalDeliveryError("trigger is disabled")
 
-    await bus.publish_inbound(
-        InboundMessage(
+    msg = InboundMessage(
         channel=trigger.channel,
         sender_id=trigger.sender_id,
         chat_id=trigger.chat_id,
@@ -99,7 +109,12 @@ async def _publish_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)
     store.record_delivery(
         trigger.id,
         status="ok",

nanobot/triggers/local_session_turns.py

@@ -41,6 +41,14 @@ def local_trigger(metadata: Mapping[str, Any] | None) -> dict[str, Any] | None:
     return automation_trigger(metadata, LOCAL_TRIGGER_AUTOMATION_SPEC)
 
 
+def local_trigger_delivery_id(metadata: Mapping[str, Any] | None) -> str | None:
+    trigger = local_trigger(metadata)
+    if not trigger:
+        return None
+    value = trigger.get("delivery_id")
+    return value if isinstance(value, str) and value else None
+
+
 def local_trigger_history_overrides(
     metadata: Mapping[str, Any] | None,
 ) -> tuple[str | None, dict[str, Any]]:

nanobot/triggers/local_turns.py

@@ -0,0 +1,136 @@
+"""Coordination for local trigger turns."""
+
+from __future__ import annotations
+
+import asyncio
+import dataclasses
+from collections.abc import Awaitable, Callable, Iterable
+
+from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.triggers.local_session_turns import local_trigger, local_trigger_delivery_id
+
+
+class LocalTriggerTurnCoordinator:
+    """Manage local trigger 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],
+    ) -> 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)
+
+    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
+
+    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 _local_trigger_id(msg: InboundMessage) -> str | None:
+    trigger = local_trigger(msg.metadata)
+    if not trigger:
+        return None
+    value = trigger.get("trigger_id")
+    return value if isinstance(value, str) and value else None

tests/agent/test_runner_injections.py

@@ -730,6 +730,56 @@ async def test_cron_turn_deferred_while_session_active(tmp_path):
     assert loop.pending_cron_job_ids_for_session(session_key) == set()
 
 
+@pytest.mark.asyncio
+async def test_local_trigger_turn_deferred_while_session_active(tmp_path):
+    """Local trigger turns wait for the active session instead of becoming injections."""
+    from nanobot.bus.events import InboundMessage
+    from nanobot.triggers.local_session_turns import LOCAL_TRIGGER_META
+
+    loop = _make_loop(tmp_path)
+    loop._dispatch = AsyncMock()  # type: ignore[method-assign]
+
+    session_key = "websocket:chat-1"
+    pending = asyncio.Queue(maxsize=20)
+    loop._pending_queues[session_key] = pending
+
+    run_task = asyncio.create_task(loop.run())
+    msg = InboundMessage(
+        channel="websocket",
+        sender_id="trigger",
+        chat_id="chat-1",
+        content="review failed CI",
+        metadata={
+            LOCAL_TRIGGER_META: {
+                "trigger_id": "trg_123",
+                "trigger_name": "CI review",
+                "delivery_id": "tdl_123",
+            },
+        },
+        session_key_override=session_key,
+    )
+    await loop.bus.publish_inbound(msg)
+
+    for _ in range(20):
+        if loop._local_trigger_turns.deferred_queues.get(session_key):
+            break
+        await asyncio.sleep(0.05)
+
+    loop.stop()
+    await asyncio.wait_for(run_task, timeout=2)
+
+    assert pending.empty()
+    assert loop._dispatch.await_count == 0
+    assert loop._local_trigger_turns.deferred_queues[session_key] == [msg]
+    assert loop.pending_local_trigger_ids_for_session(session_key) == {"trg_123"}
+
+    assert await loop._local_trigger_turns.publish_next_deferred(session_key) is True
+    queued = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=0.5)
+    assert queued is msg
+    assert session_key not in loop._local_trigger_turns.deferred_queues
+    assert loop.pending_local_trigger_ids_for_session(session_key) == set()
+
+
 @pytest.mark.asyncio
 async def test_submitted_cron_turn_reports_pending_until_completed(tmp_path):
     """Bound cron jobs remain marked pending while their session turn is in flight."""
@@ -766,6 +816,48 @@ async def test_submitted_cron_turn_reports_pending_until_completed(tmp_path):
     assert loop.pending_cron_job_ids_for_session(session_key) == set()
 
 
+@pytest.mark.asyncio
+async def test_submitted_local_trigger_turn_reports_pending_until_completed(tmp_path):
+    """Local triggers remain marked pending while their session turn is in flight."""
+    from nanobot.bus.events import InboundMessage, OutboundMessage
+    from nanobot.triggers.local_session_turns import LOCAL_TRIGGER_META
+
+    loop = _make_loop(tmp_path)
+    loop._running = True
+
+    session_key = "websocket:chat-1"
+    msg = InboundMessage(
+        channel="websocket",
+        sender_id="trigger",
+        chat_id="chat-1",
+        content="review failed CI",
+        metadata={
+            LOCAL_TRIGGER_META: {
+                "trigger_id": "trg_123",
+                "trigger_name": "CI review",
+                "delivery_id": "tdl_123",
+            },
+        },
+        session_key_override=session_key,
+    )
+
+    submit_task = asyncio.create_task(loop.submit_local_trigger_turn(msg))
+    queued = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=0.5)
+
+    assert queued is msg
+    assert loop.pending_local_trigger_ids_for_session(session_key) == {"trg_123"}
+
+    response = OutboundMessage(
+        channel="websocket",
+        chat_id="chat-1",
+        content="done",
+    )
+    loop._local_trigger_turns.complete(msg, response=response)
+
+    assert await asyncio.wait_for(submit_task, timeout=0.5) is response
+    assert loop.pending_local_trigger_ids_for_session(session_key) == set()
+
+
 @pytest.mark.asyncio
 async def test_pending_queue_preserves_overflow_for_next_injection_cycle(tmp_path):
     """Pending queue should leave overflow messages queued for later drains."""

tests/cli/test_commands.py

@@ -1986,6 +1986,118 @@ def test_gateway_bound_cron_runs_as_session_turn(
     assert msg.metadata["thread_id"] == "om_root123"
 
 
+def test_gateway_local_trigger_queue_submits_agent_turns(
+    monkeypatch,
+    tmp_path: Path,
+) -> None:
+    config = Config()
+    config.agents.defaults.workspace = str(tmp_path / "config-workspace")
+    config.agents.defaults.dream.enabled = False
+    config.gateway.heartbeat.enabled = False
+    bus = MagicMock()
+    seen: dict[str, object] = {}
+
+    _patch_cli_command_runtime(
+        monkeypatch,
+        config,
+        message_bus=lambda: bus,
+        session_manager=lambda _workspace: _FakeSessionManager(),
+        cron_service=lambda _store_path: _FakeCronService(),
+    )
+
+    class _FakeMemory:
+        def get_latest_cursor(self) -> int:
+            return 0
+
+        def get_last_dream_cursor(self) -> int:
+            return 0
+
+        def set_last_dream_cursor(self, _cursor: int) -> None:
+            return None
+
+    class _FakeContext:
+        memory = _FakeMemory()
+
+    class _FakeSessionManager:
+        def flush_all(self) -> int:
+            return 0
+
+        def list_sessions(self) -> list[dict[str, object]]:
+            return []
+
+    class _FakeCronService:
+        def __init__(self) -> None:
+            self.on_job = None
+
+        async def start(self) -> None:
+            return None
+
+        def stop(self) -> None:
+            return None
+
+        def status(self) -> dict[str, int]:
+            return {"jobs": 0}
+
+        def register_system_job(self, _job) -> None:
+            return None
+
+    class _FakeAgentLoop:
+        @classmethod
+        def from_config(cls, config, bus=None, **extra):
+            return cls(**extra)
+
+        def __init__(self, *args, **kwargs) -> None:
+            self.model = "test-model"
+            self.provider = _fake_provider()
+            self.tools = {}
+            self.context = _FakeContext()
+            self.sessions = kwargs["session_manager"]
+            self.submit_local_trigger_turn = AsyncMock()
+            seen["agent"] = self
+
+        def _schedule_background(self, _coro) -> None:
+            return None
+
+        async def run(self) -> None:
+            await asyncio.Event().wait()
+
+        async def close_mcp(self) -> None:
+            return None
+
+        def stop(self) -> None:
+            return None
+
+    class _FakeChannelManager:
+        enabled_channels: list[str] = []
+
+        def __init__(self, *_args, **_kwargs) -> None:
+            return None
+
+        async def start_all(self) -> None:
+            await asyncio.Event().wait()
+
+        async def stop_all(self) -> None:
+            return None
+
+    async def _fake_run_local_trigger_queue(**kwargs):
+        seen["local_trigger_queue_kwargs"] = kwargs
+        raise _StopGatewayError("stop")
+
+    monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
+    monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _FakeChannelManager)
+    monkeypatch.setattr(
+        "nanobot.triggers.local_runner.run_local_trigger_queue",
+        _fake_run_local_trigger_queue,
+    )
+
+    cli_commands._run_gateway(config, health_server_enabled=False)
+
+    agent = seen["agent"]
+    kwargs = seen["local_trigger_queue_kwargs"]
+    assert kwargs["bus"] is bus
+    assert kwargs["submit_turn"] is agent.submit_local_trigger_turn
+
+
 def test_gateway_workspace_override_does_not_migrate_legacy_cron(
     monkeypatch, tmp_path: Path
 ) -> None:

tests/triggers/test_local_triggers.py

@@ -128,6 +128,105 @@ async def test_local_trigger_queue_publishes_bound_inbound_message(tmp_path: Pat
     assert store.claim_deliveries() == []
 
 
+@pytest.mark.asyncio
+async def test_local_trigger_queue_waits_for_submitted_turn_before_ack(
+    tmp_path: Path,
+) -> None:
+    store = LocalTriggerStore(tmp_path)
+    trigger = store.create(
+        name="CI review",
+        channel="websocket",
+        chat_id="chat-1",
+        session_key="websocket:chat-1",
+    )
+    store.enqueue(trigger.id, "Review failed CI")
+    submitted: list[InboundMessage] = []
+    release = asyncio.Event()
+
+    async def _submit_turn(msg: InboundMessage):
+        submitted.append(msg)
+        await release.wait()
+        return None
+
+    task = asyncio.create_task(
+        run_local_trigger_queue(
+            store=store,
+            submit_turn=_submit_turn,
+            poll_interval_s=0.01,
+        )
+    )
+    try:
+        for _ in range(100):
+            if submitted:
+                break
+            await asyncio.sleep(0.01)
+
+        assert len(submitted) == 1
+        assert list(store.processing_dir.glob("*.json"))
+        stored = store.get(trigger.id)
+        assert stored is not None
+        assert stored.last_status is None
+
+        release.set()
+        for _ in range(100):
+            stored = store.get(trigger.id)
+            if stored and stored.last_status == "ok":
+                break
+            await asyncio.sleep(0.01)
+
+        assert not list(store.processing_dir.glob("*.json"))
+        stored = store.get(trigger.id)
+        assert stored is not None
+        assert stored.last_status == "ok"
+        assert store.claim_deliveries() == []
+    finally:
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+
+@pytest.mark.asyncio
+async def test_local_trigger_queue_requeues_when_submitted_turn_is_interrupted(
+    tmp_path: Path,
+) -> None:
+    store = LocalTriggerStore(tmp_path)
+    trigger = store.create(
+        name="CI review",
+        channel="websocket",
+        chat_id="chat-1",
+        session_key="websocket:chat-1",
+    )
+    store.enqueue(trigger.id, "Review failed CI")
+    started = asyncio.Event()
+
+    async def _submit_turn(_msg: InboundMessage):
+        started.set()
+        await asyncio.Future()
+
+    task = asyncio.create_task(
+        run_local_trigger_queue(
+            store=store,
+            submit_turn=_submit_turn,
+            poll_interval_s=0.01,
+        )
+    )
+    try:
+        await asyncio.wait_for(started.wait(), timeout=1)
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+        reclaimed = store.claim_deliveries()
+        assert len(reclaimed) == 1
+        assert reclaimed[0].trigger_id == trigger.id
+        assert reclaimed[0].attempts == 1
+        assert reclaimed[0].last_error == "CancelledError"
+    finally:
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+
 @pytest.mark.asyncio
 async def test_local_trigger_queue_recovers_processing_delivery_on_start(
     tmp_path: Path,