Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(trigger): add local trigger run audit records

Browse Source
This commit is contained in:
chengyongru
2026-07-02 13:32:46 +08:00
committed by Xubin Ren
parent b941233138
commit 661ab00656
7 changed files with 246 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/cli-reference.md
+2 -1
View File
@@ -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 `<workspace>/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:
nanobot/cron/service.py
+8 -11
View File
@@ -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."""
nanobot/triggers/local_runner.py
+57 -1
View File
@@ -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]:
nanobot/triggers/local_store.py
+62
View File
@@ -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
nanobot/utils/run_records.py
+53
View File
@@ -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
tests/cron/test_cron_service.py
+12
View File
@@ -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] = []
tests/triggers/test_local_triggers.py
+52 -6
View File
@@ -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):