diff --git a/nanobot/channels/discord.py b/nanobot/channels/discord.py index 6252d555..b2259045 100644 --- a/nanobot/channels/discord.py +++ b/nanobot/channels/discord.py @@ -218,6 +218,16 @@ if DISCORD_AVAILABLE: command_text = f"/model {preset}" if preset else "/model" await self._forward_slash_command(interaction, command_text) + @self.tree.command(name="trigger", description="Create a local trigger for this chat") + @app_commands.describe(name="Optional trigger name") + async def trigger_command( + interaction: discord.Interaction, + name: str | None = None, + ) -> None: + name = (name or "").strip() + command_text = f"/trigger {name}" if name else "/trigger" + await self._forward_slash_command(interaction, command_text) + @self.tree.command(name="help", description="Show available commands") async def help_command(interaction: discord.Interaction) -> None: sender_id = str(interaction.user.id) diff --git a/nanobot/channels/manager.py b/nanobot/channels/manager.py index b2ac69b8..499a0c4f 100644 --- a/nanobot/channels/manager.py +++ b/nanobot/channels/manager.py @@ -68,6 +68,7 @@ class ChannelManager: *, session_manager: "SessionManager | None" = None, cron_service: Any | None = None, + external_trigger_store: Any | None = None, webui_runtime_model_name: Callable[[], str | None] | None = None, webui_cron_pending_job_ids: Callable[[str], set[str]] | None = None, webui_static_dist: bool = True, @@ -78,6 +79,7 @@ class ChannelManager: self.bus = bus self._session_manager = session_manager self._cron_service = cron_service + self._external_trigger_store = external_trigger_store self._webui_runtime_model_name = webui_runtime_model_name self._webui_cron_pending_job_ids = webui_cron_pending_job_ids self._webui_static_dist = webui_static_dist @@ -139,6 +141,7 @@ class ChannelManager: runtime_surface=self._webui_runtime_surface, runtime_capabilities_overrides=self._webui_runtime_capabilities, cron_service=self._cron_service, + external_trigger_store=self._external_trigger_store, cron_pending_job_ids=self._webui_cron_pending_job_ids, logger=logger, ) diff --git a/nanobot/channels/telegram.py b/nanobot/channels/telegram.py index af4a3fc2..40eeef47 100644 --- a/nanobot/channels/telegram.py +++ b/nanobot/channels/telegram.py @@ -411,6 +411,7 @@ class TelegramChannel(BaseChannel): BotCommand("status", "Show bot status"), BotCommand("history", "Show recent conversation messages"), BotCommand("goal", "Start a sustained objective (long-running task)"), + BotCommand("trigger", "Create a local trigger for this chat"), BotCommand("pairing", "Manage DM pairing (approve/deny/list)"), BotCommand("model", "Switch runtime model preset"), BotCommand("skill", "List enabled skills"), @@ -423,7 +424,7 @@ class TelegramChannel(BaseChannel): # Regex for slash commands routed to AgentLoop via ``_forward_command``. # Hyphenated ``dream-*`` commands stay on a separate handler (below). TELEGRAM_BUS_SLASH_COMMAND_RE = re.compile( - r"^/(?:new|stop|restart|status|dream|history|goal|pairing|model|skill)(?:@\w+)?(?:\s+.*)?$" + r"^/(?:new|stop|restart|status|dream|history|goal|trigger|pairing|model|skill)(?:@\w+)?(?:\s+.*)?$" ) @classmethod diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index 636c3913..7e424a0a 100644 --- a/nanobot/cli/commands.py +++ b/nanobot/cli/commands.py @@ -718,6 +718,21 @@ def _load_runtime_config(config: str | None = None, workspace: str | None = None return loaded +def _read_trigger_cli_message(message: str | None) -> str: + """Read a trigger message from an argument or stdin.""" + if message and message.strip(): + return message + try: + if not sys.stdin.isatty(): + content = sys.stdin.read() + if content.strip(): + return content + except Exception: + pass + console.print("[red]Error: trigger message is required[/red]") + raise typer.Exit(1) + + def _warn_deprecated_config_keys(config_path: Path | None) -> None: """Hint users to remove obsolete keys from their config file.""" import json @@ -749,6 +764,35 @@ def _migrate_cron_store(config: "Config") -> None: shutil.move(str(legacy_path), str(new_path)) +@app.command() +def trigger( + trigger_id: str = typer.Argument(..., help="Trigger ID returned by /trigger"), + message: str | None = typer.Argument(None, help="Message to deliver; stdin is used when omitted"), + workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"), + config: str | None = typer.Option(None, "--config", "-c", help="Config file path"), +): + """Deliver a local trigger message to its bound chat session.""" + from nanobot.triggers.store import ( + ExternalTriggerStore, + TriggerDisabledError, + TriggerNotFoundError, + TriggerStoreError, + ) + + runtime_config = _load_runtime_config(config, workspace) + content = _read_trigger_cli_message(message) + store = ExternalTriggerStore(runtime_config.workspace_path) + try: + delivery = store.enqueue(trigger_id, content) + except (TriggerNotFoundError, TriggerDisabledError) as exc: + console.print(f"[red]Error: {exc}[/red]") + raise typer.Exit(1) from exc + except (TriggerStoreError, ValueError) as exc: + console.print(f"[red]Error: {exc}[/red]") + raise typer.Exit(1) from exc + console.print(f"[green]Queued[/green] {delivery.trigger_id} ({delivery.id})") + + # ============================================================================ # OpenAI-Compatible API Server # ============================================================================ @@ -865,6 +909,8 @@ def _run_gateway( from nanobot.providers.image_generation import image_gen_provider_configs from nanobot.session.manager import SessionManager from nanobot.session.webui_turns import WebuiTurnCoordinator + from nanobot.triggers.runner import run_external_trigger_queue + from nanobot.triggers.store import ExternalTriggerStore from nanobot.webui.token_usage import TokenUsageHook port = port if port is not None else config.gateway.port @@ -887,6 +933,7 @@ def _run_gateway( # Create cron service with workspace-scoped store cron_store_path = config.workspace_path / "cron" / "jobs.json" cron = CronService(cron_store_path) + trigger_store = ExternalTriggerStore(config.workspace_path) # Create agent with cron service agent = AgentLoop.from_config( @@ -907,6 +954,7 @@ def _run_gateway( sessions=session_manager, schedule_background=lambda coro: agent._schedule_background(coro), ).subscribe(runtime_events) + agent.external_trigger_store = trigger_store from nanobot.bus.events import OutboundMessage from nanobot.session.keys import session_key_for_channel @@ -1103,6 +1151,7 @@ def _run_gateway( bus, session_manager=session_manager, cron_service=cron, + external_trigger_store=trigger_store, webui_runtime_model_name=_webui_runtime_model_name, webui_cron_pending_job_ids=getattr(agent, "pending_cron_job_ids_for_session", None), webui_static_dist=webui_static_dist, @@ -1245,6 +1294,10 @@ def _run_gateway( tasks = [ asyncio.create_task(agent.run(), name="nanobot-agent-loop"), asyncio.create_task(channels.start_all(), name="nanobot-channels"), + asyncio.create_task( + run_external_trigger_queue(store=trigger_store, bus=bus), + name="nanobot-external-triggers", + ), ] if health_server_enabled: tasks.append(asyncio.create_task( diff --git a/nanobot/command/builtin.py b/nanobot/command/builtin.py index 42048549..6e95d443 100644 --- a/nanobot/command/builtin.py +++ b/nanobot/command/builtin.py @@ -81,6 +81,13 @@ BUILTIN_COMMAND_SPECS: tuple[BuiltinCommandSpec, ...] = ( "activity", "", ), + BuiltinCommandSpec( + "/trigger", + "Create local trigger", + "Create a CLI trigger bound to this chat session.", + "zap", + "[name]", + ), BuiltinCommandSpec( "/dream", "Run Dream", @@ -718,6 +725,43 @@ async def cmd_skill(ctx: CommandContext) -> OutboundMessage: metadata=dict(ctx.msg.metadata or {}), ) + +async def cmd_trigger(ctx: CommandContext) -> OutboundMessage: + """Create a local trigger bound to the current session.""" + from nanobot.triggers.store import ExternalTriggerStore + + loop = ctx.loop + workspace = getattr(loop, "workspace", None) + if workspace is None: + workspace = getattr(getattr(loop, "context", None), "workspace", None) + if workspace is None: + raise RuntimeError("workspace unavailable for trigger creation") + + store = getattr(loop, "external_trigger_store", None) + if store is None: + store = ExternalTriggerStore(workspace) + + name = ctx.args.strip() or "External trigger" + trigger = store.create( + name=name, + channel=ctx.msg.channel, + chat_id=ctx.msg.chat_id, + session_key=ctx.key, + sender_id="trigger", + origin_metadata=dict(ctx.msg.metadata or {}), + ) + command = f'nanobot trigger {trigger.id} "message"' + return OutboundMessage( + channel=ctx.msg.channel, + chat_id=ctx.msg.chat_id, + content=( + f"Trigger created: {trigger.name}\n" + f"ID: {trigger.id}\n\n" + f"Command:\n{command}" + ), + metadata={**dict(ctx.msg.metadata or {}), "render_as": "text"}, + ) + async def cmd_help(ctx: CommandContext) -> OutboundMessage: """Return available slash commands.""" return OutboundMessage( @@ -752,6 +796,8 @@ def register_builtin_commands(router: CommandRouter) -> None: router.prefix("/history ", cmd_history) router.exact("/goal", cmd_goal) router.prefix("/goal ", cmd_goal) + router.exact("/trigger", cmd_trigger) + router.prefix("/trigger ", cmd_trigger) router.exact("/dream", cmd_dream) router.exact("/dream-log", cmd_dream_log) router.prefix("/dream-log ", cmd_dream_log) diff --git a/nanobot/command/router.py b/nanobot/command/router.py index 362a0b14..fdbe1a2a 100644 --- a/nanobot/command/router.py +++ b/nanobot/command/router.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Awaitable, Callable @@ -10,6 +11,26 @@ if TYPE_CHECKING: from nanobot.session.manager import Session Handler = Callable[["CommandContext"], Awaitable["OutboundMessage | None"]] +_BOT_SUFFIX_RE = re.compile(r"^[A-Za-z0-9_]+$") + + +def normalize_command_text(text: str) -> str: + """Normalize slash-command transport variants before routing. + + Telegram and Discord-style command dispatch can produce ``/cmd@bot args``. + The bot suffix belongs to the transport, not the command name, so strip it + once at the router boundary while preserving user arguments verbatim. + """ + stripped = text.strip() + if not stripped.startswith("/"): + return stripped + first, sep, rest = stripped.partition(" ") + if "@" not in first: + return stripped + command, suffix = first.rsplit("@", 1) + if command and suffix and _BOT_SUFFIX_RE.fullmatch(suffix): + return f"{command}{sep}{rest}" if sep else command + return stripped @dataclass @@ -50,7 +71,7 @@ class CommandRouter: self._prefix.sort(key=lambda p: len(p[0]), reverse=True) def is_priority(self, text: str) -> bool: - return text.strip().lower() in self._priority + return normalize_command_text(text).lower() in self._priority def is_dispatchable_command(self, text: str) -> bool: """Check whether *text* matches any non-priority command tier (exact or prefix). @@ -58,7 +79,7 @@ class CommandRouter: Does NOT check priority tier. If this returns True, ``dispatch()`` is guaranteed to match a handler. """ - cmd = text.strip().lower() + cmd = normalize_command_text(text).lower() if cmd in self._exact: return True for pfx, _ in self._prefix: @@ -68,6 +89,7 @@ class CommandRouter: async def dispatch_priority(self, ctx: CommandContext) -> OutboundMessage | None: """Dispatch a priority command. Called from run() without the lock.""" + ctx.raw = normalize_command_text(ctx.raw) handler = self._priority.get(ctx.raw.lower()) if handler: return await handler(ctx) @@ -75,6 +97,7 @@ class CommandRouter: async def dispatch(self, ctx: CommandContext) -> OutboundMessage | None: """Try exact, then prefix handlers. Returns None if unhandled.""" + ctx.raw = normalize_command_text(ctx.raw) cmd = ctx.raw.lower() if handler := self._exact.get(cmd): diff --git a/nanobot/triggers/__init__.py b/nanobot/triggers/__init__.py new file mode 100644 index 00000000..7a47510f --- /dev/null +++ b/nanobot/triggers/__init__.py @@ -0,0 +1,19 @@ +"""Local external trigger support.""" + +from nanobot.triggers.store import ( + ExternalTriggerStore, + TriggerDisabledError, + TriggerNotFoundError, + TriggerStoreError, +) +from nanobot.triggers.types import ExternalTrigger, TriggerDelivery, TriggerRunRecord + +__all__ = [ + "ExternalTrigger", + "ExternalTriggerStore", + "TriggerDelivery", + "TriggerDisabledError", + "TriggerNotFoundError", + "TriggerRunRecord", + "TriggerStoreError", +] diff --git a/nanobot/triggers/runner.py b/nanobot/triggers/runner.py new file mode 100644 index 00000000..b7dc8c99 --- /dev/null +++ b/nanobot/triggers/runner.py @@ -0,0 +1,120 @@ +"""Gateway delivery loop for local external triggers.""" + +from __future__ import annotations + +import asyncio +import uuid +from typing import Any + +from loguru import logger + +from nanobot.bus.events import InboundMessage +from nanobot.bus.queue import MessageBus +from nanobot.triggers.store import ExternalTriggerStore +from nanobot.triggers.types import ExternalTrigger, TriggerDelivery +from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN_METADATA_KEY + +EXTERNAL_TRIGGER_META = "_external_trigger" + + +async def run_external_trigger_queue( + *, + store: ExternalTriggerStore, + bus: MessageBus, + poll_interval_s: float = 0.5, + batch_size: int = 20, +) -> None: + """Poll local trigger deliveries and publish them as normal inbound messages.""" + logger.info("External trigger queue started") + while True: + deliveries = store.claim_deliveries(limit=batch_size) + if not deliveries: + await asyncio.sleep(poll_interval_s) + continue + + for delivery in deliveries: + try: + await _publish_delivery(store, bus, delivery) + store.complete_delivery(delivery) + except asyncio.CancelledError as exc: + store.retry_delivery(delivery, str(exc) or exc.__class__.__name__) + raise + except _TerminalDeliveryError as exc: + store.record_delivery( + delivery.trigger_id, + status="error", + error=str(exc), + run_at_ms=delivery.created_at_ms, + ) + store.complete_delivery(delivery) + logger.warning( + "Trigger: dropped delivery {} for {}: {}", + delivery.id, + delivery.trigger_id, + exc, + ) + except Exception as exc: + error = str(exc) or exc.__class__.__name__ + retried = store.retry_delivery(delivery, error) + store.record_delivery( + delivery.trigger_id, + status="error", + error=error, + run_at_ms=delivery.created_at_ms, + ) + logger.exception( + "Trigger: failed delivery {} for {}{}", + delivery.id, + delivery.trigger_id, + "; queued retry" if retried else "; moved to failed queue", + ) + + +class _TerminalDeliveryError(RuntimeError): + pass + + +async def _publish_delivery( + store: ExternalTriggerStore, + bus: MessageBus, + delivery: TriggerDelivery, +) -> None: + trigger = store.get(delivery.trigger_id) + if trigger is None: + raise _TerminalDeliveryError("trigger not found") + 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, + ) + ) + store.record_delivery( + trigger.id, + status="ok", + run_at_ms=delivery.created_at_ms, + ) + + +def _delivery_metadata(trigger: ExternalTrigger, delivery: TriggerDelivery) -> dict[str, Any]: + metadata = dict(trigger.origin_metadata or {}) + metadata[EXTERNAL_TRIGGER_META] = { + "trigger_id": trigger.id, + "trigger_name": trigger.name, + "delivery_id": delivery.id, + "created_at_ms": delivery.created_at_ms, + } + if trigger.channel == "websocket": + metadata.pop(WEBUI_TURN_METADATA_KEY, None) + metadata[WEBUI_TURN_METADATA_KEY] = f"trigger:{trigger.id}:{uuid.uuid4().hex}" + source: dict[str, str] = {"kind": "trigger"} + if trigger.name: + source["label"] = trigger.name + metadata[WEBUI_MESSAGE_SOURCE_METADATA_KEY] = source + return metadata diff --git a/nanobot/triggers/store.py b/nanobot/triggers/store.py new file mode 100644 index 00000000..153119b3 --- /dev/null +++ b/nanobot/triggers/store.py @@ -0,0 +1,344 @@ +"""Workspace-scoped local trigger store and delivery queue.""" + +from __future__ import annotations + +import json +import os +import secrets +import time +import uuid +from contextlib import suppress +from pathlib import Path +from typing import Any + +from filelock import FileLock +from loguru import logger + +from nanobot.triggers.types import ExternalTrigger, TriggerDelivery, TriggerRunRecord + +_TRIGGER_ID_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" +_MAX_RUN_HISTORY = 20 +_MAX_DELIVERY_ATTEMPTS = 10 + + +class TriggerStoreError(RuntimeError): + """Base class for trigger store errors.""" + + +class TriggerNotFoundError(TriggerStoreError): + """Raised when a trigger ID does not exist.""" + + +class TriggerDisabledError(TriggerStoreError): + """Raised when a trigger is disabled.""" + + +class ExternalTriggerStore: + """Persistent local triggers for one workspace.""" + + def __init__(self, workspace_path: Path): + self.workspace_path = Path(workspace_path) + self.root = self.workspace_path / "triggers" + self.store_path = self.root / "triggers.json" + self.inbox_dir = self.root / "inbox" + self.processing_dir = self.root / "processing" + self.failed_dir = self.root / "failed" + self._lock = FileLock(str(self.root / ".lock")) + + def create( + self, + *, + name: str, + channel: str, + chat_id: str, + session_key: str, + sender_id: str = "trigger", + origin_metadata: dict[str, Any] | None = None, + ) -> ExternalTrigger: + """Create a new session-bound external trigger.""" + clean_name = _clean_name(name) + channel = channel.strip() + chat_id = chat_id.strip() + session_key = session_key.strip() + if not channel or not chat_id or not session_key: + raise ValueError("channel, chat_id, and session_key are required") + + now = _now_ms() + self._ensure_dirs() + with self._lock: + triggers = self._load_triggers_unlocked() + existing_ids = {trigger.id for trigger in triggers} + trigger_id = _new_trigger_id(existing_ids) + trigger = ExternalTrigger( + id=trigger_id, + name=clean_name, + enabled=True, + channel=channel, + chat_id=chat_id, + session_key=session_key, + sender_id=sender_id.strip() or "trigger", + origin_metadata=dict(origin_metadata or {}), + created_at_ms=now, + updated_at_ms=now, + ) + triggers.append(trigger) + self._save_triggers_unlocked(triggers) + return trigger + + def list_triggers(self, *, include_disabled: bool = False) -> list[ExternalTrigger]: + """List triggers in this workspace.""" + self._ensure_dirs() + with self._lock: + triggers = self._load_triggers_unlocked() + if not include_disabled: + triggers = [trigger for trigger in triggers if trigger.enabled] + return sorted(triggers, key=lambda trigger: (trigger.updated_at_ms, trigger.id), reverse=True) + + def list_for_session( + self, + session_key: str, + *, + include_disabled: bool = True, + ) -> list[ExternalTrigger]: + """List triggers bound to one session key.""" + return [ + trigger + for trigger in self.list_triggers(include_disabled=include_disabled) + if trigger.session_key == session_key + ] + + def get(self, trigger_id: str) -> ExternalTrigger | None: + """Return one trigger by ID.""" + self._ensure_dirs() + with self._lock: + return self._find_unlocked(self._load_triggers_unlocked(), trigger_id) + + def enable(self, trigger_id: str, *, enabled: bool) -> ExternalTrigger | None: + """Enable or disable a trigger.""" + self._ensure_dirs() + with self._lock: + triggers = self._load_triggers_unlocked() + trigger = self._find_unlocked(triggers, trigger_id) + if trigger is None: + return None + trigger.enabled = enabled + trigger.updated_at_ms = _now_ms() + self._save_triggers_unlocked(triggers) + return trigger + + def update(self, trigger_id: str, *, name: str | None = None) -> ExternalTrigger | None: + """Update mutable trigger fields.""" + self._ensure_dirs() + with self._lock: + triggers = self._load_triggers_unlocked() + trigger = self._find_unlocked(triggers, trigger_id) + if trigger is None: + return None + if name is not None: + trigger.name = _clean_name(name) + trigger.updated_at_ms = _now_ms() + self._save_triggers_unlocked(triggers) + return trigger + + def delete(self, trigger_id: str) -> bool: + """Delete a trigger by ID.""" + self._ensure_dirs() + with self._lock: + triggers = self._load_triggers_unlocked() + remaining = [trigger for trigger in triggers if trigger.id != trigger_id] + if len(remaining) == len(triggers): + return False + self._save_triggers_unlocked(remaining) + return True + + def enqueue(self, trigger_id: str, content: str) -> TriggerDelivery: + """Queue a delivery for the gateway process to consume.""" + trigger_id = trigger_id.strip() + if not content.strip(): + raise ValueError("trigger message is required") + self._ensure_dirs() + with self._lock: + trigger = self._find_unlocked(self._load_triggers_unlocked(), trigger_id) + if trigger is None: + raise TriggerNotFoundError(f"trigger not found: {trigger_id}") + if not trigger.enabled: + raise TriggerDisabledError(f"trigger is disabled: {trigger_id}") + delivery = TriggerDelivery( + id=f"tdl_{uuid.uuid4().hex[:12]}", + trigger_id=trigger_id, + content=content, + created_at_ms=_now_ms(), + ) + 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 + return delivery + + def claim_deliveries(self, *, limit: int = 20) -> list[TriggerDelivery]: + """Move pending deliveries into processing and return them.""" + self._ensure_dirs() + claimed: list[TriggerDelivery] = [] + with self._lock: + for path in sorted(self.inbox_dir.glob("*.json"))[: max(0, limit)]: + try: + data = json.loads(path.read_text(encoding="utf-8")) + delivery = TriggerDelivery.from_dict( + data.get("delivery", data), + path=self.processing_dir / path.name, + ) + except Exception: + logger.exception("Trigger: failed to parse delivery {}", path) + self._move_bad_delivery_unlocked(path) + continue + os.replace(path, delivery.path) + claimed.append(delivery) + return claimed + + def complete_delivery(self, delivery: TriggerDelivery) -> None: + """Delete a claimed delivery after it is handled.""" + if delivery.path is None: + return + self._ensure_dirs() + with self._lock: + delivery.path.unlink(missing_ok=True) + + def retry_delivery(self, delivery: TriggerDelivery, error: str) -> bool: + """Retry a claimed delivery unless it exceeded the attempt limit.""" + if delivery.path is None: + 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 + + def record_delivery( + self, + trigger_id: str, + *, + status: str, + error: str | None = None, + run_at_ms: int | None = None, + ) -> None: + """Record the latest delivery status on a trigger.""" + self._ensure_dirs() + run_at_ms = run_at_ms or _now_ms() + with self._lock: + triggers = self._load_triggers_unlocked() + trigger = self._find_unlocked(triggers, trigger_id) + if trigger is None: + return + trigger.last_run_at_ms = run_at_ms + trigger.last_status = "ok" if status == "ok" else "error" + trigger.last_error = None if status == "ok" else (error or "delivery failed") + trigger.updated_at_ms = _now_ms() + trigger.run_history.append( + TriggerRunRecord( + run_at_ms=run_at_ms, + status=trigger.last_status, + error=trigger.last_error, + ) + ) + trigger.run_history = trigger.run_history[-_MAX_RUN_HISTORY:] + self._save_triggers_unlocked(triggers) + + 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) + + def _load_triggers_unlocked(self) -> list[ExternalTrigger]: + if not self.store_path.exists(): + return [] + try: + data = json.loads(self.store_path.read_text(encoding="utf-8")) + return [ + ExternalTrigger.from_dict(raw) + for raw in data.get("triggers", []) + if isinstance(raw, dict) + ] + except Exception as exc: + backup = self.store_path.with_suffix( + self.store_path.suffix + f".corrupt-{int(time.time())}" + ) + with suppress(OSError): + os.replace(self.store_path, backup) + raise TriggerStoreError( + f"trigger store at {self.store_path} could not be loaded and was preserved " + "as a .corrupt- backup" + ) from exc + + def _save_triggers_unlocked(self, triggers: list[ExternalTrigger]) -> None: + payload = { + "version": 1, + "triggers": [trigger.to_dict() for trigger in triggers], + } + self._atomic_write(self.store_path, json.dumps(payload, indent=2, ensure_ascii=False)) + + @staticmethod + def _find_unlocked( + triggers: list[ExternalTrigger], + trigger_id: str, + ) -> ExternalTrigger | None: + return next((trigger for trigger in triggers if trigger.id == trigger_id), None) + + def _move_bad_delivery_unlocked(self, path: Path) -> None: + target = self.failed_dir / f"{path.name}.bad" + with suppress(OSError): + os.replace(path, target) + + @staticmethod + 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 + + +def _new_trigger_id(existing_ids: set[str]) -> str: + for _ in range(100): + suffix = "".join(secrets.choice(_TRIGGER_ID_ALPHABET) for _ in range(8)) + candidate = f"trg_{suffix}" + if candidate not in existing_ids: + return candidate + raise TriggerStoreError("could not allocate a unique trigger id") + + +def _clean_name(name: str) -> str: + stripped = " ".join(name.strip().split()) + return (stripped or "External trigger")[:120] + + +def _now_ms() -> int: + return int(time.time() * 1000) + + +def _delivery_payload(delivery: TriggerDelivery) -> dict[str, Any]: + return { + "version": 1, + "delivery": delivery.to_dict(), + } diff --git a/nanobot/triggers/types.py b/nanobot/triggers/types.py new file mode 100644 index 00000000..93f3d14b --- /dev/null +++ b/nanobot/triggers/types.py @@ -0,0 +1,141 @@ +"""Persistent types for local external triggers.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Literal + +TriggerStatus = Literal["ok", "error"] + + +def _get(data: dict[str, Any], camel: str, snake: str, default: Any = None) -> Any: + if camel in data: + return data[camel] + return data.get(snake, default) + + +@dataclass +class TriggerRunRecord: + """A single local trigger delivery record.""" + + run_at_ms: int + status: TriggerStatus + error: str | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "TriggerRunRecord": + return cls( + run_at_ms=int(_get(data, "runAtMs", "run_at_ms", 0)), + status=str(data.get("status") or "error"), # type: ignore[arg-type] + error=data.get("error"), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "runAtMs": self.run_at_ms, + "status": self.status, + "error": self.error, + } + + +@dataclass +class ExternalTrigger: + """A session-bound local trigger.""" + + id: str + name: str + enabled: bool + channel: str + chat_id: str + session_key: str + sender_id: str = "trigger" + origin_metadata: dict[str, Any] = field(default_factory=dict) + created_at_ms: int = 0 + updated_at_ms: int = 0 + last_run_at_ms: int | None = None + last_status: TriggerStatus | None = None + last_error: str | None = None + run_history: list[TriggerRunRecord] = field(default_factory=list) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "ExternalTrigger": + history = [ + record if isinstance(record, TriggerRunRecord) else TriggerRunRecord.from_dict(record) + for record in data.get("runHistory", data.get("run_history", [])) + if isinstance(record, (dict, TriggerRunRecord)) + ] + return cls( + id=str(data["id"]), + name=str(data.get("name") or data["id"]), + enabled=bool(data.get("enabled", True)), + channel=str(data.get("channel") or ""), + chat_id=str(_get(data, "chatId", "chat_id", "")), + session_key=str(_get(data, "sessionKey", "session_key", "")), + sender_id=str(_get(data, "senderId", "sender_id", "trigger") or "trigger"), + origin_metadata=dict(_get(data, "originMetadata", "origin_metadata", {}) or {}), + created_at_ms=int(_get(data, "createdAtMs", "created_at_ms", 0)), + updated_at_ms=int(_get(data, "updatedAtMs", "updated_at_ms", 0)), + last_run_at_ms=_get(data, "lastRunAtMs", "last_run_at_ms"), + last_status=_get(data, "lastStatus", "last_status"), # type: ignore[arg-type] + last_error=_get(data, "lastError", "last_error"), + run_history=history, + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "name": self.name, + "enabled": self.enabled, + "channel": self.channel, + "chatId": self.chat_id, + "sessionKey": self.session_key, + "senderId": self.sender_id, + "originMetadata": self.origin_metadata, + "createdAtMs": self.created_at_ms, + "updatedAtMs": self.updated_at_ms, + "lastRunAtMs": self.last_run_at_ms, + "lastStatus": self.last_status, + "lastError": self.last_error, + "runHistory": [record.to_dict() for record in self.run_history], + } + + +@dataclass +class TriggerDelivery: + """One pending local trigger delivery written by the CLI.""" + + id: str + trigger_id: str + content: str + created_at_ms: int + attempts: int = 0 + last_error: str | None = None + path: Path | None = field(default=None, compare=False, repr=False) + + @classmethod + def from_dict( + cls, + data: dict[str, Any], + *, + path: Path | None = None, + ) -> "TriggerDelivery": + return cls( + id=str(data["id"]), + trigger_id=str(_get(data, "triggerId", "trigger_id", "")), + content=str(data.get("content") or ""), + created_at_ms=int(_get(data, "createdAtMs", "created_at_ms", 0)), + attempts=int(data.get("attempts", 0)), + last_error=data.get("lastError") or data.get("last_error"), + path=path, + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "triggerId": self.trigger_id, + "content": self.content, + "createdAtMs": self.created_at_ms, + "attempts": self.attempts, + "lastError": self.last_error, + } diff --git a/nanobot/webui/gateway_services.py b/nanobot/webui/gateway_services.py index fa94b8f6..b6ae3a2b 100644 --- a/nanobot/webui/gateway_services.py +++ b/nanobot/webui/gateway_services.py @@ -26,6 +26,7 @@ class GatewayServices: workspaces: WebUIWorkspaceController session_manager: Any | None cron_service: Any | None + external_trigger_store: Any | None cron_pending_job_ids: Callable[[str], set[str]] | None @@ -42,6 +43,7 @@ def build_gateway_services( runtime_capabilities_overrides: dict[str, Any] | None, disabled_skills: set[str] | None = None, cron_service: Any | None = None, + external_trigger_store: Any | None = None, cron_pending_job_ids: Callable[[str], set[str]] | None = None, logger: Any = default_logger, ) -> GatewayServices: @@ -70,6 +72,7 @@ def build_gateway_services( skills_workspace_path=workspace_path, disabled_skills=disabled_skills, cron_service=cron_service, + external_trigger_store=external_trigger_store, cron_pending_job_ids=cron_pending_job_ids, log=logger, ) @@ -81,5 +84,6 @@ def build_gateway_services( workspaces=workspaces, session_manager=session_manager, cron_service=cron_service, + external_trigger_store=external_trigger_store, cron_pending_job_ids=cron_pending_job_ids, ) diff --git a/nanobot/webui/session_automations.py b/nanobot/webui/session_automations.py index 55e748d6..e22be27e 100644 --- a/nanobot/webui/session_automations.py +++ b/nanobot/webui/session_automations.py @@ -8,6 +8,9 @@ from typing import Any, Protocol from nanobot.cron.session_turns import CRON_HISTORY_META from nanobot.cron.types import CronJob from nanobot.session.manager import _message_preview_text +from nanobot.triggers.types import ExternalTrigger + +AutomationJob = CronJob | ExternalTrigger class _CronServiceLike(Protocol): @@ -21,6 +24,17 @@ class _CronServiceLike(Protocol): ) -> list[CronJob]: ... +class _ExternalTriggerStoreLike(Protocol): + def list_triggers(self, *, include_disabled: bool = False) -> list[ExternalTrigger]: ... + + def list_for_session( + self, + session_key: str, + *, + include_disabled: bool = True, + ) -> list[ExternalTrigger]: ... + + class _SessionManagerLike(Protocol): def read_session_file(self, key: str) -> dict[str, Any] | None: ... @@ -28,26 +42,43 @@ class _SessionManagerLike(Protocol): def session_automation_jobs( cron_service: _CronServiceLike | None, session_key: str, -) -> list[CronJob]: + *, + external_trigger_store: _ExternalTriggerStoreLike | None = None, +) -> list[AutomationJob]: """Return user automations attached to the WebUI session.""" - if cron_service is None: - return [] - return cron_service.list_bound_cron_jobs_for_session( - session_key, - include_disabled=True, - ) + jobs: list[AutomationJob] = [] + if cron_service is not None: + jobs.extend( + cron_service.list_bound_cron_jobs_for_session( + session_key, + include_disabled=True, + ) + ) + if external_trigger_store is not None: + jobs.extend( + external_trigger_store.list_for_session( + session_key, + include_disabled=True, + ) + ) + return jobs def session_automations_payload( cron_service: _CronServiceLike | None, session_key: str, *, + external_trigger_store: _ExternalTriggerStoreLike | None = None, pending_job_ids: Collection[str] | None = None, ) -> dict[str, Any]: """Return user-created automation jobs attached to a WebUI session.""" return { "jobs": serialize_automation_jobs( - session_automation_jobs(cron_service, session_key), + session_automation_jobs( + cron_service, + session_key, + external_trigger_store=external_trigger_store, + ), pending_job_ids=pending_job_ids, ) } @@ -56,11 +87,16 @@ def session_automations_payload( def all_automations_payload( cron_service: _CronServiceLike | None, *, + external_trigger_store: _ExternalTriggerStoreLike | None = None, session_manager: _SessionManagerLike | None = None, pending_job_ids: Collection[str] | None = None, ) -> dict[str, Any]: """Return all cron jobs visible to the WebUI automation manager.""" - jobs = cron_service.list_jobs(include_disabled=True) if cron_service is not None else [] + jobs: list[AutomationJob] = [] + if cron_service is not None: + jobs.extend(cron_service.list_jobs(include_disabled=True)) + if external_trigger_store is not None: + jobs.extend(external_trigger_store.list_triggers(include_disabled=True)) return { "jobs": serialize_automation_jobs( jobs, @@ -72,7 +108,7 @@ def all_automations_payload( def serialize_automation_jobs( - jobs: list[CronJob], + jobs: list[AutomationJob], *, pending_job_ids: Collection[str] | None = None, include_details: bool = False, @@ -90,12 +126,19 @@ def serialize_automation_jobs( def _serialize_job( - job: CronJob, + job: AutomationJob, *, pending: bool = False, include_details: bool = False, session_manager: _SessionManagerLike | None = None, ) -> dict[str, Any]: + if isinstance(job, ExternalTrigger): + return _serialize_trigger( + job, + include_details=include_details, + session_manager=session_manager, + ) + payload = { "id": job.id, "name": job.name, @@ -143,6 +186,66 @@ def _serialize_job( return payload +def _serialize_trigger( + trigger: ExternalTrigger, + *, + include_details: bool = False, + session_manager: _SessionManagerLike | None = None, +) -> dict[str, Any]: + command = f'nanobot trigger {trigger.id} "message"' + payload = { + "id": trigger.id, + "name": trigger.name, + "enabled": trigger.enabled, + "kind": "external_trigger", + "schedule": { + "kind": "external", + "at_ms": None, + "every_ms": None, + "expr": None, + "tz": None, + }, + "payload": { + "kind": "external_trigger", + "message": command, + "command": command, + }, + "state": { + "next_run_at_ms": None, + "last_status": trigger.last_status, + "pending": False, + }, + } + if not include_details: + return payload + + payload["protected"] = False + payload["delete_after_run"] = False + payload["created_at_ms"] = trigger.created_at_ms + payload["updated_at_ms"] = trigger.updated_at_ms + payload["state"].update( + { + "last_run_at_ms": trigger.last_run_at_ms, + "last_error": trigger.last_error, + "run_history": [ + { + "run_at_ms": record.run_at_ms, + "status": record.status, + "duration_ms": 0, + "error": record.error, + } + for record in trigger.run_history[-5:] + ], + } + ) + payload["origin"] = _trigger_origin_payload(trigger, session_manager) + payload["trigger"] = { + "id": trigger.id, + "command": command, + } + return payload + + def _origin_payload( job: CronJob, session_manager: _SessionManagerLike | None, @@ -161,6 +264,46 @@ def _origin_payload( } session_key = f"{channel}:{chat_id}" + return _websocket_origin_payload( + session_key=session_key, + channel=channel, + chat_id=chat_id, + session_manager=session_manager, + ) + + +def _trigger_origin_payload( + trigger: ExternalTrigger, + session_manager: _SessionManagerLike | None, +) -> dict[str, Any] | None: + channel = trigger.channel + chat_id = trigger.chat_id + if not channel or not chat_id: + return None + if channel != "websocket": + return { + "channel": channel, + "title": "", + "preview": "", + } + + return _websocket_origin_payload( + session_key=trigger.session_key or f"{channel}:{chat_id}", + channel=channel, + chat_id=chat_id, + session_manager=session_manager, + ) + + +def _websocket_origin_payload( + *, + session_key: str, + channel: str, + chat_id: str, + session_manager: _SessionManagerLike | None, +) -> dict[str, Any]: + title = "" + preview = "" if session_manager is not None: data = session_manager.read_session_file(session_key) if isinstance(data, dict): diff --git a/nanobot/webui/ws_http.py b/nanobot/webui/ws_http.py index 9e2981f9..83a58e97 100644 --- a/nanobot/webui/ws_http.py +++ b/nanobot/webui/ws_http.py @@ -26,6 +26,7 @@ from websockets.http11 import Response from nanobot.command.builtin import builtin_command_palette from nanobot.cron.session_turns import is_bound_cron_job from nanobot.cron.types import CronJob, CronSchedule +from nanobot.triggers.types import ExternalTrigger from nanobot.utils.subagent_channel_display import scrub_subagent_messages_for_channel from nanobot.webui.file_preview import WebUIFilePreviewError, file_preview_payload from nanobot.webui.gateway_tokens import GatewayTokenStore, token_response_payload @@ -89,6 +90,7 @@ if TYPE_CHECKING: from nanobot.bus.queue import MessageBus from nanobot.cron.service import CronService from nanobot.session.manager import SessionManager + from nanobot.triggers.store import ExternalTriggerStore def _decode_api_key(raw_key: str) -> str | None: @@ -153,6 +155,7 @@ class GatewayHTTPHandler: skills_workspace_path: Path, disabled_skills: set[str] | None = None, cron_service: CronService | None = None, + external_trigger_store: ExternalTriggerStore | None = None, cron_pending_job_ids: Callable[[str], set[str]] | None = None, log: Any = logger, ) -> None: @@ -167,6 +170,7 @@ class GatewayHTTPHandler: self.skills_workspace_path = skills_workspace_path self.disabled_skills = disabled_skills or set() self.cron_service = cron_service + self.external_trigger_store = external_trigger_store self.cron_pending_job_ids = cron_pending_job_ids self._log = log self._runtime_surface = runtime_surface @@ -490,6 +494,7 @@ class GatewayHTTPHandler: session_automations_payload( self.cron_service, decoded_key, + external_trigger_store=self.external_trigger_store, pending_job_ids=pending_job_ids, ) ) @@ -506,7 +511,11 @@ class GatewayHTTPHandler: return _http_error(404, "session not found") query = _parse_query(request.path) delete_automations = (_query_first(query, "delete_automations") or "").lower() - automation_jobs = session_automation_jobs(self.cron_service, decoded_key) + automation_jobs = session_automation_jobs( + self.cron_service, + decoded_key, + external_trigger_store=self.external_trigger_store, + ) if automation_jobs and delete_automations not in {"1", "true", "yes"}: return _http_json_response( { @@ -515,9 +524,13 @@ class GatewayHTTPHandler: "automations": serialize_automation_jobs(automation_jobs), } ) - if automation_jobs and self.cron_service is not None: + if automation_jobs: for job in automation_jobs: - self.cron_service.remove_job(job.id) + if isinstance(job, ExternalTrigger): + if self.external_trigger_store is not None: + self.external_trigger_store.delete(job.id) + elif self.cron_service is not None: + self.cron_service.remove_job(job.id) deleted = self.session_manager.delete_session(decoded_key) delete_webui_thread(decoded_key) return _http_json_response({"deleted": bool(deleted)}) @@ -554,6 +567,7 @@ class GatewayHTTPHandler: return _http_json_response( all_automations_payload( self.cron_service, + external_trigger_store=self.external_trigger_store, session_manager=self.session_manager, pending_job_ids=self._pending_cron_job_ids_for_all(), ) @@ -566,13 +580,19 @@ class GatewayHTTPHandler: ) -> Response: if not self.check_api_token(request): return _http_error(401, "Unauthorized") - if self.cron_service is None: - return _http_error(503, "cron service unavailable") + if self.cron_service is None and self.external_trigger_store is None: + return _http_error(503, "automation service unavailable") query = _parse_query(request.path) job_id = (_query_first(query, "id") or _query_first(query, "job_id") or "").strip() if not job_id: return _http_error(400, "missing automation id") + trigger = self.external_trigger_store.get(job_id) if self.external_trigger_store else None + if trigger is not None: + return self._handle_external_trigger_action(request, action, trigger) + + if self.cron_service is None: + return _http_error(404, "automation not found") job = self.cron_service.get_job(job_id) if job is None: return _http_error(404, "automation not found") @@ -618,6 +638,40 @@ class GatewayHTTPHandler: return self._handle_webui_automations(request) + def _handle_external_trigger_action( + self, + request: WsRequest, + action: str, + trigger: ExternalTrigger, + ) -> Response: + if self.external_trigger_store is None: + return _http_error(503, "trigger service unavailable") + if action == "enable": + if self.external_trigger_store.enable(trigger.id, enabled=True) is None: + return _http_error(404, "automation not found") + elif action == "disable": + if self.external_trigger_store.enable(trigger.id, enabled=False) is None: + return _http_error(404, "automation not found") + elif action == "delete": + if not self.external_trigger_store.delete(trigger.id): + return _http_error(404, "automation not found") + elif action == "run": + return _http_error(409, "external trigger requires a CLI message") + elif action == "update": + values = _automation_values_from_request(request) + if values is None: + return _http_error(400, "invalid automation update payload") + parsed = _parse_external_trigger_update(values) + if isinstance(parsed, str): + return _http_error(400, parsed) + if parsed: + if self.external_trigger_store.update(trigger.id, **parsed) is None: + return _http_error(404, "automation not found") + else: + return _http_error(404, "unknown automation action") + + return self._handle_webui_automations(request) + @staticmethod def _log_automation_run_result(task: asyncio.Task[bool]) -> None: try: @@ -830,6 +884,22 @@ def _parse_automation_update( return update +def _parse_external_trigger_update(values: dict[str, Any]) -> dict[str, Any] | str: + update: dict[str, Any] = {} + if "name" in values: + raw_name = values.get("name") + if not isinstance(raw_name, str): + return "name must be a string" + name = raw_name.strip() + if not name: + return "name cannot be empty" + update["name"] = name + forbidden = [key for key in ("message", "schedule") if key in values] + if forbidden: + return "external trigger updates only support name" + return update + + def _parse_automation_schedule(values: dict[str, Any]) -> CronSchedule | str: raw_kind = values.get("kind") if not isinstance(raw_kind, str): diff --git a/tests/channels/test_discord_channel.py b/tests/channels/test_discord_channel.py index 223d7fa4..1e3142f7 100644 --- a/tests/channels/test_discord_channel.py +++ b/tests/channels/test_discord_channel.py @@ -867,7 +867,7 @@ async def test_slash_new_is_blocked_for_disallowed_user() -> None: assert handled == [] -@pytest.mark.parametrize("slash_name", ["stop", "restart", "status", "history", "model"]) +@pytest.mark.parametrize("slash_name", ["stop", "restart", "status", "history", "model", "trigger"]) @pytest.mark.asyncio async def test_slash_commands_forward_via_handle_message(slash_name: str) -> None: channel = DiscordChannel(DiscordConfig(enabled=True, allow_from=["*"]), MessageBus()) @@ -918,6 +918,31 @@ async def test_slash_model_forwards_optional_preset() -> None: assert handled[0]["metadata"]["is_slash_command"] is True +@pytest.mark.asyncio +async def test_slash_trigger_forwards_optional_name() -> None: + channel = DiscordChannel(DiscordConfig(enabled=True, allow_from=["*"]), MessageBus()) + handled: list[dict] = [] + + async def capture_handle(**kwargs) -> None: + handled.append(kwargs) + + channel._handle_message = capture_handle # type: ignore[method-assign] + client = DiscordBotClient(channel, intents=discord.Intents.none()) + interaction = _make_interaction() + interaction.command.qualified_name = "trigger" + + trigger_cmd = client.tree.get_command("trigger") + assert trigger_cmd is not None + await trigger_cmd.callback(interaction, name="PR review") + + assert interaction.response.messages == [ + {"content": "Processing /trigger PR review...", "ephemeral": True} + ] + assert len(handled) == 1 + assert handled[0]["content"] == "/trigger PR review" + assert handled[0]["metadata"]["is_slash_command"] is True + + @pytest.mark.asyncio async def test_slash_help_returns_ephemeral_help_text() -> None: channel = DiscordChannel(DiscordConfig(enabled=True, allow_from=["*"]), MessageBus()) diff --git a/tests/channels/test_telegram_channel.py b/tests/channels/test_telegram_channel.py index 462d4d0b..fccbf9ed 100644 --- a/tests/channels/test_telegram_channel.py +++ b/tests/channels/test_telegram_channel.py @@ -1577,12 +1577,14 @@ def test_telegram_bus_slash_command_regex_matches_agent_loop_commands() -> None: assert pat.fullmatch("/history") assert pat.fullmatch("/history 5") assert pat.fullmatch("/goal ship the feature") + assert pat.fullmatch("/trigger PR review") assert pat.fullmatch("/pairing list") assert pat.fullmatch("/model fast") assert pat.fullmatch("/skill") assert pat.fullmatch("/skill@nanobot_bot") assert pat.fullmatch("/new@nanobot_bot") assert pat.fullmatch("/goal@nanobot_bot refine objective") + assert pat.fullmatch("/trigger@nanobot_bot CI summary") assert pat.fullmatch("/dream-log deadbeef") is None assert pat.fullmatch("/dream-restore deadbeef") is None @@ -1606,6 +1608,7 @@ async def test_on_help_includes_restart_command() -> None: assert "/dream" in help_text assert "/dream-log" in help_text assert "/goal" in help_text + assert "/trigger" in help_text assert "/pairing" in help_text assert "/model" in help_text assert "/dream-restore" in help_text diff --git a/tests/channels/test_websocket_http_routes.py b/tests/channels/test_websocket_http_routes.py index d767b6f1..9e5dc0f6 100644 --- a/tests/channels/test_websocket_http_routes.py +++ b/tests/channels/test_websocket_http_routes.py @@ -20,6 +20,7 @@ from nanobot.cron.service import CronService from nanobot.cron.types import CronJob, CronPayload, CronSchedule from nanobot.session.keys import UNIFIED_SESSION_KEY from nanobot.session.manager import Session, SessionManager +from nanobot.triggers.store import ExternalTriggerStore from nanobot.webui.gateway_services import GatewayServices, build_gateway_services _PORT = 29900 @@ -46,6 +47,7 @@ def _make_handler( workspace_path: Path | None = None, runtime_model_name: Any | None = None, cron_service: CronService | None = None, + external_trigger_store: ExternalTriggerStore | None = None, cron_pending_job_ids: Any | None = None, ) -> GatewayServices: config = WebSocketConfig.model_validate(cfg) if isinstance(cfg, dict) else cfg @@ -61,6 +63,7 @@ def _make_handler( runtime_surface="browser", runtime_capabilities_overrides=None, cron_service=cron_service, + external_trigger_store=external_trigger_store, cron_pending_job_ids=cron_pending_job_ids, ) @@ -74,6 +77,7 @@ def _ch( port: int = _PORT, runtime_model_name: Any | None = None, cron_service: CronService | None = None, + external_trigger_store: ExternalTriggerStore | None = None, cron_pending_job_ids: Any | None = None, **extra: Any, ) -> WebSocketChannel: @@ -93,6 +97,7 @@ def _ch( workspace_path=workspace_path, runtime_model_name=runtime_model_name, cron_service=cron_service, + external_trigger_store=external_trigger_store, cron_pending_job_ids=cron_pending_job_ids, ) return WebSocketChannel(cfg, bus, gateway=gateway) @@ -319,6 +324,50 @@ async def test_session_automations_route_ignores_unified_owner( await server_task +@pytest.mark.asyncio +async def test_session_automations_route_lists_external_triggers( + bus: MagicMock, tmp_path: Path +) -> None: + port = _free_port() + base_url = f"http://127.0.0.1:{port}" + trigger_store = ExternalTriggerStore(tmp_path) + trigger = trigger_store.create( + name="PR review", + channel="websocket", + chat_id="abc", + session_key="websocket:abc", + ) + channel = _ch( + bus, + session_manager=_seed_session(tmp_path, key="websocket:abc"), + external_trigger_store=trigger_store, + port=port, + ) + server_task = asyncio.create_task(channel.start()) + await asyncio.sleep(0.3) + try: + boot = await _http_get(f"{base_url}/webui/bootstrap") + token = boot.json()["token"] + auth = {"Authorization": f"Bearer {token}"} + + resp = await _http_get( + f"{base_url}/api/sessions/websocket%3Aabc/automations", + headers=auth, + ) + + assert resp.status_code == 200 + body = resp.json() + assert [job["id"] for job in body["jobs"]] == [trigger.id] + job = body["jobs"][0] + assert job["kind"] == "external_trigger" + assert job["schedule"]["kind"] == "external" + assert job["payload"]["kind"] == "external_trigger" + assert job["payload"]["command"] == f'nanobot trigger {trigger.id} "message"' + finally: + await channel.stop() + await server_task + + @pytest.mark.asyncio async def test_webui_skills_route_requires_token_and_hides_paths( bus: MagicMock, tmp_path: Path @@ -1080,6 +1129,86 @@ async def test_webui_automations_route_lists_all_jobs_and_allows_user_actions( await server_task +@pytest.mark.asyncio +async def test_webui_automations_route_manages_external_triggers( + bus: MagicMock, tmp_path: Path +) -> None: + port = _free_port() + base_url = f"http://127.0.0.1:{port}" + trigger_store = ExternalTriggerStore(tmp_path) + trigger = trigger_store.create( + name="PR review", + channel="websocket", + chat_id="abc", + session_key="websocket:abc", + ) + channel = _ch( + bus, + session_manager=_seed_session(tmp_path, key="websocket:abc"), + external_trigger_store=trigger_store, + port=port, + ) + server_task = asyncio.create_task(channel.start()) + await asyncio.sleep(0.3) + try: + boot = await _http_get(f"{base_url}/webui/bootstrap") + token = boot.json()["token"] + auth = {"Authorization": f"Bearer {token}"} + + listed = await _http_get(f"{base_url}/api/webui/automations", headers=auth) + assert listed.status_code == 200 + by_id = {job["id"]: job for job in listed.json()["jobs"]} + assert by_id[trigger.id]["kind"] == "external_trigger" + assert by_id[trigger.id]["trigger"]["command"] == f'nanobot trigger {trigger.id} "message"' + + disabled = await _http_get( + f"{base_url}/api/webui/automations/disable?id={trigger.id}", + headers=auth, + ) + assert disabled.status_code == 200 + stored = trigger_store.get(trigger.id) + assert stored is not None + assert stored.enabled is False + + run = await _http_get( + f"{base_url}/api/webui/automations/run?id={trigger.id}", + headers=auth, + ) + assert run.status_code == 409 + assert "CLI message" in run.text + + renamed = await _http_get( + f"{base_url}/api/webui/automations/update?id={trigger.id}", + headers={ + **auth, + "X-Nanobot-Automation-Values": json.dumps({"name": "Release review"}), + }, + ) + assert renamed.status_code == 200 + stored = trigger_store.get(trigger.id) + assert stored is not None + assert stored.name == "Release review" + + bad_update = await _http_get( + f"{base_url}/api/webui/automations/update?id={trigger.id}", + headers={ + **auth, + "X-Nanobot-Automation-Values": json.dumps({"message": "coupled"}), + }, + ) + assert bad_update.status_code == 400 + + deleted = await _http_get( + f"{base_url}/api/webui/automations/delete?id={trigger.id}", + headers=auth, + ) + assert deleted.status_code == 200 + assert trigger_store.get(trigger.id) is None + finally: + await channel.stop() + await server_task + + @pytest.mark.asyncio async def test_session_delete_blocks_when_bound_automation_exists( bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch @@ -1121,6 +1250,54 @@ async def test_session_delete_blocks_when_bound_automation_exists( await server_task +@pytest.mark.asyncio +async def test_session_delete_blocks_and_cascades_external_triggers( + bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path) + port = _free_port() + base_url = f"http://127.0.0.1:{port}" + sm = _seed_session(tmp_path, key="websocket:doomed") + trigger_store = ExternalTriggerStore(tmp_path) + trigger = trigger_store.create( + name="PR review", + channel="websocket", + chat_id="doomed", + session_key="websocket:doomed", + ) + channel = _ch( + bus, + session_manager=sm, + external_trigger_store=trigger_store, + port=port, + ) + server_task = asyncio.create_task(channel.start()) + await asyncio.sleep(0.3) + try: + boot = await _http_get(f"{base_url}/webui/bootstrap") + token = boot.json()["token"] + auth = {"Authorization": f"Bearer {token}"} + + blocked = await _http_get( + f"{base_url}/api/sessions/websocket:doomed/delete", + headers=auth, + ) + assert blocked.status_code == 200 + assert blocked.json()["blocked_by_automations"] is True + assert trigger_store.get(trigger.id) is not None + + deleted = await _http_get( + f"{base_url}/api/sessions/websocket:doomed/delete?delete_automations=true", + headers=auth, + ) + assert deleted.status_code == 200 + assert deleted.json()["deleted"] is True + assert trigger_store.get(trigger.id) is None + finally: + await channel.stop() + await server_task + + @pytest.mark.asyncio async def test_session_delete_can_cascade_bound_automations( bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch diff --git a/tests/cli/test_commands.py b/tests/cli/test_commands.py index 0b30c068..c15a9686 100644 --- a/tests/cli/test_commands.py +++ b/tests/cli/test_commands.py @@ -2516,6 +2516,38 @@ def test_serve_uses_api_config_defaults_and_workspace_override( assert seen["api_key"] == "" +def test_trigger_cli_queues_message_in_workspace( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + from nanobot.triggers.store import ExternalTriggerStore + + config_file = _write_instance_config(tmp_path) + config = Config() + config.agents.defaults.workspace = str(tmp_path / "workspace") + _patch_cli_command_runtime(monkeypatch, config) + + store = ExternalTriggerStore(config.workspace_path) + trigger = store.create( + name="Review hook", + channel="websocket", + chat_id="chat-1", + session_key="websocket:chat-1", + ) + + result = runner.invoke( + app, + ["trigger", "--config", str(config_file), trigger.id, "Review PR #4502"], + ) + + assert result.exit_code == 0 + assert f"Queued {trigger.id}" in result.stdout + deliveries = store.claim_deliveries() + assert len(deliveries) == 1 + assert deliveries[0].trigger_id == trigger.id + assert deliveries[0].content == "Review PR #4502" + + def test_serve_cli_options_override_api_config(monkeypatch, tmp_path: Path) -> None: config_file = _write_instance_config(tmp_path) config = Config() diff --git a/tests/command/test_trigger_command.py b/tests/command/test_trigger_command.py new file mode 100644 index 00000000..9c0a9213 --- /dev/null +++ b/tests/command/test_trigger_command.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from nanobot.bus.events import InboundMessage +from nanobot.command.builtin import build_help_text, register_builtin_commands +from nanobot.command.router import CommandContext, CommandRouter +from nanobot.triggers.store import ExternalTriggerStore + + +@pytest.mark.asyncio +async def test_trigger_command_creates_session_bound_local_trigger(tmp_path: Path) -> None: + router = CommandRouter() + register_builtin_commands(router) + store = ExternalTriggerStore(tmp_path) + loop = SimpleNamespace(workspace=tmp_path, external_trigger_store=store) + msg = InboundMessage( + channel="websocket", + sender_id="user", + chat_id="chat-1", + content="/trigger@nanobot_bot PR review", + metadata={"webui": True}, + ) + ctx = CommandContext( + msg=msg, + session=None, + key="websocket:chat-1", + raw="/trigger@nanobot_bot PR review", + loop=loop, + ) + + assert router.is_dispatchable_command("/trigger@nanobot_bot PR review") is True + response = await router.dispatch(ctx) + + assert response is not None + assert "Trigger created: PR review" in response.content + trigger = store.list_for_session("websocket:chat-1")[0] + assert trigger.name == "PR review" + assert trigger.channel == "websocket" + assert trigger.chat_id == "chat-1" + assert trigger.session_key == "websocket:chat-1" + assert f"nanobot trigger {trigger.id} \"message\"" in response.content + + +def test_trigger_command_is_in_help_text() -> None: + assert "/trigger [name]" in build_help_text() diff --git a/tests/triggers/test_external_triggers.py b/tests/triggers/test_external_triggers.py new file mode 100644 index 00000000..af81a524 --- /dev/null +++ b/tests/triggers/test_external_triggers.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import asyncio +from contextlib import suppress +from pathlib import Path + +import pytest + +from nanobot.bus.events import InboundMessage +from nanobot.triggers.runner import run_external_trigger_queue +from nanobot.triggers.store import ExternalTriggerStore, TriggerDisabledError +from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN_METADATA_KEY + + +def test_trigger_store_allows_multiple_triggers_per_session(tmp_path: Path) -> None: + store = ExternalTriggerStore(tmp_path) + + first = store.create( + name="PR review", + channel="websocket", + chat_id="chat-1", + session_key="websocket:chat-1", + ) + second = store.create( + name="CI summary", + channel="websocket", + chat_id="chat-1", + session_key="websocket:chat-1", + ) + + triggers = store.list_for_session("websocket:chat-1") + assert {trigger.id for trigger in triggers} == {first.id, second.id} + assert first.id.startswith("trg_") + assert second.id.startswith("trg_") + assert first.id != second.id + + +def test_enqueue_rejects_disabled_trigger(tmp_path: Path) -> None: + store = ExternalTriggerStore(tmp_path) + trigger = store.create( + name="Disabled", + channel="telegram", + chat_id="123", + session_key="telegram:123", + ) + store.enable(trigger.id, enabled=False) + + with pytest.raises(TriggerDisabledError): + store.enqueue(trigger.id, "Review PR #4502") + + +@pytest.mark.asyncio +async def test_external_trigger_queue_publishes_bound_inbound_message(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", + origin_metadata={"webui": True, WEBUI_TURN_METADATA_KEY: "old-turn"}, + ) + store.enqueue(trigger.id, "Review PR #4502") + published: list[InboundMessage] = [] + + class _Bus: + async def publish_inbound(self, msg: InboundMessage) -> None: + published.append(msg) + + task = asyncio.create_task( + run_external_trigger_queue(store=store, 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 + msg = published[0] + assert msg.channel == "websocket" + assert msg.chat_id == "chat-1" + assert msg.sender_id == "trigger" + assert msg.content == "Review PR #4502" + assert msg.session_key_override == "websocket:chat-1" + assert msg.metadata[WEBUI_TURN_METADATA_KEY].startswith(f"trigger:{trigger.id}:") + assert msg.metadata[WEBUI_TURN_METADATA_KEY] != "old-turn" + assert msg.metadata[WEBUI_MESSAGE_SOURCE_METADATA_KEY] == { + "kind": "trigger", + "label": "PR review", + } + assert msg.metadata["_external_trigger"]["trigger_id"] == trigger.id + + stored = store.get(trigger.id) + assert stored is not None + assert stored.last_status == "ok" + assert stored.last_run_at_ms is not None + assert store.claim_deliveries() == [] diff --git a/webui/src/components/DeleteConfirm.tsx b/webui/src/components/DeleteConfirm.tsx index 5b578b05..1aee2c83 100644 --- a/webui/src/components/DeleteConfirm.tsx +++ b/webui/src/components/DeleteConfirm.tsx @@ -122,6 +122,9 @@ function formatAutomationSchedule( }) : t("deleteConfirm.schedule.cron", { expr: job.schedule.expr }); } + if (job.schedule.kind === "external" || job.payload.kind === "external_trigger") { + return t("deleteConfirm.schedule.external", { defaultValue: "External trigger" }); + } return t("deleteConfirm.schedule.unknown"); } @@ -131,6 +134,9 @@ function formatAutomationNextRun( locale: string, ): string { if (!job.enabled) return t("deleteConfirm.next.disabled"); + if (job.schedule.kind === "external" || job.payload.kind === "external_trigger") { + return t("deleteConfirm.next.external", { defaultValue: "Waiting for trigger" }); + } const next = job.state.next_run_at_ms; if (!next) return t("deleteConfirm.next.none"); return t("deleteConfirm.next.label", { time: fmtDateTime(next, locale) }); diff --git a/webui/src/components/MessageBubble.tsx b/webui/src/components/MessageBubble.tsx index 776110b6..cb479a54 100644 --- a/webui/src/components/MessageBubble.tsx +++ b/webui/src/components/MessageBubble.tsx @@ -167,7 +167,7 @@ export function MessageBubble({ const reasoning = message.role === "assistant" ? message.reasoning ?? "" : ""; const reasoningStreaming = !!(message.role === "assistant" && message.reasoningStreaming); const hasReasoning = reasoning.length > 0 || reasoningStreaming; - const automationSourceLabel = message.source?.kind === "cron" + const automationSourceLabel = message.source?.kind === "cron" || message.source?.kind === "trigger" ? (message.source.label?.trim() || t("message.automationSourceFallback")) : ""; const automationTriggeredLabel = t("message.automationTriggered"); diff --git a/webui/src/components/settings/SettingsView.tsx b/webui/src/components/settings/SettingsView.tsx index 87e75dcb..4ed6710b 100644 --- a/webui/src/components/settings/SettingsView.tsx +++ b/webui/src/components/settings/SettingsView.tsx @@ -21,6 +21,7 @@ import { ChevronLeft, ChevronRight, Cloud, + Clipboard, Cpu, Database, Eye, @@ -106,6 +107,7 @@ import { updateWebSearchSettings, } from "@/lib/api"; import { notifyCliAppsChanged } from "@/lib/cli-app-events"; +import { copyTextToClipboard } from "@/lib/clipboard"; import { getHostApi } from "@/lib/runtime"; import { notifyMcpPresetsChanged } from "@/lib/mcp-preset-events"; import { fmtDateTime, relativeTime } from "@/lib/format"; @@ -3671,6 +3673,7 @@ function AutomationListItem({ const status = automationStatus(job, tx); const origin = automationOriginLabel(job, tx); const nextRun = formatAutomationNext(job, tx); + const summary = automationSummary(job, tx); return (
@@ -3696,7 +3699,7 @@ function AutomationListItem({ - {job.payload.message || tx("settings.automations.systemTask", "System-managed automation")} + {summary} @@ -3753,13 +3756,20 @@ function AutomationDetailPanel({ : null; const created = job.created_at_ms ? fmtDateTime(job.created_at_ms, locale) : null; const updated = job.updated_at_ms ? fmtDateTime(job.updated_at_ms, locale) : null; - const message = job.payload.message || tx("settings.automations.systemTask", "System-managed automation"); + const externalTrigger = isExternalTriggerAutomation(job); + const triggerCommand = automationTriggerCommand(job); + const message = automationDetailText(job, tx); + const messageLabel = externalTrigger + ? tx("settings.automations.fields.command", "Command") + : tx("settings.automations.fields.message", "Message"); const schedule = formatAutomationSchedule(job, locale, tx); const [messageExpanded, setMessageExpanded] = useState(false); + const [commandCopied, setCommandCopied] = useState(false); const messageNeedsExpansion = automationMessageNeedsExpansion(message); useEffect(() => { setMessageExpanded(false); + setCommandCopied(false); }, [job.id]); return ( @@ -3793,12 +3803,37 @@ function AutomationDetailPanel({
-
- {tx("settings.automations.fields.message", "Message")} +
+
+ {messageLabel} +
+ {externalTrigger && triggerCommand ? ( + + ) : null}
@@ -3905,7 +3940,8 @@ function AutomationActionGroup({ t(key, { defaultValue: fallback, ...(values ?? {}) }); const canManage = !job.protected; const hasLinkedChat = Boolean(job.origin); - const canRun = canManage && hasLinkedChat && job.enabled && !job.state.pending; + const externalTrigger = isExternalTriggerAutomation(job); + const canRun = canManage && hasLinkedChat && job.enabled && !job.state.pending && !externalTrigger; const toggleAction: AutomationAction = job.enabled ? "disable" : "enable"; const canToggle = canManage && (job.enabled || hasLinkedChat); const toggleBusy = actionKey === `${toggleAction}:${job.id}`; @@ -3927,14 +3963,16 @@ function AutomationActionGroup({ > - void onAction("run", job)} - > - - + {!externalTrigger ? ( + void onAction("run", job)} + > + + + ) : null} ) => t(key, { defaultValue: fallback, ...(values ?? {}) }); const [draft, setDraft] = useState(() => automationDraftFromJob(null)); + const externalTrigger = isExternalTriggerAutomation(job); useEffect(() => { setDraft(automationDraftFromJob(job)); @@ -4106,34 +4145,38 @@ function AutomationEditDialog({ /> -