From c1fd76add349da4ac7b510061b765a49a74a61d1 Mon Sep 17 00:00:00 2001 From: chengyongru <2755839590@qq.com> Date: Wed, 15 Jul 2026 00:44:39 +0800 Subject: [PATCH] refactor(prompts): share workspace override handling --- nanobot/agent/memory.py | 42 +++++++++++----------- nanobot/command/builtin.py | 21 ++--------- nanobot/utils/evaluator.py | 35 +++++++++--------- nanobot/utils/workspace_prompts.py | 58 ++++++++++++++++++++++++++++++ tests/agent/test_evaluator.py | 35 +++++++++++++++++- 5 files changed, 133 insertions(+), 58 deletions(-) create mode 100644 nanobot/utils/workspace_prompts.py diff --git a/nanobot/agent/memory.py b/nanobot/agent/memory.py index c5785b45..39ddf889 100644 --- a/nanobot/agent/memory.py +++ b/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 self.dream_prompt_file.is_file() and bool( - self.dream_prompt_file.read_text(encoding="utf-8").strip() - ) - return False + return has_workspace_prompt_override(self.dream_prompt_file) @staticmethod def default_dream_prompt() -> str: @@ -512,20 +514,19 @@ class MemoryStore: ) def _dream_template(self) -> str: - with suppress(OSError): - text = self.dream_prompt_file.read_text(encoding="utf-8") - if text.strip(): - text = text.rstrip() - if len(text) > _DREAM_PROMPT_MAX_CHARS: - if not self._dream_prompt_oversize_logged: - self._dream_prompt_oversize_logged = True - logger.warning( - "workspace Dream prompt exceeds {} chars ({}); truncating. " - "Further occurrences suppressed.", - _DREAM_PROMPT_MAX_CHARS, len(text), - ) - return truncate_text(text, _DREAM_PROMPT_MAX_CHARS) - return text + text, original_chars = load_workspace_prompt_override(self.dream_prompt_file) + if text is not None: + if ( + original_chars > WORKSPACE_PROMPT_MAX_CHARS + and not self._dream_prompt_oversize_logged + ): + self._dream_prompt_oversize_logged = True + logger.warning( + "workspace Dream prompt exceeds {} chars ({}); truncating. " + "Further occurrences suppressed.", + WORKSPACE_PROMPT_MAX_CHARS, original_chars, + ) + 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 diff --git a/nanobot/command/builtin.py b/nanobot/command/builtin.py index 534da6d6..ee8630d1 100644 --- a/nanobot/command/builtin.py +++ b/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: - 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: + if not initialize_workspace_prompt(path, store.default_dream_prompt()): 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: - 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: + if not initialize_workspace_prompt(path, default_evaluator_prompt()): 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. " diff --git a/nanobot/utils/evaluator.py b/nanobot/utils/evaluator.py index ac4dc163..4faf15db 100644 --- a/nanobot/utils/evaluator.py +++ b/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): - path = evaluator_prompt_file(workspace) - return path.is_file() and bool(path.read_text(encoding="utf-8").strip()) - return False + return has_workspace_prompt_override(evaluator_prompt_file(workspace)) 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 = 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 + text, original_chars = load_workspace_prompt_override(evaluator_prompt_file(workspace)) + if text is not None: + if original_chars > EVALUATOR_PROMPT_MAX_CHARS: + logger.warning( + "Workspace heartbeat evaluator prompt exceeds {} chars ({}); truncating.", + EVALUATOR_PROMPT_MAX_CHARS, original_chars, + ) + return text return default_evaluator_prompt() _EVALUATE_TOOL = [ diff --git a/nanobot/utils/workspace_prompts.py b/nanobot/utils/workspace_prompts.py new file mode 100644 index 00000000..5dfa2a76 --- /dev/null +++ b/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 diff --git a/tests/agent/test_evaluator.py b/tests/agent/test_evaluator.py index faf37522..eb19af98 100644 --- a/tests/agent/test_evaluator.py +++ b/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")])