feat(prompts): add operator overlays for core system prompts (#5794)

* feat(prompts): add operator overlays for core system prompts

* test(prompts): isolate SOUL forwarding test from operator config

* fix(prompts): wrap complete subagent system message with overlay

* fix(subagents): serialize prompt overlays in durable batches

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -112,6 +112,10 @@ That prompt is intended for coding agents. It tells the agent to clone the repo
 
 ### Configuration
 
+Operators can extend lead-agent, subagent, and DeerMem extraction prompts with
+literal prepend/append configuration without editing source templates. See
+[prompt overlays](backend/docs/CONFIGURATION.md#prompt-overlays).
+
 Optional per-model [`request_admission`](backend/docs/CONFIGURATION.md#model-request-admission)
 paces requests to help stay within provider request-per-minute limits.
 It is disabled by default; see the linked guide to enable it.

backend/docs/CONFIGURATION.md

@@ -2,6 +2,54 @@
 
 This guide explains how to configure DeerFlow for your environment.
 
+## Prompt overlays
+
+Operators can add instructions around existing system prompts in `config.yaml`.
+Empty or omitted extensions preserve the original prompt exactly. The built-in
+instructions remain present; extensions do not change tool authorization,
+memory admission gates, or runtime limits.
+
+```yaml
+lead_prompt_overlay:
+  prepend: "Use our organization's terminology in reports."
+  append: "Conclude with decisions and unresolved questions."
+subagents:
+  agents:
+    general-purpose:
+      prompt_overlay:
+        append: "Include evidence for research claims."
+    bash:
+      prompt_overlay:
+        prepend: "Prefer reproducible, non-interactive commands."
+memory:
+  backend_config:
+    prompt_prepend: "Prefer concise summaries of lasting preferences."
+    prompt_append: "Preserve explicit corrections without redundant wording."
+```
+
+Merge these into existing sections rather than duplicating YAML keys. Extensions
+are joined with two newlines around the assembled system text. They are literal:
+`{conversation}` and JSON braces are not interpolated. The configuration loader's
+existing `$ENV_VAR` resolution still applies to values beginning with `$`.
+Use YAML block scalars for multiline instructions.
+
+These are trusted operator instructions, never request/body context or per-user
+agent data. Lead extensions apply to the assembled default or custom lead agent;
+its SOUL, skills and runtime guidance remain intact. Subagent extensions also
+support configured custom subagents and never mutate the shared built-in
+registry. Explicit application snapshots remain isolated; subsequent assemblies
+use new settings, while already-built graphs retain their prompt.
+
+DeerMem's current `memory_update` chat prompt handles both summary updates and
+fact extraction; the historical `FACT_EXTRACTION_PROMPT` constant is not a
+separate live extraction call. Its extensions wrap the first system message
+after template rendering, leaving memory and conversation data in their original
+human message. They also work with `prompts_dir` templates, which must include
+a system message when overlays are used. Restart the memory backend/Gateway after
+changing its configuration. Other memory backends do not consume these
+DeerMem-specific fields. Existing external memory templates remain the mechanism
+for complete template replacement; this feature adds no editing endpoint or UI.
+
 ## Model request admission
 
 For request-per-minute limits, opt into pacing on each relevant `models[]`

backend/packages/harness/deerflow/agents/AGENTS.md

@@ -1,5 +1,10 @@
 ### Agent System
 
+`AppConfig.lead_prompt_overlay` wraps the fully rendered lead system prompt with
+literal operator prepend/append text. Do not template-format these extensions
+or source them from run context. The graph and assembly descriptor must see the
+same resulting prompt. Empty overlays preserve bytes.
+
 **Lead Agent** (`packages/harness/deerflow/agents/lead_agent/agent.py`):
 - `make_lead_agent(config: RunnableConfig)` is the published `langgraph.json`
   entry point; preserve its signature and bare-graph return type.

backend/packages/harness/deerflow/agents/lead_agent/prompt.py

@@ -1113,7 +1113,7 @@ def apply_prompt_template(
     # Memory and current date are injected per-turn via DynamicContextMiddleware
     # as a <system-reminder> in the first HumanMessage, keeping this prompt
     # identical across users and sessions for maximum prefix-cache reuse.
-    return SYSTEM_PROMPT_TEMPLATE.format(
+    rendered_prompt = SYSTEM_PROMPT_TEMPLATE.format(
         interaction_thinking_guidance=interaction_policy.thinking_guidance,
         clarification_system=interaction_policy.clarification_system,
         clarification_reminder=interaction_policy.clarification_reminder,
@@ -1130,3 +1130,9 @@ def apply_prompt_template(
         subagent_thinking=subagent_thinking,
         acp_section=acp_and_mounts_section,
     )
+    if app_config is None:
+        from deerflow.config import get_app_config
+
+        app_config = get_app_config()
+    overlay = getattr(app_config, "lead_prompt_overlay", None)
+    return overlay.apply(rendered_prompt) if overlay is not None else rendered_prompt

backend/packages/harness/deerflow/agents/memory/AGENTS.md

@@ -1,5 +1,11 @@
 ### Memory System
 
+DeerMem `prompt_prepend`/`prompt_append` wrap the first system message from
+`memory_update` after formatting, without touching its human data message or
+cached templates. This is the live summary/fact extraction path; do not wire
+extensions only to the legacy `FACT_EXTRACTION_PROMPT` alias. Settings belong
+to the constructed backend and require backend recreation to change.
+
 This directory owns memory capture, storage, retrieval, prompt injection, and model-driven memory tools.
 
 #### Main components

backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/config.py

@@ -46,6 +46,9 @@ class DeerMemModelConfig(BaseModel):
 class DeerMemConfig(BaseModel):
     """DeerMem-private configuration (self-contained, host-agnostic)."""
 
+    prompt_prepend: str = Field(default="", strict=True, description="Literal operator instructions prepended to the memory-update system message")
+    prompt_append: str = Field(default="", strict=True, description="Literal operator instructions appended to the memory-update system message")
+
     # ── Storage ──────────────────────────────────────────────────────────
     storage_path: str = Field(
         default="",

backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/updater.py

@@ -1387,6 +1387,16 @@ class MemoryUpdater:
             "consolidation_section": consolidation_section,
         }
         prompt = load_prompt_messages("memory_update", variables, agent_name=agent_name, prompts_dir=self._prompts_dir)
+        if config.prompt_prepend or config.prompt_append:
+            # Extend trusted instructions only, after formatting the original
+            # template. Conversation/memory data stays in its human message.
+            for index, message in enumerate(prompt):
+                if message.type == "system":
+                    content = "\n\n".join(part for part in (config.prompt_prepend, message.content, config.prompt_append) if part)
+                    prompt[index] = message.model_copy(update={"content": content})
+                    break
+            else:
+                raise ValueError("Memory prompt overlays require a system message in the configured chat template")
         return current_memory, prompt
 
     def _has_manual_facts(self, memory: dict[str, Any]) -> bool:

backend/packages/harness/deerflow/config/AGENTS.md

@@ -1,5 +1,11 @@
 ### Configuration System
 
+Operator prompt overlays: `lead_prompt_overlay` on AppConfig and
+`subagents.agents.<name>.prompt_overlay` accept literal `prepend`/`append` strings.
+The per-assembly snapshot owns these settings; no run-context override exists.
+DeerMem owns its separate `memory.backend_config.prompt_prepend`/`prompt_append`
+fields so the memory package remains host-agnostic.
+
 Custom Agent `AgentConfig.display_name` is an optional, whitespace-trimmed Unicode
 label of at most 100 Unicode code points. C0/C1 controls and bidirectional
 formatting controls (U+202A–U+202E, U+2066–U+2069) are rejected before trimming.

backend/packages/harness/deerflow/config/app_config.py

@@ -31,6 +31,7 @@ from deerflow.config.memory_config import MemoryConfig, load_memory_config_from_
 from deerflow.config.model_config import ModelConfig
 from deerflow.config.pii_redaction_config import PiiRedactionConfig
 from deerflow.config.projects_config import ProjectsConfig
+from deerflow.config.prompt_overlay import PromptOverlay
 from deerflow.config.read_before_write_config import ReadBeforeWriteConfig
 from deerflow.config.reload_boundary import format_field_description
 from deerflow.config.run_events_config import RunEventsConfig
@@ -190,6 +191,8 @@ def apply_logging_level(name: str | None) -> None:
 class AppConfig(BaseModel):
     """Config for the DeerFlow application"""
 
+    lead_prompt_overlay: PromptOverlay = Field(default_factory=PromptOverlay, description="Operator-owned literal prepend/append around the assembled lead-agent system prompt")
+
     log_level: str = Field(
         default="info",
         description=format_field_description(

backend/packages/harness/deerflow/config/prompt_overlay.py

@@ -0,0 +1,13 @@
+"""Literal operator-owned extensions around an assembled system prompt."""
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class PromptOverlay(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    prepend: str = Field(default="", strict=True)
+    append: str = Field(default="", strict=True)
+
+    def apply(self, prompt: str) -> str:
+        return "\n\n".join(part for part in (self.prepend, prompt, self.append) if part)

backend/packages/harness/deerflow/config/subagents_config.py

@@ -4,6 +4,7 @@ import logging
 
 from pydantic import BaseModel, Field
 
+from deerflow.config.prompt_overlay import PromptOverlay
 from deerflow.config.token_budget_config import TokenBudgetConfig
 
 logger = logging.getLogger(__name__)
@@ -74,6 +75,8 @@ def default_subagent_token_budget(*, summarization_enabled: bool = False) -> Tok
 class SubagentOverrideConfig(BaseModel):
     """Per-agent configuration overrides."""
 
+    prompt_overlay: PromptOverlay = Field(default_factory=PromptOverlay, description="Operator-owned literal extensions around this subagent's system prompt")
+
     timeout_seconds: int | None = Field(
         default=None,
         ge=1,

backend/packages/harness/deerflow/subagents/AGENTS.md

@@ -1,5 +1,9 @@
 ### Subagent System (`packages/harness/deerflow/subagents/`)
 
+Apply each subagent's prompt overlay after assembling its full SystemMessage.
+Registry overrides must not mutate `BUILTIN_SUBAGENTS`.
+Durable batch specs store overlays as JSON and restore them before execution.
+
 **Empty remote artifacts**: Acceptance probes treat GNU `stat -c %F` labels
 `regular file` and `regular empty file` as regular files. Empty files satisfy
 `exists`/`file_written` but fail `non-empty`; symlink leaves, directories, and
@@ -9,19 +13,17 @@ FIFOs remain rejected.
 
 `task` preserves the host incarnation, including null/missing/invalid values; never recapture it.
 
-**Direct runtime shutdown**: `SubagentRuntime.stop()` holds its lifecycle lock until the owned service stop task terminates, then propagates the first caller cancellation with any service failure/cancellation as its cause. The drain is intentionally unbounded: repository awaits and child cleanup must terminate; a timeout must not detach still-owned work. Keep terminal-outcome and repeated-cancellation coverage in `tests/test_subagent_runtime.py`.
+**Direct runtime shutdown**: `SubagentRuntime.stop()` holds its lifecycle lock until service shutdown finishes. Drain without a timeout so owned work cannot detach; propagate the first caller cancellation, chaining any service failure. Cover terminal outcomes and repeated cancellation in `tests/test_subagent_runtime.py`.
 
-**Durable batch acceptance**: `batch_task` normalizes optional per-item criteria
-before persistence (empty becomes null; 20 items × 500 neutralized characters),
-sharing `normalize_acceptance_criteria` with the executor and checker.
-Completed items reuse `acceptance_checks` through `batch_acceptance.py`, with
-owner-scoped thread paths, sandbox authorization and a client lease. Admission
-and checks share `parse_file_criterion` on the effective normalized list. Blocking
-reads drain before release on cancellation, and the batch item lease is renewed
-while checking. The nullable validated verdict survives repository queries and
-JSONL exports independently of execution status. No criteria, checker errors,
-and legacy rows have no verdict; they are not accepted by implication. Failed
-executions are not checked, and acceptance never changes automatic retry policy.
+**Durable batch acceptance**: Normalize optional `batch_task` criteria before
+persistence (empty → null; 20 items × 500 neutralized characters) with shared
+`normalize_acceptance_criteria`. Completed items use `acceptance_checks` through
+`batch_acceptance.py`, with owner-scoped paths, sandbox authorization, and a
+renewed client lease. Admission and checks share `parse_file_criterion`; drain
+blocking reads before releasing the lease on cancellation. Persist nullable
+verdicts through queries and JSONL exports: absent criteria, checker errors,
+legacy rows, and failed executions are never implicitly accepted. Acceptance
+does not change retry policy.
 
 **Built-in Agents**: `general-purpose` (all tools except `task`) and `bash` (command specialist)
 **Registry and managed definitions**: Runtime resolution is built-in → `config.yaml custom_agents` → enabled administrator-managed definitions, followed by explicit `subagents.agents.<name>` overrides. Managed definitions are deployment-wide, persist through the same `agent_storage.backend` selection as Custom Agent definitions, and remain stored but are excluded from runtime when a built-in or later-added config definition owns the same name. The default Lead Agent sees the whole enabled catalog. A Custom Agent's `allowed_subagents` is snapshotted into run metadata (`None` = all, `[]` = hard deny, list = allowlist) and must filter both prompt discovery and `task` execution; never reload caller policy from mutable agent config inside the tool.

backend/packages/harness/deerflow/subagents/batch_service.py

@@ -196,7 +196,12 @@ class SubagentBatchService:
             batch = item["batch"]
             self._item_batches[item_id] = batch["id"]
             spec = batch["execution_spec"]
-            config = SubagentConfig(**spec["subagent_config"])
+            config_data = spec["subagent_config"]
+            if "prompt_overlay" in config_data:
+                from deerflow.config.prompt_overlay import PromptOverlay
+
+                config_data = {**config_data, "prompt_overlay": PromptOverlay.model_validate(config_data["prompt_overlay"])}
+            config = SubagentConfig(**config_data)
             app_config = self._app_config or get_app_config()
             from deerflow.tools import get_available_tools
 

backend/packages/harness/deerflow/subagents/config.py

@@ -3,6 +3,8 @@
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
+from deerflow.config.prompt_overlay import PromptOverlay
+
 if TYPE_CHECKING:
     from deerflow.config.app_config import AppConfig
 
@@ -33,6 +35,7 @@ class SubagentConfig:
             effective limit is the global ``subagents.timeout_seconds`` (default
             1800 = 30 min), layered on by the registry; this 900 only applies
             when no differing global value exists.
+        prompt_overlay: Operator instructions around the complete system message.
     """
 
     name: str
@@ -44,6 +47,7 @@ class SubagentConfig:
     model: str = "inherit"
     max_turns: int = 50
     timeout_seconds: int = 900
+    prompt_overlay: PromptOverlay = field(default_factory=PromptOverlay)
 
 
 def _default_model_name(app_config: "AppConfig") -> str:

backend/packages/harness/deerflow/subagents/executor.py

@@ -1341,7 +1341,7 @@ class SubagentExecutor:
 
         messages: list[Any] = []
         if system_parts:
-            self._assembled_system_prompt = "\n\n".join(system_parts)
+            self._assembled_system_prompt = self.config.prompt_overlay.apply("\n\n".join(system_parts))
             messages.append(SystemMessage(content=self._assembled_system_prompt))
 
         if self.context_snapshot is not None:

backend/packages/harness/deerflow/subagents/registry.py

@@ -138,6 +138,10 @@ def get_subagent_config(name: str, *, app_config: Any | None = None) -> Subagent
     agent_override = subagents_config.agents.get(name)
 
     overrides = {}
+    if agent_override is not None:
+        overlay = agent_override.prompt_overlay
+        if overlay.prepend or overlay.append:
+            overrides["prompt_overlay"] = overlay
 
     # Timeout: per-agent override > global default (builtins only) > config's own value
     if agent_override is not None and agent_override.timeout_seconds is not None:

backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py

@@ -210,7 +210,7 @@ async def batch_task(
     run_id = context.get("run_id")
     submission_key = f"{run_id or thread_id}:{tool_call_id}"
     execution_spec = {
-        "subagent_config": asdict(config),
+        "subagent_config": {**asdict(config), "prompt_overlay": config.prompt_overlay.model_dump()},
         "parent_model": metadata.get("model_name"),
         "tool_groups": metadata.get("tool_groups"),
         "mcp_plugins": metadata.get("mcp_plugins"),

backend/tests/test_batch_acceptance.py

@@ -13,6 +13,7 @@ import pytest_asyncio
 
 from deerflow.config.database_config import DatabaseConfig
 from deerflow.config.paths import Paths
+from deerflow.config.prompt_overlay import PromptOverlay
 from deerflow.config.subagent_batches_config import SubagentBatchesConfig
 from deerflow.config.subagent_runtime_config import SubagentRuntimeConfig
 from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
@@ -104,16 +105,24 @@ async def _execute(env):
 async def test_tool_preserves_per_item_criteria_and_legacy_shape(env, monkeypatch):
     module = importlib.import_module("deerflow.tools.builtins.batch_task_tool")
     monkeypatch.setattr(module, "get_available_subagent_names", lambda **kwargs: ["general-purpose"])
-    monkeypatch.setattr(module, "get_subagent_config", lambda *args, **kwargs: SubagentConfig(name="general-purpose", description="Worker"))
+    monkeypatch.setattr(
+        module,
+        "get_subagent_config",
+        lambda *args, **kwargs: SubagentConfig(name="general-purpose", description="Worker", prompt_overlay=PromptOverlay(prepend="First", append="Last")),
+    )
     tools = {tool.name: tool for tool in bind_batch_tools(env.service)}
     runtime = SimpleNamespace(state={}, context={"thread_id": "thread-1", "user_id": "user-1"}, config={"metadata": {}})
     command = await tools["batch_task"].coroutine(
         runtime=runtime, title="Batch", items=[BatchTaskItem(key="one", prompt="p", acceptance_criteria=["file:../outputs/report.md exists"]), BatchTaskItem(key="two", prompt="q")], subagent_type="general-purpose", tool_call_id="c1"
     )
-    batch_id = command.update["messages"][0].additional_kwargs["subagent_batch_id"]
+    result = command.update["messages"][0]
+    assert "subagent_batch_id" in result.additional_kwargs, result.content
+    batch_id = result.additional_kwargs["subagent_batch_id"]
     items = await env.repo.list_items(batch_id, user_id="user-1")
     assert items[0]["acceptance_criteria"] == ["file:../outputs/report.md exists"]
     assert items[1]["acceptance_criteria"] is None
+    await _execute(env)
+    assert env.calls[0]["config"].prompt_overlay.apply("core") == "First\n\ncore\n\nLast"
 
 
 @pytest.mark.asyncio

backend/tests/test_prompt_overlays.py

@@ -0,0 +1,90 @@
+"""Operator overlays preserve stock prompts and never format user-like braces."""
+
+from unittest.mock import MagicMock
+
+import pytest
+from langchain_core.messages import HumanMessage, SystemMessage
+from pydantic import ValidationError
+
+from deerflow.config.prompt_overlay import PromptOverlay
+
+
+def test_default_overlay_is_byte_identical():
+    assert PromptOverlay().apply(" \nstock\n ") == " \nstock\n "
+
+
+def test_overlay_is_literal_and_keeps_core():
+    assert PromptOverlay(prepend="before {unknown}", append="after {conversation}").apply("core") == "before {unknown}\n\ncore\n\nafter {conversation}"
+
+
+def test_invalid_overlay_rejected():
+    with pytest.raises(ValidationError):
+        PromptOverlay(prepend={"not": "text"})
+
+
+def test_lead_overlay_uses_explicit_snapshot_and_does_not_accumulate(monkeypatch):
+    from deerflow.agents.lead_agent import prompt as module
+    from deerflow.config.app_config import AppConfig
+    from deerflow.config.sandbox_config import SandboxConfig
+
+    config = AppConfig(sandbox=SandboxConfig(use="test"))
+    for helper in ("get_agent_soul", "get_skills_prompt_section", "get_deferred_tools_prompt_section", "_build_acp_section", "_build_custom_mounts_section", "_build_memory_tool_section"):
+        monkeypatch.setattr(module, helper, lambda *args, **kwargs: "")
+    stock = module.apply_prompt_template(app_config=config)
+    configured = config.model_copy(update={"lead_prompt_overlay": PromptOverlay(prepend="规则 {unbound}", append="tail")})
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
+    expected = "规则 {unbound}\n\n" + stock + "\n\ntail"
+    assert module.apply_prompt_template(app_config=configured) == expected
+    assert module.apply_prompt_template(app_config=configured) == expected
+    assert module.apply_prompt_template(app_config=config) == stock
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: configured)
+    assert module.apply_prompt_template() == expected
+
+
+@pytest.mark.parametrize("name", ["general-purpose", "bash"])
+def test_builtin_overlay_does_not_mutate_registry(name):
+    from deerflow.config.subagents_config import SubagentsAppConfig
+    from deerflow.subagents.builtins import BUILTIN_SUBAGENTS
+    from deerflow.subagents.registry import get_subagent_config
+
+    original = BUILTIN_SUBAGENTS[name].system_prompt
+    config = SubagentsAppConfig(agents={name: {"prompt_overlay": {"prepend": "First rule", "append": "Operator rule"}}})
+    first = get_subagent_config(name, app_config=config)
+    second = get_subagent_config(name, app_config=config)
+    assert first.system_prompt == second.system_prompt == original
+    assert first.prompt_overlay.apply("assembled") == "First rule\n\nassembled\n\nOperator rule"
+    assert second.prompt_overlay == first.prompt_overlay
+    assert BUILTIN_SUBAGENTS[name].system_prompt == original
+    assert get_subagent_config(name, app_config=SubagentsAppConfig()).system_prompt == original
+
+
+def test_memory_overlay_only_modifies_system_authority():
+    from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
+    from deerflow.agents.memory.backends.deermem.deermem.core.updater import MemoryUpdater
+
+    storage = MagicMock()
+    base = MemoryUpdater(DeerMemConfig(), storage)
+    configured = MemoryUpdater(DeerMemConfig(prompt_prepend="before {conversation}", prompt_append="after {unknown}"), storage)
+    for updater in (base, configured):
+        updater.get_memory_data = lambda *args, **kwargs: {"facts": []}
+    messages = [HumanMessage(content="Remember my preference for green.")]
+    _, stock = base._prepare_update_prompt(messages, None, frozenset())
+    _, rendered = configured._prepare_update_prompt(messages, None, frozenset())
+    assert isinstance(rendered[0], SystemMessage)
+    assert rendered[0].content == "before {conversation}\n\n" + stock[0].content + "\n\nafter {unknown}"
+    assert rendered[1:] == stock[1:]
+    _, again = configured._prepare_update_prompt(messages, None, frozenset())
+    assert again == rendered
+    _, still_stock = base._prepare_update_prompt(messages, None, frozenset())
+    assert still_stock == stock
+
+
+def test_memory_overlay_requires_system_message(monkeypatch):
+    from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
+    from deerflow.agents.memory.backends.deermem.deermem.core import updater as module
+
+    monkeypatch.setattr(module, "load_prompt_messages", lambda *args, **kwargs: [HumanMessage(content="data")])
+    updater = module.MemoryUpdater(DeerMemConfig(prompt_append="trusted rule"), MagicMock())
+    updater.get_memory_data = lambda *args, **kwargs: {"facts": []}
+    with pytest.raises(ValueError, match="require a system message"):
+        updater._prepare_update_prompt([HumanMessage(content="conversation")], None, frozenset())

backend/tests/test_soul_prompt_injection.py

@@ -13,6 +13,8 @@ personality could otherwise close its tag and forge a framework-trusted
 from __future__ import annotations
 
 from deerflow.agents.lead_agent import prompt as prompt_module
+from deerflow.config.app_config import AppConfig
+from deerflow.config.sandbox_config import SandboxConfig
 
 # A value that breaks out of the <soul> block and forges a framework-reserved
 # block the model would read as trusted context.
@@ -58,8 +60,11 @@ def test_get_agent_soul_forwards_explicit_user_id(monkeypatch) -> None:
     assert "User-scoped soul" in result
 
 
-def test_apply_prompt_template_forwards_user_id_to_agent_soul(monkeypatch) -> None:
+def test_apply_prompt_template_forwards_user_id_to_agent_soul(monkeypatch, tmp_path) -> None:
     captured = {}
+    # Keep this offline even when the developer has an operator config locally.
+    monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(tmp_path / "missing-config.yaml"))
+    app_config = AppConfig(sandbox=SandboxConfig(use="test"))
 
     def fake_get_agent_soul(agent_name, *, user_id=None):
         captured["agent_name"] = agent_name
@@ -76,6 +81,7 @@ def test_apply_prompt_template_forwards_user_id_to_agent_soul(monkeypatch) -> No
     prompt_module.apply_prompt_template(
         agent_name="custom-agent",
         user_id="authenticated-user",
+        app_config=app_config,
     )
 
     assert captured == {

backend/tests/test_subagent_executor.py

@@ -647,6 +647,27 @@ class TestAgentConstruction:
         assert base_config.system_prompt in messages[0].content
         assert isinstance(messages[1], HumanMessage)
 
+    @pytest.mark.anyio
+    async def test_prompt_overlay_surrounds_complete_system_message(self, classes):
+        from langchain_core.messages import SystemMessage
+
+        from deerflow.config.subagents_config import SubagentsAppConfig
+        from deerflow.subagents.registry import get_subagent_config
+
+        overrides = SubagentsAppConfig(agents={"general-purpose": {"prompt_overlay": {"prepend": "Operator first", "append": "Operator last"}}})
+        config = get_subagent_config("general-purpose", app_config=overrides)
+        executor = classes["SubagentExecutor"](config=config, tools=[], thread_id="test-thread")
+
+        state, _tools, _setup = await executor._build_initial_state("Do the task")
+
+        system = state["messages"][0]
+        assert isinstance(system, SystemMessage)
+        assert system.content.startswith("Operator first\n\n")
+        assert system.content.endswith("\n\nOperator last")
+        assert config.system_prompt in system.content
+        assert "report" in system.content[len(config.system_prompt) : -len("Operator last")].lower()
+        assert executor._assembled_system_prompt == system.content
+
     @pytest.mark.anyio
     async def test_build_initial_state_inherits_background_without_execution_evidence(self, classes, base_config):
         from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage

config.example.yaml

@@ -20,7 +20,14 @@
 # ============================================================================
 # Bump this number when the config schema changes.
 # Run `make config-upgrade` to merge new fields into your local config.yaml.
-config_version: 47
+config_version: 48
+
+# Optional operator-owned literal prompt extensions (see CONFIGURATION.md):
+# lead_prompt_overlay:
+#   prepend: "Apply our organization-wide writing conventions."
+#   append: "End reports with a concise decision summary."
+# Subagents: subagents.agents.<name>.prompt_overlay.{prepend,append}
+# DeerMem: memory.backend_config.{prompt_prepend,prompt_append}
 
 # Optional per-model request pacing (inside a models[] entry):
 # request_admission:

deploy/helm/deer-flow/README.md

@@ -135,7 +135,7 @@ they resolve from the `secrets` map):
 
 ```yaml
 config: |
-  config_version: 47
+  config_version: 48
   models:
     - name: gpt-4
       use: langchain_openai:ChatOpenAI

deploy/helm/deer-flow/values.yaml

@@ -249,7 +249,7 @@ ingress:
 # -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
 #    inline literal secret values here. The default enables provisioner sandbox.
 config: |
-  config_version: 47
+  config_version: 48
   log_level: info
   recursion_limit: 100
   max_recursion_limit: 1000