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

refactor(agent): prompt_builder siblings — inline snapshot write, compact __all__/locals, tidy blank-line conventions

Browse Source
This commit is contained in:
Teknium
2026-09-02 19:05:15 -07:00
parent cd0b813ee2
commit 6e20411e4a
5 changed files with 68 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
agent/markdown_tables.py
View File

@@ -16,12 +16,7 @@ from typing import List
from wcwidth import wcswidth
__all__ = [
"is_table_divider",
"looks_like_table_row",
"realign_markdown_tables",
"split_table_row",
]
__all__ = ["is_table_divider", "looks_like_table_row", "realign_markdown_tables", "split_table_row"]
_DIVIDER_CELL_RE = re.compile(r"^\s*:?-{3,}:?\s*$")
@@ -88,8 +83,7 @@ def _render_block(rows: List[List[str]], available_width: int | None = None) ->
def _hard_break(word: str, w: int) -> List[str]:
"""Split a single over-wide word into display-width-``w`` chunks."""
out: List[str] = []
buf = ""
bw = 0
buf, bw = "", 0
for ch in word:
cw = _disp_width(ch) or 1
if bw + cw > w and buf:
@@ -116,8 +110,7 @@ def _wrap_to_width(text: str, width: int) -> List[str]:
return [""]
lines: List[str] = []
current = ""
current_w = 0
current, current_w = "", 0
def _start(word: str, ww: int) -> None:
nonlocal current, current_w
@@ -189,8 +182,7 @@ def realign_markdown_tables(text: str, available_width: int | None = None) -> st
lines = text.split("\n")
out: List[str] = []
i = 0
n = len(lines)
i, n = 0, len(lines)
while i < n:
line = lines[i]
# A table starts with a header row whose next line is a divider.

62
agent/onboarding.py
View File

@@ -25,9 +25,8 @@ PROFILE_BUILD_FLAG = "profile_build_offered"
# applied so the message matches reality; "interrupt" is the default branch.
_BUSY_INPUT_HINTS_GATEWAY = {
"queue": (
"💡 First-time tip — I queued your message instead of interrupting. "
"Send `/busy interrupt` to make new messages stop the current task "
"immediately, or `/busy status` to check. This notice won't appear again."
"💡 First-time tip — I queued your message instead of interrupting. Send `/busy interrupt` to make new messages "
"stop the current task immediately, or `/busy status` to check. This notice won't appear again."
),
"steer": (
"💡 First-time tip — I steered your message into the current run; it will arrive after the next tool "
@@ -132,21 +131,20 @@ def profile_build_directive() -> str:
connected accounts.
"""
return (
"\n\n[System note: This is the user's very first message ever. "
"After a one-sentence introduction (mention /help shows commands), "
"OFFER — do not assume — to build a short profile of them so you can "
"be more useful, and explain they can decline or do it later. If and ONLY IF they accept:\n"
" 1. Ask for whatever they're comfortable sharing (name, what they "
"do, how they like you to work). Volunteered facts come first.\n"
" 2. Before ANY external lookup, say what you intend to look up and "
"get explicit consent for that step. Never read their connected "
"accounts (email, calendar, etc.) silently — ask each time.\n"
" 3. With consent, you may use web_search to confirm public details "
"(e.g. employer, public profiles) from the data points they gave.\n"
" 4. Save each confirmed, durable fact with the memory tool using "
"target=\"user\" — keep entries compact and high-signal.\n"
"If they decline at any point, stop immediately and continue normally. "
"Keep the whole exchange light and conversational, not an interrogation.]"
"\n\n"
"[System note: This is the user's very first message ever. After a one-sentence introduction (mention /help "
"shows commands), OFFER — do not assume — to build a short profile of them so you can be more useful, and "
"explain they can decline or do it later. If and ONLY IF they accept:\n"
" 1. Ask for whatever they're comfortable sharing (name, what they do, how they like you to work). "
"Volunteered facts come first.\n"
" 2. Before ANY external lookup, say what you intend to look up and get explicit consent for that step. Never "
"read their connected accounts (email, calendar, etc.) silently — ask each time.\n"
" 3. With consent, you may use web_search to confirm public details (e.g. employer, public profiles) from the "
"data points they gave.\n"
" 4. Save each confirmed, durable fact with the memory tool using target=\"user\" — keep entries compact and "
"high-signal.\n"
"If they decline at any point, stop immediately and continue normally. Keep the whole exchange light and "
"conversational, not an interrogation.]"
)
@@ -174,12 +172,10 @@ def mark_seen(config_path: Path, flag: str) -> bool:
cfg["onboarding"] = {}
seen = cfg["onboarding"].get("seen")
if not isinstance(seen, dict):
seen = {}
cfg["onboarding"]["seen"] = seen
if seen.get(flag) is True:
return True
seen[flag] = True
atomic_config_write(config_path, cfg)
seen = cfg["onboarding"]["seen"] = {}
if seen.get(flag) is not True:
seen[flag] = True
atomic_config_write(config_path, cfg)
return True
except Exception as e:
logger.debug("onboarding: failed to mark flag %s: %s", flag, e)
@@ -187,18 +183,8 @@ def mark_seen(config_path: Path, flag: str) -> bool:
__all__ = [
"BUSY_INPUT_FLAG",
"TOOL_PROGRESS_FLAG",
"OPENCLAW_RESIDUE_FLAG",
"PROFILE_BUILD_FLAG",
"busy_input_hint_gateway",
"busy_input_hint_cli",
"tool_progress_hint_gateway",
"tool_progress_hint_cli",
"openclaw_residue_hint_cli",
"detect_openclaw_residue",
"profile_build_mode",
"profile_build_directive",
"is_seen",
"mark_seen",
"BUSY_INPUT_FLAG", "TOOL_PROGRESS_FLAG", "OPENCLAW_RESIDUE_FLAG", "PROFILE_BUILD_FLAG",
"busy_input_hint_gateway", "busy_input_hint_cli", "tool_progress_hint_gateway", "tool_progress_hint_cli",
"openclaw_residue_hint_cli", "detect_openclaw_residue", "profile_build_mode", "profile_build_directive",
"is_seen", "mark_seen",
]

5
agent/oneshot.py
View File

@@ -56,9 +56,8 @@ def _commit_message_template(variables: Dict[str, Any]) -> Tuple[str, str]:
avoid = _truncate(str(variables.get("avoid") or "").strip(), 1000)
if avoid:
parts.append(
"You already proposed the message below and the user wants a "
"different one. Write a NEW message with different wording (and, if "
"reasonable, a different emphasis or scope framing) — do not repeat "
"You already proposed the message below and the user wants a different one. Write a NEW message with "
"different wording (and, if reasonable, a different emphasis or scope framing) — do not repeat "
f"it:\n{avoid}"
)
return _COMMIT_INSTRUCTIONS, "\n\n".join(parts)

33
agent/process_bootstrap.py
View File

@@ -42,6 +42,13 @@ def _interleave_addrinfos(addrinfos: list[tuple]) -> list[tuple]:
return interleaved
def _quiet_unregister(selector, sock) -> None:
try:
selector.unregister(sock)
except Exception:
pass
def _happy_eyeballs_create_connection(
address: tuple[str, int],
timeout: Optional[float],
@@ -65,10 +72,7 @@ def _happy_eyeballs_create_connection(
deadline = None if timeout is None else time.monotonic() + max(timeout, 0.0)
next_launch = time.monotonic()
pending = list(addrinfos)
in_progress = {
0, errno.EINPROGRESS, errno.EWOULDBLOCK, errno.EALREADY, errno.EINTR,
getattr(errno, "WSAEWOULDBLOCK", 10035),
}
in_progress = {0, errno.EINPROGRESS, errno.EWOULDBLOCK, errno.EALREADY, errno.EINTR, getattr(errno, "WSAEWOULDBLOCK", 10035)}
def start_attempt(addrinfo):
family, socktype, proto, _canonname, sockaddr = addrinfo
@@ -133,10 +137,7 @@ def _happy_eyeballs_create_connection(
if winner is None:
raise last_error if last_error is not None else OSError(f"Could not connect to {host}:{port}")
try:
selector.unregister(winner)
except Exception:
pass
_quiet_unregister(selector, winner)
active.discard(winner)
winner.settimeout(timeout)
for option in socket_options or ():
@@ -145,10 +146,7 @@ def _happy_eyeballs_create_connection(
return winner
finally:
for candidate in active:
try:
selector.unregister(candidate)
except Exception:
pass
_quiet_unregister(selector, candidate)
candidate.close()
selector.close()
@@ -357,13 +355,6 @@ OpenAI = _OpenAIProxy()
__all__ = [
"OpenAI",
"_OpenAIProxy",
"_load_openai_cls",
"_SafeWriter",
"_install_safe_stdio",
"_get_proxy_from_env",
"_get_proxy_for_base_url",
"build_keepalive_http_client",
"enable_happy_eyeballs_on_client",
"OpenAI", "_OpenAIProxy", "_load_openai_cls", "_SafeWriter", "_install_safe_stdio", "_get_proxy_from_env",
"_get_proxy_for_base_url", "build_keepalive_http_client", "enable_happy_eyeballs_on_client",
]

78
agent/prompt_builder.py
View File

@@ -81,7 +81,6 @@ def _strip_yaml_frontmatter(content: str) -> str:
return content
# Constants
DEFAULT_AGENT_IDENTITY = (
# A behavior spec (sizing rule, named prohibitions, earned-depth escape hatch), not a trait list — trait
# lists change nothing. Maintainer rule: models UNDER-explore by default; never re-add an exploration-thrift line.
@@ -113,6 +112,7 @@ HERMES_AGENT_HELP_GUIDANCE_NO_SKILLS = (
"(or read it yourself if you have a way to fetch web content)."
)
def build_memory_guidance(memory_enabled: bool = True, profile_enabled: bool = True) -> str:
"""ONE memory-guidance block whose opening frame adapts to the enabled store(s); "" when both are off.
@@ -143,6 +143,7 @@ def build_memory_guidance(memory_enabled: bool = True, profile_enabled: bool = T
# Legacy aliases still imported by call sites and tests.
MEMORY_GUIDANCE = build_memory_guidance(True, True)
USER_PROFILE_GUIDANCE = build_memory_guidance(False, True)
SESSION_SEARCH_GUIDANCE = (
"When the user references something from a past conversation or you suspect relevant cross-session "
"context exists, use session_search to recall it before asking them to repeat themselves."
@@ -878,12 +879,8 @@ def build_environment_hints() -> str:
backend = (_tenv_read("TERMINAL_ENV") or "local").strip().lower()
is_remote_backend = backend in _REMOTE_TERMINAL_BACKENDS or _plugin_backend_is_remote(backend)
hints = [_remote_backend_hint(backend)] if is_remote_backend else _local_host_hints()
if is_wsl():
hints.append(WSL_ENVIRONMENT_HINT)
extra = _embedder_environment_hint()
if extra:
hints.append(extra)
return "\n\n".join(hints)
hints += [WSL_ENVIRONMENT_HINT] if is_wsl() else []
return "\n\n".join(h for h in (*hints, _embedder_environment_hint()) if h)
CONTEXT_FILE_MAX_CHARS = 20_000
@@ -935,8 +932,7 @@ def drain_truncation_warnings() -> list:
return drained
# Skills index (two-layer cache: in-process LRU, then disk snapshot)
# Skills index (two-layer cache: in-process LRU, then disk snapshot).
# One entry per profile × platform (key carries skills_dir); a multiplexing gateway needs more than a handful.
_SKILLS_PROMPT_CACHE_MAX = 32
_SKILLS_PROMPT_CACHE: OrderedDict[tuple, str] = OrderedDict()
@@ -1013,19 +1009,6 @@ def _load_skills_snapshot(skills_dir: Path) -> Optional[dict]:
return None
def _write_skills_snapshot(skills_dir: Path, skill_entries: list[dict], category_descriptions: dict[str, str]) -> None:
"""Persist skill metadata to disk for fast cold-start reuse (best-effort)."""
try:
atomic_json_write(_skills_prompt_snapshot_path(), {
"version": _SKILLS_SNAPSHOT_VERSION,
"manifest": _build_skills_manifest(skills_dir),
"skills": skill_entries,
"category_descriptions": category_descriptions,
})
except Exception as e:
logger.debug("Could not write skills prompt snapshot: %s", e)
def _build_snapshot_entry(skill_file: Path, skills_dir: Path, frontmatter: dict, description: str) -> dict:
"""Serialisable metadata dict for one skill."""
parts = skill_file.relative_to(skills_dir).parts
@@ -1287,7 +1270,6 @@ def _build_skills_system_prompt_inner(
skills_by_category: dict[str, list[tuple[str, str]]] = {}
category_descriptions: dict[str, str] = {}
# Disk snapshot (fast path) vs. full scan: both yield (entry, is_compatible) pairs so labeling runs identically.
snapshot = _load_skills_snapshot(skills_dir)
if snapshot is not None:
@@ -1317,21 +1299,21 @@ def _build_skills_system_prompt_inner(
proj_dir, iter_project_skill_files(proj_dir), hides, project_names, skills_by_category,
desc_prefix="[project] ", log_fmt="Error reading project skill %s: %s",
)
if project_names:
# Drop shadowed entries BEFORE org labeling so collision flags don't fire on intentional overrides.
visible_entries = [e for e in visible_entries if _entry_name(e) not in project_names]
_label_visible_entries(visible_entries, skills_by_category)
if snapshot is None:
# Drop shadowed entries BEFORE org labeling so collision flags don't fire on intentional overrides.
_label_visible_entries([e for e in visible_entries if _entry_name(e) not in project_names], skills_by_category)
if snapshot is None: # persist for fast cold-start reuse (best-effort)
category_descriptions.update(_read_category_descriptions(skills_dir, "Could not read skill description %s: %s"))
_write_skills_snapshot(skills_dir, [entry for entry, _ in candidates], category_descriptions)
try:
atomic_json_write(_skills_prompt_snapshot_path(), {
"version": _SKILLS_SNAPSHOT_VERSION, "manifest": _build_skills_manifest(skills_dir),
"skills": [entry for entry, _ in candidates], "category_descriptions": category_descriptions,
})
except Exception as e:
logger.debug("Could not write skills prompt snapshot: %s", e)
# External skill directories: scanned directly (read-only, small); names already indexed are skipped.
seen_skill_names: set[str] = {name for cat in skills_by_category.values() for name, _ in cat}
for ext_dir in external_dirs:
if not ext_dir.exists():
continue
for ext_dir in (d for d in external_dirs if d.exists()):
_collect_extra_skills(
ext_dir, iter_skill_index_files(ext_dir, "SKILL.md"), hides, seen_skill_names, skills_by_category,
desc_prefix="", log_fmt="Error reading external skill %s: %s",
@@ -1350,6 +1332,7 @@ def _build_skills_system_prompt_inner(
# Context files (SOUL.md, AGENTS.md, .cursorrules)
def _truncate_content(
content: str,
filename: str,
@@ -1397,10 +1380,7 @@ def load_soul_md(context_length: Optional[int] = None, home_override: "Path | No
content = soul_path.read_text(encoding="utf-8").strip()
if not content:
return None
return _truncate_content(
_scan_context_content(content, "SOUL.md"), "SOUL.md",
context_length=context_length, read_path=str(soul_path),
)
return _truncate_content(_scan_context_content(content, "SOUL.md"), "SOUL.md", context_length=context_length, read_path=str(soul_path))
except Exception as e:
logger.debug("Could not read SOUL.md from %s: %s", soul_path, e)
return None
@@ -1427,10 +1407,8 @@ def _context_section(
"""Threat-scan *content*, render it as ``## <label>``, cap it to the budget (*warn_name* labels warnings)."""
if strip_frontmatter:
content = _strip_yaml_frontmatter(content)
return _truncate_content(
f"## {label}\n\n{_scan_context_content(content, label)}", warn_name,
context_length=context_length, read_path=str(path),
)
body = f"## {label}\n\n{_scan_context_content(content, label)}"
return _truncate_content(body, warn_name, context_length=context_length, read_path=str(path))
def _load_hermes_md(cwd_path: Path, context_length: Optional[int] = None) -> str:
@@ -1479,19 +1457,15 @@ def _load_agents_md(cwd_path: Path, context_length: Optional[int] = None) -> str
if len(sections) <= 1:
return sections[0] if sections else ""
# Per-file budgets applied above; also cap the merged chain so a deep monorepo can't multiply the budget.
return _truncate_content(
"\n\n".join(sections), "AGENTS.md (directory chain)",
context_length=context_length, read_path=str(cwd_resolved / "AGENTS.md"),
)
return _truncate_content("\n\n".join(sections), "AGENTS.md (directory chain)", context_length=context_length, read_path=str(cwd_resolved / "AGENTS.md"))
def _load_claude_md(cwd_path: Path, context_length: Optional[int] = None) -> str:
"""CLAUDE.md / claude.md — cwd only."""
for name in ("CLAUDE.md", "claude.md"):
candidate = cwd_path / name
content = _read_context_file(candidate) if candidate.exists() else ""
content = _read_context_file(cwd_path / name) if (cwd_path / name).exists() else ""
if content:
return _context_section(content, name, "CLAUDE.md", candidate, context_length)
return _context_section(content, name, "CLAUDE.md", cwd_path / name, context_length)
return ""
@@ -1525,7 +1499,6 @@ def build_context_files_prompt(
from HERMES_HOME is independent and always included unless *skip_soul* (already the identity slot).
"""
cwd_path = Path(cwd if cwd is not None else os.getcwd()).resolve()
sections = []
# A FALLBACK-picked cwd inside the Hermes install tree must not gain system-prompt authority (the desktop
# default would load this repo's contributor AGENTS.md). An explicit cwd is honored verbatim.
from agent.runtime_cwd import _is_install_tree
@@ -1536,13 +1509,14 @@ def build_context_files_prompt(
"install tree (%s) — set terminal.cwd to your project directory",
cwd_path,
)
sections = []
else:
sections.append(
sections = [
_load_hermes_md(cwd_path, context_length)
or _load_agents_md(cwd_path, context_length)
or _load_claude_md(cwd_path, context_length)
or _load_cursorrules(cwd_path, context_length)
)
]
if not skip_soul:
sections.append(load_soul_md(context_length, home_override=home_override))
sections = [s for s in sections if s]