From f32007c83f9ffbba68bc0a44a3ba8439013e3613 Mon Sep 17 00:00:00 2001 From: chengyongru Date: Wed, 1 Jul 2026 17:43:34 +0800 Subject: [PATCH] fix(trigger): defer local triggers until session idle --- docs/chat-commands.md | 15 +-- docs/cli-reference.md | 12 ++- docs/concepts.md | 5 +- docs/webui.md | 8 +- nanobot/agent/cron_turns.py | 5 +- nanobot/agent/loop.py | 37 ++++++- nanobot/cli/commands.py | 6 +- nanobot/triggers/local_runner.py | 43 +++++--- nanobot/triggers/local_session_turns.py | 8 ++ nanobot/triggers/local_turns.py | 136 ++++++++++++++++++++++++ tests/agent/test_runner_injections.py | 92 ++++++++++++++++ tests/cli/test_commands.py | 112 +++++++++++++++++++ tests/triggers/test_local_triggers.py | 99 +++++++++++++++++ 13 files changed, 543 insertions(+), 35 deletions(-) create mode 100644 nanobot/triggers/local_turns.py diff --git a/docs/chat-commands.md b/docs/chat-commands.md index 02801c0e..e238b6e9 100644 --- a/docs/chat-commands.md +++ b/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: diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 02137786..9880efd5 100644 --- a/docs/cli-reference.md +++ b/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: diff --git a/docs/concepts.md b/docs/concepts.md index 6cc8a597..9065bc35 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -148,8 +148,9 @@ schedule. Create one from the target chat with `/trigger `, then call `nanobot trigger ""` 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 diff --git a/docs/webui.md b/docs/webui.md index 8f3fedfa..cc5de514 100644 --- a/docs/webui.md +++ b/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` diff --git a/nanobot/agent/cron_turns.py b/nanobot/agent/cron_turns.py index a589338c..68237eb5 100644 --- a/nanobot/agent/cron_turns.py +++ b/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: diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 61ad0614..20be7c4e 100644 --- a/nanobot/agent/loop.py +++ b/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.""" diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index 7dcb5b21..f5dabda5 100644 --- a/nanobot/cli/commands.py +++ b/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", ), ] diff --git a/nanobot/triggers/local_runner.py b/nanobot/triggers/local_runner.py index 5e213b68..e7ed70c6 100644 --- a/nanobot/triggers/local_runner.py +++ b/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,16 +101,20 @@ async def _publish_delivery( if not trigger.enabled: raise _TerminalDeliveryError("trigger is disabled") - await bus.publish_inbound( - InboundMessage( - channel=trigger.channel, - sender_id=trigger.sender_id, - chat_id=trigger.chat_id, - content=delivery.content, - metadata=_delivery_metadata(trigger, delivery), - session_key_override=trigger.session_key, - ) + msg = InboundMessage( + channel=trigger.channel, + sender_id=trigger.sender_id, + chat_id=trigger.chat_id, + content=delivery.content, + 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", diff --git a/nanobot/triggers/local_session_turns.py b/nanobot/triggers/local_session_turns.py index 771e3745..2d853bb4 100644 --- a/nanobot/triggers/local_session_turns.py +++ b/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]]: diff --git a/nanobot/triggers/local_turns.py b/nanobot/triggers/local_turns.py new file mode 100644 index 00000000..257da5ee --- /dev/null +++ b/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 diff --git a/tests/agent/test_runner_injections.py b/tests/agent/test_runner_injections.py index 637235ba..fd5ea8dd 100644 --- a/tests/agent/test_runner_injections.py +++ b/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.""" diff --git a/tests/cli/test_commands.py b/tests/cli/test_commands.py index 84334e33..db237bd1 100644 --- a/tests/cli/test_commands.py +++ b/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: diff --git a/tests/triggers/test_local_triggers.py b/tests/triggers/test_local_triggers.py index 390428e7..262aff69 100644 --- a/tests/triggers/test_local_triggers.py +++ b/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,