From 661ab0065614adc75f0bfb41709102e778ff25ff Mon Sep 17 00:00:00 2001 From: chengyongru Date: Thu, 2 Jul 2026 11:50:19 +0800 Subject: [PATCH] feat(trigger): add local trigger run audit records --- docs/cli-reference.md | 3 +- nanobot/cron/service.py | 19 ++++---- nanobot/triggers/local_runner.py | 58 ++++++++++++++++++++++++- nanobot/triggers/local_store.py | 62 +++++++++++++++++++++++++++ nanobot/utils/run_records.py | 53 +++++++++++++++++++++++ tests/cron/test_cron_service.py | 12 ++++++ tests/triggers/test_local_triggers.py | 58 ++++++++++++++++++++++--- 7 files changed, 246 insertions(+), 19 deletions(-) create mode 100644 nanobot/utils/run_records.py diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 265d3c04..a92ecf1b 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -143,7 +143,8 @@ 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. If the agent receives the delivery and the turn fails, the delivery is -marked failed instead of retried indefinitely. Run one gateway consumer per +marked failed instead of retried indefinitely. Each delivery also writes an +audit record under `/triggers/runs`. 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/nanobot/cron/service.py b/nanobot/cron/service.py index d28882d7..ccf86565 100644 --- a/nanobot/cron/service.py +++ b/nanobot/cron/service.py @@ -24,6 +24,12 @@ from nanobot.cron.types import ( CronSchedule, CronStore, ) +from nanobot.utils.run_records import ( + safe_run_record_name, +) +from nanobot.utils.run_records import ( + write_run_record as write_automation_run_record, +) class CronJobSkippedError(Exception): @@ -474,20 +480,11 @@ class CronService: @staticmethod def _safe_run_record_name(run_id: str) -> str: - return "".join(c if c.isalnum() or c in "._-" else "_" for c in run_id) + return safe_run_record_name(run_id) def write_run_record(self, run_id: str, record: dict[str, Any]) -> None: """Write an internal audit record for one cron execution.""" - name = self._safe_run_record_name(run_id) - if not name: - name = str(uuid.uuid4()) - path = self._run_records_dir / f"{name}.json" - payload = { - **record, - "run_id": run_id, - "updated_at_ms": _now_ms(), - } - self._atomic_write(path, json.dumps(payload, indent=2, ensure_ascii=False)) + write_automation_run_record(self._run_records_dir, run_id, record) async def start(self) -> None: """Start the cron service.""" diff --git a/nanobot/triggers/local_runner.py b/nanobot/triggers/local_runner.py index 751559df..55b890d1 100644 --- a/nanobot/triggers/local_runner.py +++ b/nanobot/triggers/local_runner.py @@ -50,6 +50,12 @@ async def run_local_trigger_queue( store.complete_delivery(delivery) except asyncio.CancelledError as exc: store.retry_delivery(delivery, str(exc) or exc.__class__.__name__) + _write_delivery_run_record( + store, + delivery, + status="interrupted", + error=str(exc) or exc.__class__.__name__, + ) raise except _TerminalDeliveryError as exc: store.record_delivery( @@ -58,6 +64,12 @@ async def run_local_trigger_queue( error=str(exc), run_at_ms=delivery.created_at_ms, ) + _write_delivery_run_record( + store, + delivery, + status="error", + error=str(exc), + ) store.complete_delivery(delivery) logger.warning( "Trigger: dropped delivery {} for {}: {}", @@ -73,6 +85,12 @@ async def run_local_trigger_queue( error=error, run_at_ms=delivery.created_at_ms, ) + _write_delivery_run_record( + store, + delivery, + status="error", + error=error, + ) store.complete_delivery(delivery) logger.warning( "Trigger: delivery {} for {} reached the agent but failed: {}", @@ -83,6 +101,12 @@ async def run_local_trigger_queue( except Exception as exc: error = str(exc) or exc.__class__.__name__ retried = store.retry_delivery(delivery, error) + _write_delivery_run_record( + store, + delivery, + status="retrying" if retried else "error", + error=error, + ) store.record_delivery( delivery.trigger_id, status="error", @@ -113,6 +137,7 @@ async def _deliver_delivery( if not trigger.enabled: raise _TerminalDeliveryError("trigger is disabled") + store.write_delivery_run_record(delivery, trigger=trigger, status="processing") msg = InboundMessage( channel=trigger.channel, sender_id=trigger.sender_id, @@ -121,12 +146,43 @@ async def _deliver_delivery( metadata=_delivery_metadata(trigger, delivery), session_key_override=trigger.session_key, ) - await submit_turn(msg) + response = await submit_turn(msg) store.record_delivery( trigger.id, status="ok", run_at_ms=delivery.created_at_ms, ) + _write_delivery_run_record( + store, + delivery, + trigger=trigger, + status="ok", + response=response.content if response else "", + ) + + +def _write_delivery_run_record( + store: LocalTriggerStore, + delivery: TriggerDelivery, + *, + status: str, + trigger: LocalTrigger | None = None, + error: str | None = None, + response: str | None = None, +) -> None: + try: + store.write_delivery_run_record( + delivery, + trigger=trigger, + status=status, + error=error, + response=response, + ) + except Exception: + logger.exception( + "Trigger: failed to write run record for delivery {}", + delivery.id, + ) def _delivery_metadata(trigger: LocalTrigger, delivery: TriggerDelivery) -> dict[str, Any]: diff --git a/nanobot/triggers/local_store.py b/nanobot/triggers/local_store.py index 35a5629b..69ea5290 100644 --- a/nanobot/triggers/local_store.py +++ b/nanobot/triggers/local_store.py @@ -15,6 +15,7 @@ from filelock import FileLock from loguru import logger from nanobot.triggers.local_types import LocalTrigger, TriggerDelivery, TriggerRunRecord +from nanobot.utils.run_records import write_run_record as write_automation_run_record _TRIGGER_ID_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" _MAX_RUN_HISTORY = 20 @@ -44,6 +45,7 @@ class LocalTriggerStore: self.inbox_dir = self.root / "inbox" self.processing_dir = self.root / "processing" self.failed_dir = self.root / "failed" + self.runs_dir = self.root / "runs" self._lock = FileLock(str(self.root / ".lock")) def create( @@ -175,6 +177,12 @@ class LocalTriggerStore: path = self.inbox_dir / f"{delivery.created_at_ms}-{delivery.id}.json" self._atomic_write(path, json.dumps(_delivery_payload(delivery), ensure_ascii=False)) delivery.path = path + try: + self.write_delivery_run_record(delivery, trigger=trigger, status="queued") + except BaseException: + path.unlink(missing_ok=True) + delivery.path = None + raise return delivery def claim_deliveries(self, *, limit: int = 20) -> list[TriggerDelivery]: @@ -263,11 +271,37 @@ class LocalTriggerStore: trigger.run_history = trigger.run_history[-_MAX_RUN_HISTORY:] self._save_triggers_unlocked(triggers) + def write_run_record(self, run_id: str, record: dict[str, Any]) -> Path: + """Write an internal audit record for one local trigger delivery.""" + self._ensure_dirs() + return write_automation_run_record(self.runs_dir, run_id, record) + + def write_delivery_run_record( + self, + delivery: TriggerDelivery, + *, + status: str, + trigger: LocalTrigger | None = None, + error: str | None = None, + response: str | None = None, + ) -> Path: + """Write the durable audit record for one local trigger delivery.""" + if trigger is None: + trigger = self.get(delivery.trigger_id) + record = _delivery_run_record(delivery, trigger) + record["status"] = status + if error: + record["error"] = error + if response is not None: + record["response"] = response + return self.write_run_record(delivery.id, record) + def _ensure_dirs(self) -> None: self.root.mkdir(parents=True, exist_ok=True) self.inbox_dir.mkdir(parents=True, exist_ok=True) self.processing_dir.mkdir(parents=True, exist_ok=True) self.failed_dir.mkdir(parents=True, exist_ok=True) + self.runs_dir.mkdir(parents=True, exist_ok=True) def _load_triggers_unlocked(self) -> list[LocalTrigger]: if not self.store_path.exists(): @@ -399,3 +433,31 @@ def _delivery_payload(delivery: TriggerDelivery) -> dict[str, Any]: "version": 1, "delivery": delivery.to_dict(), } + + +def _delivery_run_record( + delivery: TriggerDelivery, + trigger: LocalTrigger | None, +) -> dict[str, Any]: + record: dict[str, Any] = { + "kind": "local_trigger", + "trigger_id": delivery.trigger_id, + "delivery_id": delivery.id, + "content": delivery.content, + "created_at_ms": delivery.created_at_ms, + "attempts": delivery.attempts, + } + if delivery.last_error: + record["last_error"] = delivery.last_error + if trigger is not None: + record.update( + { + "trigger_name": trigger.name, + "session_key": trigger.session_key, + "channel": trigger.channel, + "chat_id": trigger.chat_id, + "sender_id": trigger.sender_id, + "origin_metadata": trigger.origin_metadata, + } + ) + return record diff --git a/nanobot/utils/run_records.py b/nanobot/utils/run_records.py new file mode 100644 index 00000000..6f463288 --- /dev/null +++ b/nanobot/utils/run_records.py @@ -0,0 +1,53 @@ +"""Durable JSON run records for automation executions.""" + +from __future__ import annotations + +import json +import os +import time +import uuid +from contextlib import suppress +from pathlib import Path +from typing import Any + + +def safe_run_record_name(run_id: str) -> str: + """Return a filesystem-safe filename stem for a run ID.""" + return "".join(c if c.isalnum() or c in "._-" else "_" for c in run_id) + + +def write_run_record(runs_dir: Path, run_id: str, record: dict[str, Any]) -> Path: + """Write or replace one durable automation run audit record.""" + name = safe_run_record_name(run_id) or str(uuid.uuid4()) + path = runs_dir / f"{name}.json" + payload = { + **record, + "run_id": run_id, + "updated_at_ms": _now_ms(), + } + _atomic_write(path, json.dumps(payload, indent=2, ensure_ascii=False)) + return path + + +def _now_ms() -> int: + return int(time.time() * 1000) + + +def _atomic_write(path: Path, content: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + try: + with open(tmp_path, "w", encoding="utf-8") as f: + f.write(content) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp_path, path) + with suppress(PermissionError): + fd = os.open(str(path.parent), os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + except BaseException: + tmp_path.unlink(missing_ok=True) + raise diff --git a/tests/cron/test_cron_service.py b/tests/cron/test_cron_service.py index 2072e063..9a2ed337 100644 --- a/tests/cron/test_cron_service.py +++ b/tests/cron/test_cron_service.py @@ -52,6 +52,18 @@ def test_add_job_accepts_valid_timezone(tmp_path) -> None: assert job.state.next_run_at_ms is not None +def test_write_run_record_uses_cron_runs_dir(tmp_path) -> None: + service = CronService(tmp_path / "cron" / "jobs.json") + + service.write_run_record("job:1", {"status": "queued"}) + + record_path = tmp_path / "cron" / "runs" / "job_1.json" + record = json.loads(record_path.read_text(encoding="utf-8")) + assert record["run_id"] == "job:1" + assert record["status"] == "queued" + assert record["updated_at_ms"] > 0 + + @pytest.mark.asyncio async def test_unbound_agent_jobs_are_disabled_on_add(tmp_path) -> None: called: list[str] = [] diff --git a/tests/triggers/test_local_triggers.py b/tests/triggers/test_local_triggers.py index 5541ecfe..f18ff1d9 100644 --- a/tests/triggers/test_local_triggers.py +++ b/tests/triggers/test_local_triggers.py @@ -8,7 +8,7 @@ from pathlib import Path import pytest from nanobot.agent.automation_turns import AutomationTurnError -from nanobot.bus.events import InboundMessage +from nanobot.bus.events import InboundMessage, OutboundMessage from nanobot.triggers.local_runner import run_local_trigger_queue from nanobot.triggers.local_store import LocalTriggerStore, TriggerDisabledError from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN_METADATA_KEY @@ -33,6 +33,10 @@ def _write_delivery_file(path: Path, *, trigger_id: str, delivery_id: str) -> No ) +def _read_run_record(store: LocalTriggerStore, run_id: str) -> dict: + return json.loads((store.runs_dir / f"{run_id}.json").read_text(encoding="utf-8")) + + def test_trigger_store_allows_multiple_triggers_per_session(tmp_path: Path) -> None: store = LocalTriggerStore(tmp_path) @@ -70,6 +74,34 @@ def test_enqueue_rejects_disabled_trigger(tmp_path: Path) -> None: store.enqueue(trigger.id, "Review PR #4502") +def test_enqueue_writes_trigger_run_record(tmp_path: Path) -> None: + store = LocalTriggerStore(tmp_path) + trigger = store.create( + name="PR review", + channel="websocket", + chat_id="chat-1", + session_key="websocket:chat-1", + origin_metadata={"webui": True}, + ) + + delivery = store.enqueue(trigger.id, "Review PR #4591") + + record = _read_run_record(store, delivery.id) + assert record["run_id"] == delivery.id + assert record["kind"] == "local_trigger" + assert record["status"] == "queued" + assert record["trigger_id"] == trigger.id + assert record["trigger_name"] == "PR review" + assert record["delivery_id"] == delivery.id + assert record["session_key"] == "websocket:chat-1" + assert record["channel"] == "websocket" + assert record["chat_id"] == "chat-1" + assert record["sender_id"] == "trigger" + assert record["content"] == "Review PR #4591" + assert record["origin_metadata"] == {"webui": True} + assert record["updated_at_ms"] > 0 + + def test_delete_removes_delivery_files_for_trigger(tmp_path: Path) -> None: store = LocalTriggerStore(tmp_path) trigger = store.create( @@ -140,12 +172,12 @@ async def test_local_trigger_queue_submits_bound_inbound_message(tmp_path: Path) session_key="websocket:chat-1", origin_metadata={"webui": True, WEBUI_TURN_METADATA_KEY: "old-turn"}, ) - store.enqueue(trigger.id, "Review PR #4502") + delivery = store.enqueue(trigger.id, "Review PR #4502") submitted: list[InboundMessage] = [] async def _submit_turn(msg: InboundMessage): submitted.append(msg) - return None + return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id, content="done") task = asyncio.create_task( run_local_trigger_queue(store=store, submit_turn=_submit_turn, poll_interval_s=0.01) @@ -184,6 +216,10 @@ async def test_local_trigger_queue_submits_bound_inbound_message(tmp_path: Path) assert stored.last_status == "ok" assert stored.last_run_at_ms is not None assert store.claim_deliveries() == [] + record = _read_run_record(store, delivery.id) + assert record["status"] == "ok" + assert record["response"] == "done" + assert record["trigger_id"] == trigger.id @pytest.mark.asyncio @@ -197,7 +233,7 @@ async def test_local_trigger_queue_waits_for_submitted_turn_before_ack( chat_id="chat-1", session_key="websocket:chat-1", ) - store.enqueue(trigger.id, "Review failed CI") + delivery = store.enqueue(trigger.id, "Review failed CI") submitted: list[InboundMessage] = [] release = asyncio.Event() @@ -221,6 +257,8 @@ async def test_local_trigger_queue_waits_for_submitted_turn_before_ack( assert len(submitted) == 1 assert list(store.processing_dir.glob("*.json")) + record = _read_run_record(store, delivery.id) + assert record["status"] == "processing" stored = store.get(trigger.id) assert stored is not None assert stored.last_status is None @@ -237,6 +275,8 @@ async def test_local_trigger_queue_waits_for_submitted_turn_before_ack( assert stored is not None assert stored.last_status == "ok" assert store.claim_deliveries() == [] + record = _read_run_record(store, delivery.id) + assert record["status"] == "ok" finally: task.cancel() with suppress(asyncio.CancelledError): @@ -254,7 +294,7 @@ async def test_local_trigger_queue_requeues_when_submitted_turn_is_interrupted( chat_id="chat-1", session_key="websocket:chat-1", ) - store.enqueue(trigger.id, "Review failed CI") + delivery = store.enqueue(trigger.id, "Review failed CI") started = asyncio.Event() async def _submit_turn(_msg: InboundMessage): @@ -279,6 +319,9 @@ async def test_local_trigger_queue_requeues_when_submitted_turn_is_interrupted( assert reclaimed[0].trigger_id == trigger.id assert reclaimed[0].attempts == 1 assert reclaimed[0].last_error == "CancelledError" + record = _read_run_record(store, delivery.id) + assert record["status"] == "interrupted" + assert record["attempts"] == 1 finally: task.cancel() with suppress(asyncio.CancelledError): @@ -296,7 +339,7 @@ async def test_local_trigger_queue_does_not_retry_completed_agent_failure( chat_id="chat-1", session_key="websocket:chat-1", ) - store.enqueue(trigger.id, "Review failed CI") + delivery = store.enqueue(trigger.id, "Review failed CI") started = asyncio.Event() async def _submit_turn(_msg: InboundMessage): @@ -325,6 +368,9 @@ async def test_local_trigger_queue_does_not_retry_completed_agent_failure( assert store.claim_deliveries() == [] assert not list(store.processing_dir.glob("*.json")) assert not list(store.failed_dir.glob("*.json")) + record = _read_run_record(store, delivery.id) + assert record["status"] == "error" + assert record["error"] == "model failed" finally: task.cancel() with suppress(asyncio.CancelledError):