refactor: split system prompt into per-concern extensions with extensibility hooks

- split monolithic _10_system_prompt.py into focused extensions: main (10), tools (11), mcp (12), skills (13), secrets (13), project (14)
- each extension exposes a build_prompt() function with call_extensions_async hook for plugin extensibility
- move tool prompt collection from VariablesPlugin to _11_tools_prompt using subagents.get_paths for proper directory coverage
- add {{include original}} directive to process_includes allowing prompt inheritance without copy-paste
- add agent.system.main.specifics.md for subagent-specific additions without overriding entire role
- remove redundant plugin prompt extensions (_15_text_editor, _20_code_execution) that duplicated tool collection
- add _09_text_editor_config to register per-file kwargs via agent.data instead of VariablesPlugin in prompts dir

extensions/python/system_prompt/_10_main_prompt.py

@@ -0,0 +1,24 @@
+from typing import Any
+
+from helpers.extension import Extension, call_extensions_async
+from agent import Agent, LoopData
+
+
+class MainPrompt(Extension):
+
+    async def execute(
+        self,
+        system_prompt: list[str] = [],
+        loop_data: LoopData = LoopData(),
+        **kwargs: Any,
+    ):
+        if not self.agent:
+            return
+        prompt = await build_prompt(self.agent)
+        system_prompt.append(prompt)
+
+
+async def build_prompt(agent: Agent) -> str:
+    data: dict[str, Any] = {"prompt": agent.read_prompt("agent.system.main.md")}
+    await call_extensions_async("system_prompt_main", agent=agent, data=data)
+    return data["prompt"]

extensions/python/system_prompt/_10_system_prompt.py

@@ -1,101 +0,0 @@
-from typing import Any
-from helpers.extension import Extension
-from helpers.mcp_handler import MCPConfig
-from agent import Agent, LoopData
-from helpers.settings import get_settings
-from helpers import projects, skills
-
-
-class SystemPrompt(Extension):
-
-    async def execute(
-        self,
-        system_prompt: list[str] = [],
-        loop_data: LoopData = LoopData(),
-        **kwargs: Any
-    ):
-        if not self.agent:
-            return
-
-        # append main system prompt and tools
-        main = get_main_prompt(self.agent)
-        tools = get_tools_prompt(self.agent)
-        mcp_tools = get_mcp_tools_prompt(self.agent)
-        skills = get_skills_prompt(self.agent)
-        secrets_prompt = get_secrets_prompt(self.agent)
-        project_prompt = get_project_prompt(self.agent)
-
-        system_prompt.append(main)
-        system_prompt.append(tools)
-        if mcp_tools:
-            system_prompt.append(mcp_tools)
-        if skills:
-            system_prompt.append(skills)
-        if secrets_prompt:
-            system_prompt.append(secrets_prompt)
-        if project_prompt:
-            system_prompt.append(project_prompt)
-       
-
-def get_main_prompt(agent: Agent):
-    return agent.read_prompt("agent.system.main.md")
-
-
-def get_tools_prompt(agent: Agent):
-    from plugins._model_config.helpers.model_config import get_chat_model_config
-    prompt = agent.read_prompt("agent.system.tools.md")
-    chat_cfg = get_chat_model_config(agent)
-    if chat_cfg.get("vision", False):
-        prompt += "\n\n" + agent.read_prompt("agent.system.tools_vision.md")
-    return prompt
-
-
-def get_mcp_tools_prompt(agent: Agent):
-    mcp_config = MCPConfig.get_instance()
-    if mcp_config.servers:
-        pre_progress = agent.context.log.progress
-        agent.context.log.set_progress(
-            "Collecting MCP tools"
-        )  # MCP might be initializing, better inform via progress bar
-        tools = MCPConfig.get_instance().get_tools_prompt()
-        agent.context.log.set_progress(pre_progress)  # return original progress
-        return tools
-    return ""
-
-
-def get_secrets_prompt(agent: Agent):
-    try:
-        # Use lazy import to avoid circular dependencies
-        from helpers.secrets import get_secrets_manager
-
-        secrets_manager = get_secrets_manager(agent.context)
-        secrets = secrets_manager.get_secrets_for_prompt()
-        vars = get_settings()["variables"]
-        return agent.read_prompt("agent.system.secrets.md", secrets=secrets, vars=vars)
-    except Exception as e:
-        # If secrets module is not available or has issues, return empty string
-        return ""
-
-
-def get_project_prompt(agent: Agent):
-    result = agent.read_prompt("agent.system.projects.main.md")
-    project_name = agent.context.get_data(projects.CONTEXT_DATA_KEY_PROJECT)
-    if project_name:
-        project_vars = projects.build_system_prompt_vars(project_name)
-        result += "\n\n" + agent.read_prompt(
-            "agent.system.projects.active.md", **project_vars
-        )
-    else:
-        result += "\n\n" + agent.read_prompt("agent.system.projects.inactive.md")
-    return result
-
-def get_skills_prompt(agent: Agent):
-    available = skills.list_skills(agent=agent)
-    result = []
-    for skill in available:
-        name = skill.name.strip().replace("\n", " ")[:100]
-        descr = skill.description.replace("\n", " ")[:500]
-        result.append(f"**{name}** {descr}")
-
-    if result:
-        return agent.read_prompt("agent.system.skills.md", skills="\n".join(result))

extensions/python/system_prompt/_11_tools_prompt.py

@@ -0,0 +1,59 @@
+import os
+from typing import Any
+
+from helpers.extension import Extension, call_extensions_async
+from helpers import files, subagents
+from helpers.print_style import PrintStyle
+from agent import Agent, LoopData
+
+
+TOOL_KWARGS_KEY = "_tool_prompt_kwargs"
+
+
+class ToolsPrompt(Extension):
+
+    async def execute(
+        self,
+        system_prompt: list[str] = [],
+        loop_data: LoopData = LoopData(),
+        **kwargs: Any,
+    ):
+        if not self.agent:
+            return
+        prompt = await build_prompt(self.agent)
+        system_prompt.append(prompt)
+
+
+async def build_prompt(agent: Agent) -> str:
+    # collect tool files from all prompt directories
+    prompt_dirs = subagents.get_paths(agent, "prompts")
+    tool_files = files.get_unique_filenames_in_dirs(
+        prompt_dirs, "agent.system.tool.*.md"
+    )
+
+    # per-file kwargs registered by plugin config extensions (e.g. _09_text_editor_config)
+    all_tool_kwargs: dict[str, dict[str, Any]] = agent.get_data(TOOL_KWARGS_KEY) or {}
+
+    tools: list[str] = []
+    for tool_file in tool_files:
+        try:
+            basename = os.path.basename(tool_file)
+            extra = all_tool_kwargs.get(basename, {})
+            tool = agent.read_prompt(basename, **extra)
+            tools.append(tool)
+        except Exception as e:
+            PrintStyle().error(f"Error loading tool '{tool_file}': {e}")
+
+    tools_str = "\n\n".join(tools)
+    prompt = agent.read_prompt("agent.system.tools.md", tools=tools_str)
+
+    # vision support
+    from plugins._model_config.helpers.model_config import get_chat_model_config
+
+    chat_cfg = get_chat_model_config(agent)
+    if chat_cfg.get("vision", False):
+        prompt += "\n\n" + agent.read_prompt("agent.system.tools_vision.md")
+
+    data: dict[str, Any] = {"prompt": prompt}
+    await call_extensions_async("system_prompt_tools", agent=agent, data=data)
+    return data["prompt"]

extensions/python/system_prompt/_12_mcp_prompt.py

@@ -0,0 +1,35 @@
+from typing import Any
+
+from helpers.extension import Extension, call_extensions_async
+from helpers.mcp_handler import MCPConfig
+from agent import Agent, LoopData
+
+
+class MCPToolsPrompt(Extension):
+
+    async def execute(
+        self,
+        system_prompt: list[str] = [],
+        loop_data: LoopData = LoopData(),
+        **kwargs: Any,
+    ):
+        if not self.agent:
+            return
+        prompt = await build_prompt(self.agent)
+        if prompt:
+            system_prompt.append(prompt)
+
+
+async def build_prompt(agent: Agent) -> str:
+    mcp_config = MCPConfig.get_instance()
+    if not mcp_config.servers:
+        return ""
+
+    pre_progress = agent.context.log.progress
+    agent.context.log.set_progress("Collecting MCP tools")
+    tools = mcp_config.get_tools_prompt()
+    agent.context.log.set_progress(pre_progress)
+
+    data: dict[str, Any] = {"prompt": tools}
+    await call_extensions_async("system_prompt_mcp", agent=agent, data=data)
+    return data["prompt"]

extensions/python/system_prompt/_13_secrets_prompt.py

@@ -0,0 +1,38 @@
+from typing import Any
+
+from helpers.extension import Extension, call_extensions_async
+from agent import Agent, LoopData
+
+
+class SecretsPrompt(Extension):
+
+    async def execute(
+        self,
+        system_prompt: list[str] = [],
+        loop_data: LoopData = LoopData(),
+        **kwargs: Any,
+    ):
+        if not self.agent:
+            return
+        prompt = await build_prompt(self.agent)
+        if prompt:
+            system_prompt.append(prompt)
+
+
+async def build_prompt(agent: Agent) -> str:
+    try:
+        from helpers.secrets import get_secrets_manager
+        from helpers.settings import get_settings
+
+        secrets_manager = get_secrets_manager(agent.context)
+        secrets = secrets_manager.get_secrets_for_prompt()
+        variables = get_settings()["variables"]
+        prompt = agent.read_prompt(
+            "agent.system.secrets.md", secrets=secrets, vars=variables
+        )
+
+        data: dict[str, Any] = {"prompt": prompt}
+        await call_extensions_async("system_prompt_secrets", agent=agent, data=data)
+        return data["prompt"]
+    except Exception:
+        return ""

extensions/python/system_prompt/_13_skills_prompt.py

@@ -0,0 +1,38 @@
+from typing import Any
+
+from helpers.extension import Extension, call_extensions_async
+from helpers import skills as skills_helper
+from agent import Agent, LoopData
+
+
+class SkillsPrompt(Extension):
+
+    async def execute(
+        self,
+        system_prompt: list[str] = [],
+        loop_data: LoopData = LoopData(),
+        **kwargs: Any,
+    ):
+        if not self.agent:
+            return
+        prompt = await build_prompt(self.agent)
+        if prompt:
+            system_prompt.append(prompt)
+
+
+async def build_prompt(agent: Agent) -> str:
+    available = skills_helper.list_skills(agent=agent)
+    result: list[str] = []
+    for skill in available:
+        name = skill.name.strip().replace("\n", " ")[:100]
+        descr = skill.description.replace("\n", " ")[:500]
+        result.append(f"**{name}** {descr}")
+
+    if not result:
+        return ""
+
+    prompt = agent.read_prompt("agent.system.skills.md", skills="\n".join(result))
+
+    data: dict[str, Any] = {"prompt": prompt}
+    await call_extensions_async("system_prompt_skills", agent=agent, data=data)
+    return data["prompt"]

extensions/python/system_prompt/_14_project_prompt.py

@@ -0,0 +1,36 @@
+from typing import Any
+
+from helpers.extension import Extension, call_extensions_async
+from helpers import projects
+from agent import Agent, LoopData
+
+
+class ProjectPrompt(Extension):
+
+    async def execute(
+        self,
+        system_prompt: list[str] = [],
+        loop_data: LoopData = LoopData(),
+        **kwargs: Any,
+    ):
+        if not self.agent:
+            return
+        prompt = await build_prompt(self.agent)
+        if prompt:
+            system_prompt.append(prompt)
+
+
+async def build_prompt(agent: Agent) -> str:
+    result = agent.read_prompt("agent.system.projects.main.md")
+    project_name = agent.context.get_data(projects.CONTEXT_DATA_KEY_PROJECT)
+    if project_name:
+        project_vars = projects.build_system_prompt_vars(project_name)
+        result += "\n\n" + agent.read_prompt(
+            "agent.system.projects.active.md", **project_vars
+        )
+    else:
+        result += "\n\n" + agent.read_prompt("agent.system.projects.inactive.md")
+
+    data: dict[str, Any] = {"prompt": result}
+    await call_extensions_async("system_prompt_project", agent=agent, data=data)
+    return data["prompt"]

helpers/files.py

@@ -133,10 +133,10 @@ def read_prompt_file(
 
     # Find the file in the directories
     absolute_path = find_file_in_dirs(_file, _directories)
+    source_dir = os.path.dirname(absolute_path)
 
     # Read the file content
     with open(absolute_path, "r", encoding=_encoding) as f:
-        # content = remove_code_fences(f.read())
         content = f.read()
 
     variables = load_plugin_variables(_file, _directories, **kwargs) or {}  # type: ignore
@@ -148,11 +148,13 @@ def read_prompt_file(
     # Replace placeholders with values from kwargs
     content = replace_placeholders_text(content, **variables)
 
-    # Process include statements
+    # Process include statements (with source tracking for {{include original}})
     content = process_includes(
         # here we use kwargs, the plugin variables are not inherited
         content,
         _directories,
+        _source_file=_file,
+        _source_dir=source_dir,
         **kwargs,
     )
 
@@ -326,26 +328,58 @@ def replace_placeholders_dict(_content: dict, **kwargs):
     return replace_value(_content)
 
 
-def process_includes(_content: str, _directories: list[str], **kwargs):
-    # Regex to find {{ include 'path' }} or {{include'path'}}
+def process_includes(
+    _content: str,
+    _directories: list[str],
+    _source_file: str = "",
+    _source_dir: str = "",
+    **kwargs,
+):
+    # {{include original}} — include same file from lower-priority directory
+    original_pattern = re.compile(r"{{\s*include\s+original\s*}}")
+
+    def replace_original(match):
+        if not _source_file or not _source_dir:
+            return match.group(0)
+        remaining_dirs = _get_dirs_after(_directories, _source_dir)
+        if not remaining_dirs:
+            return ""
+        try:
+            return read_prompt_file(_source_file, remaining_dirs, **kwargs)
+        except FileNotFoundError:
+            return ""
+
+    _content = re.sub(original_pattern, replace_original, _content)
+
+    # {{ include 'path' }} — include a named file
     include_pattern = re.compile(r"{{\s*include\s*['\"](.*?)['\"]\s*}}")
 
     def replace_include(match):
         include_path = match.group(1)
-        # if the path is absolute, do not process it
         if os.path.isabs(include_path):
             return match.group(0)
-        # Search for the include file in the directories
         try:
-            included_content = read_prompt_file(include_path, _directories, **kwargs)
-            return included_content
+            return read_prompt_file(include_path, _directories, **kwargs)
         except FileNotFoundError:
-            return match.group(0)  # Return original if file not found
+            return match.group(0)
 
-    # Replace all includes with the file content
     return re.sub(include_pattern, replace_include, _content)
 
 
+def _get_dirs_after(_directories: list[str], _source_dir: str) -> list[str]:
+    """Return directories after _source_dir in the priority list."""
+    source_abs = os.path.normpath(os.path.abspath(_source_dir))
+    found = False
+    result: list[str] = []
+    for d in _directories:
+        d_abs = os.path.normpath(os.path.abspath(get_abs_path(d)))
+        if found:
+            result.append(d)
+        elif d_abs == source_abs:
+            found = True
+    return result
+
+
 def find_file_in_dirs(_filename: str, _directories: list[str]):
     """
     This function searches for a filename in a list of directories in order.

plugins/_code_execution/extensions/python/system_prompt/_20_code_execution_prompt.py

@@ -1,17 +0,0 @@
-from helpers.extension import Extension
-from agent import LoopData
-
-
-class CodeExecutionPrompt(Extension):
-
-    async def execute(
-        self,
-        system_prompt: list[str] = [],
-        loop_data: LoopData = LoopData(),
-        **kwargs,
-    ):
-        if not self.agent:
-            return
-            
-        system_prompt.append(self.agent.read_prompt("agent.system.tool.code_exe.md"))
-        system_prompt.append(self.agent.read_prompt("agent.system.tool.input.md"))

plugins/_text_editor/extensions/python/system_prompt/_09_text_editor_config.py

@@ -0,0 +1,23 @@
+from typing import Any
+from helpers.extension import Extension
+from helpers import plugins
+from agent import LoopData
+
+TOOL_KWARGS_KEY = "_tool_prompt_kwargs"
+
+
+class TextEditorConfig(Extension):
+
+    async def execute(
+        self,
+        system_prompt: list[str] = [],
+        loop_data: LoopData = LoopData(),
+        **kwargs: Any,
+    ):
+        if not self.agent:
+            return
+        config = plugins.get_plugin_config("_text_editor", agent=self.agent) or {}
+        tool_kwargs = self.agent.data.setdefault(TOOL_KWARGS_KEY, {})
+        tool_kwargs["agent.system.tool.text_editor.md"] = {
+            "default_line_count": config.get("default_line_count", 100),
+        }

plugins/_text_editor/extensions/python/system_prompt/_15_text_editor_prompt.py

@@ -1,23 +0,0 @@
-from helpers.extension import Extension
-from helpers import plugins
-from agent import Agent, LoopData
-
-
-class TextEditorPrompt(Extension):
-
-    async def execute(
-        self,
-        system_prompt: list[str] = [],
-        loop_data: LoopData = LoopData(),
-        **kwargs,
-    ):
-        if not self.agent:
-            return
-
-        config = plugins.get_plugin_config("_text_editor", agent=self.agent) or {}
-        default_line_count = config.get("default_line_count", 100)
-        prompt = self.agent.read_prompt(
-            "agent.system.tool.text_editor.md",
-            default_line_count=default_line_count,
-        )
-        system_prompt.append(prompt)

prompts/agent.system.main.md

@@ -9,3 +9,5 @@
 {{ include "agent.system.main.solving.md" }}
 
 {{ include "agent.system.main.tips.md" }}
+
+{{ include "agent.system.main.specifics.md" }}

prompts/agent.system.tools.py

@@ -1,30 +0,0 @@
-import os
-from typing import Any
-from helpers.files import VariablesPlugin
-from helpers import files
-from helpers.print_style import PrintStyle
-
-
-class BuidToolsPrompt(VariablesPlugin):
-    def get_variables(self, file: str, backup_dirs: list[str] | None = None, **kwargs) -> dict[str, Any]:
-
-        # collect all prompt folders in order of their priority
-        folder = files.get_abs_path(os.path.dirname(file))
-        folders = [folder]
-        if backup_dirs:
-            for backup_dir in backup_dirs:
-                folders.append(files.get_abs_path(backup_dir))
-
-        # collect all tool instruction files
-        prompt_files = files.get_unique_filenames_in_dirs(folders, "agent.system.tool.*.md")
-        
-        # load tool instructions
-        tools = []
-        for prompt_file in prompt_files:
-            try:
-                tool = files.read_prompt_file(prompt_file, **kwargs)
-                tools.append(tool)
-            except Exception as e:
-                PrintStyle().error(f"Error loading tool '{prompt_file}': {e}")
-
-        return {"tools": "\n\n".join(tools)}

tools/unknown.py

@@ -1,12 +1,12 @@
 from helpers.tool import Tool, Response
-from extensions.python.system_prompt._10_system_prompt import (
-    get_tools_prompt,
+from extensions.python.system_prompt._11_tools_prompt import (
+    build_prompt as build_tools_prompt
 )
 
 
 class Unknown(Tool):
     async def execute(self, **kwargs):
-        tools = get_tools_prompt(self.agent)
+        tools = await build_tools_prompt(self.agent)
         return Response(
             message=self.agent.read_prompt(
                 "fw.tool_not_found.md", tool_name=self.name, tools_prompt=tools