feat(heartbeat): custom evaluator prompt

docs/configuration.md

@@ -2031,6 +2031,10 @@ The heartbeat job is backed by the same cron service as user-created reminders.
 | `gateway.heartbeat.keepRecentMessages` | `8` | Number of recent heartbeat-session messages to retain after each run. |
 | `gateway.restartMode` | `auto` | Restart strategy for `/restart`: `auto` uses `spawn` on Windows foreground runs and `exec` elsewhere. Use `exit` with Windows service wrappers such as WinSW or nssm so the service manager owns the restart. |
 
+### Custom heartbeat evaluator prompt
+
+The notification gate runs on a built-in system prompt. Advanced users can override it, but you rarely need to — it's strongly advised to first read the evaluator code and the default `evaluator.md`. To override, drop your prompt at `<workspace>/prompts/evaluator.md`. It must still instruct the model to call the `evaluate_notification` tool; otherwise the gate fails closed and stays silent.
+
 
 ## Subagent Concurrency
 

nanobot/cli/commands.py

@@ -77,7 +77,7 @@ from nanobot.cli.stream import StreamRenderer, ThinkingSpinner  # noqa: E402
 from nanobot.config.paths import get_workspace_path, is_default_workspace  # noqa: E402
 from nanobot.config.schema import Config  # noqa: E402
 from nanobot.security.network import is_loopback_host  # noqa: E402
-from nanobot.utils.evaluator import evaluate_response  # noqa: E402
+from nanobot.utils.evaluator import evaluate_response, resolve_evaluator_prompt  # noqa: E402
 from nanobot.utils.helpers import sync_workspace_templates  # noqa: E402
 from nanobot.utils.restart import (  # noqa: E402
     consume_restart_notice_from_env,
@@ -1859,21 +1859,29 @@ def _run_gateway(
             finally:
                 if isinstance(message_tool, MessageTool) and suppress_token is not None:
                     message_tool.reset_suppress_delivery(suppress_token)
-            response = resp.content if resp else ""
+
+            if not resp or not resp.content:
+                return
+
+            response = resp.content
 
             # Keep a small tail of heartbeat history so the loop stays bounded.
             session = agent.sessions.get_or_create("heartbeat")
             session.retain_recent_legal_suffix(hb_cfg.keep_recent_messages)
             agent.sessions.save(session)
 
-            if not response:
-                return None
+            evaluator_prompt = resolve_evaluator_prompt(config.workspace_path)
 
             # Fail closed: stay silent on evaluator failure instead of notifying.
             should_notify = await evaluate_response(
-                response, prompt, agent.provider, agent.model,
+                response=response,
+                task_context=prompt,
+                provider=agent.provider,
+                model=agent.model,
+                evaluator_prompt=evaluator_prompt,
                 default_notify=False,
             )
+
             if should_notify:
                 logger.info("Heartbeat: completed, delivering response")
                 await _deliver_to_channel(

nanobot/command/builtin.py

@@ -143,6 +143,14 @@ BUILTIN_COMMAND_SPECS: tuple[BuiltinCommandSpec, ...] = (
         "[init]",
         accepts_args=True,
     ),
+    BuiltinCommandSpec(
+        "/evaluator-prompt",
+        "Heartbeat evaluator",
+        "Customize the heartbeat notification gate prompt for this workspace.",
+        "file-text",
+        "[init]",
+        accepts_args=True,
+    ),
     BuiltinCommandSpec(
         "/skill",
         "List skills",
@@ -515,6 +523,64 @@ async def cmd_dream_prompt(ctx: CommandContext) -> OutboundMessage:
     )
 
 
+async def cmd_evaluator_prompt(ctx: CommandContext) -> OutboundMessage:
+    """Show or set up the workspace heartbeat evaluator prompt."""
+    from nanobot.utils.evaluator import (
+        default_evaluator_prompt,
+        evaluator_prompt_file,
+        has_evaluator_prompt_override,
+    )
+
+    workspace = ctx.loop.context.memory.workspace
+    path = evaluator_prompt_file(workspace)
+    display_path = path.relative_to(workspace).as_posix()
+    args = ctx.args.strip().lower()
+
+    if args == "init":
+        try:
+            prompt_exists_with_content = path.exists() and (
+                not path.is_file() or bool(path.read_text(encoding="utf-8").strip())
+            )
+        except OSError:
+            prompt_exists_with_content = True
+        if prompt_exists_with_content:
+            content = (
+                f"Heartbeat evaluator prompt already exists at `{display_path}`.\n\n"
+                "Edit that file, or delete/empty it to return to nanobot's default."
+            )
+        else:
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(default_evaluator_prompt() + "\n", encoding="utf-8")
+            content = (
+                f"Created heartbeat evaluator prompt at `{display_path}`.\n\n"
+                "Edit that file to control when the heartbeat notification gate speaks. "
+                "It must still instruct the model to call the `evaluate_notification` tool, "
+                "otherwise the gate fails closed and stays silent. "
+                "Delete or empty it to return to nanobot's default."
+            )
+    elif args:
+        content = "Usage: /evaluator-prompt [init]"
+    elif has_evaluator_prompt_override(workspace):
+        content = (
+            "Heartbeat evaluator prompt: custom for this workspace\n\n"
+            f"- Path: `{display_path}`\n"
+            "- Delete or empty this file to return to nanobot's default."
+        )
+    else:
+        content = (
+            "Heartbeat evaluator prompt: nanobot default\n\n"
+            f"- Editable file: `{display_path}`\n"
+            "- Run `/evaluator-prompt init` to create an editable copy."
+        )
+
+    return OutboundMessage(
+        channel=ctx.msg.channel,
+        chat_id=ctx.msg.chat_id,
+        content=content,
+        metadata={**dict(ctx.msg.metadata or {}), "render_as": "text"},
+    )
+
+
 def _format_dream_no_input_message() -> str:
     return "\n".join([
         "Dream has no conversation history to process yet.",
@@ -954,6 +1020,8 @@ def register_builtin_commands(router: CommandRouter) -> None:
     router.prefix("/dream-restore ", cmd_dream_restore)
     router.exact("/dream-prompt", cmd_dream_prompt)
     router.prefix("/dream-prompt ", cmd_dream_prompt)
+    router.exact("/evaluator-prompt", cmd_evaluator_prompt)
+    router.prefix("/evaluator-prompt ", cmd_evaluator_prompt)
     router.exact("/skill", cmd_skill)
     router.exact("/help", cmd_help)
     router.exact("/pairing", cmd_pairing)

nanobot/templates/prompts/README.md

@@ -1,11 +1,25 @@
-# Dream Memory Instructions
+# Prompt Overrides
 
-This folder is for plain-language instructions that tell Dream how to organize memory in this workspace.
+This folder holds plain-language prompt overrides for this workspace.
 
-Most users do not need to edit anything here. To guide Dream differently for this workspace, run:
+## Dream memory
+
+`dream.md` tells Dream how to organize memory in this workspace. Most users do not need to touch it. To create an editable copy, run:
 
 ```text
 /dream-prompt init
 ```
 
 That creates `prompts/dream.md`. Edit it in plain Markdown. Delete or empty it to return to nanobot's default memory behavior.
+
+## Heartbeat evaluator
+
+`evaluator.md` overrides the system prompt for the heartbeat notification gate — the model that decides whether a heartbeat result is worth delivering. This is an advanced override; you rarely need it. Before editing, read the evaluator code and the default `evaluator.md`.
+
+To create an editable copy, run:
+
+```text
+/evaluator-prompt init
+```
+
+That creates `prompts/evaluator.md`. It must still instruct the model to call the `evaluate_notification` tool; otherwise the gate fails closed and stays silent. Delete or empty the file to return to the built-in prompt.

nanobot/utils/evaluator.py

@@ -6,15 +6,58 @@ LLM call to decide whether the result warrants notifying the user.
 
 from __future__ import annotations
 
+from contextlib import suppress
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 from loguru import logger
 
+from nanobot.utils.helpers import truncate_text
 from nanobot.utils.prompt_templates import render_template
 
 if TYPE_CHECKING:
     from nanobot.providers.base import LLMProvider
 
+# Cap for a workspace-local heartbeat evaluator prompt override.
+EVALUATOR_PROMPT_MAX_CHARS = 32_000
+
+
+def evaluator_prompt_file(workspace: Path) -> Path:
+    """Path to the workspace-local heartbeat evaluator prompt override."""
+    return workspace / "prompts" / "evaluator.md"
+
+
+def has_evaluator_prompt_override(workspace: Path) -> bool:
+    """True when the workspace defines a non-empty evaluator prompt override."""
+    with suppress(OSError):
+        path = evaluator_prompt_file(workspace)
+        return path.is_file() and bool(path.read_text(encoding="utf-8").strip())
+    return False
+
+
+def default_evaluator_prompt() -> str:
+    """The built-in heartbeat notification-gate system prompt."""
+    return render_template("agent/evaluator.md", part="system", strip=True)
+
+
+def resolve_evaluator_prompt(workspace: Path) -> str:
+    """Return the active evaluator prompt: workspace override or built-in default.
+
+    Oversized overrides are truncated so a runaway file cannot blow up the
+    evaluator call.
+    """
+    with suppress(OSError):
+        text = evaluator_prompt_file(workspace).read_text(encoding="utf-8").rstrip()
+        if text:
+            if len(text) > EVALUATOR_PROMPT_MAX_CHARS:
+                logger.warning(
+                    "Workspace heartbeat evaluator prompt exceeds {} chars ({}); truncating.",
+                    EVALUATOR_PROMPT_MAX_CHARS, len(text),
+                )
+                return truncate_text(text, EVALUATOR_PROMPT_MAX_CHARS)
+            return text
+    return default_evaluator_prompt()
+
 _EVALUATE_TOOL = [
     {
         "type": "function",
@@ -44,17 +87,19 @@ async def evaluate_response(
     task_context: str,
     provider: LLMProvider,
     model: str,
-    default_notify: bool = True,
+    evaluator_prompt: str,
+    default_notify: bool = False,
 ) -> bool:
     """Decide whether a heartbeat result should be delivered to the user.
 
     On any failure, falls back to ``default_notify``. Heartbeat passes
     ``False`` to fail closed.
     """
+
     try:
         llm_response = await provider.chat_with_retry(
             messages=[
-                {"role": "system", "content": render_template("agent/evaluator.md", part="system")},
+                {"role": "system", "content": evaluator_prompt},
                 {"role": "user", "content": render_template(
                     "agent/evaluator.md",
                     part="user",
@@ -64,7 +109,7 @@ async def evaluate_response(
             ],
             tools=_EVALUATE_TOOL,
             model=model,
-            max_tokens=256,
+            max_tokens=4096,
             temperature=0.0,
         )
 

tests/agent/test_evaluator.py

@@ -31,17 +31,26 @@ def _eval_tool_call(should_notify: bool, reason: str = "") -> LLMResponse:
     )
 
 
+_EVAL_PROMPT = "You are a notification gate. Call evaluate_notification."
+
+
 @pytest.mark.asyncio
 async def test_should_notify_true() -> None:
     provider = DummyProvider([_eval_tool_call(True, "user asked to be reminded")])
-    result = await evaluate_response("Task completed with results", "check emails", provider, "m")
+    result = await evaluate_response(
+        "Task completed with results", "check emails", provider, "m",
+        evaluator_prompt=_EVAL_PROMPT,
+    )
     assert result is True
 
 
 @pytest.mark.asyncio
 async def test_should_notify_false() -> None:
     provider = DummyProvider([_eval_tool_call(False, "routine check, nothing new")])
-    result = await evaluate_response("All clear, no updates", "check status", provider, "m")
+    result = await evaluate_response(
+        "All clear, no updates", "check status", provider, "m",
+        evaluator_prompt=_EVAL_PROMPT,
+    )
     assert result is False
 
 
@@ -52,14 +61,20 @@ async def test_fallback_on_error() -> None:
             raise RuntimeError("provider down")
 
     provider = FailingProvider([])
-    result = await evaluate_response("some response", "some task", provider, "m")
+    result = await evaluate_response(
+        "some response", "some task", provider, "m",
+        evaluator_prompt=_EVAL_PROMPT, default_notify=True,
+    )
     assert result is True
 
 
 @pytest.mark.asyncio
 async def test_no_tool_call_fallback() -> None:
     provider = DummyProvider([LLMResponse(content="I think you should notify", tool_calls=[])])
-    result = await evaluate_response("some response", "some task", provider, "m")
+    result = await evaluate_response(
+        "some response", "some task", provider, "m",
+        evaluator_prompt=_EVAL_PROMPT, default_notify=True,
+    )
     assert result is True
 
 
@@ -70,12 +85,18 @@ async def test_fail_closed_on_error() -> None:
             raise RuntimeError("provider down")
 
     provider = FailingProvider([])
-    result = await evaluate_response("some", "task", provider, "m", default_notify=False)
+    result = await evaluate_response(
+        "some", "task", provider, "m",
+        evaluator_prompt=_EVAL_PROMPT, default_notify=False,
+    )
     assert result is False
 
 
 @pytest.mark.asyncio
 async def test_fail_closed_on_no_tool_call() -> None:
     provider = DummyProvider([LLMResponse(content="text only", tool_calls=[])])
-    result = await evaluate_response("some", "task", provider, "m", default_notify=False)
+    result = await evaluate_response(
+        "some", "task", provider, "m",
+        evaluator_prompt=_EVAL_PROMPT, default_notify=False,
+    )
     assert result is False

tests/command/test_builtin_evaluator_prompt.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from nanobot.bus.events import InboundMessage
+from nanobot.command.builtin import (
+    build_help_text,
+    builtin_command_palette,
+    cmd_evaluator_prompt,
+)
+from nanobot.command.router import CommandContext
+from nanobot.utils.evaluator import default_evaluator_prompt
+
+
+def _make_ctx(tmp_path, raw: str = "/evaluator-prompt", args: str = "") -> CommandContext:
+    msg = InboundMessage(channel="cli", sender_id="u1", chat_id="direct", content=raw)
+    loop = SimpleNamespace(context=SimpleNamespace(memory=SimpleNamespace(workspace=tmp_path)))
+    return CommandContext(msg=msg, session=None, key=msg.session_key, raw=raw, args=args, loop=loop)
+
+
+@pytest.mark.asyncio
+async def test_evaluator_prompt_reports_default_prompt(tmp_path) -> None:
+    out = await cmd_evaluator_prompt(_make_ctx(tmp_path))
+
+    assert "Heartbeat evaluator prompt: nanobot default" in out.content
+    assert "prompts/evaluator.md" in out.content
+    assert str(tmp_path) not in out.content
+    assert "/evaluator-prompt init" in out.content
+
+
+@pytest.mark.asyncio
+async def test_evaluator_prompt_init_copies_default_prompt(tmp_path) -> None:
+    ctx = _make_ctx(tmp_path, "/evaluator-prompt init", "init")
+
+    out = await cmd_evaluator_prompt(ctx)
+
+    prompt_file = tmp_path / "prompts" / "evaluator.md"
+    assert "Created heartbeat evaluator prompt" in out.content
+    assert "prompts/evaluator.md" in out.content
+    assert str(tmp_path) not in out.content
+    assert "evaluate_notification" in out.content
+    assert prompt_file.read_text(encoding="utf-8") == default_evaluator_prompt() + "\n"
+
+
+@pytest.mark.asyncio
+async def test_evaluator_prompt_init_does_not_overwrite_existing_prompt(tmp_path) -> None:
+    prompt_file = tmp_path / "prompts" / "evaluator.md"
+    prompt_file.parent.mkdir()
+    prompt_file.write_text("custom", encoding="utf-8")
+    ctx = _make_ctx(tmp_path, "/evaluator-prompt init", "init")
+
+    out = await cmd_evaluator_prompt(ctx)
+
+    assert "already exists" in out.content
+    assert "prompts/evaluator.md" in out.content
+    assert str(tmp_path) not in out.content
+    assert prompt_file.read_text(encoding="utf-8") == "custom"
+
+
+@pytest.mark.asyncio
+async def test_evaluator_prompt_init_recreates_empty_prompt(tmp_path) -> None:
+    prompt_file = tmp_path / "prompts" / "evaluator.md"
+    prompt_file.parent.mkdir()
+    prompt_file.write_text("  \n", encoding="utf-8")
+    ctx = _make_ctx(tmp_path, "/evaluator-prompt init", "init")
+
+    out = await cmd_evaluator_prompt(ctx)
+
+    assert "Created heartbeat evaluator prompt" in out.content
+    assert prompt_file.read_text(encoding="utf-8") == default_evaluator_prompt() + "\n"
+
+
+@pytest.mark.asyncio
+async def test_evaluator_prompt_reports_override(tmp_path) -> None:
+    prompt_file = tmp_path / "prompts" / "evaluator.md"
+    prompt_file.parent.mkdir()
+    prompt_file.write_text("custom", encoding="utf-8")
+
+    out = await cmd_evaluator_prompt(_make_ctx(tmp_path))
+
+    assert "Heartbeat evaluator prompt: custom for this workspace" in out.content
+    assert "prompts/evaluator.md" in out.content
+
+
+@pytest.mark.asyncio
+async def test_evaluator_prompt_rejects_unknown_args(tmp_path) -> None:
+    out = await cmd_evaluator_prompt(_make_ctx(tmp_path, "/evaluator-prompt nope", "nope"))
+
+    assert out.content == "Usage: /evaluator-prompt [init]"
+
+
+def test_evaluator_prompt_command_in_help_and_palette() -> None:
+    palette = builtin_command_palette()
+    entry = next(item for item in palette if item["command"] == "/evaluator-prompt")
+
+    assert entry["arg_hint"] == "[init]"
+    assert entry["lifecycle"] == "side_channel"
+    assert entry["accepts_args"] is True
+    assert "/evaluator-prompt [init]" in build_help_text()