refactor(prompts): share workspace override handling

nanobot/agent/memory.py

@@ -29,6 +29,12 @@ from nanobot.utils.helpers import (
     truncate_text_to_tokens,
 )
 from nanobot.utils.prompt_templates import render_template
+from nanobot.utils.workspace_prompts import (
+    WORKSPACE_PROMPT_MAX_CHARS,
+    has_workspace_prompt_override,
+    load_workspace_prompt_override,
+    workspace_prompt_file,
+)
 
 if TYPE_CHECKING:
     from nanobot.utils.llm_runtime import LLMRuntime
@@ -492,14 +498,10 @@ class MemoryStore:
 
     @property
     def dream_prompt_file(self) -> Path:
-        return self.workspace / "prompts" / "dream.md"
+        return workspace_prompt_file(self.workspace, "dream")
 
     def has_dream_prompt_override(self) -> bool:
-        with suppress(OSError):
+        return has_workspace_prompt_override(self.dream_prompt_file)
-            return self.dream_prompt_file.is_file() and bool(
-                self.dream_prompt_file.read_text(encoding="utf-8").strip()
-            )
-        return False
 
     @staticmethod
     def default_dream_prompt() -> str:
@@ -512,20 +514,19 @@ class MemoryStore:
         )
 
     def _dream_template(self) -> str:
-        with suppress(OSError):
+        text, original_chars = load_workspace_prompt_override(self.dream_prompt_file)
-            text = self.dream_prompt_file.read_text(encoding="utf-8")
+        if text is not None:
-            if text.strip():
+            if (
-                text = text.rstrip()
+                original_chars > WORKSPACE_PROMPT_MAX_CHARS
-                if len(text) > _DREAM_PROMPT_MAX_CHARS:
+                and not self._dream_prompt_oversize_logged
-                    if not self._dream_prompt_oversize_logged:
+            ):
-                        self._dream_prompt_oversize_logged = True
+                self._dream_prompt_oversize_logged = True
-                        logger.warning(
+                logger.warning(
-                            "workspace Dream prompt exceeds {} chars ({}); truncating. "
+                    "workspace Dream prompt exceeds {} chars ({}); truncating. "
-                            "Further occurrences suppressed.",
+                    "Further occurrences suppressed.",
-                            _DREAM_PROMPT_MAX_CHARS, len(text),
+                    WORKSPACE_PROMPT_MAX_CHARS, original_chars,
-                        )
+                )
-                    return truncate_text(text, _DREAM_PROMPT_MAX_CHARS)
+            return text
-                return text
         return self.default_dream_prompt()
 
     def build_dream_prompt(self, *, max_entries: int = 20) -> tuple[str, int] | None:
@@ -734,7 +735,6 @@ class MemoryStore:
 # that catches any new caller that forgot to set its own cap.
 _RAW_ARCHIVE_MAX_CHARS = 16_000       # fallback dump (LLM failed)
 _ARCHIVE_SUMMARY_MAX_CHARS = 8_000    # LLM-produced consolidation summary
-_DREAM_PROMPT_MAX_CHARS = 32_000      # workspace-local Dream prompt override
 _HISTORY_ENTRY_HARD_CAP = 64_000      # emergency cap in append_history
 
 

nanobot/command/builtin.py

@@ -17,6 +17,7 @@ from nanobot.bus.events import OutboundMessage
 from nanobot.command.router import CommandContext, CommandRouter
 from nanobot.utils.helpers import build_status_content
 from nanobot.utils.restart import set_restart_notice_to_env
+from nanobot.utils.workspace_prompts import initialize_workspace_prompt
 
 # WebUI protocol contract for how a slash command participates in turn state:
 # - side_channel: returns control text without starting or ending an agent turn.
@@ -480,20 +481,12 @@ async def cmd_dream_prompt(ctx: CommandContext) -> OutboundMessage:
     args = ctx.args.strip().lower()
 
     if args == "init":
-        try:
+        if not initialize_workspace_prompt(path, store.default_dream_prompt()):
-            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"Dream memory instructions already exist 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(store.default_dream_prompt() + "\n", encoding="utf-8")
             content = (
                 f"Created Dream memory instructions at `{display_path}`.\n\n"
                 "Edit that file to teach Dream how to organize memory. "
@@ -537,20 +530,12 @@ async def cmd_evaluator_prompt(ctx: CommandContext) -> OutboundMessage:
     args = ctx.args.strip().lower()
 
     if args == "init":
-        try:
+        if not initialize_workspace_prompt(path, default_evaluator_prompt()):
-            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. "

nanobot/utils/evaluator.py

@@ -6,33 +6,34 @@ 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
+from nanobot.utils.workspace_prompts import (
+    WORKSPACE_PROMPT_MAX_CHARS,
+    has_workspace_prompt_override,
+    load_workspace_prompt_override,
+    workspace_prompt_file,
+)
 
 if TYPE_CHECKING:
     from nanobot.providers.base import LLMProvider
 
 # Cap for a workspace-local heartbeat evaluator prompt override.
-EVALUATOR_PROMPT_MAX_CHARS = 32_000
+EVALUATOR_PROMPT_MAX_CHARS = WORKSPACE_PROMPT_MAX_CHARS
 
 
 def evaluator_prompt_file(workspace: Path) -> Path:
     """Path to the workspace-local heartbeat evaluator prompt override."""
-    return workspace / "prompts" / "evaluator.md"
+    return workspace_prompt_file(workspace, "evaluator")
 
 
 def has_evaluator_prompt_override(workspace: Path) -> bool:
     """True when the workspace defines a non-empty evaluator prompt override."""
-    with suppress(OSError):
+    return has_workspace_prompt_override(evaluator_prompt_file(workspace))
-        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:
@@ -46,16 +47,14 @@ def resolve_evaluator_prompt(workspace: Path) -> str:
     Oversized overrides are truncated so a runaway file cannot blow up the
     evaluator call.
     """
-    with suppress(OSError):
+    text, original_chars = load_workspace_prompt_override(evaluator_prompt_file(workspace))
-        text = evaluator_prompt_file(workspace).read_text(encoding="utf-8").rstrip()
+    if text is not None:
-        if text:
+        if original_chars > EVALUATOR_PROMPT_MAX_CHARS:
-            if len(text) > EVALUATOR_PROMPT_MAX_CHARS:
+            logger.warning(
-                logger.warning(
+                "Workspace heartbeat evaluator prompt exceeds {} chars ({}); truncating.",
-                    "Workspace heartbeat evaluator prompt exceeds {} chars ({}); truncating.",
+                EVALUATOR_PROMPT_MAX_CHARS, original_chars,
-                    EVALUATOR_PROMPT_MAX_CHARS, len(text),
+            )
-                )
+        return text
-                return truncate_text(text, EVALUATOR_PROMPT_MAX_CHARS)
-            return text
     return default_evaluator_prompt()
 
 _EVALUATE_TOOL = [

nanobot/utils/workspace_prompts.py

@@ -0,0 +1,58 @@
+"""Shared file handling for workspace-local prompt overrides."""
+
+from __future__ import annotations
+
+from contextlib import suppress
+from pathlib import Path
+
+from nanobot.utils.helpers import truncate_text
+
+WORKSPACE_PROMPT_MAX_CHARS = 32_000
+
+
+def workspace_prompt_file(workspace: Path, name: str) -> Path:
+    """Return the conventional path for a named workspace prompt override."""
+    return workspace / "prompts" / f"{name}.md"
+
+
+def load_workspace_prompt_override(
+    path: Path,
+    *,
+    max_chars: int = WORKSPACE_PROMPT_MAX_CHARS,
+) -> tuple[str | None, int]:
+    """Load and cap a non-empty UTF-8 prompt override.
+
+    Returns the loaded text and its original length. Missing, unreadable, and
+    empty files return ``(None, 0)`` so callers can fall back to their default.
+    """
+    with suppress(OSError):
+        text = path.read_text(encoding="utf-8").rstrip()
+        if text:
+            original_chars = len(text)
+            return truncate_text(text, max_chars), original_chars
+    return None, 0
+
+
+def has_workspace_prompt_override(path: Path) -> bool:
+    """Return whether a path contains a non-empty workspace prompt override."""
+    text, _original_chars = load_workspace_prompt_override(path)
+    return text is not None
+
+
+def initialize_workspace_prompt(path: Path, default_prompt: str) -> bool:
+    """Create a default prompt copy when the target is missing or empty.
+
+    Returns ``False`` without overwriting a non-empty file, non-file path, or
+    path whose current state cannot be read safely.
+    """
+    try:
+        if path.exists() and (
+            not path.is_file() or bool(path.read_text(encoding="utf-8").strip())
+        ):
+            return False
+    except OSError:
+        return False
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(default_prompt + "\n", encoding="utf-8")
+    return True

tests/agent/test_evaluator.py

@@ -1,7 +1,13 @@
 import pytest
 
 from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
-from nanobot.utils.evaluator import evaluate_response
+from nanobot.utils.evaluator import (
+    EVALUATOR_PROMPT_MAX_CHARS,
+    default_evaluator_prompt,
+    evaluate_response,
+    evaluator_prompt_file,
+    resolve_evaluator_prompt,
+)
 
 
 class DummyProvider(LLMProvider):
@@ -34,6 +40,33 @@ def _eval_tool_call(should_notify: bool, reason: str = "") -> LLMResponse:
 _EVAL_PROMPT = "You are a notification gate. Call evaluate_notification."
 
 
+def test_resolve_evaluator_prompt_uses_workspace_override(tmp_path) -> None:
+    path = evaluator_prompt_file(tmp_path)
+    path.parent.mkdir()
+    path.write_text("Custom evaluator prompt.\n", encoding="utf-8")
+
+    assert resolve_evaluator_prompt(tmp_path) == "Custom evaluator prompt."
+
+
+def test_resolve_evaluator_prompt_uses_default_for_empty_override(tmp_path) -> None:
+    path = evaluator_prompt_file(tmp_path)
+    path.parent.mkdir()
+    path.write_text("  \n", encoding="utf-8")
+
+    assert resolve_evaluator_prompt(tmp_path) == default_evaluator_prompt()
+
+
+def test_resolve_evaluator_prompt_caps_workspace_override(tmp_path) -> None:
+    path = evaluator_prompt_file(tmp_path)
+    path.parent.mkdir()
+    path.write_text("x" * (EVALUATOR_PROMPT_MAX_CHARS + 1), encoding="utf-8")
+
+    prompt = resolve_evaluator_prompt(tmp_path)
+
+    assert prompt.startswith("x" * EVALUATOR_PROMPT_MAX_CHARS)
+    assert prompt.endswith("... (truncated)")
+
+
 @pytest.mark.asyncio
 async def test_should_notify_true() -> None:
     provider = DummyProvider([_eval_tool_call(True, "user asked to be reminded")])