fix(agent): hint repeated tool results

maintainer edit: avoid optional source access after extracting automation source kind for TypeScript build.

docs/README.md

@@ -51,7 +51,7 @@ If a local `nanobot agent` session can already answer normally, you can also ask
 |---|---|---|
 | Open the bundled browser UI | [`webui.md`](./webui.md) | WebUI on port `8765`, chat workspace, Apps, Skills, Automations, and settings |
 | Connect Telegram, Discord, WeChat, Slack, and other apps | [`chat-apps.md`](./chat-apps.md) | A gateway-backed chat channel with access control |
-| Use slash commands and periodic tasks | [`chat-commands.md`](./chat-commands.md) | Pairing, model presets, heartbeat tasks, and chat-side controls |
+| Use slash commands and automations | [`chat-commands.md`](./chat-commands.md) | Pairing, model presets, local triggers, heartbeat tasks, and chat-side controls |
 | Generate images | [`image-generation.md`](./image-generation.md) | Image provider config, WebUI image mode, and artifact behavior |
 | Run several isolated bots | [`multiple-instances.md`](./multiple-instances.md) | Separate configs, workspaces, ports, and sessions |
 | Deploy outside a terminal | [`deployment.md`](./deployment.md) | Docker, systemd user services, and macOS LaunchAgent setup |

docs/chat-commands.md

@@ -16,6 +16,8 @@ These commands work inside chat channels and interactive agent sessions:
 | `/dream-restore` | List recent Dream memory versions |
 | `/dream-restore <sha>` | Restore memory to the state before a specific change |
 | `/skill` | List enabled skills and their descriptions |
+| `/trigger` | Show local trigger usage |
+| `/trigger <name>` | Create a named local trigger for the current chat/session |
 | `/pairing` | List pending pairing requests |
 | `/pairing approve <code>` | Approve a pairing code |
 | `/pairing deny <code>` | Deny a pending pairing request |
@@ -55,6 +57,65 @@ To switch presets for future turns:
 
 Preset names come from the top-level `modelPresets` config. Switching is runtime-only: it does not rewrite `config.json`, and an in-progress turn keeps using the model it started with. See [Configuration: Model presets](./configuration.md#model-presets) for setup details.
 
+## Local triggers
+
+Use `/trigger <name>` when a local script or another service should be able to
+send a message into the current chat/session later. A name is required; plain
+`/trigger` only shows the usage hint.
+
+Create the trigger from the chat where future messages should arrive:
+
+```text
+/trigger PR review
+```
+
+nanobot replies with a trigger ID and a command shaped like:
+
+```bash
+nanobot trigger trg_8K4P2Q9X "Review PR #4502"
+```
+
+Replace `"Review PR #4502"` with the message you want nanobot to receive. The
+trigger is bound to the session where it was created, so the message goes back
+to that same chat. Keep `nanobot gateway` running so trigger messages can be
+delivered. The trigger message starts an automation turn recorded in that
+session with the message you passed to the CLI; it is not treated as a normal
+user message. If that session is already running a turn, the trigger waits
+until the session is idle instead of being injected into the active turn.
+
+Trigger deliveries are stored in the workspace until their linked agent turn
+finishes successfully. If the gateway exits after claiming a delivery but before
+the turn completes, the next gateway start requeues that delivery. This is an
+at-least-once local queue: a delivery may run more than once if the process
+exits at the wrong time, so external scripts should make repeated trigger
+messages safe. If the delivery reaches the agent and the agent turn fails, the
+delivery is marked failed in Automations instead of retrying forever.
+
+For longer or generated content, omit the message argument and pipe stdin:
+
+```bash
+printf '%s\n' "Review the latest failed CI job" | nanobot trigger trg_8K4P2Q9X
+```
+
+If an external webhook should wake nanobot up, run your own small webhook
+service and have it call the trigger command after it builds the final message:
+
+```bash
+nanobot trigger <trigger-id> "<message>"
+```
+
+If you run multiple nanobot instances, pass the same config or workspace
+selector used by the gateway:
+
+```bash
+nanobot trigger --config ./bot-a/config.json trg_8K4P2Q9X "Nightly report"
+nanobot trigger --workspace ./bot-a/workspace trg_8K4P2Q9X "Nightly report"
+```
+
+Manage triggers from the WebUI Automations view. You can search, pause/resume,
+rename, delete, and copy the trigger command there. A session may have multiple
+triggers, just like it may have multiple scheduled automations.
+
 ## Periodic Tasks
 
 Periodic background checks are driven by `HEARTBEAT.md` in your workspace (`~/.nanobot/workspace/HEARTBEAT.md`). When `nanobot gateway` starts, it registers a protected heartbeat cron job by default. Every 30 minutes, that job checks the file; if it finds tasks under `## Active Tasks`, the agent executes them and delivers only results that pass the notification gate to your most recently active chat channel. If there are no active tasks, or the result is routine with nothing useful to report, the heartbeat is skipped silently.

docs/cli-reference.md

@@ -13,6 +13,7 @@ Use this page when you know what you want to run and need the command shape. For
 | Send one test message | `nanobot agent -m "Hello!"` | First proof that install, config, provider, model, and workspace all work |
 | Chat in the terminal | `nanobot agent` | Interactive local chat; exit with `exit`, `/exit`, `:q`, or `Ctrl+D` |
 | Use WebUI or chat apps | `nanobot gateway` | Keep this terminal running, or use `nanobot gateway --background` |
+| Deliver a local trigger | `nanobot trigger <id> "message"` | Created first with `/trigger <name>` in the target chat/session |
 | Serve an OpenAI-compatible API | `nanobot serve` | Starts `/v1/chat/completions`, `/v1/models`, and `/health` |
 | Check chat channel setup | `nanobot channels status` | Useful before starting `nanobot gateway` |
 | Log in to QR/OAuth-style channels | `nanobot channels login <channel>` | Used by channels such as WhatsApp and WeChat |
@@ -122,6 +123,52 @@ http://127.0.0.1:18790/health
 
 The bundled WebUI is served by the WebSocket channel, usually on port `8765`, not by the gateway health endpoint.
 
+## Local Triggers
+
+`nanobot trigger` delivers one local message to a trigger that was created from
+a chat/session with `/trigger <name>`.
+
+```bash
+nanobot trigger trg_8K4P2Q9X "Review PR #4502"
+```
+
+Keep `nanobot gateway` running so the message can be delivered to the linked
+chat/session. The message is recorded as an automation turn in that session,
+not as a normal chat message typed by the user.
+
+The command writes to a workspace-local durable queue. If `nanobot gateway` is
+not running yet, the message waits in that workspace. If the target session is
+already running a turn, the trigger waits for that session to become idle. If the
+gateway exits after claiming a delivery but before the linked turn completes,
+the next gateway start requeues that delivery. The queue is at-least-once, not
+exactly-once, so the same message can be delivered again after an interrupted
+process. If the agent receives the delivery and the turn fails, the delivery is
+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:
+
+```bash
+generate-report | nanobot trigger trg_8K4P2Q9X
+```
+
+Options:
+
+| Command | Description |
+|---|---|
+| `nanobot trigger <id> "message"` | Deliver one message through a trigger |
+| `nanobot trigger <id>` | Read the message from stdin |
+| `nanobot trigger --config <path> <id> "message"` | Use the workspace from a specific config |
+| `nanobot trigger --workspace <path> <id> "message"` | Use a specific workspace |
+
+Triggers are managed in the WebUI Automations view instead of through separate
+`list`, `revoke`, or `delete` CLI subcommands. From there you can pause/resume,
+rename, delete, search, and copy the command for each trigger.
+
+For webhooks or other external systems, run your own small service and have it
+call this CLI after it decides what message nanobot should receive.
+
 ## OpenAI-Compatible API
 
 | Command | Description |

docs/concepts.md

@@ -123,7 +123,7 @@ Tools are discovered automatically from built-in modules and plugin entry points
 - shell execution with configurable sandboxing;
 - web search and web fetch with SSRF checks;
 - MCP servers;
-- cron reminders and heartbeat tasks;
+- cron reminders, local triggers, and heartbeat tasks;
 - image generation;
 - subagents and runtime self-inspection.
 
@@ -131,14 +131,29 @@ Security-sensitive controls live in [`configuration.md#security`](./configuratio
 
 ## Background Jobs
 
-When `nanobot gateway` starts, it creates workspace-scoped cron storage at `<workspace>/cron/jobs.json` and registers system jobs:
+When `nanobot gateway` starts, it runs workspace-scoped automations and
+registers system jobs:
 
 - `dream`, when `agents.defaults.dream.enabled` is true;
 - `heartbeat`, when `gateway.heartbeat.enabled` is true.
 
 Heartbeat reads `<workspace>/HEARTBEAT.md`. If the file has tasks under `## Active Tasks`, nanobot executes them and sends only useful/actionable results to the most recently active chat target. Routine "nothing changed" results are suppressed.
 
-User-created reminders use the same cron service but are not the same as the protected heartbeat system job. They run as scheduled turns in their origin chat/session and normally deliver the result back to that channel.
+User-created reminders use the same cron service but are not the same as the
+protected heartbeat system job. They run as scheduled turns in their origin
+chat/session and normally deliver the result back to that channel.
+
+Local triggers are also session-bound, but they do not have their own
+schedule. Create one from the target chat with `/trigger <name>`, then call
+`nanobot trigger <id> "<message>"` when a local script or external service wants
+nanobot to respond in that session. Webhook servers, third-party auth, and
+event-to-message formatting stay outside nanobot. Trigger deliveries are stored
+in the workspace until the linked agent turn finishes successfully. If the
+target session is busy, the trigger waits until that session is idle instead of
+being injected into the active turn. The message is recorded as an automation
+turn in that session. Delivery is at-least-once, so external systems should
+tolerate repeated trigger messages; a delivery that reaches the agent but fails
+is marked failed rather than retried forever.
 
 ## Where to Go Next
 

docs/webui.md

@@ -56,7 +56,7 @@ Enter `tokenIssueSecret` when the WebUI asks for a password.
 | Composer | Send text, images, voice input, slash commands, and `@` mentions for Apps or MCP presets |
 | Apps | Install, test, update, and use local CLI App adapters and MCP presets |
 | Skills | Inspect available built-in and workspace skills before relying on them |
-| Automations | Review, search, run, pause, edit, and delete scheduled agent turns |
+| Automations | Review, search, run, pause, edit, and delete scheduled and local-trigger agent turns |
 | Settings | Adjust models, providers, image generation, voice, web tools, runtime, and safety options |
 
 ## Chat Workspace
@@ -116,10 +116,30 @@ to perform that task.
 
 ## Automations
 
-Automations are scheduled agent turns. They should be created from the chat,
-channel, or session where they are supposed to run so nanobot keeps the correct
-target context. When an automation runs, it normally delivers the result back to
-that linked chat.
+Automations are agent turns that run later in a linked chat/session. They should
+be created from the chat, channel, or session where they are supposed to run so
+nanobot keeps the correct target context. When an automation runs, it normally
+delivers the result back to that linked chat.
+
+There are two user-facing automation types:
+
+- Scheduled automations, created by the agent's cron tool, run at a time,
+  interval, or cron expression.
+- Local triggers, created with `/trigger <name>`, run when you call a local
+  command such as `nanobot trigger trg_8K4P2Q9X "Review PR #4502"`.
+
+If a GitHub webhook, CI system, or another service should wake nanobot up, keep
+that webhook/service outside nanobot and have it call the trigger command with
+the final message.
+
+Trigger deliveries use the same workspace as the gateway. They survive gateway
+restarts and are requeued if the process exits before the linked turn completes.
+If the linked session is already running a turn, the local trigger waits until
+that session is idle instead of being injected into the active turn. This is an
+at-least-once local queue, so repeated delivery is possible after an interrupted
+process. A delivered trigger is recorded as an automation turn in the linked
+session; if the agent receives it but the turn fails, Automations marks the run
+failed instead of retrying indefinitely.
 
 For recurring background checks that should stay quiet unless there is something
 useful to report, use the protected heartbeat job by editing `HEARTBEAT.md`
@@ -128,18 +148,25 @@ instead of creating a chat automation.
 Use the Automations view to:
 
 - Filter by all, active, paused, needs-attention, or system jobs.
-- Search by task name, message, linked chat, schedule, or status.
+- Search by task name, message, trigger command, linked chat, schedule, or status.
 - Sort by next run, last run, updated time, or name.
-- Run now, pause or resume, edit, or delete user-created automations.
+- Run scheduled automations now.
+- Pause or resume, rename, or delete user-created automations.
+- Copy the CLI command for local triggers.
 - Inspect protected system automations without changing them.
 
 Search accepts plain text and field filters such as `name:backup`,
-`chat:WeChat`, `schedule:09:30`, `cron:"0 23 * * *"`, and `status:paused`.
+`chat:WeChat`, `schedule:09:30`, `cron:"0 23 * * *"`, `trigger`, and
+`status:paused`.
 
 An automation without a linked chat cannot be enabled or run from the WebUI,
 because nanobot would not know where to deliver the scheduled turn. Recreate it
 from the target chat or channel so the automation has complete context.
 
+Local triggers do not have a WebUI "Run now" action because each run needs a
+message. Use the copied `nanobot trigger ...` command and replace `"message"`
+with the content that should be delivered.
+
 ## Settings
 
 Settings is the control surface for the browser session and gateway-backed

nanobot/agent/automation_turns.py

@@ -0,0 +1,145 @@
+"""Shared coordination for session-bound automation turns."""
+
+from __future__ import annotations
+
+import asyncio
+import dataclasses
+from collections.abc import Awaitable, Callable, Iterable
+
+from nanobot.bus.events import InboundMessage, OutboundMessage
+
+
+class AutomationTurnError(RuntimeError):
+    """Raised when an automation turn reaches the agent and finishes with an error."""
+
+
+async def publish_next_deferred_turn(
+    *,
+    deferred_queues: dict[str, list[InboundMessage]],
+    publish_inbound: Callable[[InboundMessage], Awaitable[None]],
+    session_key: str,
+) -> bool:
+    """Publish the next deferred automation turn for a session."""
+    queue = deferred_queues.get(session_key)
+    if not queue:
+        return False
+    msg = queue.pop(0)
+    if not queue:
+        deferred_queues.pop(session_key, None)
+    await publish_inbound(msg)
+    return True
+
+
+class AutomationTurnCoordinator:
+    """Manage automation turns without mixing them into live injections."""
+
+    def __init__(
+        self,
+        *,
+        publish_inbound: Callable[[InboundMessage], Awaitable[None]],
+        dispatch: Callable[[InboundMessage], Awaitable[object]],
+        is_running: Callable[[], bool],
+        turn_id: Callable[[InboundMessage], str | None],
+        pending_id: Callable[[InboundMessage], str | None],
+        should_defer_turn: Callable[[InboundMessage, str, Iterable[str]], bool],
+        missing_id_error: str,
+        duplicate_id_error: Callable[[str], str],
+        deferred_queues: dict[str, list[InboundMessage]] | None = None,
+    ) -> None:
+        self._publish_inbound = publish_inbound
+        self._dispatch = dispatch
+        self._is_running = is_running
+        self._turn_id = turn_id
+        self._pending_id = pending_id
+        self._should_defer_turn = should_defer_turn
+        self._missing_id_error = missing_id_error
+        self._duplicate_id_error = duplicate_id_error
+        self.deferred_queues = deferred_queues if deferred_queues is not None else {}
+        self._waiters: dict[str, asyncio.Future[OutboundMessage | None]] = {}
+        self._pending_messages_by_turn_id: dict[str, InboundMessage] = {}
+
+    async def submit(self, msg: InboundMessage) -> OutboundMessage | None:
+        """Submit an automation turn and wait for its session response."""
+        turn_id = self._turn_id(msg)
+        if not turn_id:
+            raise ValueError(self._missing_id_error)
+        if turn_id in self._waiters:
+            raise RuntimeError(self._duplicate_id_error(turn_id))
+
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[OutboundMessage | None] = loop.create_future()
+        self._waiters[turn_id] = future
+        self._pending_messages_by_turn_id[turn_id] = msg
+        try:
+            if self._is_running():
+                await self._publish_inbound(msg)
+            else:
+                await self._dispatch(msg)
+            try:
+                return await future
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                raise AutomationTurnError(str(exc) or exc.__class__.__name__) from exc
+        finally:
+            self._waiters.pop(turn_id, None)
+            self._pending_messages_by_turn_id.pop(turn_id, None)
+
+    def defer_if_active(
+        self,
+        msg: InboundMessage,
+        *,
+        session_key: str,
+        active_session_keys: Iterable[str],
+    ) -> bool:
+        """Defer an automation turn when its target session is already active."""
+        if not self._should_defer_turn(msg, session_key, active_session_keys):
+            return False
+        pending_msg = msg
+        if session_key != msg.session_key:
+            pending_msg = dataclasses.replace(
+                msg,
+                session_key_override=session_key,
+            )
+        self.deferred_queues.setdefault(session_key, []).append(pending_msg)
+        return True
+
+    def complete(
+        self,
+        msg: InboundMessage,
+        *,
+        response: OutboundMessage | None = None,
+        error: BaseException | None = None,
+    ) -> None:
+        turn_id = self._turn_id(msg)
+        if not turn_id:
+            return
+        future = self._waiters.get(turn_id)
+        if future is None or future.done():
+            return
+        if error is not None:
+            future.set_exception(error)
+        else:
+            future.set_result(response)
+
+    def pending_ids_for_session(self, session_key: str) -> set[str]:
+        """Return automation IDs that are waiting for or running in *session_key*."""
+        pending_ids: set[str] = set()
+        for msg in self.deferred_queues.get(session_key, []):
+            pending_id = self._pending_id(msg)
+            if pending_id:
+                pending_ids.add(pending_id)
+        for msg in self._pending_messages_by_turn_id.values():
+            if msg.session_key != session_key:
+                continue
+            pending_id = self._pending_id(msg)
+            if pending_id:
+                pending_ids.add(pending_id)
+        return pending_ids
+
+    async def publish_next_deferred(self, session_key: str) -> bool:
+        return await publish_next_deferred_turn(
+            deferred_queues=self.deferred_queues,
+            publish_inbound=self._publish_inbound,
+            session_key=session_key,
+        )

nanobot/agent/cron_turns.py

@@ -2,11 +2,10 @@
 
 from __future__ import annotations
 
-import asyncio
-import dataclasses
 from collections.abc import Awaitable, Callable, Iterable
 
-from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.agent.automation_turns import AutomationTurnCoordinator
+from nanobot.bus.events import InboundMessage
 from nanobot.cron.session_turns import (
     cron_run_id,
     cron_trigger,
@@ -14,7 +13,7 @@ from nanobot.cron.session_turns import (
 )
 
 
-class CronTurnCoordinator:
+class CronTurnCoordinator(AutomationTurnCoordinator):
     """Manage scheduled cron turns without mixing them into live injections."""
 
     def __init__(
@@ -23,115 +22,31 @@ class CronTurnCoordinator:
         publish_inbound: Callable[[InboundMessage], Awaitable[None]],
         dispatch: Callable[[InboundMessage], Awaitable[object]],
         is_running: Callable[[], bool],
+        deferred_queues: dict[str, list[InboundMessage]] | None = None,
     ) -> None:
-        self._publish_inbound = publish_inbound
-        self._dispatch = dispatch
-        self._is_running = is_running
-        self.deferred_queues: dict[str, list[InboundMessage]] = {}
-        self._waiters: dict[str, asyncio.Future[OutboundMessage | None]] = {}
-        self._pending_messages_by_run_id: dict[str, InboundMessage] = {}
-
-    async def submit(self, msg: InboundMessage) -> OutboundMessage | None:
-        """Submit a scheduled cron turn and wait for its session response."""
-        run_id = cron_run_id(msg.metadata)
-        if not run_id:
-            raise ValueError("cron turn metadata must include a run_id")
-        if run_id in self._waiters:
-            raise RuntimeError(f"cron run {run_id!r} is already pending")
-
-        loop = asyncio.get_running_loop()
-        future: asyncio.Future[OutboundMessage | None] = loop.create_future()
-        self._waiters[run_id] = future
-        self._pending_messages_by_run_id[run_id] = msg
-        try:
-            if self._is_running():
-                await self._publish_inbound(msg)
-            else:
-                await self._dispatch(msg)
-            return await future
-        finally:
-            self._waiters.pop(run_id, None)
-            self._pending_messages_by_run_id.pop(run_id, None)
-
-    def should_defer(
-        self,
-        msg: InboundMessage,
-        *,
-        session_key: str,
-        active_session_keys: Iterable[str],
-    ) -> bool:
-        return (
-            defer_cron_until_session_idle(msg.metadata)
-            and session_key in active_session_keys
+        super().__init__(
+            publish_inbound=publish_inbound,
+            dispatch=dispatch,
+            is_running=is_running,
+            turn_id=lambda msg: cron_run_id(msg.metadata),
+            pending_id=_cron_job_id,
+            should_defer_turn=_should_defer_cron_turn,
+            missing_id_error="cron turn metadata must include a run_id",
+            duplicate_id_error=lambda run_id: f"cron run {run_id!r} is already pending",
+            deferred_queues=deferred_queues,
         )
 
-    def defer_if_active(
-        self,
-        msg: InboundMessage,
-        *,
-        session_key: str,
-        active_session_keys: Iterable[str],
-    ) -> bool:
-        """Defer a cron turn when its target session is already active."""
-        if not self.should_defer(
-            msg,
-            session_key=session_key,
-            active_session_keys=active_session_keys,
-        ):
-            return False
-        pending_msg = msg
-        if session_key != msg.session_key:
-            pending_msg = dataclasses.replace(
-                msg,
-                session_key_override=session_key,
-            )
-        self.defer(session_key, pending_msg)
-        return True
-
-    def complete(
-        self,
-        msg: InboundMessage,
-        *,
-        response: OutboundMessage | None = None,
-        error: BaseException | None = None,
-    ) -> None:
-        run_id = cron_run_id(msg.metadata)
-        if not run_id:
-            return
-        future = self._waiters.get(run_id)
-        if future is None or future.done():
-            return
-        if error is not None:
-            future.set_exception(error)
-        else:
-            future.set_result(response)
-
-    def defer(self, session_key: str, msg: InboundMessage) -> None:
-        self.deferred_queues.setdefault(session_key, []).append(msg)
-
     def pending_job_ids_for_session(self, session_key: str) -> set[str]:
         """Return cron jobs that are waiting for or running in *session_key*."""
-        job_ids: set[str] = set()
-        for msg in self.deferred_queues.get(session_key, []):
-            job_id = _cron_job_id(msg)
-            if job_id:
-                job_ids.add(job_id)
-        for msg in self._pending_messages_by_run_id.values():
-            if msg.session_key != session_key:
-                continue
-            job_id = _cron_job_id(msg)
-            if job_id:
-                job_ids.add(job_id)
-        return job_ids
+        return self.pending_ids_for_session(session_key)
 
-    async def publish_next_deferred(self, session_key: str) -> None:
-        queue = self.deferred_queues.get(session_key)
-        if not queue:
-            return
-        msg = queue.pop(0)
-        if not queue:
-            self.deferred_queues.pop(session_key, None)
-        await self._publish_inbound(msg)
+
+def _should_defer_cron_turn(
+    msg: InboundMessage,
+    session_key: str,
+    active_session_keys: Iterable[str],
+) -> bool:
+    return defer_cron_until_session_idle(msg.metadata) and session_key in active_session_keys
 
 
 def _cron_job_id(msg: InboundMessage) -> str | None:

nanobot/agent/loop.py

@@ -18,6 +18,7 @@ from loguru import logger
 from nanobot.agent import context as agent_context
 from nanobot.agent import model_presets as preset_helpers
 from nanobot.agent.autocompact import AutoCompact
+from nanobot.agent.automation_turns import publish_next_deferred_turn
 from nanobot.agent.context import ContextBuilder
 from nanobot.agent.cron_turns import CronTurnCoordinator
 from nanobot.agent.hook import AgentHook, CompositeHook
@@ -47,9 +48,6 @@ from nanobot.bus.runtime_events import (
 )
 from nanobot.command import CommandContext, CommandRouter, register_builtin_commands
 from nanobot.config.schema import AgentDefaults, ModelPresetConfig
-from nanobot.cron.session_turns import (
-    cron_history_overrides,
-)
 from nanobot.providers.base import LLMProvider
 from nanobot.providers.factory import ProviderSnapshot
 from nanobot.security.workspace_access import (
@@ -58,6 +56,7 @@ from nanobot.security.workspace_access import (
     reset_workspace_scope,
 )
 from nanobot.session import turn_continuation
+from nanobot.session.automation_turns import automation_history_overrides
 from nanobot.session.goal_state import (
     goal_state_runtime_lines,
     runner_wall_llm_timeout_s,
@@ -69,6 +68,7 @@ from nanobot.session.manager import (
     SessionManager,
     replay_max_messages_for_context,
 )
+from nanobot.triggers.local_turns import LocalTriggerTurnCoordinator
 from nanobot.utils.document import extract_documents, reference_non_image_attachments
 from nanobot.utils.helpers import image_placeholder_text
 from nanobot.utils.helpers import truncate_text as truncate_text_fn
@@ -226,6 +226,7 @@ class AgentLoop:
         runtime_events: RuntimeEventBus | None = None,
         runtime_model_publisher: Callable[[str, str | None], None] | None = None,
         restart_mode: str = "auto",
+        local_trigger_store: Any | None = None,
     ):
         from nanobot.config.schema import ToolsConfig
 
@@ -273,6 +274,7 @@ class AgentLoop:
         ):
             self._image_generation_provider_configs["openrouter"] = image_generation_provider_config
         self.cron_service = cron_service
+        self.local_trigger_store = local_trigger_store
         self.restrict_to_workspace = restrict_to_workspace
         self.workspace_scopes = WorkspaceScopeResolver(
             default_workspace=workspace,
@@ -317,10 +319,22 @@ class AgentLoop:
         # When a session has an active task, new messages for that session
         # are routed here instead of creating a new task.
         self._pending_queues: dict[str, asyncio.Queue] = {}
+        self._deferred_automation_turns: dict[str, list[InboundMessage]] = {}
         self._cron_turns = CronTurnCoordinator(
             publish_inbound=self.bus.publish_inbound,
             dispatch=self._dispatch,
             is_running=lambda: self._running,
+            deferred_queues=self._deferred_automation_turns,
+        )
+        self._local_trigger_turns = LocalTriggerTurnCoordinator(
+            publish_inbound=self.bus.publish_inbound,
+            dispatch=self._dispatch,
+            is_running=lambda: self._running,
+            deferred_queues=self._deferred_automation_turns,
+        )
+        self._automation_turn_coordinators = (
+            ("cron", self._cron_turns),
+            ("local trigger", self._local_trigger_turns),
         )
         # NANOBOT_MAX_CONCURRENT_REQUESTS: <=0 means unlimited; default 3.
         _max = int(os.environ.get("NANOBOT_MAX_CONCURRENT_REQUESTS", "3"))
@@ -591,9 +605,22 @@ class AgentLoop:
     async def submit_cron_turn(self, msg: InboundMessage) -> OutboundMessage | None:
         return await self._cron_turns.submit(msg)
 
+    async def submit_local_trigger_turn(self, msg: InboundMessage) -> OutboundMessage | None:
+        return await self._local_trigger_turns.submit(msg)
+
     def pending_cron_job_ids_for_session(self, session_key: str) -> set[str]:
         return self._cron_turns.pending_job_ids_for_session(session_key)
 
+    def pending_local_trigger_ids_for_session(self, session_key: str) -> set[str]:
+        return self._local_trigger_turns.pending_trigger_ids_for_session(session_key)
+
+    async def _publish_next_deferred_automation_turn(self, session_key: str) -> None:
+        await publish_next_deferred_turn(
+            deferred_queues=self._deferred_automation_turns,
+            publish_inbound=self.bus.publish_inbound,
+            session_key=session_key,
+        )
+
     def _persist_user_message_early(
         self,
         msg: InboundMessage,
@@ -612,10 +639,10 @@ class AgentLoop:
             extra: dict[str, Any] = ({"media": list(media_paths)} if media_paths else {}) | agent_context.session_extra(msg.metadata)
             extra.update(kwargs)
             text = msg.content if isinstance(msg.content, str) else ""
-            text_override, cron_extra = cron_history_overrides(msg.metadata)
+            text_override, automation_extra = automation_history_overrides(msg.metadata)
             if text_override is not None:
                 text = text_override
-            extra.update(cron_extra)
+            extra.update(automation_extra)
             session.add_message("user", text, **extra)
             self._mark_pending_user_turn(session)
             self.sessions.save(session)
@@ -923,15 +950,21 @@ class AgentLoop:
                         self.commands.dispatch_priority,
                     )
                     continue
-                if self._cron_turns.defer_if_active(
+                deferred = False
+                for label, coordinator in self._automation_turn_coordinators:
+                    if coordinator.defer_if_active(
                         msg,
                         session_key=effective_key,
                         active_session_keys=self._pending_queues.keys(),
                     ):
                         logger.info(
-                        "Deferred cron turn for active session {}",
+                            "Deferred {} turn for active session {}",
+                            label,
                             effective_key,
                         )
+                        deferred = True
+                        break
+                if deferred:
                     continue
                 # If this session already has an active pending queue (i.e. a task
                 # is processing this session), route the message there for mid-turn
@@ -1054,12 +1087,11 @@ class AgentLoop:
                             session_key=session_key,
                             metadata=msg.metadata,
                         )
-                    self._cron_turns.complete(msg, response=response)
+                    for _, coordinator in self._automation_turn_coordinators:
+                        coordinator.complete(msg, response=response)
                 except asyncio.CancelledError:
-                    self._cron_turns.complete(
-                        msg,
-                        error=asyncio.CancelledError(),
-                    )
+                    for _, coordinator in self._automation_turn_coordinators:
+                        coordinator.complete(msg, error=asyncio.CancelledError())
                     logger.info("Task cancelled for session {}", session_key)
                     # Preserve partial context from the interrupted turn so
                     # the user does not lose tool results and assistant
@@ -1098,7 +1130,8 @@ class AgentLoop:
                             session_key=session_key,
                             metadata=msg.metadata,
                         )
-                    self._cron_turns.complete(msg, error=exc)
+                    for _, coordinator in self._automation_turn_coordinators:
+                        coordinator.complete(msg, error=exc)
                 finally:
                     # Drain any messages still in the pending queue and re-publish
                     # them to the bus so they are processed as fresh inbound messages
@@ -1129,14 +1162,14 @@ class AgentLoop:
                             msg, session_key, "idle"
                         )
                         self._runtime_events().clear_turn(session_key)
-                    await self._cron_turns.publish_next_deferred(session_key)
+                    await self._publish_next_deferred_automation_turn(session_key)
         finally:
             if pending is None:
                 await self._runtime_events().run_status_changed(
                     msg, session_key, "idle"
                 )
                 self._runtime_events().clear_turn(session_key)
-                await self._cron_turns.publish_next_deferred(session_key)
+                await self._publish_next_deferred_automation_turn(session_key)
 
     async def close_mcp(self) -> None:
         """Drain pending background archives, then close MCP connections."""
@@ -1450,7 +1483,7 @@ class AgentLoop:
             # message.  Mark messages with _command so get_history can filter
             # them out of LLM context.  /new is excluded because it
             # intentionally clears the session.
-            if raw.lower() != "/new":
+            if cmd_ctx.raw.lower() != "/new":
                 ctx.user_persisted_early = self._persist_user_message_early(
                     ctx.msg, ctx.session, _command=True
                 )

nanobot/agent/runner.py

@@ -52,6 +52,7 @@ from nanobot.utils.runtime import (
     build_length_recovery_message,
     is_blank_text,
     repeated_external_lookup_error,
+    repeated_tool_result_hint,
     repeated_workspace_violation_error,
 )
 
@@ -351,6 +352,7 @@ class AgentRunner:
         stop_reason = "completed"
         tool_events: list[dict[str, str]] = []
         external_lookup_counts: dict[str, int] = {}
+        repeated_result_counts: dict[str, int] = {}
         # Per-turn throttle for repeated attempts against the same outside target.
         workspace_violation_counts: dict[str, int] = {}
         empty_content_retries = 0
@@ -468,17 +470,29 @@ class AgentRunner:
                 context.tool_results = list(results)
                 context.tool_events = list(new_events)
                 completed_tool_results: list[dict[str, Any]] = []
-                for tool_call, result in zip(response.tool_calls, results):
-                    tool_message = {
-                        "role": "tool",
-                        "tool_call_id": tool_call.id,
-                        "name": tool_call.name,
-                        "content": self.context_governor.normalize_tool_result(
+                for tool_call, result, event in zip(response.tool_calls, results, new_events):
+                    content = self.context_governor.normalize_tool_result(
                         governance_config,
                         tool_call.id,
                         tool_call.name,
                         result,
-                        ),
+                    )
+                    if event.get("status") == "ok":
+                        result_hint = repeated_tool_result_hint(
+                            tool_call.name,
+                            content,
+                            repeated_result_counts,
+                        )
+                        if result_hint:
+                            if isinstance(content, str):
+                                content = content + result_hint
+                            elif isinstance(content, list):
+                                content = [*content, {"type": "text", "text": result_hint.strip()}]
+                    tool_message = {
+                        "role": "tool",
+                        "tool_call_id": tool_call.id,
+                        "name": tool_call.name,
+                        "content": content,
                     }
                     messages.append(tool_message)
                     completed_tool_results.append(tool_message)
@@ -1135,7 +1149,10 @@ class AgentRunner:
             if spec.concurrent_tools and len(batch) > 1:
                 batch_results = await asyncio.gather(*(
                     self._run_tool(
-                        spec, tool_call, external_lookup_counts, workspace_violation_counts,
+                        spec,
+                        tool_call,
+                        external_lookup_counts,
+                        workspace_violation_counts,
                     )
                     for tool_call in batch
                 ))
@@ -1144,7 +1161,10 @@ class AgentRunner:
                 batch_results = []
                 for tool_call in batch:
                     result = await self._run_tool(
-                        spec, tool_call, external_lookup_counts, workspace_violation_counts,
+                        spec,
+                        tool_call,
+                        external_lookup_counts,
+                        workspace_violation_counts,
                     )
                     tool_results.append(result)
                     batch_results.append(result)

nanobot/agent/tools/shell.py

@@ -8,7 +8,6 @@ import re
 import shutil
 import sys
 from contextlib import suppress
-from copy import deepcopy
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -16,7 +15,7 @@ from typing import Any
 from loguru import logger
 from pydantic import Field
 
-from nanobot.agent.tools.base import Schema, Tool, ToolResult, tool_parameters
+from nanobot.agent.tools.base import Tool, ToolResult, tool_parameters
 from nanobot.agent.tools.context import current_request_session_key
 from nanobot.agent.tools.exec_session import (
     DEFAULT_EXEC_SESSION_MANAGER,
@@ -74,9 +73,12 @@ class _PreparedCommand:
     login: bool
 
 
-_EXEC_TOOL_PARAMETERS = tool_parameters_schema(
+@tool_parameters(
+    tool_parameters_schema(
         command=StringSchema("The shell command to execute"),
+        cmd=StringSchema("Compatibility alias for command"),
         working_dir=StringSchema("Optional working directory for the command"),
+        workdir=StringSchema("Compatibility alias for working_dir"),
         timeout=IntegerSchema(
             60,
             description=(
@@ -115,14 +117,7 @@ _EXEC_TOOL_PARAMETERS = tool_parameters_schema(
             maximum=MAX_OUTPUT_CHARS,
             nullable=True,
         ),
-)
-
-_EXEC_TOOL_COMPAT_PARAMETERS = deepcopy(_EXEC_TOOL_PARAMETERS)
-_EXEC_TOOL_COMPAT_PARAMETERS["properties"].update(
-    {
-        "cmd": StringSchema("Compatibility alias for command").to_json_schema(),
-        "workdir": StringSchema("Compatibility alias for working_dir").to_json_schema(),
-        "max_output_tokens": IntegerSchema(
+        max_output_tokens=IntegerSchema(
             description=(
                 "Compatibility alias for max_output_chars. The current runtime "
                 "uses a character budget."
@@ -130,12 +125,9 @@ _EXEC_TOOL_COMPAT_PARAMETERS["properties"].update(
             minimum=1000,
             maximum=MAX_OUTPUT_CHARS,
             nullable=True,
-        ).to_json_schema(),
-    }
+        ),
+    )
 )
-
-
-@tool_parameters(_EXEC_TOOL_PARAMETERS)
 class ExecTool(Tool):
     """Tool to execute shell commands."""
     _scopes = {"core", "subagent"}
@@ -252,18 +244,6 @@ class ExecTool(Tool):
     def exclusive(self) -> bool:
         return True
 
-    def cast_params(self, params: dict[str, Any]) -> dict[str, Any]:
-        return self._cast_object(params, _EXEC_TOOL_COMPAT_PARAMETERS)
-
-    def validate_params(self, params: dict[str, Any]) -> list[str]:
-        if not isinstance(params, dict):
-            return [f"parameters must be an object, got {type(params).__name__}"]
-        return Schema.validate_json_schema_value(
-            params,
-            {**_EXEC_TOOL_COMPAT_PARAMETERS, "type": "object"},
-            "",
-        )
-
     async def execute(
         self, command: str | None = None, cmd: str | None = None,
         working_dir: str | None = None, workdir: str | None = None,

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 named local trigger for this chat")
+            @app_commands.describe(name="Trigger name")
+            async def trigger_command(
+                interaction: discord.Interaction,
+                name: str,
+            ) -> None:
+                name = name.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)

nanobot/channels/manager.py

@@ -68,8 +68,10 @@ class ChannelManager:
         *,
         session_manager: "SessionManager | None" = None,
         cron_service: Any | None = None,
+        local_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_local_trigger_pending_ids: Callable[[str], set[str]] | None = None,
         webui_static_dist: bool = True,
         webui_runtime_surface: str = "browser",
         webui_runtime_capabilities: dict[str, Any] | None = None,
@@ -78,8 +80,10 @@ class ChannelManager:
         self.bus = bus
         self._session_manager = session_manager
         self._cron_service = cron_service
+        self._local_trigger_store = local_trigger_store
         self._webui_runtime_model_name = webui_runtime_model_name
         self._webui_cron_pending_job_ids = webui_cron_pending_job_ids
+        self._webui_local_trigger_pending_ids = webui_local_trigger_pending_ids
         self._webui_static_dist = webui_static_dist
         self._webui_runtime_surface = webui_runtime_surface
         self._webui_runtime_capabilities = dict(webui_runtime_capabilities or {})
@@ -139,7 +143,9 @@ class ChannelManager:
                         runtime_surface=self._webui_runtime_surface,
                         runtime_capabilities_overrides=self._webui_runtime_capabilities,
                         cron_service=self._cron_service,
+                        local_trigger_store=self._local_trigger_store,
                         cron_pending_job_ids=self._webui_cron_pending_job_ids,
+                        local_trigger_pending_ids=self._webui_local_trigger_pending_ids,
                         logger=logger,
                     )
                     kwargs["gateway"] = gateway

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 named local trigger"),
         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

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.local_store import (
+        LocalTriggerStore,
+        TriggerDisabledError,
+        TriggerNotFoundError,
+        TriggerStoreError,
+    )
+
+    runtime_config = _load_runtime_config(config, workspace)
+    content = _read_trigger_cli_message(message)
+    store = LocalTriggerStore(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.local_runner import run_local_trigger_queue
+    from nanobot.triggers.local_store import LocalTriggerStore
     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 = LocalTriggerStore(config.workspace_path)
 
     # Create agent with cron service
     agent = AgentLoop.from_config(
@@ -901,13 +948,13 @@ def _run_gateway(
         runtime_events=runtime_events,
         provider_signature=provider_snapshot.signature,
         hooks=[TokenUsageHook(timezone_name=config.agents.defaults.timezone)],
+        local_trigger_store=trigger_store,
     )
     WebuiTurnCoordinator(
         bus=bus,
         sessions=session_manager,
         schedule_background=lambda coro: agent._schedule_background(coro),
     ).subscribe(runtime_events)
-
     from nanobot.bus.events import OutboundMessage
     from nanobot.session.keys import session_key_for_channel
 
@@ -1103,8 +1150,14 @@ def _run_gateway(
         bus,
         session_manager=session_manager,
         cron_service=cron,
+        local_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_local_trigger_pending_ids=getattr(
+            agent,
+            "pending_local_trigger_ids_for_session",
+            None,
+        ),
         webui_static_dist=webui_static_dist,
         webui_runtime_surface=webui_runtime_surface,
         webui_runtime_capabilities=webui_runtime_capabilities,
@@ -1245,6 +1298,13 @@ 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_local_trigger_queue(
+                        store=trigger_store,
+                        submit_turn=getattr(agent, "submit_local_trigger_turn", None),
+                    ),
+                    name="nanobot-local-triggers",
+                ),
             ]
             if health_server_enabled:
                 tasks.append(asyncio.create_task(

nanobot/command/builtin.py

@@ -81,6 +81,13 @@ BUILTIN_COMMAND_SPECS: tuple[BuiltinCommandSpec, ...] = (
         "activity",
         "<goal>",
     ),
+    BuiltinCommandSpec(
+        "/trigger",
+        "Create named local trigger",
+        "Create a named CLI trigger bound to this chat session.",
+        "zap",
+        "<name>",
+    ),
     BuiltinCommandSpec(
         "/dream",
         "Run Dream",
@@ -718,6 +725,61 @@ 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."""
+    name = ctx.args.strip()
+    if not name:
+        return OutboundMessage(
+            channel=ctx.msg.channel,
+            chat_id=ctx.msg.chat_id,
+            content=(
+                "Usage: /trigger <name>\n\n"
+                "Create a named local trigger bound to this chat session."
+            ),
+            metadata={**dict(ctx.msg.metadata or {}), "render_as": "text"},
+        )
+
+    from nanobot.triggers.local_store import LocalTriggerStore
+
+    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, "local_trigger_store", None)
+    if store is None:
+        store = LocalTriggerStore(workspace)
+
+    from nanobot.session.keys import UNIFIED_SESSION_KEY
+
+    session_key = (
+        ctx.msg.session_key
+        if ctx.key == UNIFIED_SESSION_KEY
+        else ctx.key
+    )
+    trigger = store.create(
+        name=name,
+        channel=ctx.msg.channel,
+        chat_id=ctx.msg.chat_id,
+        session_key=session_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 +814,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)

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):

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."""

nanobot/cron/session_turns.py

@@ -5,16 +5,43 @@ from __future__ import annotations
 from typing import Any, Mapping
 
 from nanobot.cron.types import CronJob
+from nanobot.session.automation_turns import (
+    AutomationTurnSpec,
+    automation_history_overrides_for_spec,
+    automation_trigger,
+)
 
 CRON_TRIGGER_META = "_cron_trigger"
 CRON_DEFER_UNTIL_IDLE_META = "_cron_defer_until_session_idle"
 CRON_HISTORY_META = "_cron_turn"
 
 
+def _cron_history_text(trigger: Mapping[str, Any]) -> str | None:
+    persist_content = trigger.get("persist_content")
+    return (
+        persist_content
+        if isinstance(persist_content, str) and persist_content.strip()
+        else None
+    )
+
+
+CRON_AUTOMATION_SPEC = AutomationTurnSpec(
+    kind="cron",
+    trigger_meta_key=CRON_TRIGGER_META,
+    legacy_history_meta_key=CRON_HISTORY_META,
+    history_fields={
+        "cron_job_id": "job_id",
+        "cron_job_name": "job_name",
+        "cron_run_id": "run_id",
+        "cron_prompt_ref": "prompt_ref",
+    },
+    text_builder=_cron_history_text,
+)
+
+
 def cron_trigger(metadata: Mapping[str, Any] | None) -> dict[str, Any] | None:
     """Return structured cron trigger metadata when present."""
-    raw = (metadata or {}).get(CRON_TRIGGER_META)
-    return raw if isinstance(raw, dict) else None
+    return automation_trigger(metadata, CRON_AUTOMATION_SPEC)
 
 
 def is_cron_turn(metadata: Mapping[str, Any] | None) -> bool:
@@ -38,22 +65,7 @@ def cron_run_id(metadata: Mapping[str, Any] | None) -> str | None:
 
 def cron_history_overrides(metadata: Mapping[str, Any] | None) -> tuple[str | None, dict[str, Any]]:
     """Return session-history text/metadata overrides for a cron turn."""
-    trigger = cron_trigger(metadata)
-    if not trigger:
-        return None, {}
-    persist_content = trigger.get("persist_content")
-    text = (
-        persist_content
-        if isinstance(persist_content, str) and persist_content.strip()
-        else None
-    )
-    return text, {
-        CRON_HISTORY_META: True,
-        "cron_job_id": trigger.get("job_id"),
-        "cron_job_name": trigger.get("job_name"),
-        "cron_run_id": trigger.get("run_id"),
-        "cron_prompt_ref": trigger.get("prompt_ref"),
-    }
+    return automation_history_overrides_for_spec(metadata, CRON_AUTOMATION_SPEC)
 
 
 def is_bound_cron_job(job: CronJob) -> bool:

nanobot/session/automation_turns.py

@@ -0,0 +1,92 @@
+"""Shared handling for session-bound automation turns."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, field
+from functools import lru_cache
+from typing import Any
+
+AUTOMATION_HISTORY_META = "_automation_turn"
+
+
+@dataclass(frozen=True)
+class AutomationTurnSpec:
+    """Source-specific wiring for one session-bound automation turn type."""
+
+    kind: str
+    trigger_meta_key: str
+    legacy_history_meta_key: str | None = None
+    history_fields: Mapping[str, str] = field(default_factory=dict)
+    text_builder: Callable[[Mapping[str, Any]], str | None] | None = None
+
+
+def automation_trigger(
+    metadata: Mapping[str, Any] | None,
+    spec: AutomationTurnSpec,
+) -> dict[str, Any] | None:
+    """Return source trigger metadata for *spec* when present."""
+    raw = (metadata or {}).get(spec.trigger_meta_key)
+    return raw if isinstance(raw, dict) else None
+
+
+def automation_history_overrides_for_spec(
+    metadata: Mapping[str, Any] | None,
+    spec: AutomationTurnSpec,
+) -> tuple[str | None, dict[str, Any]]:
+    """Return hidden session-history text/metadata overrides for *spec*."""
+    trigger = automation_trigger(metadata, spec)
+    if not trigger:
+        return None, {}
+
+    details: dict[str, Any] = {"kind": spec.kind}
+    extra: dict[str, Any] = {AUTOMATION_HISTORY_META: details}
+    if spec.legacy_history_meta_key:
+        extra[spec.legacy_history_meta_key] = True
+    for history_key, trigger_key in spec.history_fields.items():
+        value = trigger.get(trigger_key)
+        extra[history_key] = value
+        details[history_key] = value
+
+    text = spec.text_builder(trigger) if spec.text_builder else None
+    return text, extra
+
+
+@lru_cache(maxsize=1)
+def _automation_specs() -> tuple[AutomationTurnSpec, ...]:
+    # Source modules import the generic helpers above, so keep spec loading lazy.
+    from nanobot.cron.session_turns import CRON_AUTOMATION_SPEC
+    from nanobot.triggers.local_session_turns import LOCAL_TRIGGER_AUTOMATION_SPEC
+
+    return (CRON_AUTOMATION_SPEC, LOCAL_TRIGGER_AUTOMATION_SPEC)
+
+
+def automation_history_overrides(
+    metadata: Mapping[str, Any] | None,
+) -> tuple[str | None, dict[str, Any]]:
+    """Return session-history text/metadata overrides for supported automation turns."""
+    for spec in _automation_specs():
+        text, extra = automation_history_overrides_for_spec(metadata, spec)
+        if extra:
+            return text, extra
+    return None, {}
+
+
+def is_automation_history_message(message: Mapping[str, Any] | None) -> bool:
+    """True for hidden automation trigger records in session history."""
+    if not message:
+        return False
+    marker = message.get(AUTOMATION_HISTORY_META)
+    if marker is True or isinstance(marker, Mapping):
+        return True
+    return any(
+        spec.legacy_history_meta_key
+        and message.get(spec.legacy_history_meta_key) is True
+        for spec in _automation_specs()
+    )
+
+
+def is_automation_kind(value: Any) -> bool:
+    return isinstance(value, str) and (
+        value == "trigger" or any(spec.kind == value for spec in _automation_specs())
+    )

nanobot/session/webui_turns.py

@@ -30,8 +30,8 @@ from nanobot.bus.runtime_events import (
     TurnCompleted,
     TurnRunStatusChanged,
 )
-from nanobot.cron.session_turns import CRON_HISTORY_META
 from nanobot.providers.base import LLMProvider
+from nanobot.session.automation_turns import is_automation_history_message
 from nanobot.session.goal_state import goal_state_ws_blob
 from nanobot.session.manager import Session, SessionManager
 from nanobot.utils.helpers import strip_think, truncate_text
@@ -77,7 +77,7 @@ def _title_inputs(session: Session) -> tuple[str, str]:
     for message in session.messages:
         if message.get("_command") is True:
             continue
-        if message.get(CRON_HISTORY_META) is True:
+        if is_automation_history_message(message):
             continue
         role = message.get("role")
         content = message.get("content")

nanobot/triggers/__init__.py

@@ -0,0 +1,19 @@
+"""Local trigger support."""
+
+from nanobot.triggers.local_store import (
+    LocalTriggerStore,
+    TriggerDisabledError,
+    TriggerNotFoundError,
+    TriggerStoreError,
+)
+from nanobot.triggers.local_types import LocalTrigger, TriggerDelivery, TriggerRunRecord
+
+__all__ = [
+    "LocalTrigger",
+    "LocalTriggerStore",
+    "TriggerDelivery",
+    "TriggerDisabledError",
+    "TriggerNotFoundError",
+    "TriggerRunRecord",
+    "TriggerStoreError",
+]

nanobot/triggers/local_runner.py

@@ -0,0 +1,209 @@
+"""Gateway delivery loop for local triggers."""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from loguru import logger
+
+from nanobot.agent.automation_turns import AutomationTurnError
+from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.triggers.local_session_turns import LOCAL_TRIGGER_META
+from nanobot.triggers.local_store import LocalTriggerStore
+from nanobot.triggers.local_types import LocalTrigger, TriggerDelivery
+from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN_METADATA_KEY
+
+
+async def run_local_trigger_queue(
+    *,
+    store: LocalTriggerStore,
+    submit_turn: Callable[[InboundMessage], Awaitable[OutboundMessage | None]] | None = None,
+    poll_interval_s: float = 0.5,
+    batch_size: int = 20,
+) -> None:
+    """Poll local trigger deliveries and submit them as session turns."""
+    if submit_turn is None:
+        raise ValueError("run_local_trigger_queue requires submit_turn")
+    logger.info("Local trigger queue started")
+    recovered = store.recover_processing_deliveries()
+    if recovered:
+        logger.warning(
+            "Trigger: recovered {} interrupted delivery file(s) from processing",
+            recovered,
+        )
+    while True:
+        deliveries = store.claim_deliveries(limit=batch_size)
+        if not deliveries:
+            await asyncio.sleep(poll_interval_s)
+            continue
+
+        for delivery in deliveries:
+            try:
+                await _deliver_delivery(
+                    store,
+                    delivery,
+                    submit_turn=submit_turn,
+                )
+                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(
+                    delivery.trigger_id,
+                    status="error",
+                    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 {}: {}",
+                    delivery.id,
+                    delivery.trigger_id,
+                    exc,
+                )
+            except AutomationTurnError as exc:
+                error = str(exc) or exc.__class__.__name__
+                store.record_delivery(
+                    delivery.trigger_id,
+                    status="error",
+                    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: {}",
+                    delivery.id,
+                    delivery.trigger_id,
+                    error,
+                )
+            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",
+                    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 _deliver_delivery(
+    store: LocalTriggerStore,
+    delivery: TriggerDelivery,
+    *,
+    submit_turn: Callable[[InboundMessage], Awaitable[OutboundMessage | None]],
+) -> 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")
+
+    store.write_delivery_run_record(delivery, trigger=trigger, status="processing")
+    msg = InboundMessage(
+        channel=trigger.channel,
+        sender_id=trigger.sender_id,
+        chat_id=trigger.chat_id,
+        content=delivery.content,
+        metadata=_delivery_metadata(trigger, delivery),
+        session_key_override=trigger.session_key,
+    )
+    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]:
+    metadata = dict(trigger.origin_metadata or {})
+    metadata[LOCAL_TRIGGER_META] = {
+        "trigger_id": trigger.id,
+        "trigger_name": trigger.name,
+        "delivery_id": delivery.id,
+        "created_at_ms": delivery.created_at_ms,
+        "persist_content": _history_content(trigger, delivery),
+    }
+    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": "local_trigger"}
+        if trigger.name:
+            source["label"] = trigger.name
+        metadata[WEBUI_MESSAGE_SOURCE_METADATA_KEY] = source
+    return metadata
+
+
+def _history_content(trigger: LocalTrigger, delivery: TriggerDelivery) -> str:
+    label = trigger.name.strip() if trigger.name else trigger.id
+    return f"Local trigger received: {label}\n\n{delivery.content}"

nanobot/triggers/local_session_turns.py

@@ -0,0 +1,62 @@
+"""Shared metadata helpers for local trigger session turns."""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+from nanobot.session.automation_turns import (
+    AutomationTurnSpec,
+    automation_history_overrides_for_spec,
+    automation_trigger,
+)
+
+LOCAL_TRIGGER_META = "_local_trigger"
+
+
+def _local_trigger_history_text(trigger: Mapping[str, Any]) -> str:
+    persist_content = trigger.get("persist_content")
+    if isinstance(persist_content, str) and persist_content.strip():
+        return persist_content
+    name = trigger.get("trigger_name")
+    trigger_id = trigger.get("trigger_id")
+    label = name if isinstance(name, str) and name.strip() else trigger_id
+    return (
+        f"Local trigger received: {label}"
+        if isinstance(label, str) and label.strip()
+        else "Local trigger received"
+    )
+
+
+LOCAL_TRIGGER_AUTOMATION_SPEC = AutomationTurnSpec(
+    kind="local_trigger",
+    trigger_meta_key=LOCAL_TRIGGER_META,
+    history_fields={
+        "trigger_id": "trigger_id",
+        "trigger_name": "trigger_name",
+        "trigger_delivery_id": "delivery_id",
+    },
+    text_builder=_local_trigger_history_text,
+)
+
+
+def local_trigger(metadata: Mapping[str, Any] | None) -> dict[str, Any] | None:
+    """Return structured local trigger metadata when present."""
+    return automation_trigger(metadata, LOCAL_TRIGGER_AUTOMATION_SPEC)
+
+
+def local_trigger_delivery_id(metadata: Mapping[str, Any] | None) -> str | None:
+    trigger = local_trigger(metadata)
+    if not trigger:
+        return None
+    value = trigger.get("delivery_id")
+    return value if isinstance(value, str) and value else None
+
+
+def local_trigger_history_overrides(
+    metadata: Mapping[str, Any] | None,
+) -> tuple[str | None, dict[str, Any]]:
+    """Return session-history text/metadata overrides for a local trigger turn."""
+    return automation_history_overrides_for_spec(
+        metadata,
+        LOCAL_TRIGGER_AUTOMATION_SPEC,
+    )

nanobot/triggers/local_store.py

@@ -0,0 +1,474 @@
+"""Workspace-scoped local trigger store and delivery queue."""
+
+from __future__ import annotations
+
+import errno
+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.local_types import LocalTrigger, TriggerDelivery, TriggerRunRecord
+from nanobot.utils.helpers import truncate_text
+from nanobot.utils.run_records import write_run_record as write_automation_run_record
+
+_TRIGGER_ID_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
+_MAX_RUN_HISTORY = 20
+_MAX_DELIVERY_ATTEMPTS = 10
+_RUN_RECORD_TEXT_MAX_CHARS = 4000
+_PROCESSING_RECOVERY_ERROR = "delivery was recovered from interrupted processing"
+
+
+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 LocalTriggerStore:
+    """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.runs_dir = self.root / "runs"
+        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,
+    ) -> LocalTrigger:
+        """Create a new session-bound local 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 = LocalTrigger(
+                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[LocalTrigger]:
+        """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[LocalTrigger]:
+        """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) -> LocalTrigger | 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) -> LocalTrigger | 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) -> LocalTrigger | 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."""
+        trigger_id = trigger_id.strip()
+        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)
+            self._delete_delivery_files_for_trigger_unlocked(trigger_id)
+            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
+            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]:
+        """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 recover_processing_deliveries(self) -> int:
+        """Requeue deliveries left in processing by an interrupted gateway."""
+        self._ensure_dirs()
+        recovered = 0
+        with self._lock:
+            for path in sorted(self.processing_dir.glob("*.json")):
+                try:
+                    data = json.loads(path.read_text(encoding="utf-8"))
+                    delivery = TriggerDelivery.from_dict(
+                        data.get("delivery", data),
+                        path=path,
+                    )
+                except Exception:
+                    logger.exception("Trigger: failed to parse processing delivery {}", path)
+                    self._move_bad_delivery_unlocked(path)
+                    continue
+                if self._retry_delivery_unlocked(delivery, _PROCESSING_RECOVERY_ERROR):
+                    recovered += 1
+        return recovered
+
+    def complete_delivery(self, delivery: TriggerDelivery) -> None:
+        """Delete a claimed delivery after it is handled."""
+        if delivery.path is None:
+            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:
+            return self._retry_delivery_unlocked(delivery, error)
+
+    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 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"] = _run_record_text(error)
+        if response is not None:
+            record["response"] = _run_record_text(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():
+            return []
+        try:
+            data = json.loads(self.store_path.read_text(encoding="utf-8"))
+            return [
+                LocalTrigger.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-<ts> backup"
+            ) from exc
+
+    def _save_triggers_unlocked(self, triggers: list[LocalTrigger]) -> 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[LocalTrigger],
+        trigger_id: str,
+    ) -> LocalTrigger | 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)
+
+    def _retry_delivery_unlocked(self, delivery: TriggerDelivery, error: str) -> bool:
+        if delivery.path is None:
+            return False
+        if delivery.attempts + 1 >= _MAX_DELIVERY_ATTEMPTS:
+            delivery.attempts += 1
+            delivery.last_error = error
+            failed = self.failed_dir / delivery.path.name
+            self._atomic_write(failed, json.dumps(_delivery_payload(delivery), ensure_ascii=False))
+            delivery.path.unlink(missing_ok=True)
+            return False
+        delivery.attempts += 1
+        delivery.last_error = error
+        target = self.inbox_dir / delivery.path.name
+        self._atomic_write(target, json.dumps(_delivery_payload(delivery), ensure_ascii=False))
+        delivery.path.unlink(missing_ok=True)
+        return True
+
+    def _delete_delivery_files_for_trigger_unlocked(self, trigger_id: str) -> None:
+        for directory in (self.inbox_dir, self.processing_dir, self.failed_dir):
+            for path in directory.iterdir():
+                if not path.is_file():
+                    continue
+                if self._delivery_file_trigger_id(path) != trigger_id:
+                    continue
+                try:
+                    path.unlink(missing_ok=True)
+                except OSError as exc:
+                    logger.warning(
+                        "Trigger: failed to delete delivery file {} for deleted trigger {}: {}",
+                        path,
+                        trigger_id,
+                        exc,
+                    )
+
+    @staticmethod
+    def _delivery_file_trigger_id(path: Path) -> str | None:
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except Exception:
+            return None
+        raw = data.get("delivery", data) if isinstance(data, dict) else None
+        if not isinstance(raw, dict):
+            return None
+        trigger_id = raw.get("triggerId", raw.get("trigger_id", ""))
+        return str(trigger_id) if trigger_id else None
+
+    @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:
+                    try:
+                        os.fsync(fd)
+                    except OSError as exc:
+                        if exc.errno != errno.EINVAL:
+                            raise
+                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 "Local 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(),
+    }
+
+
+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": _run_record_text(delivery.content),
+        "created_at_ms": delivery.created_at_ms,
+        "attempts": delivery.attempts,
+    }
+    if delivery.last_error:
+        record["last_error"] = _run_record_text(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
+
+
+def _run_record_text(value: str) -> str:
+    return truncate_text(value, _RUN_RECORD_TEXT_MAX_CHARS)

nanobot/triggers/local_turns.py

@@ -0,0 +1,55 @@
+"""Coordination for local trigger turns."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable, Iterable
+
+from nanobot.agent.automation_turns import AutomationTurnCoordinator
+from nanobot.bus.events import InboundMessage
+from nanobot.triggers.local_session_turns import local_trigger, local_trigger_delivery_id
+
+
+class LocalTriggerTurnCoordinator(AutomationTurnCoordinator):
+    """Manage local trigger turns without mixing them into live injections."""
+
+    def __init__(
+        self,
+        *,
+        publish_inbound: Callable[[InboundMessage], Awaitable[None]],
+        dispatch: Callable[[InboundMessage], Awaitable[object]],
+        is_running: Callable[[], bool],
+        deferred_queues: dict[str, list[InboundMessage]] | None = None,
+    ) -> None:
+        super().__init__(
+            publish_inbound=publish_inbound,
+            dispatch=dispatch,
+            is_running=is_running,
+            turn_id=lambda msg: local_trigger_delivery_id(msg.metadata),
+            pending_id=_local_trigger_id,
+            should_defer_turn=_should_defer_local_trigger_turn,
+            missing_id_error="local trigger turn metadata must include a delivery_id",
+            duplicate_id_error=lambda delivery_id: (
+                f"local trigger delivery {delivery_id!r} is already pending"
+            ),
+            deferred_queues=deferred_queues,
+        )
+
+    def pending_trigger_ids_for_session(self, session_key: str) -> set[str]:
+        """Return local triggers waiting for or running in *session_key*."""
+        return self.pending_ids_for_session(session_key)
+
+
+def _should_defer_local_trigger_turn(
+    msg: InboundMessage,
+    session_key: str,
+    active_session_keys: Iterable[str],
+) -> bool:
+    return local_trigger(msg.metadata) is not None and session_key in active_session_keys
+
+
+def _local_trigger_id(msg: InboundMessage) -> str | None:
+    trigger = local_trigger(msg.metadata)
+    if not trigger:
+        return None
+    value = trigger.get("trigger_id")
+    return value if isinstance(value, str) and value else None

nanobot/triggers/local_types.py

@@ -0,0 +1,141 @@
+"""Persistent types for local 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 LocalTrigger:
+    """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]) -> "LocalTrigger":
+        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,
+        }

nanobot/utils/run_records.py

@@ -0,0 +1,58 @@
+"""Durable JSON run records for automation executions."""
+
+from __future__ import annotations
+
+import errno
+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:
+                try:
+                    os.fsync(fd)
+                except OSError as exc:
+                    if exc.errno != errno.EINVAL:
+                        raise
+            finally:
+                os.close(fd)
+    except BaseException:
+        tmp_path.unlink(missing_ok=True)
+        raise

nanobot/utils/runtime.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import re
 from pathlib import Path
 from typing import Any
@@ -10,7 +11,7 @@ from loguru import logger
 
 from nanobot.utils.helpers import stringify_text_blocks
 
-_MAX_REPEAT_EXTERNAL_LOOKUPS = 2
+_MAX_REPEAT_ATTEMPTS = 2
 
 # Third same-target workspace violation in a turn escalates to "stop retrying".
 _MAX_REPEAT_WORKSPACE_VIOLATIONS = 2
@@ -103,6 +104,14 @@ def external_lookup_signature(tool_name: str, arguments: Any) -> str | None:
     return None
 
 
+def _over_repeat_budget(signature: str | None, seen_counts: dict[str, int]) -> int | None:
+    if signature is None:
+        return None
+    count = seen_counts.get(signature, 0) + 1
+    seen_counts[signature] = count
+    return count if count > _MAX_REPEAT_ATTEMPTS else None
+
+
 def repeated_external_lookup_error(
     tool_name: str,
     arguments: Any,
@@ -110,11 +119,8 @@ def repeated_external_lookup_error(
 ) -> str | None:
     """Block repeated external lookups after a small retry budget."""
     signature = external_lookup_signature(tool_name, arguments)
-    if signature is None:
-        return None
-    count = seen_counts.get(signature, 0) + 1
-    seen_counts[signature] = count
-    if count <= _MAX_REPEAT_EXTERNAL_LOOKUPS:
+    count = _over_repeat_budget(signature, seen_counts)
+    if count is None:
         return None
     logger.warning(
         "Blocking repeated external lookup {} on attempt {}",
@@ -127,6 +133,33 @@ def repeated_external_lookup_error(
     )
 
 
+def repeated_tool_result_hint(
+    tool_name: str,
+    result: Any,
+    seen_counts: dict[str, int],
+) -> str | None:
+    """Hint when a successful tool keeps returning the exact same text in one turn."""
+    if isinstance(result, str):
+        text = result
+    elif isinstance(result, list):
+        text = stringify_text_blocks(result)
+    else:
+        text = None
+    if text is None:
+        return None
+    digest = hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()
+    signature = f"tool_result:{tool_name}:{len(text)}:{digest}"
+    count = _over_repeat_budget(signature, seen_counts)
+    if count is None:
+        return None
+    logger.warning("Hinting repeated {} result on attempt {}", tool_name, count)
+    return (
+        f"\n\n[Repeated {tool_name} result: this exact output has already been "
+        "returned in this turn. Use the existing evidence, or change the tool input "
+        "if you need new information.]"
+    )
+
+
 # Workspace-boundary violations are soft errors, with per-target throttling.
 
 _OUTSIDE_PATH_PATTERN = re.compile(r"(?:^|[\s|>'\"])((?:/[^\s\"'>;|<]+)|(?:~[^\s\"'>;|<]+))")

nanobot/webui/gateway_services.py

@@ -26,7 +26,9 @@ class GatewayServices:
     workspaces: WebUIWorkspaceController
     session_manager: Any | None
     cron_service: Any | None
+    local_trigger_store: Any | None
     cron_pending_job_ids: Callable[[str], set[str]] | None
+    local_trigger_pending_ids: Callable[[str], set[str]] | None
 
 
 def build_gateway_services(
@@ -42,7 +44,9 @@ def build_gateway_services(
     runtime_capabilities_overrides: dict[str, Any] | None,
     disabled_skills: set[str] | None = None,
     cron_service: Any | None = None,
+    local_trigger_store: Any | None = None,
     cron_pending_job_ids: Callable[[str], set[str]] | None = None,
+    local_trigger_pending_ids: Callable[[str], set[str]] | None = None,
     logger: Any = default_logger,
 ) -> GatewayServices:
     tokens = GatewayTokenStore()
@@ -70,7 +74,9 @@ def build_gateway_services(
         skills_workspace_path=workspace_path,
         disabled_skills=disabled_skills,
         cron_service=cron_service,
+        local_trigger_store=local_trigger_store,
         cron_pending_job_ids=cron_pending_job_ids,
+        local_trigger_pending_ids=local_trigger_pending_ids,
         log=logger,
     )
     return GatewayServices(
@@ -81,5 +87,7 @@ def build_gateway_services(
         workspaces=workspaces,
         session_manager=session_manager,
         cron_service=cron_service,
+        local_trigger_store=local_trigger_store,
         cron_pending_job_ids=cron_pending_job_ids,
+        local_trigger_pending_ids=local_trigger_pending_ids,
     )

nanobot/webui/session_automations.py

@@ -5,9 +5,12 @@ from __future__ import annotations
 from collections.abc import Collection
 from typing import Any, Protocol
 
-from nanobot.cron.session_turns import CRON_HISTORY_META
 from nanobot.cron.types import CronJob
+from nanobot.session.automation_turns import is_automation_history_message
 from nanobot.session.manager import _message_preview_text
+from nanobot.triggers.local_types import LocalTrigger
+
+AutomationJob = CronJob | LocalTrigger
 
 
 class _CronServiceLike(Protocol):
@@ -21,6 +24,17 @@ class _CronServiceLike(Protocol):
     ) -> list[CronJob]: ...
 
 
+class _LocalTriggerStoreLike(Protocol):
+    def list_triggers(self, *, include_disabled: bool = False) -> list[LocalTrigger]: ...
+
+    def list_for_session(
+        self,
+        session_key: str,
+        *,
+        include_disabled: bool = True,
+    ) -> list[LocalTrigger]: ...
+
+
 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]:
+    *,
+    local_trigger_store: _LocalTriggerStoreLike | 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(
+    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 local_trigger_store is not None:
+        jobs.extend(
+            local_trigger_store.list_for_session(
+                session_key,
+                include_disabled=True,
+            )
+        )
+    return jobs
 
 
 def session_automations_payload(
     cron_service: _CronServiceLike | None,
     session_key: str,
     *,
+    local_trigger_store: _LocalTriggerStoreLike | 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,
+                local_trigger_store=local_trigger_store,
+            ),
             pending_job_ids=pending_job_ids,
         )
     }
@@ -56,11 +87,16 @@ def session_automations_payload(
 def all_automations_payload(
     cron_service: _CronServiceLike | None,
     *,
+    local_trigger_store: _LocalTriggerStoreLike | 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 local_trigger_store is not None:
+        jobs.extend(local_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,20 @@ 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, LocalTrigger):
+        return _serialize_trigger(
+            job,
+            pending=pending,
+            include_details=include_details,
+            session_manager=session_manager,
+        )
+
     payload = {
         "id": job.id,
         "name": job.name,
@@ -143,6 +187,67 @@ def _serialize_job(
     return payload
 
 
+def _serialize_trigger(
+    trigger: LocalTrigger,
+    *,
+    pending: bool = False,
+    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": "local_trigger",
+        "schedule": {
+            "kind": "local",
+            "at_ms": None,
+            "every_ms": None,
+            "expr": None,
+            "tz": None,
+        },
+        "payload": {
+            "kind": "local_trigger",
+            "message": command,
+            "command": command,
+        },
+        "state": {
+            "next_run_at_ms": None,
+            "last_status": trigger.last_status,
+            "pending": pending,
+        },
+    }
+    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 +266,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: LocalTrigger,
+    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):
@@ -183,7 +328,7 @@ def _session_preview(messages: Any) -> str:
     for message in messages:
         if not isinstance(message, dict):
             continue
-        if message.get(CRON_HISTORY_META) is True:
+        if is_automation_history_message(message):
             continue
         text = _message_preview_text(message)
         if not text:

nanobot/webui/session_list_index.py

@@ -16,7 +16,7 @@ from typing import Any
 from loguru import logger
 
 from nanobot.config.paths import get_webui_dir
-from nanobot.cron.session_turns import CRON_HISTORY_META
+from nanobot.session.automation_turns import is_automation_history_message
 from nanobot.session.manager import (
     _SESSION_LIST_PREVIEW_MAX_CHARS,
     _SESSION_LIST_PREVIEW_MAX_RECORDS,
@@ -154,7 +154,7 @@ def _preview_from_messages(messages: list[dict[str, Any]]) -> str:
             or scanned_chars > _SESSION_LIST_PREVIEW_MAX_CHARS
         ):
             break
-        if item.get(CRON_HISTORY_META) is True:
+        if is_automation_history_message(item):
             continue
         text = _message_preview_text(item)
         if not text:
@@ -216,7 +216,7 @@ def _latest_updated_at(stored: str | None, activity: str | None) -> str | None:
 
 
 def _visible_message_timestamp(item: dict[str, Any]) -> str | None:
-    if item.get(CRON_HISTORY_META) is True:
+    if is_automation_history_message(item):
         return None
     if item.get("role") not in _VISIBLE_TRANSCRIPT_ROLES:
         return None
@@ -296,7 +296,9 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
                     ):
                         preview_done = True
                         continue
-                    if item.get(CRON_HISTORY_META) is True:
+                    if item.get("_type") == "metadata":
+                        continue
+                    if is_automation_history_message(item):
                         continue
                     text = _message_preview_text(item)
                     if not text:

nanobot/webui/transcript.py

@@ -17,7 +17,7 @@ from urllib.parse import unquote, urlparse
 from loguru import logger
 
 from nanobot.config.paths import get_webui_dir
-from nanobot.cron.session_turns import CRON_HISTORY_META
+from nanobot.session.automation_turns import is_automation_history_message, is_automation_kind
 from nanobot.session.manager import SessionManager
 from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN_METADATA_KEY
 
@@ -598,9 +598,12 @@ def normalize_webui_turn_id(value: Any) -> str:
 
 def webui_message_source(metadata: dict[str, Any] | None) -> dict[str, str] | None:
     raw = (metadata or {}).get(WEBUI_MESSAGE_SOURCE_METADATA_KEY)
-    if not isinstance(raw, dict) or raw.get("kind") != "cron":
+    if not isinstance(raw, dict):
         return None
-    source: dict[str, str] = {"kind": "cron"}
+    kind = raw.get("kind")
+    if not is_automation_kind(kind):
+        return None
+    source: dict[str, str] = {"kind": kind}
     label = raw.get("label")
     if isinstance(label, str) and label.strip():
         source["label"] = label.strip()
@@ -779,6 +782,8 @@ def write_session_messages_as_transcript(
     target_chat_id = _chat_id_from_session_key(target_key)
     rows: list[dict[str, Any]] = []
     for msg in messages:
+        if is_automation_history_message(msg):
+            continue
         role = msg.get("role")
         content = msg.get("content")
         text = content if isinstance(content, str) else ""
@@ -855,7 +860,7 @@ def _session_user_event(
 ) -> dict[str, Any] | None:
     if message.get("role") != "user":
         return None
-    if message.get(CRON_HISTORY_META) is True:
+    if is_automation_history_message(message):
         return None
     content = message.get("content")
     text = content if isinstance(content, str) else ""
@@ -1271,9 +1276,12 @@ def replay_transcript_to_ui_messages(
 
     def _source_fields(rec: dict[str, Any]) -> dict[str, Any]:
         source = rec.get("source")
-        if not isinstance(source, dict) or source.get("kind") != "cron":
+        if not isinstance(source, dict):
             return {}
-        out: dict[str, Any] = {"source": {"kind": "cron"}}
+        kind = source.get("kind")
+        if not is_automation_kind(kind):
+            return {}
+        out: dict[str, Any] = {"source": {"kind": kind}}
         label = source.get("label")
         if isinstance(label, str) and label.strip():
             out["source"]["label"] = label.strip()

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.local_types import LocalTrigger
 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.local_store import LocalTriggerStore
 
 
 def _decode_api_key(raw_key: str) -> str | None:
@@ -153,7 +155,9 @@ class GatewayHTTPHandler:
         skills_workspace_path: Path,
         disabled_skills: set[str] | None = None,
         cron_service: CronService | None = None,
+        local_trigger_store: LocalTriggerStore | None = None,
         cron_pending_job_ids: Callable[[str], set[str]] | None = None,
+        local_trigger_pending_ids: Callable[[str], set[str]] | None = None,
         log: Any = logger,
     ) -> None:
         self.config = config
@@ -167,7 +171,9 @@ class GatewayHTTPHandler:
         self.skills_workspace_path = skills_workspace_path
         self.disabled_skills = disabled_skills or set()
         self.cron_service = cron_service
+        self.local_trigger_store = local_trigger_store
         self.cron_pending_job_ids = cron_pending_job_ids
+        self.local_trigger_pending_ids = local_trigger_pending_ids
         self._log = log
         self._runtime_surface = runtime_surface
 
@@ -483,13 +489,12 @@ class GatewayHTTPHandler:
             return _http_error(400, "invalid session key")
         if not _is_websocket_channel_session_key(decoded_key):
             return _http_error(404, "session not found")
-        pending_job_ids: set[str] = set()
-        if self.cron_pending_job_ids is not None:
-            pending_job_ids = self.cron_pending_job_ids(decoded_key)
+        pending_job_ids = self._pending_automation_ids_for_session(decoded_key)
         return _http_json_response(
             session_automations_payload(
                 self.cron_service,
                 decoded_key,
+                local_trigger_store=self.local_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,
+            local_trigger_store=self.local_trigger_store,
+        )
         if automation_jobs and delete_automations not in {"1", "true", "yes"}:
             return _http_json_response(
                 {
@@ -515,8 +524,12 @@ 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:
+                if isinstance(job, LocalTrigger):
+                    if self.local_trigger_store is not None:
+                        self.local_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)
@@ -548,14 +561,37 @@ class GatewayHTTPHandler:
                 pending.update(self.cron_pending_job_ids(session_key))
         return pending
 
+    def _pending_local_trigger_ids_for_all(self) -> set[str]:
+        if self.local_trigger_store is None or self.local_trigger_pending_ids is None:
+            return set()
+        pending: set[str] = set()
+        for trigger in self.local_trigger_store.list_triggers(include_disabled=True):
+            session_key = trigger.session_key
+            if not session_key and trigger.channel and trigger.chat_id:
+                session_key = f"{trigger.channel}:{trigger.chat_id}"
+            if session_key:
+                pending.update(self.local_trigger_pending_ids(session_key))
+        return pending
+
+    def _pending_automation_ids_for_session(self, session_key: str) -> set[str]:
+        pending: set[str] = set()
+        if self.cron_pending_job_ids is not None:
+            pending.update(self.cron_pending_job_ids(session_key))
+        if self.local_trigger_pending_ids is not None:
+            pending.update(self.local_trigger_pending_ids(session_key))
+        return pending
+
     def _handle_webui_automations(self, request: WsRequest) -> Response:
         if not self.check_api_token(request):
             return _http_error(401, "Unauthorized")
+        pending_job_ids = self._pending_cron_job_ids_for_all()
+        pending_job_ids.update(self._pending_local_trigger_ids_for_all())
         return _http_json_response(
             all_automations_payload(
                 self.cron_service,
+                local_trigger_store=self.local_trigger_store,
                 session_manager=self.session_manager,
-                pending_job_ids=self._pending_cron_job_ids_for_all(),
+                pending_job_ids=pending_job_ids,
             )
         )
 
@@ -566,13 +602,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.local_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.local_trigger_store.get(job_id) if self.local_trigger_store else None
+        if trigger is not None:
+            return self._handle_local_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 +660,40 @@ class GatewayHTTPHandler:
 
         return self._handle_webui_automations(request)
 
+    def _handle_local_trigger_action(
+        self,
+        request: WsRequest,
+        action: str,
+        trigger: LocalTrigger,
+    ) -> Response:
+        if self.local_trigger_store is None:
+            return _http_error(503, "trigger service unavailable")
+        if action == "enable":
+            if self.local_trigger_store.enable(trigger.id, enabled=True) is None:
+                return _http_error(404, "automation not found")
+        elif action == "disable":
+            if self.local_trigger_store.enable(trigger.id, enabled=False) is None:
+                return _http_error(404, "automation not found")
+        elif action == "delete":
+            if not self.local_trigger_store.delete(trigger.id):
+                return _http_error(404, "automation not found")
+        elif action == "run":
+            return _http_error(409, "local 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_local_trigger_update(values)
+            if isinstance(parsed, str):
+                return _http_error(400, parsed)
+            if parsed:
+                if self.local_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 +906,22 @@ def _parse_automation_update(
     return update
 
 
+def _parse_local_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 "local 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):

tests/agent/test_loop_save_turn.py

@@ -18,6 +18,7 @@ from nanobot.bus.outbound_events import (
 from nanobot.bus.queue import MessageBus
 from nanobot.cron.session_turns import CRON_HISTORY_META, CRON_TRIGGER_META
 from nanobot.providers.base import LLMResponse
+from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
 from nanobot.session.goal_state import GOAL_STATE_KEY
 from nanobot.session.manager import Session, SessionManager
 from nanobot.session.turn_continuation import (
@@ -33,6 +34,7 @@ from nanobot.session.webui_turns import (
     clean_generated_title,
     maybe_generate_webui_title,
 )
+from nanobot.triggers.local_session_turns import LOCAL_TRIGGER_META
 from nanobot.utils.llm_runtime import LLMRuntime
 
 
@@ -101,6 +103,13 @@ def test_persist_cron_turn_uses_distinct_history_marker(tmp_path: Path) -> None:
     assert persisted is True
     message = session.messages[-1]
     assert message["content"] == "Scheduled cron job triggered: Daily check"
+    assert message[AUTOMATION_HISTORY_META] == {
+        "kind": "cron",
+        "cron_job_id": "job-1",
+        "cron_job_name": "Daily check",
+        "cron_run_id": "job-1:1",
+        "cron_prompt_ref": prompt_ref,
+    }
     assert message[CRON_HISTORY_META] is True
     assert CRON_TRIGGER_META not in message
     assert message["cron_job_id"] == "job-1"
@@ -109,6 +118,63 @@ def test_persist_cron_turn_uses_distinct_history_marker(tmp_path: Path) -> None:
     assert message["cron_prompt_ref"] == prompt_ref
 
 
+def test_persist_local_trigger_turn_uses_hidden_automation_marker(tmp_path: Path) -> None:
+    loop = _make_full_loop(tmp_path)
+    session = loop.sessions.get_or_create("websocket:auto")
+
+    persisted = loop._persist_user_message_early(
+        InboundMessage(
+            channel="websocket",
+            sender_id="trigger",
+            chat_id="auto",
+            content="Review PR #4502",
+            metadata={
+                LOCAL_TRIGGER_META: {
+                    "trigger_id": "trg_123",
+                    "trigger_name": "PR review",
+                    "delivery_id": "tdel_456",
+                    "created_at_ms": 1_700_000_000_000,
+                    "persist_content": "Local trigger received: PR review\n\nReview PR #4502",
+                }
+            },
+        ),
+        session,
+    )
+
+    assert persisted is True
+    message = session.messages[-1]
+    assert message["content"] == "Local trigger received: PR review\n\nReview PR #4502"
+    assert message[AUTOMATION_HISTORY_META] == {
+        "kind": "local_trigger",
+        "trigger_id": "trg_123",
+        "trigger_name": "PR review",
+        "trigger_delivery_id": "tdel_456",
+    }
+    assert LOCAL_TRIGGER_META not in message
+    assert message["trigger_id"] == "trg_123"
+    assert message["trigger_name"] == "PR review"
+    assert message["trigger_delivery_id"] == "tdel_456"
+
+
+@pytest.mark.asyncio
+async def test_new_with_bot_suffix_does_not_persist_command(tmp_path: Path) -> None:
+    loop = _make_full_loop(tmp_path)
+
+    response = await loop._process_message(
+        InboundMessage(
+            channel="websocket",
+            sender_id="user",
+            chat_id="chat-1",
+            content="/new@nanobot_bot",
+        )
+    )
+
+    assert response is not None
+    assert response.content == "New session started."
+    session = loop.sessions.get_or_create("websocket:chat-1")
+    assert session.messages == []
+
+
 def test_clean_generated_title_strips_reasoning_tags() -> None:
     assert clean_generated_title("<think>reasoning</think> WebUI polish") == "WebUI polish"
     assert clean_generated_title("Title: <think> The user said hello") == ""

tests/agent/test_runner_injections.py

@@ -730,6 +730,56 @@ async def test_cron_turn_deferred_while_session_active(tmp_path):
     assert loop.pending_cron_job_ids_for_session(session_key) == set()
 
 
+@pytest.mark.asyncio
+async def test_local_trigger_turn_deferred_while_session_active(tmp_path):
+    """Local trigger turns wait for the active session instead of becoming injections."""
+    from nanobot.bus.events import InboundMessage
+    from nanobot.triggers.local_session_turns import LOCAL_TRIGGER_META
+
+    loop = _make_loop(tmp_path)
+    loop._dispatch = AsyncMock()  # type: ignore[method-assign]
+
+    session_key = "websocket:chat-1"
+    pending = asyncio.Queue(maxsize=20)
+    loop._pending_queues[session_key] = pending
+
+    run_task = asyncio.create_task(loop.run())
+    msg = InboundMessage(
+        channel="websocket",
+        sender_id="trigger",
+        chat_id="chat-1",
+        content="review failed CI",
+        metadata={
+            LOCAL_TRIGGER_META: {
+                "trigger_id": "trg_123",
+                "trigger_name": "CI review",
+                "delivery_id": "tdl_123",
+            },
+        },
+        session_key_override=session_key,
+    )
+    await loop.bus.publish_inbound(msg)
+
+    for _ in range(20):
+        if loop._local_trigger_turns.deferred_queues.get(session_key):
+            break
+        await asyncio.sleep(0.05)
+
+    loop.stop()
+    await asyncio.wait_for(run_task, timeout=2)
+
+    assert pending.empty()
+    assert loop._dispatch.await_count == 0
+    assert loop._local_trigger_turns.deferred_queues[session_key] == [msg]
+    assert loop.pending_local_trigger_ids_for_session(session_key) == {"trg_123"}
+
+    assert await loop._local_trigger_turns.publish_next_deferred(session_key) is True
+    queued = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=0.5)
+    assert queued is msg
+    assert session_key not in loop._local_trigger_turns.deferred_queues
+    assert loop.pending_local_trigger_ids_for_session(session_key) == set()
+
+
 @pytest.mark.asyncio
 async def test_submitted_cron_turn_reports_pending_until_completed(tmp_path):
     """Bound cron jobs remain marked pending while their session turn is in flight."""
@@ -766,6 +816,48 @@ async def test_submitted_cron_turn_reports_pending_until_completed(tmp_path):
     assert loop.pending_cron_job_ids_for_session(session_key) == set()
 
 
+@pytest.mark.asyncio
+async def test_submitted_local_trigger_turn_reports_pending_until_completed(tmp_path):
+    """Local triggers remain marked pending while their session turn is in flight."""
+    from nanobot.bus.events import InboundMessage, OutboundMessage
+    from nanobot.triggers.local_session_turns import LOCAL_TRIGGER_META
+
+    loop = _make_loop(tmp_path)
+    loop._running = True
+
+    session_key = "websocket:chat-1"
+    msg = InboundMessage(
+        channel="websocket",
+        sender_id="trigger",
+        chat_id="chat-1",
+        content="review failed CI",
+        metadata={
+            LOCAL_TRIGGER_META: {
+                "trigger_id": "trg_123",
+                "trigger_name": "CI review",
+                "delivery_id": "tdl_123",
+            },
+        },
+        session_key_override=session_key,
+    )
+
+    submit_task = asyncio.create_task(loop.submit_local_trigger_turn(msg))
+    queued = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=0.5)
+
+    assert queued is msg
+    assert loop.pending_local_trigger_ids_for_session(session_key) == {"trg_123"}
+
+    response = OutboundMessage(
+        channel="websocket",
+        chat_id="chat-1",
+        content="done",
+    )
+    loop._local_trigger_turns.complete(msg, response=response)
+
+    assert await asyncio.wait_for(submit_task, timeout=0.5) is response
+    assert loop.pending_local_trigger_ids_for_session(session_key) == set()
+
+
 @pytest.mark.asyncio
 async def test_pending_queue_preserves_overflow_for_next_injection_cycle(tmp_path):
     """Pending queue should leave overflow messages queued for later drains."""

tests/agent/test_runner_tool_execution.py

@@ -465,3 +465,83 @@ async def test_runner_blocks_repeated_external_fetches():
         if msg.get("role") == "tool" and msg.get("tool_call_id") == "call_3"
     ][0]
     assert "repeated external lookup blocked" in blocked_tool_message["content"]
+
+
+@pytest.mark.asyncio
+async def test_runner_hints_repeated_tool_results():
+    provider = MagicMock()
+    captured_final_call: list[dict] = []
+    call_count = {"n": 0}
+
+    async def chat_with_retry(*, messages, **kwargs):
+        call_count["n"] += 1
+        if call_count["n"] <= 3:
+            return LLMResponse(
+                content="reading",
+                tool_calls=[ToolCallRequest(
+                    id=f"call_{call_count['n']}",
+                    name="grep",
+                    arguments={"pattern": "TODO", "path": "nanobot"},
+                )],
+                usage={},
+            )
+        captured_final_call[:] = messages
+        return LLMResponse(content="done", tool_calls=[], usage={})
+
+    provider.chat_with_retry = chat_with_retry
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+    tools.execute = AsyncMock(return_value="file content")
+
+    result = await AgentRunner(provider).run(AgentRunSpec(
+        initial_messages=[{"role": "user", "content": "review code"}],
+        tools=tools,
+        model="test-model",
+        max_iterations=4,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+    ))
+
+    assert result.final_content == "done"
+    assert tools.execute.await_count == 3
+    hinted_tool_message = [
+        msg for msg in captured_final_call
+        if msg.get("role") == "tool" and msg.get("tool_call_id") == "call_3"
+    ][0]
+    assert "Repeated grep result" in hinted_tool_message["content"]
+
+
+@pytest.mark.asyncio
+async def test_runner_does_not_hint_different_tool_results():
+    provider = MagicMock()
+    call_count = {"n": 0}
+
+    async def chat_with_retry(*, messages, **kwargs):
+        call_count["n"] += 1
+        if call_count["n"] <= 3:
+            return LLMResponse(
+                content="reading",
+                tool_calls=[ToolCallRequest(
+                    id=f"call_{call_count['n']}",
+                    name="grep",
+                    arguments={"pattern": "TODO", "path": "nanobot"},
+                )],
+                usage={},
+            )
+        return LLMResponse(content="done", tool_calls=[], usage={})
+
+    provider.chat_with_retry = chat_with_retry
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+    tools.execute = AsyncMock(side_effect=["first result", "second result", "third result"])
+
+    result = await AgentRunner(provider).run(AgentRunSpec(
+        initial_messages=[{"role": "user", "content": "review code"}],
+        tools=tools,
+        model="test-model",
+        max_iterations=4,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+    ))
+
+    assert result.final_content == "done"
+    assert tools.execute.await_count == 3
+    assert all("Repeated grep result" not in str(msg.get("content", "")) for msg in result.messages)

tests/channels/test_discord_channel.py

@@ -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_required_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())

tests/channels/test_telegram_channel.py

@@ -1577,12 +1577,15 @@ 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")
+    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 +1609,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

tests/channels/test_websocket_channel.py

@@ -2975,3 +2975,49 @@ def test_handle_webui_thread_get_does_not_backfill_cron_internal_prompt(
     body = json.loads(resp.body.decode())
     assert [message["role"] for message in body["messages"]] == ["assistant"]
     assert [message["content"] for message in body["messages"]] == ["提醒已经到期。"]
+
+
+def test_handle_webui_thread_get_does_not_backfill_trigger_internal_prompt(
+    tmp_path,
+    monkeypatch,
+) -> None:
+    from urllib.parse import quote
+
+    from websockets.datastructures import Headers
+    from websockets.http11 import Request
+
+    from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
+    from nanobot.webui.transcript import append_transcript_object
+
+    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
+    workspace = tmp_path / "workspace"
+    sessions = SessionManager(workspace)
+    key = "websocket:c-trigger"
+    session = sessions.get_or_create(key)
+    session.add_message(
+        "user",
+        "Local trigger received: PR review",
+        **{AUTOMATION_HISTORY_META: {"kind": "local_trigger", "trigger_id": "trg_123"}},
+    )
+    session.add_message("assistant", "PR #4502 已经开始 review。")
+    sessions.save(session)
+    append_transcript_object(
+        key,
+        {"event": "message", "chat_id": "c-trigger", "text": "PR #4502 已经开始 review。"},
+    )
+
+    bus = MagicMock()
+    channel = WebSocketChannel(
+        {"enabled": True, "allowFrom": ["*"]},
+        bus,
+        gateway=_basic_handler(bus, session_manager=sessions, workspace_path=workspace),
+    )
+    channel.gateway.tokens.api_tokens["tok"] = time.monotonic() + 300.0
+    enc = quote(key, safe="")
+    req = Request(f"/api/sessions/{enc}/webui-thread", Headers([("Authorization", "Bearer tok")]))
+    resp = channel.gateway.http._handle_webui_thread_get(req, enc)
+
+    assert resp.status_code == 200
+    body = json.loads(resp.body.decode())
+    assert [message["role"] for message in body["messages"]] == ["assistant"]
+    assert [message["content"] for message in body["messages"]] == ["PR #4502 已经开始 review。"]

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.local_store import LocalTriggerStore
 from nanobot.webui.gateway_services import GatewayServices, build_gateway_services
 
 _PORT = 29900
@@ -46,7 +47,9 @@ def _make_handler(
     workspace_path: Path | None = None,
     runtime_model_name: Any | None = None,
     cron_service: CronService | None = None,
+    local_trigger_store: LocalTriggerStore | None = None,
     cron_pending_job_ids: Any | None = None,
+    local_trigger_pending_ids: Any | None = None,
 ) -> GatewayServices:
     config = WebSocketConfig.model_validate(cfg) if isinstance(cfg, dict) else cfg
     workspace = workspace_path or Path.cwd()
@@ -61,7 +64,9 @@ def _make_handler(
         runtime_surface="browser",
         runtime_capabilities_overrides=None,
         cron_service=cron_service,
+        local_trigger_store=local_trigger_store,
         cron_pending_job_ids=cron_pending_job_ids,
+        local_trigger_pending_ids=local_trigger_pending_ids,
     )
 
 
@@ -74,7 +79,9 @@ def _ch(
     port: int = _PORT,
     runtime_model_name: Any | None = None,
     cron_service: CronService | None = None,
+    local_trigger_store: LocalTriggerStore | None = None,
     cron_pending_job_ids: Any | None = None,
+    local_trigger_pending_ids: Any | None = None,
     **extra: Any,
 ) -> WebSocketChannel:
     cfg: dict[str, Any] = {
@@ -93,7 +100,9 @@ def _ch(
         workspace_path=workspace_path,
         runtime_model_name=runtime_model_name,
         cron_service=cron_service,
+        local_trigger_store=local_trigger_store,
         cron_pending_job_ids=cron_pending_job_ids,
+        local_trigger_pending_ids=local_trigger_pending_ids,
     )
     return WebSocketChannel(cfg, bus, gateway=gateway)
 
@@ -319,6 +328,54 @@ async def test_session_automations_route_ignores_unified_owner(
         await server_task
 
 
+@pytest.mark.asyncio
+async def test_session_automations_route_lists_local_triggers(
+    bus: MagicMock, tmp_path: Path
+) -> None:
+    port = _free_port()
+    base_url = f"http://127.0.0.1:{port}"
+    trigger_store = LocalTriggerStore(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"),
+        local_trigger_store=trigger_store,
+        local_trigger_pending_ids=lambda key: (
+            {trigger.id} if key == "websocket:abc" else set()
+        ),
+        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"] == "local_trigger"
+        assert job["schedule"]["kind"] == "local"
+        assert job["payload"]["kind"] == "local_trigger"
+        assert job["payload"]["command"] == f'nanobot trigger {trigger.id} "message"'
+        assert job["state"]["pending"] is True
+    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 +1137,93 @@ 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_local_triggers(
+    bus: MagicMock, tmp_path: Path
+) -> None:
+    port = _free_port()
+    base_url = f"http://127.0.0.1:{port}"
+    trigger_store = LocalTriggerStore(tmp_path)
+    trigger = trigger_store.create(
+        name="PR review",
+        channel="websocket",
+        chat_id="abc",
+        session_key="websocket:abc",
+    )
+    delivery = trigger_store.enqueue(trigger.id, "Review queued PR")
+    assert delivery.path is not None
+    channel = _ch(
+        bus,
+        session_manager=_seed_session(tmp_path, key="websocket:abc"),
+        local_trigger_store=trigger_store,
+        local_trigger_pending_ids=lambda key: (
+            {trigger.id} if key == "websocket:abc" else set()
+        ),
+        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"] == "local_trigger"
+        assert by_id[trigger.id]["state"]["pending"] is True
+        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
+        assert not delivery.path.exists()
+    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 +1265,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_local_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 = LocalTriggerStore(tmp_path)
+    trigger = trigger_store.create(
+        name="PR review",
+        channel="websocket",
+        chat_id="doomed",
+        session_key="websocket:doomed",
+    )
+    channel = _ch(
+        bus,
+        session_manager=sm,
+        local_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

tests/cli/test_commands.py

@@ -1986,6 +1986,122 @@ def test_gateway_bound_cron_runs_as_session_turn(
     assert msg.metadata["thread_id"] == "om_root123"
 
 
+def test_gateway_local_trigger_queue_submits_agent_turns(
+    monkeypatch,
+    tmp_path: Path,
+) -> None:
+    config = Config()
+    config.agents.defaults.workspace = str(tmp_path / "config-workspace")
+    config.agents.defaults.dream.enabled = False
+    config.gateway.heartbeat.enabled = False
+    bus = MagicMock()
+    seen: dict[str, object] = {}
+
+    _patch_cli_command_runtime(
+        monkeypatch,
+        config,
+        message_bus=lambda: bus,
+        session_manager=lambda _workspace: _FakeSessionManager(),
+        cron_service=lambda _store_path: _FakeCronService(),
+    )
+
+    class _FakeMemory:
+        def get_latest_cursor(self) -> int:
+            return 0
+
+        def get_last_dream_cursor(self) -> int:
+            return 0
+
+        def set_last_dream_cursor(self, _cursor: int) -> None:
+            return None
+
+    class _FakeContext:
+        memory = _FakeMemory()
+
+    class _FakeSessionManager:
+        def flush_all(self) -> int:
+            return 0
+
+        def list_sessions(self) -> list[dict[str, object]]:
+            return []
+
+    class _FakeCronService:
+        def __init__(self) -> None:
+            self.on_job = None
+
+        async def start(self) -> None:
+            return None
+
+        def stop(self) -> None:
+            return None
+
+        def status(self) -> dict[str, int]:
+            return {"jobs": 0}
+
+        def register_system_job(self, _job) -> None:
+            return None
+
+    class _FakeAgentLoop:
+        @classmethod
+        def from_config(cls, config, bus=None, **extra):
+            seen["agent_from_config_kwargs"] = extra
+            return cls(**extra)
+
+        def __init__(self, *args, **kwargs) -> None:
+            self.model = "test-model"
+            self.provider = _fake_provider()
+            self.tools = {}
+            self.context = _FakeContext()
+            self.sessions = kwargs["session_manager"]
+            self.submit_local_trigger_turn = AsyncMock()
+            seen["agent"] = self
+
+        def _schedule_background(self, _coro) -> None:
+            return None
+
+        async def run(self) -> None:
+            await asyncio.Event().wait()
+
+        async def close_mcp(self) -> None:
+            return None
+
+        def stop(self) -> None:
+            return None
+
+    class _FakeChannelManager:
+        enabled_channels: list[str] = []
+
+        def __init__(self, *_args, **_kwargs) -> None:
+            return None
+
+        async def start_all(self) -> None:
+            await asyncio.Event().wait()
+
+        async def stop_all(self) -> None:
+            return None
+
+    async def _fake_run_local_trigger_queue(**kwargs):
+        seen["local_trigger_queue_kwargs"] = kwargs
+        raise _StopGatewayError("stop")
+
+    monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
+    monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _FakeChannelManager)
+    monkeypatch.setattr(
+        "nanobot.triggers.local_runner.run_local_trigger_queue",
+        _fake_run_local_trigger_queue,
+    )
+
+    cli_commands._run_gateway(config, health_server_enabled=False)
+
+    agent = seen["agent"]
+    agent_kwargs = seen["agent_from_config_kwargs"]
+    kwargs = seen["local_trigger_queue_kwargs"]
+    assert "local_trigger_store" in agent_kwargs
+    assert kwargs["store"] is agent_kwargs["local_trigger_store"]
+    assert "bus" not in kwargs
+    assert kwargs["submit_turn"] is agent.submit_local_trigger_turn
+
+
 def test_gateway_workspace_override_does_not_migrate_legacy_cron(
     monkeypatch, tmp_path: Path
 ) -> None:
@@ -2516,6 +2632,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.local_store import LocalTriggerStore
+
+    config_file = _write_instance_config(tmp_path)
+    config = Config()
+    config.agents.defaults.workspace = str(tmp_path / "workspace")
+    _patch_cli_command_runtime(monkeypatch, config)
+
+    store = LocalTriggerStore(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()

tests/command/test_trigger_command.py

@@ -0,0 +1,109 @@
+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.session.keys import UNIFIED_SESSION_KEY
+from nanobot.triggers.local_store import LocalTriggerStore
+
+
+@pytest.mark.asyncio
+async def test_trigger_command_creates_session_bound_local_trigger(tmp_path: Path) -> None:
+    router = CommandRouter()
+    register_builtin_commands(router)
+    store = LocalTriggerStore(tmp_path)
+    loop = SimpleNamespace(workspace=tmp_path, local_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
+
+
+@pytest.mark.asyncio
+async def test_trigger_command_binds_inbound_session_when_unified_session_is_active(
+    tmp_path: Path,
+) -> None:
+    router = CommandRouter()
+    register_builtin_commands(router)
+    store = LocalTriggerStore(tmp_path)
+    loop = SimpleNamespace(workspace=tmp_path, local_trigger_store=store)
+    msg = InboundMessage(
+        channel="websocket",
+        sender_id="user",
+        chat_id="chat-1",
+        content="/trigger PR review",
+        session_key_override="websocket:chat-1:thread-a",
+    )
+    ctx = CommandContext(
+        msg=msg,
+        session=None,
+        key=UNIFIED_SESSION_KEY,
+        raw="/trigger PR review",
+        loop=loop,
+    )
+
+    response = await router.dispatch(ctx)
+
+    assert response is not None
+    trigger = store.list_for_session("websocket:chat-1:thread-a")[0]
+    assert trigger.session_key == "websocket:chat-1:thread-a"
+    assert store.list_for_session(UNIFIED_SESSION_KEY) == []
+
+
+@pytest.mark.asyncio
+async def test_trigger_command_without_name_returns_usage_only(tmp_path: Path) -> None:
+    router = CommandRouter()
+    register_builtin_commands(router)
+    store = LocalTriggerStore(tmp_path)
+    loop = SimpleNamespace(workspace=tmp_path, local_trigger_store=store)
+    msg = InboundMessage(
+        channel="websocket",
+        sender_id="user",
+        chat_id="chat-1",
+        content="/trigger@nanobot_bot",
+        metadata={"webui": True},
+    )
+    ctx = CommandContext(
+        msg=msg,
+        session=None,
+        key="websocket:chat-1",
+        raw="/trigger@nanobot_bot",
+        loop=loop,
+    )
+
+    response = await router.dispatch(ctx)
+
+    assert response is not None
+    assert "Usage: /trigger <name>" in response.content
+    assert store.list_for_session("websocket:chat-1") == []
+
+
+def test_trigger_command_is_in_help_text() -> None:
+    assert "/trigger <name>" in build_help_text()

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] = []

tests/tools/test_exec_session_tools.py

@@ -11,7 +11,7 @@ from nanobot.agent.tools.exec_session import (
     ListExecSessionsTool,
     WriteStdinTool,
 )
-from nanobot.agent.tools.registry import ToolRegistry, is_tool_error_result
+from nanobot.agent.tools.registry import is_tool_error_result
 from nanobot.agent.tools.shell import ExecTool
 
 
@@ -53,38 +53,6 @@ def test_exec_accepts_command_aliases(tmp_path):
     assert "Exit code: 0" in result
 
 
-def test_exec_schema_hides_compatibility_aliases():
-    props = ExecTool().parameters["properties"]
-
-    assert "command" in props
-    assert "working_dir" in props
-    assert "max_output_chars" in props
-    assert "cmd" not in props
-    assert "workdir" not in props
-    assert "max_output_tokens" not in props
-
-
-def test_exec_registry_accepts_hidden_compatibility_aliases(tmp_path):
-    async def run() -> str:
-        registry = ToolRegistry()
-        registry.register(ExecTool(working_dir="/", timeout=5))
-        command = _python_command("import os; print(os.getcwd()); print('A' * 2000)")
-        return await registry.execute(
-            "exec",
-            {
-                "cmd": command,
-                "workdir": str(tmp_path),
-                "max_output_tokens": 1000,
-            },
-        )
-
-    result = asyncio.run(run())
-
-    assert str(tmp_path) in result
-    assert "chars truncated" in result
-    assert "Exit code: 0" in result
-
-
 def test_exec_returns_completed_session_output_when_yield_time_ms_is_used(tmp_path):
     async def run() -> str:
         manager = ExecSessionManager()

tests/triggers/test_local_triggers.py

@@ -0,0 +1,492 @@
+from __future__ import annotations
+
+import asyncio
+import errno
+import json
+import os
+from contextlib import suppress
+from pathlib import Path
+
+import pytest
+
+from nanobot.agent.automation_turns import AutomationTurnError
+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
+
+
+def _write_delivery_file(path: Path, *, trigger_id: str, delivery_id: str) -> None:
+    path.write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "delivery": {
+                    "id": delivery_id,
+                    "triggerId": trigger_id,
+                    "content": "queued",
+                    "createdAtMs": 1,
+                    "attempts": 0,
+                    "lastError": None,
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+
+
+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)
+
+    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_trigger_store_atomic_writes_ignore_unsupported_directory_fsync(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Shared folders may allow opening directories but reject directory fsync."""
+    store = LocalTriggerStore(tmp_path)
+    real_open = os.open
+    real_close = os.close
+    real_fsync = os.fsync
+    directory_fds: set[int] = set()
+
+    def fake_open(path: str, flags: int, *args: object, **kwargs: object) -> int:
+        fd = real_open(path, flags, *args, **kwargs)
+        if Path(path).name in {"triggers", "runs"}:
+            directory_fds.add(fd)
+        return fd
+
+    def fake_fsync(fd: int) -> None:
+        if fd in directory_fds:
+            raise OSError(errno.EINVAL, "Invalid argument")
+        real_fsync(fd)
+
+    def fake_close(fd: int) -> None:
+        directory_fds.discard(fd)
+        real_close(fd)
+
+    monkeypatch.setattr(os, "open", fake_open)
+    monkeypatch.setattr(os, "close", fake_close)
+    monkeypatch.setattr(os, "fsync", fake_fsync)
+
+    trigger = store.create(
+        name="Shared folder safe",
+        channel="websocket",
+        chat_id="chat-1",
+        session_key="websocket:chat-1",
+    )
+
+    assert store.get(trigger.id) is not None
+    delivery = store.enqueue(trigger.id, "queued from shared folder")
+    record = _read_run_record(store, delivery.id)
+    assert record["content"] == "queued from shared folder"
+
+
+def test_enqueue_rejects_disabled_trigger(tmp_path: Path) -> None:
+    store = LocalTriggerStore(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")
+
+
+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_delivery_run_record_truncates_large_content_and_response(tmp_path: Path) -> None:
+    store = LocalTriggerStore(tmp_path)
+    trigger = store.create(
+        name="Large audit",
+        channel="websocket",
+        chat_id="chat-1",
+        session_key="websocket:chat-1",
+    )
+    large_content = "content-" * 1000
+    large_response = "response-" * 1000
+
+    delivery = store.enqueue(trigger.id, large_content)
+    queued_record = _read_run_record(store, delivery.id)
+    assert queued_record["content"].startswith("content-")
+    assert queued_record["content"].endswith("\n... (truncated)")
+    assert len(queued_record["content"]) < len(large_content)
+
+    store.write_delivery_run_record(
+        delivery,
+        trigger=trigger,
+        status="ok",
+        response=large_response,
+    )
+
+    final_record = _read_run_record(store, delivery.id)
+    assert final_record["content"].endswith("\n... (truncated)")
+    assert final_record["response"].startswith("response-")
+    assert final_record["response"].endswith("\n... (truncated)")
+    assert len(final_record["response"]) < len(large_response)
+
+
+def test_delete_removes_delivery_files_for_trigger(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",
+    )
+    other = store.create(
+        name="CI summary",
+        channel="websocket",
+        chat_id="chat-2",
+        session_key="websocket:chat-2",
+    )
+    inbox = store.inbox_dir / "1-tdl_inbox.json"
+    processing = store.processing_dir / "2-tdl_processing.json"
+    failed = store.failed_dir / "3-tdl_failed.json"
+    other_inbox = store.inbox_dir / "4-tdl_other.json"
+    _write_delivery_file(inbox, trigger_id=trigger.id, delivery_id="tdl_inbox")
+    _write_delivery_file(processing, trigger_id=trigger.id, delivery_id="tdl_processing")
+    _write_delivery_file(failed, trigger_id=trigger.id, delivery_id="tdl_failed")
+    _write_delivery_file(other_inbox, trigger_id=other.id, delivery_id="tdl_other")
+
+    assert store.delete(trigger.id) is True
+
+    assert store.get(trigger.id) is None
+    assert not inbox.exists()
+    assert not processing.exists()
+    assert not failed.exists()
+    assert other_inbox.exists()
+    assert store.get(other.id) is not None
+
+
+def test_recover_processing_deliveries_requeues_claimed_delivery(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",
+    )
+    store.enqueue(trigger.id, "Review PR #4591")
+
+    claimed = store.claim_deliveries()
+    assert len(claimed) == 1
+    assert claimed[0].path is not None
+    assert claimed[0].path.parent.name == "processing"
+    assert LocalTriggerStore(tmp_path).claim_deliveries() == []
+
+    restarted = LocalTriggerStore(tmp_path)
+    assert restarted.recover_processing_deliveries() == 1
+
+    reclaimed = restarted.claim_deliveries()
+    assert len(reclaimed) == 1
+    assert reclaimed[0].trigger_id == trigger.id
+    assert reclaimed[0].content == "Review PR #4591"
+    assert reclaimed[0].attempts == 1
+    assert reclaimed[0].last_error == "delivery was recovered from interrupted processing"
+
+
+@pytest.mark.asyncio
+async def test_local_trigger_queue_submits_bound_inbound_message(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, WEBUI_TURN_METADATA_KEY: "old-turn"},
+    )
+    delivery = store.enqueue(trigger.id, "Review PR #4502")
+    submitted: list[InboundMessage] = []
+
+    async def _submit_turn(msg: InboundMessage):
+        submitted.append(msg)
+        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)
+    )
+    try:
+        for _ in range(100):
+            if submitted:
+                break
+            await asyncio.sleep(0.01)
+    finally:
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+    assert len(submitted) == 1
+    msg = submitted[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": "local_trigger",
+        "label": "PR review",
+    }
+    assert msg.metadata["_local_trigger"]["trigger_id"] == trigger.id
+    assert (
+        msg.metadata["_local_trigger"]["persist_content"]
+        == "Local trigger received: PR review\n\nReview PR #4502"
+    )
+
+    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() == []
+    record = _read_run_record(store, delivery.id)
+    assert record["status"] == "ok"
+    assert record["response"] == "done"
+    assert record["trigger_id"] == trigger.id
+
+
+@pytest.mark.asyncio
+async def test_local_trigger_queue_waits_for_submitted_turn_before_ack(
+    tmp_path: Path,
+) -> None:
+    store = LocalTriggerStore(tmp_path)
+    trigger = store.create(
+        name="CI review",
+        channel="websocket",
+        chat_id="chat-1",
+        session_key="websocket:chat-1",
+    )
+    delivery = store.enqueue(trigger.id, "Review failed CI")
+    submitted: list[InboundMessage] = []
+    release = asyncio.Event()
+
+    async def _submit_turn(msg: InboundMessage):
+        submitted.append(msg)
+        await release.wait()
+        return None
+
+    task = asyncio.create_task(
+        run_local_trigger_queue(
+            store=store,
+            submit_turn=_submit_turn,
+            poll_interval_s=0.01,
+        )
+    )
+    try:
+        for _ in range(100):
+            if submitted:
+                break
+            await asyncio.sleep(0.01)
+
+        assert len(submitted) == 1
+        assert list(store.processing_dir.glob("*.json"))
+        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
+
+        release.set()
+        for _ in range(100):
+            stored = store.get(trigger.id)
+            if stored and stored.last_status == "ok":
+                break
+            await asyncio.sleep(0.01)
+
+        assert not list(store.processing_dir.glob("*.json"))
+        stored = store.get(trigger.id)
+        assert stored is not None
+        assert stored.last_status == "ok"
+        assert store.claim_deliveries() == []
+        record = _read_run_record(store, delivery.id)
+        assert record["status"] == "ok"
+    finally:
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+
+@pytest.mark.asyncio
+async def test_local_trigger_queue_requeues_when_submitted_turn_is_interrupted(
+    tmp_path: Path,
+) -> None:
+    store = LocalTriggerStore(tmp_path)
+    trigger = store.create(
+        name="CI review",
+        channel="websocket",
+        chat_id="chat-1",
+        session_key="websocket:chat-1",
+    )
+    delivery = store.enqueue(trigger.id, "Review failed CI")
+    started = asyncio.Event()
+
+    async def _submit_turn(_msg: InboundMessage):
+        started.set()
+        await asyncio.Future()
+
+    task = asyncio.create_task(
+        run_local_trigger_queue(
+            store=store,
+            submit_turn=_submit_turn,
+            poll_interval_s=0.01,
+        )
+    )
+    try:
+        await asyncio.wait_for(started.wait(), timeout=1)
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+        reclaimed = store.claim_deliveries()
+        assert len(reclaimed) == 1
+        assert reclaimed[0].trigger_id == trigger.id
+        assert reclaimed[0].attempts == 1
+        assert reclaimed[0].last_error == "CancelledError"
+        record = _read_run_record(store, delivery.id)
+        assert record["status"] == "interrupted"
+        assert record["attempts"] == 1
+    finally:
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+
+@pytest.mark.asyncio
+async def test_local_trigger_queue_does_not_retry_completed_agent_failure(
+    tmp_path: Path,
+) -> None:
+    store = LocalTriggerStore(tmp_path)
+    trigger = store.create(
+        name="CI review",
+        channel="websocket",
+        chat_id="chat-1",
+        session_key="websocket:chat-1",
+    )
+    delivery = store.enqueue(trigger.id, "Review failed CI")
+    started = asyncio.Event()
+
+    async def _submit_turn(_msg: InboundMessage):
+        started.set()
+        raise AutomationTurnError("model failed")
+
+    task = asyncio.create_task(
+        run_local_trigger_queue(
+            store=store,
+            submit_turn=_submit_turn,
+            poll_interval_s=0.01,
+        )
+    )
+    try:
+        await asyncio.wait_for(started.wait(), timeout=1)
+        for _ in range(100):
+            stored = store.get(trigger.id)
+            if stored and stored.last_status == "error":
+                break
+            await asyncio.sleep(0.01)
+
+        stored = store.get(trigger.id)
+        assert stored is not None
+        assert stored.last_status == "error"
+        assert stored.last_error == "model failed"
+        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):
+            await task
+
+
+@pytest.mark.asyncio
+async def test_local_trigger_queue_recovers_processing_delivery_on_start(
+    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",
+    )
+    store.enqueue(trigger.id, "Review PR #4591")
+    assert len(store.claim_deliveries()) == 1
+    submitted: list[InboundMessage] = []
+
+    async def _submit_turn(msg: InboundMessage):
+        submitted.append(msg)
+        return None
+
+    restarted = LocalTriggerStore(tmp_path)
+    task = asyncio.create_task(
+        run_local_trigger_queue(store=restarted, submit_turn=_submit_turn, poll_interval_s=0.01)
+    )
+    try:
+        for _ in range(100):
+            if submitted:
+                break
+            await asyncio.sleep(0.01)
+    finally:
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+    assert len(submitted) == 1
+    assert submitted[0].content == "Review PR #4591"
+    assert submitted[0].metadata["_local_trigger"]["trigger_id"] == trigger.id
+    assert restarted.claim_deliveries() == []

tests/utils/test_repeated_tool_result_hint.py

@@ -0,0 +1,60 @@
+"""Tests for repeated tool-result hints."""
+
+from __future__ import annotations
+
+from nanobot.utils.runtime import (
+    repeated_external_lookup_error,
+    repeated_tool_result_hint,
+)
+
+
+def test_repeated_tool_result_hints_after_two_identical_results():
+    counts: dict[str, int] = {}
+
+    assert repeated_tool_result_hint("grep", "same result", counts) is None
+    assert repeated_tool_result_hint("grep", "same result", counts) is None
+    third = repeated_tool_result_hint("grep", "same result", counts)
+
+    assert third is not None
+    assert "Repeated grep result" in third
+
+
+def test_repeated_tool_result_ignores_different_results():
+    counts: dict[str, int] = {}
+
+    assert repeated_tool_result_hint("grep", "first", counts) is None
+    assert repeated_tool_result_hint("grep", "second", counts) is None
+    assert repeated_tool_result_hint("grep", "third", counts) is None
+
+
+def test_repeated_tool_result_is_per_tool():
+    counts: dict[str, int] = {}
+
+    repeated_tool_result_hint("grep", "same", counts)
+    repeated_tool_result_hint("grep", "same", counts)
+
+    assert repeated_tool_result_hint("read_file", "same", counts) is None
+
+
+def test_repeated_tool_result_handles_text_blocks():
+    counts: dict[str, int] = {}
+    result = [{"type": "text", "text": "same result"}]
+
+    repeated_tool_result_hint("mcp", result, counts)
+    repeated_tool_result_hint("mcp", result, counts)
+    third = repeated_tool_result_hint("mcp", result, counts)
+
+    assert third is not None
+    assert "Repeated mcp result" in third
+
+
+def test_repeated_external_lookup_still_blocks_after_two_attempts():
+    counts: dict[str, int] = {}
+    arguments = {"url": "https://example.com"}
+
+    repeated_external_lookup_error("web_fetch", arguments, counts)
+    repeated_external_lookup_error("web_fetch", arguments, counts)
+    third = repeated_external_lookup_error("web_fetch", arguments, counts)
+
+    assert third is not None
+    assert "repeated external lookup blocked" in third

tests/utils/test_webui_transcript.py

@@ -473,6 +473,42 @@ def test_replay_reused_turn_id_after_turn_end_starts_new_turn(tmp_path, monkeypa
     assert msgs[2]["source"] == {"kind": "cron", "label": "drink water"}
 
 
+def test_replay_preserves_local_trigger_source_metadata(tmp_path, monkeypatch) -> None:
+    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
+    key = "websocket:t-local-trigger-source"
+    append_transcript_object(
+        key,
+        {
+            "event": "message",
+            "chat_id": "t-local-trigger-source",
+            "text": "PR #4502 review started.",
+            "source": {"kind": "local_trigger", "label": "PR review"},
+        },
+    )
+
+    msgs = replay_transcript_to_ui_messages(read_transcript_lines(key))
+
+    assert msgs[0]["source"] == {"kind": "local_trigger", "label": "PR review"}
+
+
+def test_replay_preserves_legacy_trigger_source_metadata(tmp_path, monkeypatch) -> None:
+    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
+    key = "websocket:t-trigger-source"
+    append_transcript_object(
+        key,
+        {
+            "event": "message",
+            "chat_id": "t-trigger-source",
+            "text": "PR #4502 review started.",
+            "source": {"kind": "trigger", "label": "PR review"},
+        },
+    )
+
+    msgs = replay_transcript_to_ui_messages(read_transcript_lines(key))
+
+    assert msgs[0]["source"] == {"kind": "trigger", "label": "PR review"}
+
+
 def test_build_response_restores_session_users_for_legacy_transcript(
     tmp_path,
     monkeypatch,

tests/webui/test_session_list_index.py

@@ -6,6 +6,7 @@ from pathlib import Path
 
 import nanobot.webui.session_list_index as session_list_index
 from nanobot.cron.session_turns import CRON_HISTORY_META
+from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
 from nanobot.session.manager import SessionManager
 
 
@@ -88,6 +89,20 @@ def test_webui_session_list_skips_cron_internal_user_preview(tmp_path: Path) ->
     assert list_webui_sessions(manager)[0]["preview"] == "提醒已经到期。"
 
 
+def test_webui_session_list_skips_trigger_internal_user_preview(tmp_path: Path) -> None:
+    manager = SessionManager(tmp_path)
+    session = manager.get_or_create("websocket:trigger-preview")
+    session.add_message(
+        "user",
+        "Local trigger received: PR review",
+        **{AUTOMATION_HISTORY_META: {"kind": "local_trigger", "trigger_id": "trg_123"}},
+    )
+    session.add_message("assistant", "PR #4502 已经开始 review。")
+    manager.save(session)
+
+    assert list_webui_sessions(manager)[0]["preview"] == "PR #4502 已经开始 review。"
+
+
 def test_webui_session_list_uses_webui_transcript_activity_for_sort(
     tmp_path: Path,
     monkeypatch,

webui/src/components/DeleteConfirm.tsx

@@ -122,6 +122,9 @@ function formatAutomationSchedule(
         })
       : t("deleteConfirm.schedule.cron", { expr: job.schedule.expr });
   }
+  if (job.schedule.kind === "local" || job.payload.kind === "local_trigger") {
+    return t("deleteConfirm.schedule.local", { defaultValue: "Local 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 === "local" || job.payload.kind === "local_trigger") {
+    return t("deleteConfirm.next.local", { 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) });

webui/src/components/MessageBubble.tsx

@@ -167,8 +167,14 @@ 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"
-    ? (message.source.label?.trim() || t("message.automationSourceFallback"))
+  const automationSourceKind = message.source?.kind;
+  const automationSourceName = message.source?.label?.trim();
+  const automationSourceLabel = (
+    automationSourceKind === "cron"
+    || automationSourceKind === "local_trigger"
+    || automationSourceKind === "trigger"
+  )
+    ? (automationSourceName || t("message.automationSourceFallback"))
     : "";
   const automationTriggeredLabel = t("message.automationTriggered");
 

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 (
     <div role="listitem">
@@ -3696,7 +3699,7 @@ function AutomationListItem({
             </span>
           </span>
           <span className="mt-1.5 line-clamp-2 text-[12px] leading-5 text-muted-foreground">
-            {job.payload.message || tx("settings.automations.systemTask", "System-managed automation")}
+            {summary}
           </span>
           <span className="mt-2.5 flex min-w-0 items-center gap-2 text-[11.5px] leading-none text-muted-foreground">
             <span className="truncate" title={formatAutomationNextTitle(job, locale, tx)}>
@@ -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 localTrigger = isLocalTriggerAutomation(job);
+  const triggerCommand = automationTriggerCommand(job);
+  const message = automationDetailText(job, tx);
+  const messageLabel = localTrigger
+    ? 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({
       <div className="grid min-h-0 min-w-0 flex-1 overflow-hidden lg:grid-cols-[minmax(0,1fr)_14.5rem]">
         <div className="min-h-0 min-w-0 space-y-3 overflow-y-auto overscroll-contain p-4 sm:p-5">
           <section className="rounded-[20px] border border-border/35 bg-background/62 px-4 py-3.5 shadow-[inset_0_1px_0_rgba(255,255,255,0.58)] dark:border-white/10 dark:bg-background/24">
+            <div className="flex items-center justify-between gap-3">
               <div className="text-[11px] font-medium leading-none text-muted-foreground/75">
-              {tx("settings.automations.fields.message", "Message")}
+                {messageLabel}
+              </div>
+              {localTrigger && triggerCommand ? (
+                <Button
+                  type="button"
+                  variant="ghost"
+                  size="sm"
+                  className="h-7 shrink-0 rounded-full px-2 text-[11.5px]"
+                  onClick={() => {
+                    void copyTextToClipboard(triggerCommand).then((ok) => {
+                      if (ok) setCommandCopied(true);
+                    });
+                  }}
+                >
+                  {commandCopied ? (
+                    <Check className="mr-1.5 h-3.5 w-3.5" aria-hidden />
+                  ) : (
+                    <Clipboard className="mr-1.5 h-3.5 w-3.5" aria-hidden />
+                  )}
+                  {commandCopied
+                    ? tx("settings.automations.commandCopied", "Copied")
+                    : tx("settings.automations.copyCommand", "Copy")}
+                </Button>
+              ) : null}
             </div>
             <div
               className={cn(
                 "mt-3 whitespace-pre-wrap break-words text-[13px] leading-6 text-foreground/85",
+                localTrigger && "font-mono text-[12.5px]",
                 !messageExpanded && messageNeedsExpansion && "line-clamp-6",
               )}
             >
@@ -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 localTrigger = isLocalTriggerAutomation(job);
+  const canRun = canManage && hasLinkedChat && job.enabled && !job.state.pending && !localTrigger;
   const toggleAction: AutomationAction = job.enabled ? "disable" : "enable";
   const canToggle = canManage && (job.enabled || hasLinkedChat);
   const toggleBusy = actionKey === `${toggleAction}:${job.id}`;
@@ -3927,6 +3963,7 @@ function AutomationActionGroup({
       >
         <Pencil className="h-4 w-4" aria-hidden />
       </AppsActionButton>
+      {!localTrigger ? (
         <AppsActionButton
           ariaLabel={tx("settings.automations.runNow", "Run now")}
           busy={actionKey === `run:${job.id}`}
@@ -3935,6 +3972,7 @@ function AutomationActionGroup({
         >
           <PlayCircle className="h-4 w-4" aria-hidden />
         </AppsActionButton>
+      ) : null}
       <AppsActionButton
         ariaLabel={
           job.enabled
@@ -4057,6 +4095,7 @@ function AutomationEditDialog({
   const tx = (key: string, fallback: string, values?: Record<string, unknown>) =>
     t(key, { defaultValue: fallback, ...(values ?? {}) });
   const [draft, setDraft] = useState<AutomationEditDraft>(() => automationDraftFromJob(null));
+  const localTrigger = isLocalTriggerAutomation(job);
 
   useEffect(() => {
     setDraft(automationDraftFromJob(job));
@@ -4106,6 +4145,7 @@ function AutomationEditDialog({
                 />
               </label>
 
+              {!localTrigger ? (
                 <label className="block space-y-1.5">
                   <span className="text-[12px] font-medium text-muted-foreground">
                     {tx("settings.automations.fields.message", "Message")}
@@ -4116,7 +4156,9 @@ function AutomationEditDialog({
                     className="min-h-[160px] resize-none rounded-[12px] text-[13px] leading-5"
                   />
                 </label>
+              ) : null}
 
+              {!localTrigger ? (
                 <div className="space-y-2">
                   <span className="text-[12px] font-medium text-muted-foreground">
                     {tx("settings.automations.fields.scheduleType", "Schedule type")}
@@ -4132,8 +4174,9 @@ function AutomationEditDialog({
                     }
                   />
                 </div>
+              ) : null}
 
-              {draft.scheduleKind === "every" ? (
+              {!localTrigger && draft.scheduleKind === "every" ? (
                 <div className="grid gap-2 sm:grid-cols-[minmax(0,1fr)_10rem]">
                   <label className="block space-y-1.5">
                     <span className="text-[12px] font-medium text-muted-foreground">
@@ -4174,7 +4217,7 @@ function AutomationEditDialog({
                 </div>
               ) : null}
 
-              {draft.scheduleKind === "cron" ? (
+              {!localTrigger && draft.scheduleKind === "cron" ? (
                 <div className="grid gap-2 sm:grid-cols-[minmax(0,1fr)_12rem]">
                   <label className="block space-y-1.5">
                     <span className="text-[12px] font-medium text-muted-foreground">
@@ -4201,7 +4244,7 @@ function AutomationEditDialog({
                 </div>
               ) : null}
 
-              {draft.scheduleKind === "at" ? (
+              {!localTrigger && draft.scheduleKind === "at" ? (
                 <label className="block space-y-1.5">
                   <span className="text-[12px] font-medium text-muted-foreground">
                     {tx("settings.automations.fields.runAt", "Run at")}
@@ -4266,7 +4309,7 @@ function AutomationDeleteDialog({
           <DialogDescription>
             {tx(
               "settings.automations.deleteDescription",
-              "This removes {{name}} from the cron store. Past chat messages stay in the session.",
+              "This removes {{name}} from automations. Past chat messages stay in the session.",
               { name: job?.name || job?.id || "" },
             )}
           </DialogDescription>
@@ -4297,6 +4340,34 @@ function AutomationDeleteDialog({
   );
 }
 
+function isLocalTriggerAutomation(job: SessionAutomationJob | null): boolean {
+  if (!job) return false;
+  return job.kind === "local_trigger"
+    || job.payload.kind === "local_trigger"
+    || job.schedule.kind === "local";
+}
+
+function automationTriggerCommand(job: SessionAutomationJob): string {
+  return job.trigger?.command || job.payload.command || job.payload.message || "";
+}
+
+function automationSummary(
+  job: SessionAutomationJob,
+  tx: (key: string, fallback: string, values?: Record<string, unknown>) => string,
+): string {
+  if (isLocalTriggerAutomation(job)) {
+    return automationTriggerCommand(job) || tx("settings.automations.localTrigger", "Local trigger");
+  }
+  return job.payload.message || tx("settings.automations.systemTask", "System-managed automation");
+}
+
+function automationDetailText(
+  job: SessionAutomationJob,
+  tx: (key: string, fallback: string, values?: Record<string, unknown>) => string,
+): string {
+  return automationSummary(job, tx);
+}
+
 function automationNeedsAttention(job: SessionAutomationJob): boolean {
   return job.state.last_status === "error";
 }
@@ -4308,6 +4379,7 @@ function automationStatusKey(
   if (job.state.pending) return "running";
   if (!job.enabled) return "paused";
   if (job.state.last_status === "error") return "failed";
+  if (isLocalTriggerAutomation(job)) return "active";
   if (job.delete_after_run && !job.state.next_run_at_ms && job.state.last_status === "ok") {
     return "completed";
   }
@@ -4371,6 +4443,7 @@ function automationEditDraftError(
   tx: (key: string, fallback: string, values?: Record<string, unknown>) => string,
 ): string | null {
   if (!draft.name.trim()) return tx("settings.automations.validation.nameRequired", "Name is required.");
+  if (isLocalTriggerAutomation(job)) return null;
   if (!draft.message.trim()) {
     return tx("settings.automations.validation.messageRequired", "Message is required.");
   }
@@ -4400,6 +4473,10 @@ function automationUpdatePayloadFromDraft(
   job: SessionAutomationJob | null,
 ): AutomationUpdatePayload | string {
   const name = draft.name.trim();
+  if (isLocalTriggerAutomation(job)) {
+    if (!name) return "invalid";
+    return { name };
+  }
   const message = draft.message.trim();
   if (!name || !message) return "invalid";
   const payload: AutomationUpdatePayload = { name, message };
@@ -4517,7 +4594,7 @@ function automationSearchParts(
   const scheduleParts = automationScheduleSearchParts(job);
   if (field === "id") return [job.id];
   if (field === "name") return [job.name, job.id];
-  if (field === "message") return [job.payload.message];
+  if (field === "message") return [job.payload.message, job.payload.command, job.trigger?.command];
   if (field === "chat") return originParts;
   if (field === "cron" || field === "schedule") return scheduleParts;
   if (field === "status") return [automationStatusKey(job), job.enabled ? "enabled" : "disabled"];
@@ -4525,6 +4602,9 @@ function automationSearchParts(
     job.id,
     job.name,
     job.payload.message,
+    job.payload.command,
+    job.trigger?.command,
+    isLocalTriggerAutomation(job) ? "trigger local" : null,
     ...scheduleParts,
     automationStatusKey(job),
     ...originParts,
@@ -4714,6 +4794,9 @@ function formatAutomationSchedule(
         })
       : tx("settings.automations.schedule.cron", "Cron {{expr}}", { expr: job.schedule.expr });
   }
+  if (isLocalTriggerAutomation(job)) {
+    return tx("settings.automations.schedule.local", "Local trigger");
+  }
   return tx("settings.automations.schedule.custom", "Custom schedule");
 }
 
@@ -4768,6 +4851,9 @@ function formatAutomationNext(
 ): string {
   if (!job.enabled) return tx("settings.automations.next.paused", "Paused");
   if (job.state.pending) return tx("settings.automations.next.pending", "Running now");
+  if (isLocalTriggerAutomation(job)) {
+    return tx("settings.automations.next.local", "Waiting for trigger");
+  }
   if (!job.state.next_run_at_ms) return tx("settings.automations.next.none", "No next run");
   return relativeTime(job.state.next_run_at_ms);
 }

webui/src/components/thread/SessionInfoPopover.tsx

@@ -152,6 +152,9 @@ function AutomationRow({ job, now }: { job: SessionAutomationJob; now: number })
 
 function formatSchedule(job: SessionAutomationJob, t: TFunction) {
   const locale = currentLocale();
+  if (isLocalTriggerAutomation(job)) {
+    return t("thread.sessionInfo.schedule.local");
+  }
   if (job.schedule.kind === "at" && job.schedule.at_ms) {
     return t("thread.sessionInfo.schedule.at", { time: fmtDateTime(job.schedule.at_ms, locale) });
   }
@@ -179,6 +182,9 @@ function formatNextRun(job: SessionAutomationJob, t: TFunction, now: number) {
   if (job.state.pending) {
     return { label: t("thread.sessionInfo.next.pending"), title: "" };
   }
+  if (isLocalTriggerAutomation(job)) {
+    return { label: t("thread.sessionInfo.next.local"), title: "" };
+  }
   const next = job.state.next_run_at_ms;
   if (!next) {
     return { label: t("thread.sessionInfo.next.none"), title: "" };
@@ -189,6 +195,12 @@ function formatNextRun(job: SessionAutomationJob, t: TFunction, now: number) {
   };
 }
 
+function isLocalTriggerAutomation(job: SessionAutomationJob): boolean {
+  return job.kind === "local_trigger"
+    || job.payload.kind === "local_trigger"
+    || job.schedule.kind === "local";
+}
+
 function relativeTimeFrom(value: number, now: number, locale: string): string {
   let delta = (value - now) / 1000;
   const formatter = new Intl.RelativeTimeFormat(locale, { numeric: "auto" });

webui/src/i18n/locales/en/common.json

@@ -493,6 +493,7 @@
       "emptyHint": "Create one from where it should run so nanobot keeps the right context.",
       "oneShot": "One-time",
       "systemTask": "System-managed automation",
+      "localTrigger": "Local trigger",
       "labels": {
         "schedule": "Schedule",
         "next": "Next",
@@ -551,11 +552,13 @@
         "weekdaysAt": "Weekdays at {{time}}",
         "hourlyAt": "Hourly at :{{minute}}",
         "hourlyWindow": "Hourly {{start}}-{{end}} at :{{minute}}",
+        "local": "Local trigger",
         "custom": "Custom schedule"
       },
       "next": {
         "paused": "Paused",
         "pending": "Running now",
+        "local": "Waiting for trigger",
         "none": "No next run"
       },
       "message": {
@@ -692,11 +695,13 @@
       "every": "Every {{duration}}",
       "cron": "Cron {{expr}}",
       "cronWithTz": "Cron {{expr}} · {{tz}}",
+      "local": "Local trigger",
       "unknown": "Custom schedule"
     },
     "next": {
       "label": "Next: {{time}}",
       "disabled": "Paused",
+      "local": "Waiting for trigger",
       "none": "No next run"
     }
   },
@@ -791,12 +796,14 @@
         "every": "Every {{duration}}",
         "cron": "Cron {{expr}}",
         "cronWithTz": "Cron {{expr}} · {{tz}}",
+        "local": "Local trigger",
         "unknown": "Custom schedule"
       },
       "next": {
         "label": "{{time}}",
         "pending": "Runs shortly",
         "disabled": "Paused",
+        "local": "Waiting for trigger",
         "none": "No next run"
       }
     },
@@ -918,6 +925,10 @@
             "title": "Long-running goal",
             "description": "Tell the agent to treat this as a sustained multi-step goal."
           },
+          "trigger": {
+            "title": "Create local trigger",
+            "description": "Create a CLI trigger bound to this chat session."
+          },
           "help": {
             "title": "Show help",
             "description": "List available slash commands."

webui/src/i18n/locales/es/common.json

@@ -493,6 +493,7 @@
       "emptyHint": "Créala desde donde debe ejecutarse para que nanobot conserve el contexto correcto.",
       "oneShot": "Una vez",
       "systemTask": "Automatización administrada por el sistema",
+      "localTrigger": "Activador local",
       "labels": {
         "schedule": "Programación",
         "next": "Siguiente",
@@ -551,11 +552,13 @@
         "weekdaysAt": "Días laborables a las {{time}}",
         "hourlyAt": "Cada hora en :{{minute}}",
         "hourlyWindow": "Cada hora {{start}}-{{end}} en :{{minute}}",
+        "local": "Activador local",
         "custom": "Programación personalizada"
       },
       "next": {
         "paused": "Pausada",
         "pending": "Ejecutándose ahora",
+        "local": "Esperando activador",
         "none": "Sin próxima ejecución"
       },
       "message": {
@@ -692,11 +695,13 @@
       "every": "Cada {{duration}}",
       "cron": "Cron {{expr}}",
       "cronWithTz": "Cron {{expr}} · {{tz}}",
+      "local": "Activador local",
       "unknown": "Programación personalizada"
     },
     "next": {
       "label": "Siguiente: {{time}}",
       "disabled": "Pausada",
+      "local": "Esperando activador",
       "none": "Sin próxima ejecución"
     }
   },
@@ -791,12 +796,14 @@
         "every": "Cada {{duration}}",
         "cron": "Cron {{expr}}",
         "cronWithTz": "Cron {{expr}} · {{tz}}",
+        "local": "Activador local",
         "unknown": "Programación personalizada"
       },
       "next": {
         "label": "Siguiente {{time}}",
         "pending": "Se ejecutará pronto",
         "disabled": "En pausa",
+        "local": "Esperando activador",
         "none": "Sin próxima ejecución"
       }
     },
@@ -908,6 +915,10 @@
             "title": "Objetivo a largo plazo",
             "description": "Indica al agente que trate esto como un objetivo sostenido en varios pasos."
           },
+          "trigger": {
+            "title": "Crear trigger local",
+            "description": "Crea un trigger de CLI vinculado a esta sesion de chat."
+          },
           "help": {
             "title": "Mostrar ayuda",
             "description": "Lista los comandos slash disponibles."

webui/src/i18n/locales/fr/common.json

@@ -493,6 +493,7 @@
       "emptyHint": "Créez-la depuis son point d'exécution pour que nanobot conserve le bon contexte.",
       "oneShot": "Ponctuelle",
       "systemTask": "Automatisation gérée par le système",
+      "localTrigger": "Déclencheur local",
       "labels": {
         "schedule": "Planning",
         "next": "Prochaine",
@@ -551,11 +552,13 @@
         "weekdaysAt": "Jours ouvrés à {{time}}",
         "hourlyAt": "Toutes les heures à :{{minute}}",
         "hourlyWindow": "Toutes les heures {{start}}-{{end}} à :{{minute}}",
+        "local": "Déclencheur local",
         "custom": "Planning personnalisé"
       },
       "next": {
         "paused": "En pause",
         "pending": "En cours d’exécution",
+        "local": "En attente du déclencheur",
         "none": "Aucune prochaine exécution"
       },
       "message": {
@@ -692,11 +695,13 @@
       "every": "Tous les {{duration}}",
       "cron": "Cron {{expr}}",
       "cronWithTz": "Cron {{expr}} · {{tz}}",
+      "local": "Déclencheur local",
       "unknown": "Planification personnalisée"
     },
     "next": {
       "label": "Prochaine exécution : {{time}}",
       "disabled": "En pause",
+      "local": "En attente du déclencheur",
       "none": "Aucune prochaine exécution"
     }
   },
@@ -791,12 +796,14 @@
         "every": "Toutes les {{duration}}",
         "cron": "Cron {{expr}}",
         "cronWithTz": "Cron {{expr}} · {{tz}}",
+        "local": "Déclencheur local",
         "unknown": "Planification personnalisée"
       },
       "next": {
         "label": "Prochaine {{time}}",
         "pending": "Exécution imminente",
         "disabled": "En pause",
+        "local": "En attente du déclencheur",
         "none": "Aucune prochaine exécution"
       }
     },
@@ -908,6 +915,10 @@
             "title": "Objectif long terme",
             "description": "Demandez à l’agent de traiter ceci comme un objectif multi‑étapes durable."
           },
+          "trigger": {
+            "title": "Créer un trigger local",
+            "description": "Crée un trigger CLI lié à cette session de chat."
+          },
           "help": {
             "title": "Afficher l’aide",
             "description": "Lister les commandes slash disponibles."

webui/src/i18n/locales/id/common.json

@@ -493,6 +493,7 @@
       "emptyHint": "Buat dari tempat tugas ini berjalan agar nanobot menyimpan konteks yang tepat.",
       "oneShot": "Satu kali",
       "systemTask": "Automasi yang dikelola sistem",
+      "localTrigger": "Pemicu lokal",
       "labels": {
         "schedule": "Jadwal",
         "next": "Berikutnya",
@@ -551,11 +552,13 @@
         "weekdaysAt": "Hari kerja pukul {{time}}",
         "hourlyAt": "Setiap jam pada :{{minute}}",
         "hourlyWindow": "Setiap jam {{start}}-{{end}} pada :{{minute}}",
+        "local": "Pemicu lokal",
         "custom": "Jadwal khusus"
       },
       "next": {
         "paused": "Dijeda",
         "pending": "Sedang berjalan",
+        "local": "Menunggu pemicu",
         "none": "Tidak ada jadwal berikutnya"
       },
       "message": {
@@ -692,11 +695,13 @@
       "every": "Setiap {{duration}}",
       "cron": "Cron {{expr}}",
       "cronWithTz": "Cron {{expr}} · {{tz}}",
+      "local": "Pemicu lokal",
       "unknown": "Jadwal khusus"
     },
     "next": {
       "label": "Berikutnya: {{time}}",
       "disabled": "Dijeda",
+      "local": "Menunggu pemicu",
       "none": "Tidak ada jadwal berikutnya"
     }
   },
@@ -791,12 +796,14 @@
         "every": "Setiap {{duration}}",
         "cron": "Cron {{expr}}",
         "cronWithTz": "Cron {{expr}} · {{tz}}",
+        "local": "Pemicu lokal",
         "unknown": "Jadwal khusus"
       },
       "next": {
         "label": "Berikutnya {{time}}",
         "pending": "Segera berjalan",
         "disabled": "Dijeda",
+        "local": "Menunggu pemicu",
         "none": "Tidak ada jadwal berikutnya"
       }
     },
@@ -908,6 +915,10 @@
             "title": "Tujuan jangka panjang",
             "description": "Instruksikan agen memperlakukan ini sebagai tujuan multi-langkah yang berkelanjutan."
           },
+          "trigger": {
+            "title": "Buat trigger lokal",
+            "description": "Buat trigger CLI yang terikat ke sesi chat ini."
+          },
           "help": {
             "title": "Tampilkan bantuan",
             "description": "Daftar perintah slash yang tersedia."

webui/src/i18n/locales/ja/common.json

@@ -493,6 +493,7 @@
       "emptyHint": "実行元から作成すると、nanobot が正しいコンテキストを保持できます。",
       "oneShot": "一回限り",
       "systemTask": "システム管理の自動タスク",
+      "localTrigger": "ローカルトリガー",
       "labels": {
         "schedule": "スケジュール",
         "next": "次回",
@@ -551,11 +552,13 @@
         "weekdaysAt": "平日 {{time}}",
         "hourlyAt": "毎時 :{{minute}}",
         "hourlyWindow": "{{start}}-{{end}} の毎時 :{{minute}}",
+        "local": "ローカルトリガー",
         "custom": "カスタムスケジュール"
       },
       "next": {
         "paused": "一時停止",
         "pending": "実行中",
+        "local": "トリガー待ち",
         "none": "次回実行なし"
       },
       "message": {
@@ -692,11 +695,13 @@
       "every": "{{duration}} ごと",
       "cron": "Cron {{expr}}",
       "cronWithTz": "Cron {{expr}} · {{tz}}",
+      "local": "ローカルトリガー",
       "unknown": "カスタムスケジュール"
     },
     "next": {
       "label": "次回: {{time}}",
       "disabled": "一時停止中",
+      "local": "トリガー待ち",
       "none": "次回実行なし"
     }
   },
@@ -791,12 +796,14 @@
         "every": "{{duration}}ごと",
         "cron": "Cron {{expr}}",
         "cronWithTz": "Cron {{expr}} · {{tz}}",
+        "local": "ローカルトリガー",
         "unknown": "カスタムスケジュール"
       },
       "next": {
         "label": "次回 {{time}}",
         "pending": "まもなく実行",
         "disabled": "一時停止",
+        "local": "トリガー待ち",
         "none": "次回実行なし"
       }
     },
@@ -908,6 +915,10 @@
             "title": "長期目標",
             "description": "持続的な複数ステップの目標として扱うようエージェントに伝えます。"
           },
+          "trigger": {
+            "title": "ローカルトリガーを作成",
+            "description": "このチャットセッションに紐づく CLI トリガーを作成します。"
+          },
           "help": {
             "title": "ヘルプを表示",
             "description": "利用可能なスラッシュコマンドを一覧表示します。"

webui/src/i18n/locales/ko/common.json

@@ -493,6 +493,7 @@
       "emptyHint": "실행될 위치에서 만들면 nanobot이 올바른 컨텍스트를 유지합니다.",
       "oneShot": "일회성",
       "systemTask": "시스템 관리 자동화",
+      "localTrigger": "로컬 트리거",
       "labels": {
         "schedule": "일정",
         "next": "다음",
@@ -551,11 +552,13 @@
         "weekdaysAt": "평일 {{time}}",
         "hourlyAt": "매시간 :{{minute}}",
         "hourlyWindow": "{{start}}-{{end}} 사이 매시간 :{{minute}}",
+        "local": "로컬 트리거",
         "custom": "사용자 지정 일정"
       },
       "next": {
         "paused": "일시 중지",
         "pending": "실행 중",
+        "local": "트리거 대기 중",
         "none": "다음 실행 없음"
       },
       "message": {
@@ -692,11 +695,13 @@
       "every": "{{duration}}마다",
       "cron": "Cron {{expr}}",
       "cronWithTz": "Cron {{expr}} · {{tz}}",
+      "local": "로컬 트리거",
       "unknown": "사용자 지정 일정"
     },
     "next": {
       "label": "다음: {{time}}",
       "disabled": "일시 중지됨",
+      "local": "트리거 대기 중",
       "none": "다음 실행 없음"
     }
   },
@@ -791,12 +796,14 @@
         "every": "{{duration}}마다",
         "cron": "Cron {{expr}}",
         "cronWithTz": "Cron {{expr}} · {{tz}}",
+        "local": "로컬 트리거",
         "unknown": "사용자 지정 일정"
       },
       "next": {
         "label": "다음 {{time}}",
         "pending": "곧 실행됨",
         "disabled": "일시 중지됨",
+        "local": "트리거 대기 중",
         "none": "다음 실행 없음"
       }
     },
@@ -908,6 +915,10 @@
             "title": "장기 목표",
             "description": "에이전트에게 지속적인 다단계 목표로 처리하도록 지시합니다."
           },
+          "trigger": {
+            "title": "로컬 트리거 만들기",
+            "description": "이 채팅 세션에 연결된 CLI 트리거를 만듭니다."
+          },
           "help": {
             "title": "도움말 보기",
             "description": "사용 가능한 슬래시 명령을 나열합니다."

webui/src/i18n/locales/vi/common.json

@@ -493,6 +493,7 @@
       "emptyHint": "Tạo từ nơi tác vụ sẽ chạy để nanobot giữ đúng ngữ cảnh.",
       "oneShot": "Một lần",
       "systemTask": "Tự động hóa do hệ thống quản lý",
+      "localTrigger": "Trình kích hoạt cục bộ",
       "labels": {
         "schedule": "Lịch",
         "next": "Tiếp theo",
@@ -551,11 +552,13 @@
         "weekdaysAt": "Ngày làm việc lúc {{time}}",
         "hourlyAt": "Mỗi giờ tại :{{minute}}",
         "hourlyWindow": "Mỗi giờ {{start}}-{{end}} tại :{{minute}}",
+        "local": "Trình kích hoạt cục bộ",
         "custom": "Lịch tùy chỉnh"
       },
       "next": {
         "paused": "Đã tạm dừng",
         "pending": "Đang chạy",
+        "local": "Đang chờ kích hoạt",
         "none": "Không có lần chạy tiếp theo"
       },
       "message": {
@@ -692,11 +695,13 @@
       "every": "Mỗi {{duration}}",
       "cron": "Cron {{expr}}",
       "cronWithTz": "Cron {{expr}} · {{tz}}",
+      "local": "Trình kích hoạt cục bộ",
       "unknown": "Lịch tùy chỉnh"
     },
     "next": {
       "label": "Tiếp theo: {{time}}",
       "disabled": "Đã tạm dừng",
+      "local": "Đang chờ kích hoạt",
       "none": "Không có lần chạy tiếp theo"
     }
   },
@@ -791,12 +796,14 @@
         "every": "Mỗi {{duration}}",
         "cron": "Cron {{expr}}",
         "cronWithTz": "Cron {{expr}} · {{tz}}",
+        "local": "Trình kích hoạt cục bộ",
         "unknown": "Lịch tùy chỉnh"
       },
       "next": {
         "label": "Tiếp theo {{time}}",
         "pending": "Sắp chạy",
         "disabled": "Đã tạm dừng",
+        "local": "Đang chờ kích hoạt",
         "none": "Không có lần chạy tiếp theo"
       }
     },
@@ -908,6 +915,10 @@
             "title": "Mục tiêu dài hạn",
             "description": "Yêu cầu agent xử lý đây là mục tiêu nhiều bước kéo dài."
           },
+          "trigger": {
+            "title": "Tạo trigger cục bộ",
+            "description": "Tạo trigger CLI gắn với phiên chat này."
+          },
           "help": {
             "title": "Hiển thị trợ giúp",
             "description": "Liệt kê các lệnh slash có sẵn."

webui/src/i18n/locales/zh-CN/common.json

@@ -493,6 +493,7 @@
       "emptyHint": "请从它应该运行的来源处创建，这样 nanobot 才能保留正确上下文。",
       "oneShot": "一次性",
       "systemTask": "系统管理的自动任务",
+      "localTrigger": "本地触发器",
       "labels": {
         "schedule": "计划",
         "next": "下次",
@@ -551,11 +552,13 @@
         "weekdaysAt": "工作日 {{time}}",
         "hourlyAt": "每小时第 {{minute}} 分钟",
         "hourlyWindow": "{{start}}-{{end}} 点每小时第 {{minute}} 分钟",
+        "local": "本地触发器",
         "custom": "自定义计划"
       },
       "next": {
         "paused": "已暂停",
         "pending": "正在运行",
+        "local": "等待触发",
         "none": "没有下次运行"
       },
       "message": {
@@ -692,11 +695,13 @@
       "every": "每 {{duration}}",
       "cron": "Cron {{expr}}",
       "cronWithTz": "Cron {{expr}} · {{tz}}",
+      "local": "本地触发器",
       "unknown": "自定义计划"
     },
     "next": {
       "label": "下次：{{time}}",
       "disabled": "已暂停",
+      "local": "等待触发",
       "none": "没有下次运行"
     }
   },
@@ -791,12 +796,14 @@
         "every": "每 {{duration}}",
         "cron": "Cron {{expr}}",
         "cronWithTz": "Cron {{expr}} · {{tz}}",
+        "local": "本地触发器",
         "unknown": "自定义计划"
       },
       "next": {
         "label": "下次 {{time}}",
         "pending": "即将执行",
         "disabled": "已暂停",
+        "local": "等待触发",
         "none": "没有下次运行"
       }
     },
@@ -917,6 +924,10 @@
             "title": "长期目标",
             "description": "让助手把当前请求当作需要多步骤持续推进的目标。"
           },
+          "trigger": {
+            "title": "创建本地触发器",
+            "description": "创建绑定到当前聊天会话的 CLI 触发器。"
+          },
           "help": {
             "title": "查看帮助",
             "description": "列出可用的斜杠命令。"

webui/src/i18n/locales/zh-TW/common.json

@@ -493,6 +493,7 @@
       "emptyHint": "請從它應該執行的來源處建立，這樣 nanobot 才能保留正確上下文。",
       "oneShot": "一次性",
       "systemTask": "系統管理的自動任務",
+      "localTrigger": "本機觸發器",
       "labels": {
         "schedule": "排程",
         "next": "下次",
@@ -551,11 +552,13 @@
         "weekdaysAt": "工作日 {{time}}",
         "hourlyAt": "每小時第 {{minute}} 分鐘",
         "hourlyWindow": "{{start}}-{{end}} 點每小時第 {{minute}} 分鐘",
+        "local": "本機觸發器",
         "custom": "自訂排程"
       },
       "next": {
         "paused": "已暫停",
         "pending": "正在執行",
+        "local": "等待觸發",
         "none": "沒有下次執行"
       },
       "message": {
@@ -692,11 +695,13 @@
       "every": "每 {{duration}}",
       "cron": "Cron {{expr}}",
       "cronWithTz": "Cron {{expr}} · {{tz}}",
+      "local": "本機觸發器",
       "unknown": "自訂計畫"
     },
     "next": {
       "label": "下次：{{time}}",
       "disabled": "已暫停",
+      "local": "等待觸發",
       "none": "沒有下次執行"
     }
   },
@@ -791,12 +796,14 @@
         "every": "每 {{duration}}",
         "cron": "Cron {{expr}}",
         "cronWithTz": "Cron {{expr}} · {{tz}}",
+        "local": "本機觸發器",
         "unknown": "自訂計畫"
       },
       "next": {
         "label": "下次 {{time}}",
         "pending": "即將執行",
         "disabled": "已暫停",
+        "local": "等待觸發",
         "none": "沒有下次執行"
       }
     },
@@ -908,6 +915,10 @@
             "title": "長期目標",
             "description": "請助理把這則請求當成需要多步驟持續推進的目標。"
           },
+          "trigger": {
+            "title": "建立本機觸發器",
+            "description": "建立綁定到目前聊天工作階段的 CLI 觸發器。"
+          },
           "help": {
             "title": "查看說明",
             "description": "列出可用的斜線命令。"

webui/src/lib/types.ts

@@ -32,7 +32,7 @@ export interface UIMediaAttachment {
   name?: string;
 }
 
-export interface UIMessageSource { kind: "cron"; label?: string; }
+export interface UIMessageSource { kind: "cron" | "local_trigger" | "trigger" | string; label?: string; }
 
 export interface UIMessage {
   id: string;
@@ -104,8 +104,9 @@ export interface SessionAutomationJob {
   delete_after_run?: boolean;
   created_at_ms?: number | null;
   updated_at_ms?: number | null;
+  kind?: "local_trigger" | "cron" | string;
   schedule: {
-    kind: "at" | "every" | "cron" | string;
+    kind: "at" | "every" | "cron" | "local" | string;
     at_ms?: number | null;
     every_ms?: number | null;
     expr?: string | null;
@@ -113,7 +114,8 @@ export interface SessionAutomationJob {
   };
   payload: {
     message: string;
-    kind?: "agent_turn" | "system_event" | string;
+    kind?: "agent_turn" | "system_event" | "local_trigger" | string;
+    command?: string;
   };
   state: {
     next_run_at_ms?: number | null;
@@ -135,6 +137,10 @@ export interface SessionAutomationJob {
     title?: string;
     preview?: string;
   } | null;
+  trigger?: {
+    id: string;
+    command: string;
+  };
 }
 
 export interface SessionAutomationsPayload { jobs: SessionAutomationJob[]; }

webui/src/tests/i18n.test.tsx

@@ -21,6 +21,7 @@ const SLASH_COMMAND_KEYS = [
   "dream_log",
   "dream_restore",
   "goal",
+  "trigger",
   "help",
   "pairing",
 ];