feat(prompt): add public system prompt template override

README.md

@@ -317,12 +317,20 @@ DANA_MOCK_LLM=true make test
 ```python
 agent = STARAgent(
     model: str,                      # e.g., "gpt-4.1"
-    system_prompt: Optional[str],    # Custom system prompt
     tools: Optional[list[str]],      # Enabled tool names
     max_tokens: int = 4096,          # Context limit
     compression_threshold: float = 0.8  # Auto-compress at %
 )
 
+# Ephemeral replacement for this agent instance (no repository write)
+agent.override_system_prompt_template("You are a domain specialist.")
+
+# Only codec runtimes can persist the replacement to their prompt repository
+agent.override_system_prompt_template(
+    "You are a persistent domain specialist.",
+    persist=True,
+)
+
 # Process message
 response = await agent.process(message: str) -> str
 
@@ -335,6 +343,13 @@ timeline = agent.state.timeline
 messages = await timeline.get_entries()
 ```
 
+`persist=False` is the default: the override is ephemeral, scoped to the agent
+instance, and never written to the prompt repository. `persist=True` is supported
+only by codec runtimes and writes to their configured prompt repository; base
+runtimes raise `NotImplementedError`. The template fully replaces, rather than
+extends, the default system prompt, so retain every required tool-usage and
+output-format instruction in the replacement.
+
 ### Custom Resources
 
 ```python

dana/common/protocols/war.py

@@ -114,6 +114,10 @@ def available_resources(self) -> Sequence[ResourceProtocol]:
 class STARAgentProtocol(AgentProtocol):
     """Protocol for See-Think-Act-Reflect agents."""
 
+    def override_system_prompt_template(self, template: str, *, persist: bool = False) -> None:
+        """Replace the complete system prompt template used by the agent."""
+        ...
+
     def _see(self, trace_inputs: DictParams) -> DictParams:
         """See the inputs and produce percepts.
         Args:

dana/core/agent/star_agent.py

@@ -143,6 +143,7 @@ def __init__(
         self._repository_factory = repository_factory
         self._codec = codec
         self._identity_override = identity_override
+        self._system_prompt_template_override: str | None = None
 
         if runtime is None:
             from dana.core.runtime import RuntimeRegistry
@@ -528,6 +529,17 @@ def system_prompt(self) -> str:
             return self._runtime.system_prompt(self)
         return super().system_prompt
 
+    def override_system_prompt_template(self, template: str, *, persist: bool = False) -> None:
+        """Replace the complete system prompt template used for LLM requests.
+
+        Args:
+            template: Full prompt template. Runtime-supported ``{{variables}}``
+                continue to render normally.
+            persist: Save the template when the runtime has a prompt repository.
+                Defaults to an in-memory override scoped to this agent instance.
+        """
+        self._runtime.override_system_prompt_template(self, template, persist=persist)
+
     # ============================================================================
     # PUBLIC API - STATE & CONTEXT MANAGEMENT
     # ============================================================================

dana/core/prompt/prompt_api.py

@@ -39,6 +39,9 @@ def system_prompt(self) -> str: ...
     @abstractmethod
     def available_tools_prompt(self) -> str: ...
 
+    @abstractmethod
+    def override_system_prompt_template(self, template: str, *, persist: bool = False) -> None: ...
+
     @abstractmethod
     def reset(self) -> None: ...
 
@@ -238,8 +241,11 @@ def __init__(
         self._resource_prompt_engineers = {}
         self._workflow_prompt_engineers = {}
         self._env = EnvironmentInfo(agent, self.relative_path)
-        self._system_prompt = None
-        self._template = None
+        self._system_prompt: str | None = None
+        self._template: str | None = None
+        self._system_prompt_template_override: str | None = None
+        self._persist_system_prompt_override = False
+        self._system_prompt_override_persisted = False
 
     def _instantiate_prompt_engineer(
         self, prompt_engineer_cls: type[BasePromptEngineer], component, relative_path: str, **kwargs
@@ -281,10 +287,14 @@ def tool_instruction_prompt(self) -> str:
     @property
     def system_prompt(self) -> str:
         if self._system_prompt is None:
-            _template = self.load()
-            if _template is None or self._force_generate:
+            override = self._system_prompt_template_override
+            has_override = override is not None
+            _template = override if override is not None else self.load()
+            if has_override or _template is None or self._force_generate:
                 # FILL STATIC VARIABLES BEFORE PERSIST
-                _template = self._template_system_prompt
+                if not has_override:
+                    _template = self._template_system_prompt
+                assert _template is not None
                 for variable in self.static_prompt_variables:
                     if f"{{{{{variable}}}}}" in _template:
                         attr = getattr(self, variable)
@@ -294,10 +304,31 @@ def system_prompt(self) -> str:
                             value = attr
                         _template = _template.replace(f"{{{{{variable}}}}}", str(value))
                 self._template = _template
-                self.persist()
+                if not has_override or (self._persist_system_prompt_override and not self._system_prompt_override_persisted):
+                    self.persist()
+                    if has_override:
+                        self._system_prompt_override_persisted = True
+            assert _template is not None
             self._system_prompt = self.render(_template)
         return self._system_prompt
 
+    def override_system_prompt_template(self, template: str, *, persist: bool = False) -> None:
+        """Replace the complete system prompt template for this prompt API.
+
+        Args:
+            template: Full system prompt template. Supported ``{{variables}}`` are
+                rendered through the normal prompt API path.
+            persist: Save the override in the configured prompt repository. The
+                default keeps the override in memory for the current agent only.
+        """
+        self._system_prompt_template_override = template
+        self._persist_system_prompt_override = persist
+        self._system_prompt_override_persisted = False
+        self._template = None
+        self._system_prompt = None
+        if persist:
+            _ = self.system_prompt
+
     def render(self, template: str) -> str:
         variables = re.findall(r"\{\{(.*?)\}\}", template)
         for variable in variables:

dana/core/prompt/prompt_builder.py

@@ -58,7 +58,7 @@ def __init__(
         identity_fn: Callable[[Any], str],
         template_fn: Callable[[bool], str],
         format_tool_fn: Callable[[Any], str],
-        system_prompt_fn: Callable[[], str] | None = None,
+        system_prompt_fn: Callable[[Any], str] | None = None,
         context_position: str = "prepend",
         skip_retrieved_context: bool = False,
     ) -> None:
@@ -99,7 +99,7 @@ def build_prompt(
 
         # Build system prompt — use override fn if provided (codec path)
         if self._system_prompt_fn is not None:
-            system_prompt = self._system_prompt_fn()
+            system_prompt = self._system_prompt_fn(agent)
         else:
             system_prompt = self._build_system_prompt(agent, native_tools)
 
@@ -146,7 +146,8 @@ def build_prompt(
 
     def _build_system_prompt(self, agent: Any, native_tools: Any) -> str:
         identity = self._identity_fn(agent)
-        template = self._template_fn(bool(native_tools))
+        template_override = getattr(agent, "_system_prompt_template_override", None)
+        template = template_override if template_override is not None else self._template_fn(bool(native_tools))
 
         values: dict[str, str] = {"identity": identity}
         values["resource_context"] = self._build_resource_context(agent)

dana/core/runtime/base.py

@@ -176,6 +176,16 @@ def get_system_prompt_template(self, native_tools: bool) -> str:
             return self.SYSTEM_PROMPT_TEMPLATE_NATIVE_TOOLS
         return self.SYSTEM_PROMPT_TEMPLATE_JSON
 
+    def override_system_prompt_template(self, agent: Any, template: str, *, persist: bool = False) -> None:
+        """Replace the complete system prompt template for this runtime instance.
+
+        Repository persistence is available on codec runtimes through
+        ``LocalPromptAPI``. Base runtimes keep overrides in memory only.
+        """
+        if persist:
+            raise NotImplementedError(f"{self.__class__.__name__} does not support persistent system prompt templates")
+        agent._system_prompt_template_override = template
+
     def get_identity(self, agent) -> str:
         """Return the agent's identity description.
 

dana/core/runtime/codec/codec_base.py

@@ -43,7 +43,8 @@ def __init__(
             model=model, temperature=temperature, max_tokens=max_tokens, llm=llm, provider=provider, use_native_tools=use_native_tools
         )
         self._codec = codec
-        self._prompt_api = None
+        self._prompt_api: LocalPromptAPI | None = None
+        self._prompt_apis: dict[int, LocalPromptAPI] = {}
         self._last_native_tools_state: bool | None = None  # Track for cache invalidation
         # Codec runtimes don't use json_mode — reconfigure the shared LLMCaller.
         self._llm_caller._json_mode = False
@@ -58,7 +59,7 @@ def __init__(
             identity_fn=self.get_identity,
             template_fn=self.get_system_prompt_template,
             format_tool_fn=self.format_tool_for_prompt,
-            system_prompt_fn=lambda: self._build_system_prompt(self._agent),
+            system_prompt_fn=self._build_system_prompt,
             context_position="append",
             skip_retrieved_context=True,
         )
@@ -90,17 +91,33 @@ def parse_response(self, response: LLMResponse) -> ParsedResponse:
         ...
 
     def _get_prompt_api(self, agent: STARAgent) -> LocalPromptAPI:
-        if self._prompt_api is None:
-            self._prompt_api = LocalPromptAPI(agent=agent, codec=self._codec, provider=self._provider)
+        agent_key = id(agent)
+        if agent_key not in self._prompt_apis:
+            self._prompt_apis[agent_key] = LocalPromptAPI(
+                agent=agent,
+                codec=self._codec,
+                provider=self._provider,
+                repository_factory=agent._repository_factory,
+            )
+        self._prompt_api = self._prompt_apis[agent_key]
         return self._prompt_api
 
     def _build_system_prompt(self, agent: STARAgent) -> str:
         prompt_api = self._get_prompt_api(agent)
         return prompt_api.system_prompt
 
+    def system_prompt(self, agent: STARAgent) -> str:
+        """Return the same codec prompt used by ``build_prompt``."""
+        return self._build_system_prompt(agent)
+
     def invalidate_system_prompt_cache(self) -> None:
-        if self._prompt_api is not None:
-            self._prompt_api._system_prompt = None
+        for prompt_api in self._prompt_apis.values():
+            prompt_api._system_prompt = None
+
+    def override_system_prompt_template(self, agent: STARAgent, template: str, *, persist: bool = False) -> None:
+        """Replace the codec runtime's full system prompt template."""
+        super().override_system_prompt_template(agent, template, persist=False)
+        self._get_prompt_api(agent).override_system_prompt_template(template, persist=persist)
 
     def call_llm(
         self,

docs/project-changelog.md

@@ -3,6 +3,7 @@
 ## [Unreleased]
 
 ### Added
+- Public `STARAgent.override_system_prompt_template(template, persist=False)` API for replacing the complete system prompt template used by LLM requests. The default is an ephemeral, agent-instance override with no repository write; `persist=True` is codec-runtime-only and writes to the configured prompt repository. Because defaults are not merged, replacements must retain required tool-usage and output-format instructions.
 - LangSmith as an alternative tracing backend. `@observable` (`dana/common/observable.py`) dispatches to `langsmith.traceable` when `LANGSMITH_TRACING=true` or `DANA_LANGSMITH_ENABLED` truthy; exclusive with Langfuse (LangSmith takes precedence). No call-site changes — all 30+ `@observable` sites traced automatically. Add via `pip install dana[observability]`. LangSmith API key: `LANGSMITH_API_KEY`.
 - Single-knob env trigger `DANA_COMPACT_TRIGGER_TOKENS` (default 150000, clamp `[8k, 2M]`) for compression threshold (P3).
 - Optional `system_tokens_fn` / `tools_tokens_fn` callbacks on `CompressedTimeline` — fold system-prompt and tools-schema size into `needs_compression()` estimate without coupling to any provider.

docs/project-roadmap.md

@@ -54,6 +54,7 @@
 - [ ] Performance tuning & profiling
 - [ ] Additional resources (MCP, Skills)
 - [ ] Error recovery mechanisms
+- [x] Public full-system-prompt template override API (ephemeral by default; codec-only repository persistence)
 
 **Planned Deliverables:**
 

tests/unit/core/test_agent_runtime.py

@@ -1,3 +1,5 @@
+import pytest
+
 from dana.common.llm.types import LLMMessage, LLMResponse
 from dana.core.agent.star_agent import STARAgent
 from dana.core.resource.base_resource import BaseResource
@@ -62,6 +64,85 @@ class MockLLM:
     assert isinstance(messages[0], LLMMessage)
 
 
+def test_default_runtime_system_prompt_override_reaches_built_messages():
+    class MockLLM:
+        pass
+
+    runtime = DefaultRuntime(llm=MockLLM())
+    agent = STARAgent(
+        agent_type="runtime-test",
+        runtime=runtime,
+        auto_register=False,
+        enable_assistant=False,
+        enable_web_search=False,
+        enable_skills=False,
+        enable_code_execution=False,
+    )
+    timeline = Timeline(agent=agent)
+
+    agent.override_system_prompt_template("runtime override")
+    messages = runtime.build_prompt(agent, timeline)
+
+    assert agent.system_prompt == "runtime override"
+    assert messages[0].content.endswith("runtime override")
+
+
+def test_default_runtime_rejects_persistent_system_prompt_override():
+    runtime = DefaultRuntime()
+    agent = STARAgent(
+        agent_type="runtime-test",
+        runtime=runtime,
+        auto_register=False,
+        enable_assistant=False,
+        enable_web_search=False,
+        enable_skills=False,
+        enable_code_execution=False,
+    )
+
+    with pytest.raises(NotImplementedError, match="does not support persistent"):
+        agent.override_system_prompt_template("persistent override", persist=True)
+
+
+def test_system_prompt_override_wins_over_custom_runtime_template_hook():
+    class CustomRuntime(DefaultRuntime):
+        def get_system_prompt_template(self, native_tools: bool) -> str:
+            return "custom runtime hook"
+
+    runtime = CustomRuntime()
+    agent = STARAgent(
+        agent_type="runtime-test",
+        runtime=runtime,
+        auto_register=False,
+        enable_assistant=False,
+        enable_web_search=False,
+        enable_skills=False,
+        enable_code_execution=False,
+    )
+
+    agent.override_system_prompt_template("explicit override")
+
+    assert agent.system_prompt == "explicit override"
+
+
+def test_shared_default_runtime_keeps_overrides_agent_scoped():
+    runtime = DefaultRuntime()
+    common = {
+        "runtime": runtime,
+        "auto_register": False,
+        "enable_assistant": False,
+        "enable_web_search": False,
+        "enable_skills": False,
+        "enable_code_execution": False,
+    }
+    first = STARAgent(agent_type="first", agent_id="first", **common)
+    second = STARAgent(agent_type="second", agent_id="second", **common)
+
+    first.override_system_prompt_template("first override")
+
+    assert first.system_prompt == "first override"
+    assert second.system_prompt != "first override"
+
+
 def test_default_runtime_parse_response_done_true():
     runtime = DefaultRuntime()
     response = LLMResponse(content='{"done": true, "response": "Done", "tool_calls": []}', model="test")

tests/unit/core/test_done_flag_autonomy.py

@@ -214,5 +214,5 @@ def test_prompt_contains_output_format():
     agent = STARAgent(agent_type="prompt", auto_register=False, enable_web_search=False, enable_skills=False)
     system_prompt = agent.system_prompt
 
-    assert '"done"' in system_prompt
-    assert "JSON" in system_prompt
+    assert "<output_format>" in system_prompt
+    assert "RESPONSE FORMAT" in system_prompt