From 1ed2c9a21333737772818ee5b4bf566e154a601c Mon Sep 17 00:00:00 2001 From: chengyongru Date: Tue, 30 Jun 2026 14:08:54 +0800 Subject: [PATCH] fix(trigger): recover interrupted deliveries --- docs/chat-commands.md | 6 +++ docs/cli-reference.md | 7 +++ docs/concepts.md | 5 +- docs/webui.md | 5 ++ nanobot/triggers/runner.py | 6 +++ nanobot/triggers/store.py | 52 ++++++++++++++----- tests/triggers/test_external_triggers.py | 66 ++++++++++++++++++++++++ 7 files changed, 133 insertions(+), 14 deletions(-) diff --git a/docs/chat-commands.md b/docs/chat-commands.md index 96729be9..e93c1f8b 100644 --- a/docs/chat-commands.md +++ b/docs/chat-commands.md @@ -81,6 +81,12 @@ 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. +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. + For longer or generated content, omit the message argument and pipe stdin: ```bash diff --git a/docs/cli-reference.md b/docs/cli-reference.md index d8e27e28..2f8da0db 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -135,6 +135,13 @@ nanobot trigger trg_8K4P2Q9X "Review PR #4502" 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. + Use stdin when another local process generates the message: ```bash diff --git a/docs/concepts.md b/docs/concepts.md index da39c80a..c782cdd0 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -147,7 +147,10 @@ External triggers are also session-bound, but they do not have their own 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. +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 +systems should tolerate repeated trigger messages. ## Where to Go Next diff --git a/docs/webui.md b/docs/webui.md index f1172b27..7a36bcba 100644 --- a/docs/webui.md +++ b/docs/webui.md @@ -132,6 +132,11 @@ If a GitHub webhook, CI system, or another service should wake nanobot up, keep 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. + For recurring background checks that should stay quiet unless there is something useful to report, use the protected heartbeat job by editing `HEARTBEAT.md` instead of creating a chat automation. diff --git a/nanobot/triggers/runner.py b/nanobot/triggers/runner.py index ef75dccc..b1eea405 100644 --- a/nanobot/triggers/runner.py +++ b/nanobot/triggers/runner.py @@ -25,6 +25,12 @@ async def run_external_trigger_queue( ) -> None: """Poll local trigger deliveries and publish them as normal inbound messages.""" logger.info("External trigger queue started") + recovered = store.recover_processing_deliveries() + if recovered: + logger.warning( + "Trigger: recovered {} interrupted delivery file(s) from processing", + recovered, + ) while True: deliveries = store.claim_deliveries(limit=batch_size) if not deliveries: diff --git a/nanobot/triggers/store.py b/nanobot/triggers/store.py index 153119b3..5f081ed5 100644 --- a/nanobot/triggers/store.py +++ b/nanobot/triggers/store.py @@ -19,6 +19,7 @@ from nanobot.triggers.types import ExternalTrigger, TriggerDelivery, TriggerRunR _TRIGGER_ID_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" _MAX_RUN_HISTORY = 20 _MAX_DELIVERY_ATTEMPTS = 10 +_PROCESSING_RECOVERY_ERROR = "delivery was recovered from interrupted processing" class TriggerStoreError(RuntimeError): @@ -194,6 +195,26 @@ class ExternalTriggerStore: claimed.append(delivery) return claimed + def recover_processing_deliveries(self) -> int: + """Requeue deliveries left in processing by an interrupted gateway.""" + self._ensure_dirs() + recovered = 0 + with self._lock: + for path in sorted(self.processing_dir.glob("*.json")): + try: + data = json.loads(path.read_text(encoding="utf-8")) + delivery = TriggerDelivery.from_dict( + data.get("delivery", data), + path=path, + ) + except Exception: + logger.exception("Trigger: failed to parse processing delivery {}", path) + self._move_bad_delivery_unlocked(path) + continue + if self._retry_delivery_unlocked(delivery, _PROCESSING_RECOVERY_ERROR): + recovered += 1 + return recovered + def complete_delivery(self, delivery: TriggerDelivery) -> None: """Delete a claimed delivery after it is handled.""" if delivery.path is None: @@ -208,19 +229,7 @@ class ExternalTriggerStore: return False self._ensure_dirs() with self._lock: - if delivery.attempts + 1 >= _MAX_DELIVERY_ATTEMPTS: - delivery.attempts += 1 - delivery.last_error = error - failed = self.failed_dir / delivery.path.name - self._atomic_write(failed, json.dumps(_delivery_payload(delivery), ensure_ascii=False)) - delivery.path.unlink(missing_ok=True) - return False - delivery.attempts += 1 - delivery.last_error = error - target = self.inbox_dir / delivery.path.name - self._atomic_write(target, json.dumps(_delivery_payload(delivery), ensure_ascii=False)) - delivery.path.unlink(missing_ok=True) - return True + return self._retry_delivery_unlocked(delivery, error) def record_delivery( self, @@ -298,6 +307,23 @@ class ExternalTriggerStore: with suppress(OSError): os.replace(path, target) + def _retry_delivery_unlocked(self, delivery: TriggerDelivery, error: str) -> bool: + if delivery.path is None: + return False + if delivery.attempts + 1 >= _MAX_DELIVERY_ATTEMPTS: + delivery.attempts += 1 + delivery.last_error = error + failed = self.failed_dir / delivery.path.name + self._atomic_write(failed, json.dumps(_delivery_payload(delivery), ensure_ascii=False)) + delivery.path.unlink(missing_ok=True) + return False + delivery.attempts += 1 + delivery.last_error = error + target = self.inbox_dir / delivery.path.name + self._atomic_write(target, json.dumps(_delivery_payload(delivery), ensure_ascii=False)) + delivery.path.unlink(missing_ok=True) + return True + @staticmethod def _atomic_write(path: Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) diff --git a/tests/triggers/test_external_triggers.py b/tests/triggers/test_external_triggers.py index af81a524..d4450653 100644 --- a/tests/triggers/test_external_triggers.py +++ b/tests/triggers/test_external_triggers.py @@ -49,6 +49,33 @@ def test_enqueue_rejects_disabled_trigger(tmp_path: Path) -> None: store.enqueue(trigger.id, "Review PR #4502") +def test_recover_processing_deliveries_requeues_claimed_delivery(tmp_path: Path) -> None: + store = ExternalTriggerStore(tmp_path) + trigger = store.create( + name="PR review", + channel="websocket", + chat_id="chat-1", + session_key="websocket:chat-1", + ) + store.enqueue(trigger.id, "Review PR #4591") + + claimed = store.claim_deliveries() + assert len(claimed) == 1 + assert claimed[0].path is not None + assert claimed[0].path.parent.name == "processing" + assert ExternalTriggerStore(tmp_path).claim_deliveries() == [] + + restarted = ExternalTriggerStore(tmp_path) + assert restarted.recover_processing_deliveries() == 1 + + reclaimed = restarted.claim_deliveries() + assert len(reclaimed) == 1 + assert reclaimed[0].trigger_id == trigger.id + assert reclaimed[0].content == "Review PR #4591" + assert reclaimed[0].attempts == 1 + assert reclaimed[0].last_error == "delivery was recovered from interrupted processing" + + @pytest.mark.asyncio async def test_external_trigger_queue_publishes_bound_inbound_message(tmp_path: Path) -> None: store = ExternalTriggerStore(tmp_path) @@ -99,3 +126,42 @@ async def test_external_trigger_queue_publishes_bound_inbound_message(tmp_path: assert stored.last_status == "ok" assert stored.last_run_at_ms is not None assert store.claim_deliveries() == [] + + +@pytest.mark.asyncio +async def test_external_trigger_queue_recovers_processing_delivery_on_start( + tmp_path: Path, +) -> None: + store = ExternalTriggerStore(tmp_path) + trigger = store.create( + name="PR review", + channel="websocket", + chat_id="chat-1", + session_key="websocket:chat-1", + ) + store.enqueue(trigger.id, "Review PR #4591") + assert len(store.claim_deliveries()) == 1 + published: list[InboundMessage] = [] + + class _Bus: + async def publish_inbound(self, msg: InboundMessage) -> None: + published.append(msg) + + restarted = ExternalTriggerStore(tmp_path) + task = asyncio.create_task( + run_external_trigger_queue(store=restarted, bus=_Bus(), poll_interval_s=0.01) + ) + try: + for _ in range(100): + if published: + break + await asyncio.sleep(0.01) + finally: + task.cancel() + with suppress(asyncio.CancelledError): + await task + + assert len(published) == 1 + assert published[0].content == "Review PR #4591" + assert published[0].metadata["_external_trigger"]["trigger_id"] == trigger.id + assert restarted.claim_deliveries() == []