hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 30 Packages Projects Releases Wiki Activity

refactor(agent/skills): drop intra-function blank lines, fold remaining single-use guards

Browse Source
This commit is contained in:
Teknium
2026-09-02 19:51:22 -07:00
parent 18dee0f8e2
commit bdf1e0bf98
5 changed files with 42 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
agent/skill_bundles.py
View File

@@ -32,9 +32,7 @@ _bundles_cache_mtime: Optional[float] = None
def _bundles_dir() -> Path:
"""Bundles directory: ``HERMES_BUNDLES_DIR`` override (tests) or ``<HERMES_HOME>/skill-bundles``."""
override = os.environ.get("HERMES_BUNDLES_DIR")
if override:
return Path(override).expanduser()
return get_hermes_home() / "skill-bundles"
return Path(override).expanduser() if override else get_hermes_home() / "skill-bundles"
def _iter_bundle_files() -> List[Path]:
@@ -63,10 +61,8 @@ def _load_bundle_file(path: Path) -> Optional[Dict[str, Any]]:
except yaml.YAMLError as exc:
logger.warning("Invalid YAML in bundle %s: %s", path, exc)
return None
def _skip(reason: str) -> None:
logger.warning("Bundle %s %s; skipping", path, reason)
if not isinstance(data, dict):
return _skip("is not a mapping")
name = str(data.get("name") or path.stem).strip()
@@ -135,10 +131,7 @@ def list_bundles() -> List[Dict[str, Any]]:
def build_bundle_invocation_message(
cmd_key: str,
user_instruction: str = "",
task_id: str | None = None,
platform: str | None = None,
cmd_key: str, user_instruction: str = "", task_id: str | None = None, platform: str | None = None,
) -> Optional[Tuple[str, List[str], List[str]]]:
"""Build the user message for a bundle invocation: ``(message,
loaded_skill_names, missing_skill_names)`` or ``None`` if the bundle wasn't
@@ -148,10 +141,8 @@ def build_bundle_invocation_message(
info = get_skill_bundles().get(cmd_key)
if not info:
return None
# Late import keeps skill_bundles cheap to import (no tools/* at import time).
from agent.skill_commands import _disabled_skill_names, _load_skill_blocks, _load_skill_payload, _scaffold_header
bundle_name = info["name"]
loaded_names, missing, disabled, skill_blocks = _load_skill_blocks(
[(skill_id or "").strip() for skill_id in info["skills"]],
@@ -162,7 +153,6 @@ def build_bundle_invocation_message(
)
if not skill_blocks:
return None
header = _scaffold_header(
f'"{bundle_name}" skill bundle',
loaded_names,
@@ -186,13 +176,7 @@ def bundle_path_for(name: str) -> Path:
return _bundles_dir() / f"{slug}.yaml"
def save_bundle(
name: str,
skills: List[str],
description: str = "",
instruction: str = "",
overwrite: bool = False,
) -> Path:
def save_bundle(name: str, skills: List[str], description: str = "", instruction: str = "", overwrite: bool = False) -> Path:
"""Write a bundle to disk and refresh the cache. Raises ``FileExistsError``
if the target exists and not ``overwrite``; ``ValueError`` for unusable inputs."""
name = (name or "").strip()
@@ -201,17 +185,12 @@ def save_bundle(
cleaned_skills = [str(s).strip() for s in skills if str(s).strip()]
if not cleaned_skills:
raise ValueError("Bundle must reference at least one skill")
path = bundle_path_for(name)
if path.exists() and not overwrite:
raise FileExistsError(f"Bundle already exists at {path}")
path.parent.mkdir(parents=True, exist_ok=True)
payload: Dict[str, Any] = {"name": name, "skills": cleaned_skills}
if description:
payload["description"] = description
if instruction:
payload["instruction"] = instruction
payload.update({k: v for k, v in (("description", description), ("instruction", instruction)) if v})
path.write_text(yaml.safe_dump(payload, sort_keys=False, allow_unicode=True), encoding="utf-8")
scan_bundles()
return path

19
agent/skill_commands.py
View File

@@ -56,8 +56,7 @@ SKILL_EXCERPT_JOINT = "\x1e"
def slugify_skill_name(name: str) -> str:
"""Normalize a skill/bundle name to a ``/command`` slug (``Foo Bar`` -> ``foo-bar``);
strips chars (``+``, ``/``) that would make invalid Telegram command names."""
cmd = name.lower().replace(" ", "-").replace("_", "-")
cmd = _SKILL_INVALID_CHARS.sub("", cmd)
cmd = _SKILL_INVALID_CHARS.sub("", name.lower().replace(" ", "-").replace("_", "-"))
return _SKILL_MULTI_HYPHEN.sub("-", cmd).strip("-")
@@ -114,9 +113,7 @@ def _cut_after(message: str, marker: str, stop_marker: str, find) -> Optional[st
marker_idx = find(marker)
if marker_idx < 0:
return None
text = message[marker_idx + len(marker):]
idx = text.find(stop_marker)
return (text[:idx] if idx >= 0 else text).strip() or None
return message[marker_idx + len(marker):].split(stop_marker, 1)[0].strip() or None
def _resolve_skill_commands_platform() -> Optional[str]:
@@ -125,7 +122,6 @@ def _resolve_skill_commands_platform() -> Optional[str]:
own ``skills.platform_disabled`` view."""
try:
from gateway.session_context import get_session_env
resolved_platform = os.getenv("HERMES_PLATFORM") or get_session_env("HERMES_SESSION_PLATFORM")
except Exception:
resolved_platform = os.getenv("HERMES_PLATFORM")
@@ -136,7 +132,6 @@ def _resolve_skill_commands_home() -> str:
"""Effective Hermes home the scan is scoped to (profiles carry their own
``skills.external_dirs``, so a profile switch must invalidate the cache)."""
from hermes_constants import get_hermes_home
return str(get_hermes_home())
@@ -148,7 +143,6 @@ def _load_skill_payload(skill_identifier: str, task_id: str | None = None) -> tu
try:
from tools.skills_tool import _skills_dir, skill_view
from agent.skill_utils import normalize_skill_lookup_name
normalized = normalize_skill_lookup_name(raw_identifier)
loaded_skill = json.loads(skill_view(normalized, task_id=task_id, preprocess=False))
except Exception:
@@ -174,7 +168,6 @@ def _inject_skill_config(loaded_skill: dict[str, Any], parts: list[str]) -> None
values so the agent needn't read config.yaml. Any failure leaves the message without it."""
try:
from agent.skill_utils import extract_skill_config_vars, parse_frontmatter, resolve_skill_config_values
raw_content = str(loaded_skill.get("raw_content") or loaded_skill.get("content") or "")
frontmatter, _ = parse_frontmatter(raw_content)
resolved = resolve_skill_config_values(extract_skill_config_vars(frontmatter))
@@ -230,7 +223,6 @@ def _build_skill_message(
) -> str:
"""Format a loaded skill into a user/system message payload."""
from tools.skills_tool import _skills_dir
# Preprocess first so downstream blocks see the expanded content.
content = preprocess_skill_content(
str(loaded_skill.get("content") or ""), skill_dir, session_id, skills_cfg=_load_skills_config(),
@@ -245,7 +237,6 @@ def _build_skill_message(
"`templates/config.yaml`) against that directory, then run them "
"with the terminal tool using the absolute path.",
]
_inject_skill_config(loaded_skill, parts)
setup_note = _setup_note(loaded_skill)
if setup_note:
@@ -322,7 +313,6 @@ _SCAN_SKIP_PARTS = {'.git', '.github', '.hub', '.archive'}
def _scan_skill_md(skill_md: Path, disabled: set, seen_names: set, commands: Dict[str, Dict[str, Any]], resolve_command) -> None:
"""Register one SKILL.md in *commands* (no-op when filtered or colliding)."""
from tools.skills_tool import _parse_frontmatter, skill_matches_platform, skill_matches_environment
if any(part in _SCAN_SKIP_PARTS for part in skill_md.parts):
return
frontmatter, body = _parse_frontmatter(skill_md.read_text(encoding='utf-8'))
@@ -382,7 +372,6 @@ def scan_skill_commands() -> Dict[str, Dict[str, Any]]:
from hermes_cli.commands import resolve_command
disabled = _get_disabled_skill_names()
seen_names: set = set()
# Precedence: project (through the quarantine chokepoint) > local > external.
# Resolve the local dir at call time: import-time SKILLS_DIR is frozen to
# the launch home, but a multiplexed profile scope may have changed it.
@@ -501,9 +490,7 @@ def split_stacked_skill_commands(rest: str) -> tuple[list[str], str]:
stripped = remaining.lstrip()
if not stripped.startswith("/"):
break
parts = stripped.split(None, 1)
token = parts[0]
tail = parts[1] if len(parts) > 1 else ""
token, tail = (stripped.split(None, 1) + [""])[:2]
cmd_key = resolve_skill_command_key(token.lstrip("/"))
if cmd_key is None or cmd_key in keys:
break

4
agent/skill_preprocessing.py
View File

@@ -23,7 +23,6 @@ def load_skills_config() -> dict:
"""Load the ``skills`` section of config.yaml (best-effort)."""
try:
from hermes_cli.config import load_config_readonly
skills_cfg = (load_config_readonly() or {}).get("skills")
if isinstance(skills_cfg, dict):
return skills_cfg
@@ -69,7 +68,6 @@ def run_inline_shell(command: str, cwd: Path | None, timeout: int) -> str:
if isinstance(exc, RuntimeError) and "live-system guard: blocked os.kill" in str(exc):
return f"[inline-shell timeout after {timeout}s: {command}]"
return f"[inline-shell error: {exc}]"
output = (completed.stdout or "").rstrip("\n") or (completed.stderr or "").rstrip("\n")
if len(output) > _INLINE_SHELL_MAX_OUTPUT:
output = output[:_INLINE_SHELL_MAX_OUTPUT] + "...[truncated]"
@@ -80,11 +78,9 @@ def expand_inline_shell(content: str, skill_dir: Path | None, timeout: int) -> s
"""Replace every !`cmd` snippet with its stdout, run with the skill dir as CWD."""
if "!`" not in content:
return content
def _replace(match: re.Match) -> str:
cmd = match.group(1).strip()
return run_inline_shell(cmd, skill_dir, timeout) if cmd else ""
return _INLINE_SHELL_RE.sub(_replace, content)

41
agent/skill_utils.py
View File

@@ -93,7 +93,6 @@ def yaml_load(content: str):
if _yaml_load_fn is None:
import functools
import yaml
_yaml_load_fn = functools.partial(yaml.load, Loader=getattr(yaml, "CSafeLoader", None) or yaml.SafeLoader)
return _yaml_load_fn(content)
@@ -106,7 +105,6 @@ def parse_frontmatter(content: str) -> Tuple[Dict[str, Any], str]:
end_match = re.search(r"\n---\s*\n", content[3:]) if content.startswith("---") else None
if not end_match:
return {}, content
yaml_content = content[3 : end_match.start() + 3]
body = content[end_match.end() + 3 :]
frontmatter: Dict[str, Any] = {}
@@ -126,16 +124,13 @@ def skill_matches_platform_list(platforms: Any) -> bool:
"""Return True when *platforms* is compatible with the current OS."""
if not platforms:
return True
if not isinstance(platforms, list):
platforms = [platforms]
current = sys.platform
running_in_termux = is_termux()
for platform in platforms:
for platform in platforms if isinstance(platforms, list) else [platforms]:
normalized = str(platform).lower().strip()
mapped = PLATFORM_MAP.get(normalized, normalized)
# Termux is a Linux userland on Android: accept linux-tagged skills
# whether sys.platform is "linux" (pre-3.13) or "android" (3.13+).
if current.startswith(mapped) or (running_in_termux and mapped in ("linux", "termux", "android")):
if sys.platform.startswith(mapped) or (running_in_termux and mapped in ("linux", "termux", "android")):
return True
return False
@@ -159,7 +154,6 @@ def _detect_kanban() -> bool:
if os.getenv("HERMES_KANBAN_TASK") or os.getenv("HERMES_KANBAN_BOARD"):
try:
from agent.delegation_context import is_dispatcher_owned_worker_context
owned = is_dispatcher_owned_worker_context()
except Exception:
owned = True
@@ -167,7 +161,6 @@ def _detect_kanban() -> bool:
return True
try:
from tools.kanban_tools import _profile_has_kanban_toolset
return bool(_profile_has_kanban_toolset())
except Exception:
return False
@@ -176,7 +169,6 @@ def _detect_kanban() -> bool:
def _detect_docker() -> bool:
try:
from hermes_constants import is_container
return is_container()
except Exception:
return False
@@ -239,7 +231,6 @@ def _load_raw_config() -> Dict[str, Any]:
cached = _RAW_CONFIG_CACHE.get(cache_key) if cache_key is not None else None
if cached is not None:
return cached
try:
parsed = yaml_load(config_path.read_text(encoding="utf-8"))
except Exception as e:
@@ -247,7 +238,6 @@ def _load_raw_config() -> Dict[str, Any]:
return {}
if not isinstance(parsed, dict):
return {}
if cache_key is not None:
_RAW_CONFIG_CACHE.clear()
_RAW_CONFIG_CACHE[cache_key] = parsed
@@ -274,7 +264,6 @@ def _expand_path(entry: str) -> Path:
def _home_relative(p: Path) -> Path:
"""Anchor a relative config path at HERMES_HOME; absolute paths pass through."""
from hermes_constants import get_hermes_home
return p if p.is_absolute() else get_hermes_home() / p
@@ -289,7 +278,6 @@ def get_disabled_skill_names(platform: str | None = None) -> Set[str]:
skills_cfg = _skills_cfg()
if skills_cfg is None:
return set()
from gateway.session_context import get_session_env
resolved_platform = platform or os.getenv("HERMES_PLATFORM") or get_session_env("HERMES_SESSION_PLATFORM")
disabled = _normalize_string_set(skills_cfg.get("disabled"))
@@ -354,7 +342,6 @@ def get_external_skills_dirs() -> List[Path]:
skills_cfg = _skills_cfg()
if skills_cfg is None:
return []
local_skills = get_skills_dir().resolve()
result: List[Path] = []
for entry in _config_str_list(skills_cfg.get("external_dirs")):
@@ -394,14 +381,12 @@ def display_skill_create_dir() -> str:
"""User-facing path where new skills are created (``~/`` shorthand when
possible); tool schema descriptions and prompts follow ``skills.create_dir``."""
from hermes_constants import display_hermes_home
create_dir = get_skill_create_dir()
if create_dir is None:
return f"{display_hermes_home()}/skills/"
try:
if create_dir.is_relative_to(Path.home()):
return "~/" + create_dir.relative_to(Path.home()).as_posix() + "/"
except ValueError:
return create_dir.as_posix() + "/"
return create_dir.as_posix() + "/"
def get_all_skills_dirs() -> List[Path]:
@@ -433,7 +418,6 @@ def find_project_root(start: Optional[Path] = None) -> Optional[Path]:
try:
if start is None:
from agent.runtime_cwd import scope_terminal_cwd
env_cwd = scope_terminal_cwd()
start = Path(env_cwd) if env_cwd else Path.cwd()
cur = Path(start).resolve()
@@ -478,8 +462,7 @@ def _candidate_project_skills_dirs(root: Path) -> List[Path]:
(HERMES_HOME itself may live inside a git checkout)."""
local_skills = get_skills_dir().resolve()
dirs: List[Path] = []
for sub in PROJECT_SKILLS_SUBDIRS:
cand = root / sub
for cand in (root / sub for sub in PROJECT_SKILLS_SUBDIRS):
try:
if cand.is_dir() and cand.resolve() != local_skills:
dirs.append(cand.resolve())
@@ -505,10 +488,8 @@ def get_project_skills_dirs() -> List[Path]:
def get_untrusted_project_skills_root() -> Optional[Tuple[Path, int]]:
"""(root, skill_count) when cwd's project has skills but is NOT trusted, else None."""
root = _current_project_root(trusted=False)
if root is None:
return None
count = 0
for d in _candidate_project_skills_dirs(root):
for d in _candidate_project_skills_dirs(root) if root is not None else ():
try:
count += sum(1 for _ in iter_skill_index_files(d, "SKILL.md"))
except OSError:
@@ -540,7 +521,6 @@ def is_quarantined_project_skill(skill_md) -> bool:
try:
from tools.skills_guard import scan_skill_cached
from hermes_constants import get_hermes_home
result, _prov = scan_skill_cached(
skill_dir, source=_PROJECT_SCAN_SOURCE, cache_dir=get_hermes_home() / "cache" / "project_skill_scans",
)
@@ -567,11 +547,9 @@ def normalize_skill_lookup_name(identifier: str) -> str:
raw_identifier = (identifier or "").strip()
if not raw_identifier:
return raw_identifier
identifier_path = Path(raw_identifier).expanduser()
if not identifier_path.is_absolute():
return raw_identifier.lstrip("/")
# Resolve the primary root via tools.skills_tool at CALL time: tests patch
# ``tools.skills_tool.SKILLS_DIR`` and skill_view() enforces ``_skills_dir()``
# (which follows the live profile-scoped HERMES_HOME), so normalization
@@ -581,14 +559,12 @@ def normalize_skill_lookup_name(identifier: str) -> str:
primary_root = _skills_tool._skills_dir()
except Exception:
primary_root = get_skills_dir()
trusted_roots = [primary_root]
for getter in (get_project_skills_dirs, get_external_skills_dirs):
try:
trusted_roots.extend(getter())
except Exception:
pass
# Prefer the lexical path under a trusted root before resolving symlinks:
# ~/.hermes/skills/<name> may be a symlink to a checkout elsewhere, and
# resolving first would turn that trusted path into one skill_view rejects.
@@ -713,13 +689,12 @@ def resolve_skill_config_values(config_vars: List[Dict[str, Any]]) -> Dict[str,
config = _load_raw_config()
resolved: Dict[str, Any] = {}
for var in config_vars:
logical_key = var["key"]
value = _resolve_dotpath(config, f"{SKILL_CONFIG_PREFIX}.{logical_key}")
value = _resolve_dotpath(config, f"{SKILL_CONFIG_PREFIX}.{var['key']}")
if value is None or (isinstance(value, str) and not value.strip()):
value = var.get("default", "")
if isinstance(value, str) and ("~" in value or "${" in value):
value = os.path.expanduser(os.path.expandvars(value))
resolved[logical_key] = value
resolved[var["key"]] = value
return resolved

90
agent/system_prompt.py
View File

@@ -34,8 +34,7 @@ _PLUGIN_SECTION_FRAME_RE = re.compile(
r"<!-- hermes-plugin-section-chars:(?P<chars>[0-9]{1,4}) -->\n\n",
re.MULTILINE,
)
_GATE_WORDS = {"true": True, "always": True, "yes": True, "on": True,
"false": False, "never": False, "no": False, "off": False}
_GATE_WORDS = {**dict.fromkeys(("true", "always", "yes", "on"), True), **dict.fromkeys(("false", "never", "no", "off"), False)}
def _ra():
@@ -70,12 +69,9 @@ def _resolve_platform_hint(agent: Any, platform_key: str, default_hint: str) ->
spec = {"append": spec}
if not isinstance(spec, dict):
return default_hint
replace_text = spec.get("replace")
base = replace_text.strip() if isinstance(replace_text, str) and replace_text.strip() else default_hint
append_text = spec.get("append")
if isinstance(append_text, str) and append_text.strip():
return f"{base}\n\n{append_text.strip()}".strip()
return base
replace_text, append_text = (v.strip() if isinstance(v, str) else "" for v in (spec.get("replace"), spec.get("append")))
base = replace_text or default_hint
return f"{base}\n\n{append_text}".strip() if append_text else base
_TUI_EMBEDDED_PANE_CLARIFIER = (
@@ -106,7 +102,6 @@ def _plugin_session_info(agent: Any) -> Dict[str, str]:
def _ambient_plugin_profile_name() -> str:
from hermes_cli.profiles import get_active_profile_name
return str(get_active_profile_name() or "default")
@@ -134,7 +129,6 @@ def _frozen_plugin_prompt_sections(agent: Any) -> tuple:
else:
try:
from hermes_cli.plugins import render_system_prompt_sections
rendered = tuple(render_system_prompt_sections(_plugin_session_info(agent)))
except Exception as exc:
rendered = getattr(agent, "_plugin_system_prompt_sections_previous", None)
@@ -155,7 +149,6 @@ def _restore_plugin_prompt_sections(prompt: str) -> tuple:
MAX_SYSTEM_PROMPT_SECTION_CHARS, PLUGIN_SECTIONS_END, PLUGIN_SECTIONS_START,
RenderedPluginSystemPromptSection, format_system_prompt_sections,
)
start = prompt.rfind(PLUGIN_SECTIONS_START)
end = prompt.find(PLUGIN_SECTIONS_END, start + len(PLUGIN_SECTIONS_START)) if start >= 0 else -1
if end < 0:
@@ -164,7 +157,6 @@ def _restore_plugin_prompt_sections(prompt: str) -> tuple:
if not prompt[after_end:].startswith("\n\nConversation started:"):
return ()
framed = prompt[start:after_end]
restored = []
for match in _PLUGIN_SECTION_FRAME_RE.finditer(framed):
content_len = int(match.group("chars"))
@@ -184,7 +176,6 @@ def restore_plugin_prompt_sections(agent: Any, prompt: str) -> None:
def _plugin_section_blocks(sections: tuple, position: str) -> List[str]:
from hermes_cli.plugins import format_system_prompt_sections
block = format_system_prompt_sections([s for s in sections if s.position == position])
return [block] if block else []
@@ -198,7 +189,6 @@ def _session_start_like(agent: Any, now: Any) -> Any:
then ``now``. Stamps are box-local wall-clock: attach that zone first, then
convert to ``now``'s zone so the date matches the per-turn clock."""
from datetime import datetime
def _to_display_tz(dt: Any) -> Any:
if dt.tzinfo is None:
try:
@@ -211,7 +201,6 @@ def _session_start_like(agent: Any, now: Any) -> Any:
except (ValueError, OSError):
pass
return dt
session_id = getattr(agent, "session_id", None)
db = getattr(agent, "_session_db", None)
try:
@@ -237,7 +226,6 @@ def _agent_home(agent: Any) -> Optional[Path]:
ContextVar, where ambient resolution would leak the launch profile."""
try:
from hermes_constants import get_hermes_home_override
override = get_hermes_home_override()
if override:
return Path(override)
@@ -253,7 +241,7 @@ def _agent_home(agent: Any) -> Optional[Path]:
def _agent_skills_dir(agent: Any) -> Optional[Path]:
"""The agent's own ``<home>/skills`` dir, or None to use ambient home."""
home = _agent_home(agent)
return (home / "skills") if home is not None else None
return home / "skills" if home is not None else None
def _profile_name_for_home(home: Path) -> str:
@@ -263,7 +251,6 @@ def _profile_name_for_home(home: Path) -> str:
would misreport as "default"."""
try:
from hermes_constants import get_default_hermes_root
rel = home.resolve().relative_to((get_default_hermes_root() / "profiles").resolve())
return rel.parts[0] if rel.parts else "default"
except (ValueError, OSError):
@@ -304,7 +291,6 @@ def _skills_prompt(agent: Any, _r: Any) -> str:
avail_toolsets = {_r.get_toolset_for_tool(tool_name) for tool_name in agent.valid_tool_names} - {None, ""}
try:
from agent.coding_context import coding_compact_skill_categories
_compact_cats = coding_compact_skill_categories(platform=agent.platform, cwd=resolve_context_cwd())
except Exception:
_compact_cats = frozenset()
@@ -323,7 +309,6 @@ def _bot_mode_parts(agent: Any) -> List[str]:
parts: List[str] = []
try:
from tools.bot_mode_probe import BOT_CHAT_TITLE, epoch_line, get_bot_mode_protocol_section
_title = str(getattr(agent, "_session_title_hint", "") or "").strip()
if not _title:
_sdb = getattr(agent, "_session_db", None)
@@ -343,7 +328,6 @@ def _bot_mode_parts(agent: Any) -> List[str]:
def _ambient_file_safety_profile_name() -> str:
from agent.file_safety import _resolve_active_profile_name
return _resolve_active_profile_name()
@@ -489,24 +473,22 @@ def _identity_parts(agent: Any, _r: Any, ctx_len: Optional[int]) -> Tuple[List[s
"""SOUL.md (primary identity; cron keeps the persona while skipping cwd
instructions, scoped to the agent's OWN home) or the default identity.
Returns ``(parts, soul_loaded)``."""
if agent.load_soul_identity or not agent.skip_context_files:
_soul_content = _r.load_soul_md(ctx_len, home_override=_agent_home(agent))
if _soul_content:
return [_soul_content], True
return [DEFAULT_AGENT_IDENTITY], False
wants_soul = agent.load_soul_identity or not agent.skip_context_files
_soul_content = _r.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)
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:
if getattr(agent, "_task_completion_guidance", True):
parts.append(TASK_COMPLETION_GUIDANCE)
if getattr(agent, "_parallel_tool_call_guidance", True):
parts.append(PARALLEL_TOOL_CALL_GUIDANCE)
_tool_block = _tool_guidance_block(agent)
if _tool_block:
parts.append(_tool_block)
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 not agent.valid_tool_names:
return parts
# Steering only lands inside tool results, so only reachable with tools.
@@ -517,8 +499,7 @@ def _guidance_parts(agent: Any) -> List[str]:
# 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)
_model_lower = (agent.model or "").lower()
if "gemini" in _model_lower or "gemma" in _model_lower:
if any(g in (agent.model or "").lower() for g in ("gemini", "gemma")):
parts.append(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
@@ -531,7 +512,7 @@ def _alibaba_identity_part(agent: Any) -> List[str]:
the real identity so the agent can answer correctly."""
if agent.provider != "alibaba":
return []
_model_short = agent.model.split("/")[-1] if "/" in agent.model else agent.model
_model_short = agent.model.rsplit("/", 1)[-1]
return [
f"You are powered by the model named {_model_short}. "
f"The exact model ID is {agent.model}. "
@@ -545,7 +526,6 @@ def _coding_parts(agent: Any) -> Tuple[List[str], List[str], List[str]]:
without tools or when probing fails (it must never block prompt build)."""
try:
from agent.coding_context import coding_system_prompt_parts
if agent.valid_tool_names:
return coding_system_prompt_parts(
platform=agent.platform, cwd=resolve_context_cwd(),
@@ -564,15 +544,12 @@ def _post_workspace_parts(agent: Any) -> List[str]:
if getattr(agent, "_environment_probe", True):
try:
from tools.env_probe import get_environment_probe_line
_probe_line = get_environment_probe_line()
if _probe_line:
parts.append(_probe_line)
parts.append(get_environment_probe_line())
except Exception:
pass # Probe failure must never block prompt build.
if getattr(agent, "_bot_mode_protocol", True):
parts.extend(_bot_mode_parts(agent))
parts.append(_active_profile_line(agent))
parts.append(_platform_hint(agent))
parts += [_active_profile_line(agent), _platform_hint(agent)]
return parts
@@ -585,15 +562,13 @@ def _context_files_part(agent: Any, _r: Any, ctx_len: Optional[int], soul_loaded
if agent.skip_context_files:
return []
launch_artifact = getattr(agent, "_context_cwd_is_launch_artifact", False)
context_files_prompt = _r.build_context_files_prompt(
cwd=None if launch_artifact else resolve_context_cwd(), skip_soul=soul_loaded,
context_length=ctx_len,
allow_install_tree_fallback=agent.platform in ("cli", "tui"),
home_override=_agent_home(agent))
return [context_files_prompt] if context_files_prompt else []
return [_r.build_context_files_prompt(
cwd=None if launch_artifact else resolve_context_cwd(), skip_soul=soul_loaded, context_length=ctx_len,
allow_install_tree_fallback=agent.platform in ("cli", "tui"), home_override=_agent_home(agent))]
def _join_tier(parts: List[str]) -> str:
def _join_tier(parts: List[Optional[str]]) -> str:
"""Join non-empty parts; None/blank entries are dropped."""
return "\n\n".join(p.strip() for p in parts if p and p.strip())
@@ -607,7 +582,6 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
# Model context window scales the context-file caps; stable per conversation.
_cc_len = getattr(getattr(agent, "context_compressor", None), "context_length", None)
_ctx_len = _cc_len if isinstance(_cc_len, int) and _cc_len > 0 else None
# ── Stable tier ────────────────────────────────────────────────
stable_parts, _soul_loaded = _identity_parts(agent, _r, _ctx_len)
# The skill_view() pointer dangles without skill tools OR without the
@@ -616,25 +590,19 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
_help_guidance_slot = len(stable_parts)
stable_parts.append(HERMES_AGENT_HELP_GUIDANCE_NO_SKILLS)
stable_parts.extend(_guidance_parts(agent))
skills_prompt = _skills_prompt(agent, _r)
# Skill-pointer variant requires BOTH skill_view AND the hermes-agent skill
# 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.extend(_alibaba_identity_part(agent))
_env_hints = _r.build_environment_hints()
if _env_hints:
stable_parts.append(_env_hints)
stable_parts.append(_r.build_environment_hints())
# Coding posture: operating brief stays in the stable prefix; the live
# git/workspace snapshot sits behind its own cache boundary, and the blocks
# below it must keep their historical post-snapshot position.
coding_prefix_parts, coding_workspace_parts, coding_trailing_parts = _coding_parts(agent)
stable_parts.extend(coding_prefix_parts)
post_workspace_parts = _post_workspace_parts(agent)
# ── Context tier (cwd-dependent, may change between sessions) ─
context_parts: List[str] = []
(context_parts if coding_workspace_parts else stable_parts).extend(
@@ -644,17 +612,14 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
if system_message is not None:
context_parts.append(system_message)
context_parts.extend(_context_files_part(agent, _r, _ctx_len, _soul_loaded))
# ── Volatile tier (most likely to differ on a rebuild; kept last so the stable prefix stays reusable) ──
# Skills are runtime-mutable, so the index leads the volatile band: on a longest-prefix
# backend an unchanged index stays inside the reused prefix; a changed one re-prefills from here.
volatile_parts: List[str] = [skills_prompt] if skills_prompt else []
volatile_parts.extend(_memory_parts(agent))
volatile_parts: List[str] = [skills_prompt, *_memory_parts(agent)]
# Plugin sections are confined to one coarse anchor in the volatile tail so
# a resumed process can reconstruct the stable prefix without re-running plugins.
volatile_parts.extend(_plugin_section_blocks(_frozen_plugin_prompt_sections(agent), "after_memory"))
volatile_parts.append(_timestamp_line(agent))
return {
"stable": _join_tier(stable_parts),
"context": _join_tier(context_parts),
@@ -667,12 +632,11 @@ def build_system_prompt(agent: Any, system_message: Optional[str] = None) -> str
only rebuilt after compression. Tiers are ordered stable -> context ->
volatile so implicit longest-prefix caches keep the unchanged scaffold."""
parts = build_system_prompt_parts(agent, system_message=system_message)
joined = "\n\n".join(p for p in (parts["stable"], parts["context"], parts["volatile"]) if p)
agent._cached_system_prompt_static = parts["stable"]
# Surface context-file truncation warnings in chat, not only in logs.
for warning in drain_truncation_warnings():
agent._emit_status(warning)
return joined
return "\n\n".join(p for p in (parts["stable"], parts["context"], parts["volatile"]) if p)
def invalidate_system_prompt(agent: Any) -> None: