Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions agent/agent_init.py
Original file line number Diff line number Diff line change
Expand Up @@ -1408,6 +1408,8 @@ def _apply_agent_section(agent, _agent_cfg):
"environment_probe", "bot_mode_protocol",
):
setattr(agent, f"_{_key}", bool(_agent_section.get(_key, True)))
from agent.prompt_overrides import normalize_overrides
agent._prompt_overrides = normalize_overrides(_agent_section.get("prompt_overrides", {}))
# Warm the probe (~0.5s of subprocesses) off-thread so the first prompt build finds it cached.
if agent._environment_probe:
with suppress(Exception):
Expand Down
174 changes: 174 additions & 0 deletions agent/prompt_overrides.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,174 @@
"""Declarative, cache-safe overrides for named system-prompt fragments.

Named fragments assembled into the system prompt
carry a stable string *key* (see :data:`FRAGMENT_KEYS`). Users reshape any
of them from ``config.yaml`` under ``agent.prompt_overrides`` without editing
source:

.. code-block:: yaml

agent:
prompt_overrides:
task_completion: {mode: replace, text: "..."}
tool_use_enforcement: {mode: append, text: "..."}
google_operational: {mode: remove}
# shorthand: a bare string means replace
steer_channel: "..."

Design contract — **overrides are pure data, resolved once at prompt-build
time**. There is no callable hook and no conditional logic, by design: the
assembled prompt must stay a deterministic function of (agent, config) so it
is byte-stable across turns and the upstream prefix cache stays warm. Project
context and runtime environment facts are outside this named-fragment surface. With no
overrides configured the output is byte-identical to the un-overridden prompt.

A fragment override only takes effect when that fragment is actually emitted
this session (e.g. ``model_identity`` only ships for Alibaba). ``append``/``prepend`` to a fragment that isn't present this session
silently no-ops — there is nothing to append to.
"""

from __future__ import annotations

import logging
from typing import Any, Dict, Optional

logger = logging.getLogger(__name__)

VALID_MODES = ("replace", "append", "prepend", "remove")

# Canonical registry of every override-addressable fragment key. Keep in sync
# with the ``_fragment(...)`` call sites in ``agent/system_prompt.py``. Surfaced to
# users for discovery (``hermes`` docs / tooling) — an override map keyed by
# string is only usable if the keys are enumerable.
FRAGMENT_KEYS: Dict[str, str] = {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This registry is documented as canonical, but agent/system_prompt.py:230 in this PR still directly appends PARALLEL_TOOL_CALL_GUIDANCE. Add a key and route that block through emit(), or narrow the feature/docs claim so the omission is explicit.

"identity": "Agent identity slot (SOUL.md content, or the default identity).",
"hermes_help": "Pointer to the hermes-agent skill/docs for questions about Hermes itself.",
"task_completion": "Universal 'finish the job' / no-fabrication guidance (all models).",
"parallel_tool_call_guidance": "Universal guidance to batch independent tool calls.",
"tool_guidance": "Composed per-tool behavioral guidance (memory, session_search, skills, kanban).",
"steer_channel": "Note explaining the mid-turn steer channel in tool results.",
"tool_use_enforcement": "Tells the model to actually call tools instead of describing intent.",
"google_operational": "Gemini/Gemma operational directives (absolute paths, parallel calls, etc.).",
"execution_discipline": "GPT/Codex/Grok execution discipline (tool persistence, verification).",
"skills": "Skills system-prompt block (available skills and how to load them).",
"model_identity": "Explicit model-identity line (Alibaba Coding Plan API name workaround).",
"environment_probe": "Local Python/pip/uv/PEP-668 toolchain probe line.",
"active_profile": "Active Hermes profile note and cross-profile write guidance.",
"platform_hints": "Platform-specific operational hints (from PLATFORM_HINTS or a plugin).",
"coding_brief": "Coding posture guidance before project context.",
"coding_workspace": "Pinned workspace snapshot after project context.",
"coding_instructions": "Configured coding instructions after the workspace snapshot.",
}


def normalize_overrides(raw: Any) -> Dict[str, Dict[str, str]]:
"""Validate and normalize ``agent.prompt_overrides`` config into a clean map.

Accepts the loosely-typed YAML value and returns ``{key: {"mode": ...,
"text": ...}}`` containing only well-formed entries for known fragment
keys. Malformed entries are dropped with a warning rather than raising —
a bad override should never block prompt assembly.
"""
if not raw:
return {}
if not isinstance(raw, dict):
logger.warning(
"agent.prompt_overrides must be a mapping of fragment-key -> override; "
"got %s. Ignoring.", type(raw).__name__,
)
return {}

result: Dict[str, Dict[str, str]] = {}
for key, spec in raw.items():
if key not in FRAGMENT_KEYS:
logger.warning(
"agent.prompt_overrides: unknown fragment key %r (ignored). "
"Valid keys: %s", key, ", ".join(sorted(FRAGMENT_KEYS)),
)
continue

# Shorthand: a bare string is a full replacement.
if isinstance(spec, str):
result[key] = {"mode": "replace", "text": spec}
continue

if not isinstance(spec, dict):
logger.warning(
"agent.prompt_overrides[%r] must be a string or mapping; got %s. "
"Ignored.", key, type(spec).__name__,
)
continue

mode = str(spec.get("mode", "replace")).lower().strip()
if mode not in VALID_MODES:
logger.warning(
"agent.prompt_overrides[%r]: invalid mode %r (valid: %s). Ignored.",
key, mode, ", ".join(VALID_MODES),
)
continue

if mode == "remove":
result[key] = {"mode": "remove", "text": ""}
continue

text = spec.get("text", "")
if text is None:
text = ""
if not isinstance(text, str):
logger.warning(
"agent.prompt_overrides[%r].text must be a string; got %s. Ignored.",
key, type(text).__name__,
)
continue
if mode in ("append", "prepend") and not text.strip():
logger.warning(
"agent.prompt_overrides[%r]: %s mode with empty text is a no-op. "
"Ignored.", key, mode,
)
continue
result[key] = {"mode": mode, "text": text}

return result


def apply_fragment_override(
overrides: Optional[Dict[str, Dict[str, str]]],
key: str,
text: Optional[str],
) -> Optional[str]:
"""Apply any configured override for ``key`` to ``text``.

Returns the (possibly transformed) fragment text, or ``None`` when the
fragment should be dropped (``remove`` mode, or an empty result). ``text``
is the default fragment content Hermes would emit with no override.
"""
if not overrides:
return text
spec = overrides.get(key)
if not spec:
return text

mode = spec.get("mode", "replace")
if mode == "remove":
return None

new = spec.get("text", "")
if mode == "replace":
return new
if mode == "append":
if not text or not text.strip():
return text
return f"{text}\n\n{new}"
if mode == "prepend":
if not text or not text.strip():
return text
return f"{new}\n\n{text}"
return text


__all__ = [
"FRAGMENT_KEYS",
"VALID_MODES",
"normalize_overrides",
"apply_fragment_override",
]
68 changes: 50 additions & 18 deletions agent/system_prompt.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@
from utils import is_truthy_value

logger = logging.getLogger(__name__)
_OVERRIDDEN_WORKSPACE_PIN = "<!-- Hermes coding workspace present -->"
_WORKSPACE_PIN_SENTINEL = "pinned coding workspace"
_PLUGIN_SECTION_FRAME_RE = re.compile(
r"^## Plugin Context: (?P<id>[a-z0-9][a-z0-9._-]{0,127})\n<!-- hermes-plugin-section-chars:(?P<chars>[0-9]{1,4}) -->\n\n",
re.MULTILINE,
Expand Down Expand Up @@ -546,35 +548,42 @@ def _identity_parts(agent: Any, ctx_len: Optional[int]) -> Tuple[List[str], bool
Returns ``(parts, soul_loaded)``."""
wants_soul = agent.load_soul_identity or not agent.skip_context_files
_soul_content = _pb.load_soul_md(ctx_len, home_override=_agent_home(agent)) if wants_soul else None
return ([_soul_content], True) if _soul_content else ([DEFAULT_AGENT_IDENTITY], False)
return ([_fragment(agent, "identity", _soul_content)], True) if _soul_content else (
[_fragment(agent, "identity", DEFAULT_AGENT_IDENTITY)], False)


def _fragment(agent: Any, key: str, text: Optional[str]) -> Optional[str]:
"""Apply a configured override when the named fragment is emitted."""
from agent.prompt_overrides import apply_fragment_override
return apply_fragment_override(getattr(agent, "_prompt_overrides", None), key, text)


def _guidance_parts(agent: Any) -> List[str]:
"""Universal + tool-aware + model-gated guidance blocks, each gated by its config.yaml key."""
parts: List[str] = []
if agent.valid_tool_names:
parts += [
text for flag, text in (
("_task_completion_guidance", TASK_COMPLETION_GUIDANCE),
("_parallel_tool_call_guidance", PARALLEL_TOOL_CALL_GUIDANCE),
) if getattr(agent, flag, True)
]
parts.append(_tool_guidance_block(agent)) # None/empty entries are dropped by _join_tier
if getattr(agent, "_task_completion_guidance", True):
parts.append(_fragment(agent, "task_completion", TASK_COMPLETION_GUIDANCE))
if getattr(agent, "_parallel_tool_call_guidance", True):
parts.append(_fragment(agent, "parallel_tool_call_guidance", PARALLEL_TOOL_CALL_GUIDANCE))
guidance = _tool_guidance_block(agent)
if guidance:
parts.append(_fragment(agent, "tool_guidance", guidance))
if not agent.valid_tool_names:
return parts
# Steering only lands inside tool results, so only reachable with tools.
parts.append(STEER_CHANNEL_NOTE)
parts.append(_fragment(agent, "steer_channel", STEER_CHANNEL_NOTE))
# agent.tool_use_enforcement / agent.execution_guidance: "auto" (default)
# matches the hardcoded model lists; true/false force; a list gives custom
# model-name substrings. Execution guidance is an independent gate so
# DeepSeek/Kimi/Qwen-class models get it even with enforcement off.
if _model_gate(agent._tool_use_enforcement, agent.model, TOOL_USE_ENFORCEMENT_MODELS):
parts.append(TOOL_USE_ENFORCEMENT_GUIDANCE)
parts.append(_fragment(agent, "tool_use_enforcement", TOOL_USE_ENFORCEMENT_GUIDANCE))
if any(g in (agent.model or "").lower() for g in ("gemini", "gemma")):
parts.append(GOOGLE_MODEL_OPERATIONAL_GUIDANCE)
parts.append(_fragment(agent, "google_operational", GOOGLE_MODEL_OPERATIONAL_GUIDANCE))
if _model_gate(getattr(agent, "_execution_guidance", "auto"), agent.model, EXECUTION_GUIDANCE_MODELS):
from agent.prompt_builder import execution_guidance_text
parts.append(execution_guidance_text())
parts.append(_fragment(agent, "execution_discipline", execution_guidance_text()))
return parts


Expand All @@ -584,12 +593,12 @@ def _alibaba_identity_part(agent: Any) -> List[str]:
if agent.provider != "alibaba":
return []
_model_short = agent.model.rsplit("/", 1)[-1]
return [
return [_fragment(agent, "model_identity",
f"You are powered by the model named {_model_short}. "
f"The exact model ID is {agent.model}. "
f"When asked what model you are, always answer based on this information, "
f"not on any model name returned by the API."
]
)]


def _workspace_pin_key() -> str:
Expand Down Expand Up @@ -654,9 +663,16 @@ def _seed_workspace_pin(agent: Any, key: str) -> None:
# coding posture, or with tools off) leaves the pin open so this build captures one.
if block:
agent._frozen_workspace_snapshot = (key, block)
elif stored_cwd == key and (getattr(agent, "_prompt_overrides", None) or {}).get("coding_workspace", {}).get("mode") in {"replace", "remove"}:
# A replace/remove override hides the snapshot bytes. Its renderer-owned
# marker preserves only the fact that a workspace block was emitted.
from agent.surface_switch import split_runtime_boundary
_identity, boundary, runtime = split_runtime_boundary(prompt)
if boundary and runtime.endswith(f"\n\n{_OVERRIDDEN_WORKSPACE_PIN}\n\n{_pb.RUNTIME_ENVIRONMENT_END}"):
agent._frozen_workspace_snapshot = (key, _WORKSPACE_PIN_SENTINEL)


def _coding_parts(agent: Any) -> Tuple[List[str], List[str], List[str]]:
def _coding_parts(agent: Any) -> Tuple[List[Optional[str]], List[Optional[str]], List[Optional[str]]]:
"""``(prefix, workspace, trailing)`` coding-posture blocks; all empty
without tools or when probing fails (it must never block prompt build).

Expand All @@ -682,7 +698,10 @@ def _coding_parts(agent: Any) -> Tuple[List[str], List[str], List[str]]:
valid_tool_names=agent.valid_tool_names, workspace_block=replay)
if replay is None:
agent._frozen_workspace_snapshot = (cwd_key, parts[1][0] if parts[1] else "")
return parts
return tuple(
[_fragment(agent, key, text) for text in group]
for key, group in zip(("coding_brief", "coding_workspace", "coding_instructions"), parts)
)
except Exception:
pass
return [], [], []
Expand All @@ -696,12 +715,17 @@ def _post_workspace_parts(agent: Any) -> List[str]:
if getattr(agent, "_environment_probe", True):
try:
from tools.env_probe import get_environment_probe_line
parts.append(get_environment_probe_line())
probe = get_environment_probe_line()
if probe:
parts.append(_fragment(agent, "environment_probe", probe))
except Exception:
pass # Probe failure must never block prompt build.
if getattr(agent, "_bot_mode_protocol", True):
parts.extend(_bot_mode_parts(agent))
parts += [_active_profile_line(agent), platform_hint(agent)]
parts.append(_fragment(agent, "active_profile", _active_profile_line(agent)))
hint = platform_hint(agent)
if hint:
parts.append(_fragment(agent, "platform_hints", hint))
return parts


Expand Down Expand Up @@ -748,6 +772,9 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
# in the rendered index (pure string check — inherits the index's stability).
if "skill_view" in (agent.valid_tool_names or set()) and "- hermes-agent:" in skills_prompt:
stable_parts[_help_guidance_slot] = HERMES_AGENT_HELP_GUIDANCE
stable_parts[_help_guidance_slot] = _fragment(agent, "hermes_help", stable_parts[_help_guidance_slot])
if skills_prompt:
skills_prompt = _fragment(agent, "skills", skills_prompt)
stable_parts.extend(_alibaba_identity_part(agent))
# Pinned skills are per-agent constants (resolved once), so they live in the stable prefix.
stable_parts.extend(_auto_load_parts(agent))
Expand Down Expand Up @@ -783,6 +810,11 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
if environment_hints:
# Embedder hints are prose too; reserve the delimiter for the renderer.
environment_hints = environment_hints.replace(_pb.RUNTIME_ENVIRONMENT_HEADING, "> " + _pb.RUNTIME_ENVIRONMENT_HEADING)
environment_hints = environment_hints.replace(_OVERRIDDEN_WORKSPACE_PIN, "> " + _OVERRIDDEN_WORKSPACE_PIN)
workspace_override = (getattr(agent, "_prompt_overrides", None) or {}).get("coding_workspace", {})
if coding_workspace_parts and workspace_override.get("mode") in {"replace", "remove"}:
environment_hints += f"\n\n{_OVERRIDDEN_WORKSPACE_PIN}"
if environment_hints:
volatile_parts.append(f"{_pb.RUNTIME_ENVIRONMENT_HEADING}\n\n{environment_hints}\n\n{_pb.RUNTIME_ENVIRONMENT_END}")
return {"stable": _join_tier(stable_parts), "context": _join_tier(context_parts), "volatile": _join_tier(volatile_parts)}

Expand Down
22 changes: 22 additions & 0 deletions cli-config.yaml.example
Original file line number Diff line number Diff line change
Expand Up @@ -1264,6 +1264,28 @@ agent:
# Default 3.
# max_verify_nudges: 3

# Per-fragment system-prompt overrides.
# Reshape named prompt fragments without editing Hermes
# source. Each entry is `mode: replace|append|prepend|remove` + `text`
# (a bare string is shorthand for `replace`). Pure data, resolved once at
# prompt-build time — never busts the prefix cache. With no entries the
# assembled prompt is byte-identical to the default.
#
# Project context and runtime environment facts are outside this named-fragment surface.
# Fragment keys: identity, hermes_help, task_completion,
# parallel_tool_call_guidance, tool_guidance,
# steer_channel, tool_use_enforcement,
# google_operational, execution_discipline, skills, model_identity,
# environment_probe, active_profile, platform_hints, coding_brief,
# coding_workspace, coding_instructions
#
# An override only applies when that fragment is actually emitted this
# session; append/prepend to an absent fragment is a no-op.
# prompt_overrides:
# task_completion: { mode: append, text: "..." }
# google_operational: { mode: remove }
# steer_channel: "..." # bare string = replace

# Enable verbose logging
verbose: false

Expand Down
2 changes: 2 additions & 0 deletions hermes_cli/config_defaults.py
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,8 @@ def _aux(timeout, *, reasoning_effort=True, **extra):
# read-only commands) into one batched turn; the runtime already runs them concurrently. ~70
# cached tokens. False disables.
"parallel_tool_call_guidance": True,
# Named system-prompt fragments can be replaced, extended, or removed.
"prompt_overrides": {},
# Toolchain probe: surfaces Python/pip/uv/PEP-668 state in the system prompt only when
# something non-default is detected (no pip module, pip/python mismatch, PEP 668 without
# uv); zero tokens when clean. Skipped for docker/modal/ssh backends (own probe).
Expand Down
Loading