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

Merge branch 'simp/r3-06-p2-PB' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:42:43 -07:00
parent 21ab5b1c9f 123d962dc7
commit 4dcf8ce7cf
5 changed files with 179 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/markdown_tables.py
+9 -25
View File
@@ -28,14 +28,9 @@ def _disp_width(s: str) -> int:
return max(wcswidth(s), 0)
def _pad_to_width(s: str, target: int) -> str:
return s + " " * max(0, target - _disp_width(s))
def split_table_row(row: str) -> List[str]:
"""Split ``| a | b | c |`` into ``["a", "b", "c"]`` with trims."""
s = row.strip().removeprefix("|").removesuffix("|")
return [c.strip() for c in s.split("|")]
return [c.strip() for c in row.strip().removeprefix("|").removesuffix("|").split("|")]
def is_table_divider(row: str) -> bool:
@@ -66,14 +61,12 @@ def _render_block(rows: List[List[str]], available_width: int | None = None) ->
ncols = max(len(r) for r in rows)
rows = [r + [""] * (ncols - len(r)) for r in rows]
widths = [max(_MIN_COL_WIDTH, *(_disp_width(r[c]) for r in rows)) for c in range(ncols)]
# `| ` + cell + ` ` per column, plus the closing `|`.
horizontal_width = sum(widths) + 3 * ncols + 1
if available_width is not None and horizontal_width > max(available_width, 20):
if available_width is not None and sum(widths) + 3 * ncols + 1 > max(available_width, 20):
return _render_vertical(rows, ncols, available_width)
def _row(cells: List[str]) -> str:
return "| " + " | ".join(_pad_to_width(c, widths[k]) for k, c in enumerate(cells)) + " |"
return "| " + " | ".join(c + " " * max(0, widths[k] - _disp_width(c)) for k, c in enumerate(cells)) + " |"
out = [_row(rows[0]), "|" + "|".join("-" * (w + 2) for w in widths) + "|"]
out.extend(_row(r) for r in rows[1:])
@@ -90,11 +83,8 @@ def _hard_break(word: str, w: int) -> List[str]:
out.append(buf)
buf, bw = ch, cw
else:
buf += ch
bw += cw
if buf:
out.append(buf)
return out
buf, bw = buf + ch, bw + cw
return out + [buf] if buf else out
def _wrap_to_width(text: str, width: int) -> List[str]:
@@ -108,7 +98,6 @@ def _wrap_to_width(text: str, width: int) -> List[str]:
words = text.split()
if not words:
return [""]
lines: List[str] = []
current, current_w = "", 0
@@ -145,18 +134,14 @@ def _render_vertical(rows: List[List[str]], ncols: int, available_width: int) ->
"""
if not rows:
return []
headers = rows[0] + [""] * (ncols - len(rows[0]))
labels = [h or f"Column {i + 1}" for i, h in enumerate(headers)]
labels = [h or f"Column {i + 1}" for i, h in enumerate(rows[0] + [""] * (ncols - len(rows[0])))]
separator = "─" * (max(20, min(40, available_width - 2)) if available_width else 30)
indent = " "
cont_budget = max(10, available_width - _disp_width(indent))
cont_budget = max(10, available_width - 2) # continuation lines are indented two spaces
out: List[str] = []
for ri, row in enumerate(rows[1:]):
if ri > 0:
out.append(separator)
for ci in range(ncols):
label = labels[ci]
for ci, label in enumerate(labels):
value = row[ci] if ci < len(row) else ""
if not value:
out.append(f"{label}:")
@@ -166,7 +151,7 @@ def _render_vertical(rows: List[List[str]], ncols: int, available_width: int) ->
# Re-flow continuation text at the wider continuation budget.
for cl in _wrap_to_width(" ".join(wrapped[1:]), cont_budget) if len(wrapped) > 1 else ():
if cl.strip():
out.append(f"{indent}{cl}")
out.append(f" {cl}")
return out
@@ -179,7 +164,6 @@ def realign_markdown_tables(text: str, available_width: int | None = None) -> st
"""
if "|" not in text:
return text
lines = text.split("\n")
out: List[str] = []
i, n = 0, len(lines)
agent/onboarding.py
+4 -9
View File
@@ -76,17 +76,13 @@ def busy_input_hint_cli(mode: str) -> str:
def tool_progress_hint_gateway() -> str:
return (
"💡 First-time tip — that tool took a while and I'm streaming every step. If the progress messages "
"feel noisy, send `/verbose` to cycle modes (all → new → off). This notice won't appear again."
)
return ("💡 First-time tip — that tool took a while and I'm streaming every step. If the progress messages "
"feel noisy, send `/verbose` to cycle modes (all → new → off). This notice won't appear again.")
def tool_progress_hint_cli() -> str:
return (
"(tip) That tool ran for a while. Use /verbose to cycle tool-progress "
"display modes (all -> new -> off -> verbose). This tip only shows once."
)
return ("(tip) That tool ran for a while. Use /verbose to cycle tool-progress "
"display modes (all -> new -> off -> verbose). This tip only shows once.")
def openclaw_residue_hint_cli() -> str:
@@ -162,7 +158,6 @@ def mark_seen(config_path: Path, flag: str) -> bool:
except Exception as e: # pragma: no cover — dependency issue
logger.debug("onboarding: failed to import yaml/utils: %s", e)
return False
try:
cfg: dict = {}
if config_path.exists():
agent/plugin_llm.py
+42 -117
View File
@@ -21,9 +21,6 @@ from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, Uni
logger = logging.getLogger(__name__)
# -- public dataclasses -------------------------------------------------------
@dataclass
class PluginLlmTextInput:
"""Text block in a structured input list."""
@@ -89,9 +86,6 @@ class PluginLlmStructuredResult:
audit: Dict[str, Any] = field(default_factory=dict)
# -- trust gate ---------------------------------------------------------------
@dataclass(frozen=True)
class _TrustPolicy:
"""Resolved trust gate for one plugin's LLM access."""
@@ -111,14 +105,8 @@ class _TrustPolicy:
# The ``allow_*_override`` config keys; each is a same-named ``_TrustPolicy`` field.
_OVERRIDE_FLAGS = (
"allow_provider_override", "allow_model_override", "allow_agent_id_override",
"allow_profile_override", "allow_task_override",
)
def _normalize_ref(raw: str) -> str:
return (raw or "").strip().lower()
_OVERRIDE_FLAGS = ("allow_provider_override", "allow_model_override", "allow_agent_id_override",
"allow_profile_override", "allow_task_override")
def _coerce_allowlist(raw: Any) -> tuple[Optional[frozenset], bool]:
@@ -126,7 +114,7 @@ def _coerce_allowlist(raw: Any) -> tuple[Optional[frozenset], bool]:
``allow_any``; missing / non-list → ``(None, False)`` = no allowlist."""
if not isinstance(raw, list):
return None, False
normalized = [_normalize_ref(item) for item in raw if isinstance(item, str)]
normalized = [item.strip().lower() for item in raw if isinstance(item, str)]
return frozenset(item for item in normalized if item and item != "*"), "*" in normalized
@@ -135,19 +123,15 @@ def _resolve_trust_policy(plugin_id: str) -> _TrustPolicy:
restrictive). Resolved per call so config edits apply without a restart."""
if not plugin_id:
return _TrustPolicy(plugin_id="")
try:
from hermes_cli.config import load_config_readonly
config = load_config_readonly() or {}
llm_cfg: Any = (load_config_readonly() or {}).get("plugins")
except Exception: # pragma: no cover — config IO failure
return _TrustPolicy(plugin_id=plugin_id)
llm_cfg: Any = config.get("plugins")
llm_cfg = None
for key in ("entries", plugin_id, "llm"):
llm_cfg = llm_cfg.get(key) if isinstance(llm_cfg, dict) else None
if not isinstance(llm_cfg, dict):
return _TrustPolicy(plugin_id=plugin_id)
allowed_models, allow_any_model = _coerce_allowlist(llm_cfg.get("allowed_models"))
allowed_providers, allow_any_provider = _coerce_allowlist(llm_cfg.get("allowed_providers"))
return _TrustPolicy(
@@ -163,9 +147,7 @@ class PluginLlmTrustError(PermissionError):
def _denied(plugin_id: str, what: str, flag: str) -> PluginLlmTrustError:
return PluginLlmTrustError(
f"Plugin {plugin_id!r} cannot {what} "
f"(set plugins.entries.{plugin_id}.llm.{flag} to true to allow)."
)
f"Plugin {plugin_id!r} cannot {what} (set plugins.entries.{plugin_id}.llm.{flag} to true to allow).")
def _gate_ref_override(policy: _TrustPolicy, kind: str, requested: str) -> str:
@@ -173,12 +155,9 @@ def _gate_ref_override(policy: _TrustPolicy, kind: str, requested: str) -> str:
if not getattr(policy, f"allow_{kind}_override"):
raise _denied(policy.plugin_id, f"override the {kind}", f"allow_{kind}_override")
allowed = getattr(policy, f"allowed_{kind}s")
if not getattr(policy, f"allow_any_{kind}") and allowed is not None and _normalize_ref(requested) not in allowed:
raise PluginLlmTrustError(
f"Plugin {policy.plugin_id!r} {kind} override "
f"{requested!r} is not in plugins.entries."
f"{policy.plugin_id}.llm.allowed_{kind}s."
)
if not getattr(policy, f"allow_any_{kind}") and allowed is not None and requested.strip().lower() not in allowed:
raise PluginLlmTrustError(f"Plugin {policy.plugin_id!r} {kind} override {requested!r} is not in "
f"plugins.entries.{policy.plugin_id}.llm.allowed_{kind}s.")
return requested.strip()
@@ -199,8 +178,7 @@ def _check_overrides(
for kind, requested in (("agent_id", requested_agent_id), ("profile", requested_profile)):
if requested and not getattr(policy, f"allow_{kind}_override"):
raise _denied(policy.plugin_id, _FLAG_ONLY_OVERRIDES[kind], f"allow_{kind}_override")
final_profile = requested_profile.strip() if requested_profile else None
return final_provider, final_model, requested_agent_id, final_profile
return final_provider, final_model, requested_agent_id, requested_profile.strip() if requested_profile else None
def _resolve_task_ownership(plugin_id: str) -> tuple[frozenset, frozenset]:
@@ -214,10 +192,8 @@ def _resolve_task_ownership(plugin_id: str) -> tuple[frozenset, frozenset]:
builtin: set = set()
try:
from hermes_cli.plugins import get_plugin_auxiliary_tasks
owned = {
e.get("key") for e in get_plugin_auxiliary_tasks()
if e.get("plugin") == plugin_id and isinstance(e.get("key"), str) and e.get("key")
}
owned = {e.get("key") for e in get_plugin_auxiliary_tasks()
if e.get("plugin") == plugin_id and isinstance(e.get("key"), str) and e.get("key")}
except Exception: # pragma: no cover — registry unavailable
pass
try:
@@ -239,21 +215,15 @@ def _check_task(policy: _TrustPolicy, *, plugin_id: str, requested_task: Optiona
task = (requested_task or "").strip()
if not task or task.lower() == "auto":
return None
owned, builtin = _resolve_task_ownership(plugin_id)
if task in owned or (task in builtin and policy.allow_task_override):
return task
if task in builtin:
logger.warning(
"plugin_llm task routing denied: plugin %r requested built-in "
"auxiliary task %r without plugins.entries.%s.llm.allow_task_override",
plugin_id, task, plugin_id,
)
logger.warning("plugin_llm task routing denied: plugin %r requested built-in "
"auxiliary task %r without plugins.entries.%s.llm.allow_task_override", plugin_id, task, plugin_id)
raise _denied(plugin_id, f"route through the built-in auxiliary task {task!r}", "allow_task_override")
logger.warning(
"plugin_llm task routing denied: plugin %r requested auxiliary task %r it did not register",
plugin_id, task,
)
logger.warning("plugin_llm task routing denied: plugin %r requested auxiliary task %r it did not register",
plugin_id, task)
raise PluginLlmTrustError(
f"Plugin {plugin_id!r} cannot route through auxiliary task {task!r} — a "
f"plugin may only pass a task key it registered itself via "
@@ -262,9 +232,6 @@ def _check_task(policy: _TrustPolicy, *, plugin_id: str, requested_task: Optiona
)
# -- input normalization ------------------------------------------------------
def _normalize_input_block(block: PluginLlmInput) -> Dict[str, Any]:
"""Coerce a structured input block to a plain dict. Unknown shapes raise ``ValueError``."""
if isinstance(block, PluginLlmTextInput):
@@ -280,17 +247,14 @@ def _normalize_input_block(block: PluginLlmInput) -> Dict[str, Any]:
raise ValueError(f"Unsupported input block: {type(block).__name__}")
kind = block.get("type")
if kind == "text":
text = block.get("text")
if not isinstance(text, str):
if not isinstance(block.get("text"), str):
raise ValueError("text input block requires 'text' string")
return {"type": "text", "text": text}
return {"type": "text", "text": block["text"]}
if kind == "image":
if "data" not in block and not block.get("url"):
raise ValueError("image input block requires 'data' bytes or 'url'")
return {
"type": "image", "data": block.get("data"), "url": block.get("url"),
"mime_type": block.get("mime_type") or "image/png", "file_name": block.get("file_name") or "",
}
return {"type": "image", "data": block.get("data"), "url": block.get("url"),
"mime_type": block.get("mime_type") or "image/png", "file_name": block.get("file_name") or ""}
raise ValueError(f"Unknown input block type: {kind!r}")
@@ -301,8 +265,7 @@ def _image_part(norm: Dict[str, Any]) -> Dict[str, Any]:
data = norm.get("data") or b""
if not isinstance(data, (bytes, bytearray)):
raise ValueError("image input 'data' must be bytes")
b64 = base64.b64encode(data).decode("ascii")
url = f"data:{norm.get('mime_type') or 'image/png'};base64,{b64}"
url = f"data:{norm.get('mime_type') or 'image/png'};base64,{base64.b64encode(data).decode('ascii')}"
return {"type": "image_url", "image_url": {"url": url}}
@@ -316,13 +279,10 @@ def _build_structured_messages(
messages: List[Dict[str, Any]] = []
sys_parts: List[str] = [system_prompt.strip()] if system_prompt else []
if json_mode or json_schema is not None:
sys_parts.append(
"Respond with a single JSON object that matches the requested shape. "
"Do not include prose or markdown fences."
)
sys_parts.append("Respond with a single JSON object that matches the requested shape. "
"Do not include prose or markdown fences.")
if sys_parts:
messages.append({"role": "system", "content": "\n\n".join(sys_parts)})
header = instructions.strip()
if schema_name:
header = f"{header}\n\nSchema name: {schema_name}"
@@ -340,9 +300,6 @@ def _build_structured_messages(
return messages
# -- JSON parsing / response extraction --------------------------------------
_FENCE_RE = re.compile(r"```(?:json)?\s*(.+?)```", re.DOTALL | re.IGNORECASE)
@@ -352,20 +309,16 @@ def _strip_code_fences(text: str) -> str:
return match.group(1).strip() if match else text.strip()
def _parse_structured_text(
*, text: str, json_mode: bool, json_schema: Optional[Any]
) -> tuple[Optional[Any], str]:
def _parse_structured_text(*, text: str, json_mode: bool, json_schema: Optional[Any]) -> tuple[Optional[Any], str]:
"""``(parsed, content_type)``: ``"json"`` when parsing (and schema validation, if
given) succeeded, ``"text"`` otherwise. Schema violations raise ``ValueError``;
a missing ``jsonschema`` package skips validation with a debug log."""
if not (json_mode or json_schema is not None) or not text:
return None, "text"
try:
parsed = json.loads(_strip_code_fences(text))
except (json.JSONDecodeError, ValueError):
return None, "text"
if json_schema is not None:
try:
import jsonschema # type: ignore[import-untyped]
@@ -374,7 +327,6 @@ def _parse_structured_text(
logger.debug("jsonschema unavailable; skipping schema validation")
except jsonschema.ValidationError as exc: # type: ignore[attr-defined]
raise ValueError(f"Plugin LLM structured output did not match schema: {exc.message}") from exc
return parsed, "json"
@@ -397,7 +349,6 @@ def _extract_usage(response: Any) -> PluginLlmUsage:
except (TypeError, ValueError):
pass
return 0
inp, out = _g("prompt_tokens", "input_tokens"), _g("completion_tokens", "output_tokens")
return PluginLlmUsage(
input_tokens=inp, output_tokens=out, total_tokens=_g("total_tokens") or (inp + out),
@@ -413,11 +364,8 @@ def _extract_text(response: Any) -> str:
if isinstance(content, str):
return content
if isinstance(content, list):
texts = (
(part.get("text") if part.get("type") == "text" else None)
if isinstance(part, dict) else getattr(part, "text", None)
for part in content
)
texts = ((part.get("text") if part.get("type") == "text" else None)
if isinstance(part, dict) else getattr(part, "text", None) for part in content)
return "".join(t for t in texts if isinstance(t, str))
except (AttributeError, IndexError, TypeError):
pass
@@ -433,10 +381,8 @@ def _main_config_value(reader: str, default: str) -> str:
return default
def _resolve_attribution(
*, provider_override: Optional[str], model_override: Optional[str], response: Any,
route_info: Optional[Dict[str, str]] = None,
) -> tuple[str, str]:
def _resolve_attribution(*, provider_override: Optional[str], model_override: Optional[str], response: Any,
route_info: Optional[Dict[str, str]] = None) -> tuple[str, str]:
"""``(provider, model)`` to record on the result.
Provider: route selected by ``auxiliary_client`` > explicit override > current
@@ -450,9 +396,6 @@ def _resolve_attribution(
return provider, route_info.get("model") or model_override or _main_config_value("_read_main_model", "default")
# -- PluginLlm facade ---------------------------------------------------------
def _json_response_format(*, json_mode: bool, json_schema: Optional[Any]) -> Optional[Dict[str, Any]]:
"""``extra_body.response_format``; falls back to ``json_object`` without a
schema so schema-blind providers still get a hint."""
@@ -473,10 +416,8 @@ def _structured_spec(
raise ValueError(f"{name} requires non-empty instructions")
if not input:
raise ValueError(f"{name} requires at least one input block")
return dict(
instructions=instructions, inputs=list(input), system_prompt=system_prompt,
json_mode=json_mode, json_schema=json_schema, schema_name=schema_name,
)
return dict(instructions=instructions, inputs=list(input), system_prompt=system_prompt,
json_mode=json_mode, json_schema=json_schema, schema_name=schema_name)
class PluginLlm:
@@ -498,8 +439,6 @@ class PluginLlm:
self._sync_caller = sync_caller
self._async_caller = async_caller
# -- public API -----------------------------------------------------------
def complete(
self, messages: List[Dict[str, Any]], *, provider: Optional[str] = None, model: Optional[str] = None,
temperature: Optional[float] = None, max_tokens: Optional[int] = None,
@@ -552,8 +491,6 @@ class PluginLlm:
agent, kw = self._gate(provider, model, agent_id, profile, task, None, temperature, max_tokens, timeout, spec)
return self._finish("acomplete_structured", agent, kw, await self._invoke_async(kw), purpose, spec)
# -- shared core ----------------------------------------------------------
def _gate(
self, provider: Optional[str], model: Optional[str], agent_id: Optional[str], profile: Optional[str],
task: Optional[str], messages: Optional[List[Dict[str, Any]]], temperature: Optional[float],
@@ -567,18 +504,15 @@ class PluginLlm:
policy = self._policy_loader(self._plugin_id)
eff_task = _check_task(policy, plugin_id=self._plugin_id, requested_task=task)
eff_provider, eff_model, eff_agent, eff_profile = _check_overrides(
policy, requested_provider=provider, requested_model=model,
requested_agent_id=agent_id, requested_profile=profile,
)
policy, requested_provider=provider, requested_model=model, requested_agent_id=agent_id,
requested_profile=profile)
extra_body = None
if spec is not None:
messages = _build_structured_messages(**spec)
extra_body = _json_response_format(json_mode=spec["json_mode"], json_schema=spec["json_schema"])
return eff_agent, dict(
messages=messages, provider_override=eff_provider, model_override=eff_model,
profile_override=eff_profile, temperature=temperature, max_tokens=max_tokens,
timeout=timeout, extra_body=extra_body, task=eff_task,
)
return eff_agent, dict(messages=messages, provider_override=eff_provider, model_override=eff_model,
profile_override=eff_profile, temperature=temperature, max_tokens=max_tokens,
timeout=timeout, extra_body=extra_body, task=eff_task)
def _finish(
self, name: str, agent_id: Optional[str], kw: Dict[str, Any], invoked: tuple[str, str, Any],
@@ -605,8 +539,6 @@ class PluginLlm:
logger.info(fmt + "tokens=%d", *log_args, usage.total_tokens)
return cls(**fields, audit=audit)
# -- host invocation ------------------------------------------------------
@staticmethod
def _host_kwargs(kw: Dict[str, Any]) -> tuple[Dict[str, Any], Optional[Dict[str, str]]]:
"""Call kwargs → ``call_llm`` kwargs. The auth profile rides in
@@ -616,19 +548,14 @@ class PluginLlm:
if kw["profile_override"]:
merged_extra.setdefault("metadata", {})["auth_profile"] = kw["profile_override"]
route_info: Optional[Dict[str, str]] = {} if kw["task"] else None
return dict(
task=kw["task"], provider=kw["provider_override"], model=kw["model_override"],
messages=kw["messages"], temperature=kw["temperature"], max_tokens=kw["max_tokens"],
timeout=kw["timeout"], extra_body=merged_extra or None, route_info=route_info,
), route_info
return dict(task=kw["task"], provider=kw["provider_override"], model=kw["model_override"],
messages=kw["messages"], temperature=kw["temperature"], max_tokens=kw["max_tokens"],
timeout=kw["timeout"], extra_body=merged_extra or None, route_info=route_info), route_info
@staticmethod
def _attributed(kw: Dict[str, Any], response: Any, route_info: Optional[Dict[str, str]]) -> tuple[str, str, Any]:
provider, model = _resolve_attribution(
provider_override=kw["provider_override"], model_override=kw["model_override"],
response=response, route_info=route_info,
)
return provider, model, response
return (*_resolve_attribution(provider_override=kw["provider_override"], model_override=kw["model_override"],
response=response, route_info=route_info), response)
def _invoke_sync(self, kw: Dict[str, Any]) -> tuple[str, str, Any]:
"""Host ``call_llm`` (lazy import: circular deps at plugin discovery) →
@@ -649,10 +576,8 @@ class PluginLlm:
return self._attributed(kw, await async_call_llm(**call_kw), route_info)
def make_plugin_llm_for_test(
*, plugin_id: str, policy: _TrustPolicy, sync_caller: Optional[Callable[..., Any]] = None,
async_caller: Optional[Callable[..., Awaitable[Any]]] = None,
) -> PluginLlm:
def make_plugin_llm_for_test(*, plugin_id: str, policy: _TrustPolicy, sync_caller: Optional[Callable[..., Any]] = None,
async_caller: Optional[Callable[..., Awaitable[Any]]] = None) -> PluginLlm:
""":class:`PluginLlm` with an injected policy and caller (no config.yaml, no
provider). Not part of the public plugin API."""
return PluginLlm(plugin_id=plugin_id, policy_loader=lambda _pid: policy, sync_caller=sync_caller, async_caller=async_caller)
agent/process_bootstrap.py
+18 -49
View File
@@ -30,15 +30,12 @@ def _interleave_addrinfos(addrinfos: list[tuple]) -> list[tuple]:
seen: set[tuple] = set()
for addrinfo in addrinfos:
family, socktype, proto, _canonname, sockaddr = addrinfo
marker = (family, socktype, proto, sockaddr)
if marker not in seen:
seen.add(marker)
if (family, socktype, proto, sockaddr) not in seen:
seen.add((family, socktype, proto, sockaddr))
queues.setdefault(family, []).append(addrinfo)
interleaved: list[tuple] = []
while any(queues.values()):
for queue in queues.values():
if queue:
interleaved.append(queue.pop(0))
interleaved.extend(queue.pop(0) for queue in queues.values() if queue)
return interleaved
@@ -49,12 +46,8 @@ def _quiet_unregister(selector, sock) -> None:
pass
def _happy_eyeballs_create_connection(
address: tuple[str, int],
timeout: Optional[float],
source_address: Optional[tuple[str, int]] = None,
socket_options=(),
):
def _happy_eyeballs_create_connection(address: tuple[str, int], timeout: Optional[float],
source_address: Optional[tuple[str, int]] = None, socket_options=()):
"""RFC 8305-style connect: staggered non-blocking attempts across families.
``socket.create_connection`` tries addresses serially, so broken-but-
@@ -101,7 +94,6 @@ def _happy_eyeballs_create_connection(
now = time.monotonic()
if deadline is not None and now >= deadline:
raise socket.timeout("timed out")
if pending and now >= next_launch:
try:
winner = start_attempt(pending.pop(0))
@@ -113,12 +105,10 @@ def _happy_eyeballs_create_connection(
if winner is not None:
break
next_launch = now + _HAPPY_EYEBALLS_DELAY_SECONDS
wait_timeout = None if deadline is None else max(0.0, deadline - now)
if pending:
until_launch = max(0.0, next_launch - now)
wait_timeout = until_launch if wait_timeout is None else min(wait_timeout, until_launch)
for key, _mask in selector.select(wait_timeout):
candidate = key.fileobj
error_code = candidate.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
@@ -136,7 +126,6 @@ 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}")
_quiet_unregister(selector, winner)
active.discard(winner)
winner.settimeout(timeout)
@@ -160,22 +149,17 @@ class _HappyEyeballsSyncBackend:
def _default_backend(self):
if self._fallback is None:
from httpcore import SyncBackend
self._fallback = SyncBackend()
return self._fallback
def connect_tcp(
self, host: str, port: int, timeout: Optional[float] = None,
local_address: Optional[str] = None, socket_options=None,
):
def connect_tcp(self, host: str, port: int, timeout: Optional[float] = None, local_address: Optional[str] = None,
socket_options=None):
from httpcore import ConnectError, ConnectTimeout
from httpcore._backends.sync import SyncStream
source_address = None if local_address is None else (local_address, 0)
try:
sock = _happy_eyeballs_create_connection(
(host, port), timeout, source_address=source_address, socket_options=socket_options or ()
)
sock = _happy_eyeballs_create_connection((host, port), timeout, source_address=source_address,
socket_options=socket_options or ())
except socket.timeout as exc:
raise ConnectTimeout(str(exc)) from exc
except OSError as exc:
@@ -201,11 +185,8 @@ def _enable_happy_eyeballs(transport, skip_pool_types: tuple = ()) -> None:
serial backend. Pools of ``skip_pool_types`` (proxies) are left alone.
"""
pool = getattr(transport, "_pool", None)
if pool is None or not hasattr(pool, "_network_backend"):
return
if skip_pool_types and isinstance(pool, skip_pool_types):
return
pool._network_backend = _HappyEyeballsSyncBackend()
if pool is not None and hasattr(pool, "_network_backend") and not (skip_pool_types and isinstance(pool, skip_pool_types)):
pool._network_backend = _HappyEyeballsSyncBackend()
def enable_happy_eyeballs_on_client(client) -> None:
@@ -217,14 +198,11 @@ def enable_happy_eyeballs_on_client(client) -> None:
"""
try:
import httpcore
proxy_pool_types = tuple(
t for t in (getattr(httpcore, "HTTPProxy", None), getattr(httpcore, "SOCKSProxy", None)) if t is not None
)
t for t in (getattr(httpcore, "HTTPProxy", None), getattr(httpcore, "SOCKSProxy", None)) if t is not None)
except Exception:
return
transports = [getattr(client, "_transport", None), *(getattr(client, "_mounts", None) or {}).values()]
for transport in transports:
for transport in (getattr(client, "_transport", None), *(getattr(client, "_mounts", None) or {}).values()):
_enable_happy_eyeballs(transport, proxy_pool_types)
@@ -232,8 +210,7 @@ def _load_openai_cls() -> type:
"""Import and cache ``openai.OpenAI``."""
global _OPENAI_CLS_CACHE
if _OPENAI_CLS_CACHE is None:
from openai import OpenAI as _cls
_OPENAI_CLS_CACHE = _cls
from openai import OpenAI as _OPENAI_CLS_CACHE
return _OPENAI_CLS_CACHE
@@ -292,25 +269,18 @@ class _SafeWriter:
def _get_proxy_from_env() -> Optional[str]:
"""First configured proxy URL from HTTPS_PROXY / HTTP_PROXY / ALL_PROXY (any case), or None."""
for key in ("HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "https_proxy", "http_proxy", "all_proxy"):
value = os.environ.get(key, "").strip()
if value:
return normalize_proxy_url(value)
return None
keys = ("HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "https_proxy", "http_proxy", "all_proxy")
return next((normalize_proxy_url(v) for k in keys if (v := os.environ.get(k, "").strip())), None)
def _get_proxy_for_base_url(base_url: Optional[str]) -> Optional[str]:
"""Env-configured proxy unless NO_PROXY excludes this base URL."""
proxy = _get_proxy_from_env()
host = base_url_hostname(base_url) if proxy and base_url else ""
if not host:
return proxy
try:
if urllib.request.proxy_bypass_environment(host):
return None
return None if host and urllib.request.proxy_bypass_environment(host) else proxy
except Exception:
pass
return proxy
return proxy
def build_keepalive_http_client(base_url: str = "", *, async_mode: bool = False, verify: Any = True) -> Optional[Any]:
@@ -324,7 +294,6 @@ def build_keepalive_http_client(base_url: str = "", *, async_mode: bool = False,
"""
try:
import httpx
proxy = _get_proxy_for_base_url(base_url)
limits = httpx.Limits(max_keepalive_connections=20, max_connections=100, keepalive_expiry=20.0)
timeout = httpx.Timeout(connect=15.0, read=None, write=15.0, pool=10.0) # read=None for SSE streaming
agent/prompt_builder.py
+106 -205
View File
@@ -50,10 +50,7 @@ def _scan_context_content(content: str, filename: str) -> str:
def _find_git_root(start: Path) -> Optional[Path]:
"""Nearest ancestor (or *start* itself) containing ``.git``, else None."""
current = start.resolve()
for parent in [current, *current.parents]:
if (parent / ".git").exists():
return parent
return None
return next((p for p in (current, *current.parents) if (p / ".git").exists()), None)
def _find_hermes_md(cwd: Path) -> Optional[Path]:
@@ -62,23 +59,17 @@ def _find_hermes_md(cwd: Path) -> Optional[Path]:
current = cwd.resolve()
# No git root: cwd only — walking parents could pick up a file planted in /tmp, /home, etc.
for directory in [current, *current.parents] if stop_at else [current]:
for name in (".hermes.md", "HERMES.md"):
if (directory / name).is_file():
return directory / name
if directory == stop_at:
break
found = next((directory / n for n in (".hermes.md", "HERMES.md") if (directory / n).is_file()), None)
if found or directory == stop_at:
return found
return None
def _strip_yaml_frontmatter(content: str) -> str:
"""Drop optional ``---`` YAML frontmatter so only the markdown body is injected."""
content = content.lstrip("\ufeff")
if content.startswith("---"):
end = content.find("\n---", 3)
if end != -1:
body = content[end + 4:].lstrip("\n")
return body if body else content
return content
end = content.find("\n---", 3) if content.startswith("---") else -1
return (content[end + 4:].lstrip("\n") or content) if end != -1 else content
DEFAULT_AGENT_IDENTITY = (
@@ -254,7 +245,6 @@ TOOL_USE_ENFORCEMENT_GUIDANCE = (
"user. Responses that only describe intentions without acting are not acceptable."
)
# Model name substrings that trigger tool-use enforcement guidance.
TOOL_USE_ENFORCEMENT_MODELS = ("gpt", "codex", "gemini", "gemma", "grok", "glm", "qwen", "deepseek")
# Models that receive OPENAI_MODEL_EXECUTION_GUIDANCE when agent.execution_guidance is "auto" (agentic-eval
@@ -371,6 +361,8 @@ def execution_guidance_text(valid_tool_names=None) -> str:
text = text.replace("- Current facts (weather, news, versions) → use web_search\n", "")
text = text.replace("(search_files, web_search, read_file, etc.)", "(search_files, read_file, etc.)")
return text
# Gemini/Gemma-specific operational guidance, adapted from OpenCode's gemini.txt.
# Injected alongside TOOL_USE_ENFORCEMENT_GUIDANCE when the model is Gemini or Gemma.
GOOGLE_MODEL_OPERATIONAL_GUIDANCE = (
@@ -675,7 +667,6 @@ def _plugin_backend_attr(backend: str, attr: str, default=None):
"""*attr* of a plugin-registered terminal backend, fail-soft (unknown backend / raising plugin -> *default*)."""
try:
from agent.terminal_env_registry import provider_flag
return provider_flag(backend, attr, default)
except Exception:
return default
@@ -683,19 +674,16 @@ def _plugin_backend_attr(backend: str, attr: str, default=None):
def _plugin_backend_is_remote(backend: str) -> bool:
"""Whether a plugin-registered terminal backend runs commands remotely (unknown names are local)."""
if not backend or backend in _REMOTE_TERMINAL_BACKENDS or backend == "local":
return False
return bool(_plugin_backend_attr(backend, "is_remote", False))
return bool(backend and backend != "local" and backend not in _REMOTE_TERMINAL_BACKENDS
and _plugin_backend_attr(backend, "is_remote", False))
def _windows_marketing_version() -> str:
""""10"/"11" (``platform.release()`` says 10 for both; 11 is build >= 22000)."""
try:
build = sys.getwindowsversion().build # type: ignore[attr-defined]
return "11" if build >= 22000 else "10"
return "11" if sys.getwindowsversion().build >= 22000 else "10" # type: ignore[attr-defined]
except Exception:
import platform
return platform.release()
@@ -748,19 +736,13 @@ def _run_backend_probe(env_type: str, terminal_tool) -> str:
"""Execute the probe command inside a freshly built backend; "" when it yields nothing."""
config = terminal_tool._get_env_config()
# Mirrors tools/terminal_tool.py's live-command assembly (`_create_environment` is the factory).
image_key = _BACKEND_IMAGE_KEYS.get(env_type)
env = terminal_tool._create_environment(
env_type=env_type,
image=config.get(image_key, "") if image_key else "",
cwd=config.get("cwd", ""),
env_type=env_type, image=config.get(_BACKEND_IMAGE_KEYS[env_type], "") if env_type in _BACKEND_IMAGE_KEYS else "", cwd=config.get("cwd", ""),
timeout=config.get("timeout", 180),
ssh_config=terminal_tool._ssh_config_from_config(config) if env_type == "ssh" else None,
container_config=(
{k: config.get(k, d) for k, d in _CONTAINER_CONFIG_DEFAULTS}
if terminal_tool._is_container_backend(env_type) else None
),
task_id="prompt-backend-probe",
host_cwd=config.get("host_cwd"),
container_config=({k: config.get(k, d) for k, d in _CONTAINER_CONFIG_DEFAULTS}
if terminal_tool._is_container_backend(env_type) else None),
task_id="prompt-backend-probe", host_cwd=config.get("host_cwd"),
)
result = env.execute(_BACKEND_PROBE_CMD, timeout=4)
if result.get("returncode") != 0:
@@ -783,20 +765,19 @@ def _format_backend_probe(output: str) -> str:
def _probe_remote_backend(env_type: str) -> str | None:
"""Describe the active non-local backend via a live probe; None if it failed (cached, failures included)."""
cache_key = (env_type, _tenv_read("TERMINAL_CWD", ""))
cached = _BACKEND_PROBE_CACHE.get(cache_key)
if cached is not None:
return cached or None
formatted = ""
try:
import tools.terminal_tool as terminal_tool # heavy; only needed for non-local backends
except Exception as e:
logger.debug("Backend probe unavailable (import failed): %s", e)
else:
formatted = _BACKEND_PROBE_CACHE.get(cache_key)
if formatted is None:
formatted = ""
try:
formatted = _format_backend_probe(_run_backend_probe(env_type, terminal_tool))
import tools.terminal_tool as terminal_tool # heavy; only needed for non-local backends
except Exception as e:
logger.debug("Backend probe failed: %s", e)
_BACKEND_PROBE_CACHE[cache_key] = formatted
logger.debug("Backend probe unavailable (import failed): %s", e)
else:
try:
formatted = _format_backend_probe(_run_backend_probe(env_type, terminal_tool))
except Exception as e:
logger.debug("Backend probe failed: %s", e)
_BACKEND_PROBE_CACHE[cache_key] = formatted
return formatted or None
@@ -815,25 +796,20 @@ def _local_host_hints() -> list[str]:
host_lines.append(f"Current working directory: {resolve_agent_cwd()}")
except OSError:
pass
native_windows = sys.platform == "win32" and not is_wsl()
if native_windows:
host_lines.append(
"Note: on Windows, the machine hostname (e.g. from `hostname` or uname) is NOT the username. "
"Use the 'User home directory' above to construct paths under C:\\Users\\<user>\\, never the hostname."
)
hints = ["\n".join(host_lines)]
if not (sys.platform == "win32" and not is_wsl()):
return ["\n".join(host_lines)]
host_lines.append(
"Note: on Windows, the machine hostname (e.g. from `hostname` or uname) is NOT the username. "
"Use the 'User home directory' above to construct paths under C:\\Users\\<user>\\, never the hostname."
)
# Windows-local terminal runs bash, not PowerShell — without this the model issues PowerShell syntax.
if native_windows:
hints.append(_WINDOWS_BASH_SHELL_HINT)
return hints
return ["\n".join(host_lines), _WINDOWS_BASH_SHELL_HINT]
def _remote_backend_hint(backend: str) -> str:
"""Backend-only block for remote/sandbox backends (host info deliberately suppressed)."""
lead = (
f"Terminal backend: {backend}. Your `terminal`, `read_file`, `write_file`, `patch`, and "
f"`search_files` tools all operate inside "
)
lead = (f"Terminal backend: {backend}. Your `terminal`, `read_file`, `write_file`, `patch`, and "
f"`search_files` tools all operate inside ")
probe = _probe_remote_backend(backend)
if probe:
return lead + (
@@ -856,7 +832,6 @@ def _config_readonly(what: str) -> dict:
"""config.yaml as a dict, or {} when unreadable (logged at debug with *what* for context)."""
try:
from hermes_cli.config import load_config_readonly
return load_config_readonly()
except Exception as e:
logger.debug("Could not read %s from config: %s", what, e)
@@ -866,10 +841,8 @@ def _config_readonly(what: str) -> dict:
def _embedder_environment_hint() -> str:
"""Embedder-supplied environment description: HERMES_ENVIRONMENT_HINT (container ENV)
wins over config.yaml ``agent.environment_hint``. Read once at prompt-build time."""
extra = (os.getenv("HERMES_ENVIRONMENT_HINT") or "").strip()
if not extra:
extra = str((_config_readonly("agent.environment_hint").get("agent", {}) or {}).get("environment_hint", "")).strip()
return extra
return (os.getenv("HERMES_ENVIRONMENT_HINT") or "").strip() or str(
(_config_readonly("agent.environment_hint").get("agent", {}) or {}).get("environment_hint", "")).strip()
def build_environment_hints() -> str:
@@ -905,9 +878,7 @@ def _dynamic_context_file_max_chars(context_length: Optional[int]) -> int:
def _get_context_file_max_chars(context_length: Optional[int] = None) -> int:
"""Context-file truncation limit: explicit config.yaml ``context_file_max_chars`` wins, else the dynamic cap."""
val = _config_readonly("context_file_max_chars").get("context_file_max_chars")
if isinstance(val, (int, float)) and val > 0:
return int(val)
return _dynamic_context_file_max_chars(context_length)
return int(val) if isinstance(val, (int, float)) and val > 0 else _dynamic_context_file_max_chars(context_length)
# Truncation warnings for run_agent to surface. A ContextVar so concurrent gateway prompt builds cannot
@@ -917,13 +888,6 @@ _truncation_warnings: "contextvars.ContextVar[Optional[list]]" = contextvars.Con
)
def _record_truncation_warning(msg: str) -> None:
warnings = _truncation_warnings.get()
if warnings is None:
_truncation_warnings.set(warnings := [])
warnings.append(msg)
def drain_truncation_warnings() -> list:
"""Return and clear any truncation warnings accumulated in this context."""
warnings = _truncation_warnings.get() or []
@@ -949,10 +913,9 @@ def clear_skills_system_prompt_cache(*, clear_snapshot: bool = False) -> None:
"""Drop the in-process skills prompt cache (and optionally the disk snapshot)."""
with _SKILLS_PROMPT_CACHE_LOCK:
_SKILLS_PROMPT_CACHE.clear()
if not clear_snapshot:
return
try:
_skills_prompt_snapshot_path().unlink(missing_ok=True)
if clear_snapshot:
_skills_prompt_snapshot_path().unlink(missing_ok=True)
except OSError as e:
logger.debug("Could not remove skills prompt snapshot: %s", e)
@@ -976,35 +939,26 @@ def _build_skills_manifest(skills_dir: Path) -> dict[str, list[int]]:
dirs.remove(ORG_MIRROR_DIR_NAME)
elif root == org_root:
dirs[:] = [d for d in dirs if d == active_org]
dirs[:] = [
d for d in dirs
if d not in EXCLUDED_SKILL_DIRS and not (has_skill_md and d in SKILL_SUPPORT_DIRS)
]
dirs[:] = [d for d in dirs if d not in EXCLUDED_SKILL_DIRS and not (has_skill_md and d in SKILL_SUPPORT_DIRS)]
for filename in ("SKILL.md", "DESCRIPTION.md"):
path = os.path.join(root, filename)
try:
st = os.stat(path) if filename in files else None
if filename in files:
st = os.stat(path)
manifest[path[prefix_len:]] = [st.st_mtime_ns, st.st_size]
except OSError:
st = None
if st is not None:
manifest[path[prefix_len:]] = [st.st_mtime_ns, st.st_size]
pass
return manifest
def _load_skills_snapshot(skills_dir: Path) -> Optional[dict]:
"""The disk snapshot if it exists, is current-version, and its manifest still matches."""
snapshot_path = _skills_prompt_snapshot_path()
if not snapshot_path.exists():
return None
try:
snapshot = json.loads(snapshot_path.read_text(encoding="utf-8"))
except Exception:
snapshot = json.loads(_skills_prompt_snapshot_path().read_text(encoding="utf-8"))
except Exception: # missing, unreadable or corrupt -> rebuild
return None
if (
isinstance(snapshot, dict)
and snapshot.get("version") == _SKILLS_SNAPSHOT_VERSION
and snapshot.get("manifest") == _build_skills_manifest(skills_dir)
):
if (isinstance(snapshot, dict) and snapshot.get("version") == _SKILLS_SNAPSHOT_VERSION
and snapshot.get("manifest") == _build_skills_manifest(skills_dir)):
return snapshot
return None
@@ -1016,18 +970,13 @@ def _build_snapshot_entry(skill_file: Path, skills_dir: Path, frontmatter: dict,
org_id: str | None = None
if len(parts) >= 3 and parts[0] == ORG_MIRROR_DIR_NAME:
org_id, parts = parts[1], parts[2:]
if len(parts) >= 2:
skill_name, category = parts[-2], ("/".join(parts[:-2]) if len(parts) > 2 else parts[0])
else:
skill_name, category = skill_file.parent.name, "general"
skill_name = skill_file.parent.name # == parts[-2] whenever a parent component exists
category = "general" if len(parts) < 2 else "/".join(parts[:-2]) if len(parts) > 2 else parts[0]
platforms = frontmatter.get("platforms") or []
platforms = [platforms] if isinstance(platforms, str) else platforms
entry = {
"skill_name": skill_name,
"category": category,
"frontmatter_name": str(frontmatter.get("name", skill_name)),
"description": description,
"platforms": [str(p).strip() for p in platforms if str(p).strip()],
"skill_name": skill_name, "category": category, "frontmatter_name": str(frontmatter.get("name", skill_name)),
"description": description, "platforms": [str(p).strip() for p in platforms if str(p).strip()],
"conditions": extract_skill_conditions(frontmatter),
}
if org_id:
@@ -1054,9 +1003,7 @@ def _parse_skill_file(skill_file: Path) -> tuple[bool, dict, str]:
def _skill_should_show(
conditions: dict,
available_tools: "set[str] | None",
available_toolsets: "set[str] | None",
conditions: dict, available_tools: "set[str] | None", available_toolsets: "set[str] | None",
session_platform: "str | None" = None,
) -> bool:
"""False if the skill's conditional activation rules exclude it."""
@@ -1089,10 +1036,8 @@ def _current_session_platform_hint() -> str:
def build_skills_system_prompt(
available_tools: "set[str] | None" = None,
available_toolsets: "set[str] | None" = None,
compact_categories: "frozenset[str] | None" = None,
skills_dir_override: "Path | None" = None,
available_tools: "set[str] | None" = None, available_toolsets: "set[str] | None" = None,
compact_categories: "frozenset[str] | None" = None, skills_dir_override: "Path | None" = None,
) -> str:
"""Compact skill index for the system prompt.
@@ -1114,7 +1059,8 @@ def build_skills_system_prompt(
project_dirs = get_project_skills_dirs()
if not skills_dir.exists() and not external_dirs and not project_dirs:
return ""
return _build_skills_system_prompt_inner(skills_dir, external_dirs, available_tools, available_toolsets, compact_categories, project_dirs)
return _build_skills_system_prompt_inner(
skills_dir, external_dirs, available_tools, available_toolsets, compact_categories, project_dirs)
finally:
if _home_token is not None:
reset_hermes_home_override(_home_token)
@@ -1139,14 +1085,8 @@ def _read_category_descriptions(root: Path, log_fmt: str) -> dict[str, str]:
def _collect_extra_skills(
root: Path,
skill_files,
hides,
claimed: set[str],
skills_by_category: dict[str, list[tuple[str, str]]],
*,
desc_prefix: str,
log_fmt: str,
root: Path, skill_files, hides, claimed: set[str], skills_by_category: dict[str, list[tuple[str, str]]],
*, desc_prefix: str, log_fmt: str,
) -> None:
"""Add visible skills from a project/external dir; names already in *claimed* are skipped."""
for skill_file in skill_files:
@@ -1180,10 +1120,8 @@ def _label_visible_entries(visible_entries: list[dict], skills_by_category: dict
def _render_skills_index(
skills_by_category: dict[str, list[tuple[str, str]]],
category_descriptions: dict[str, str],
compact_categories: "frozenset[str] | None",
available_tools: "set[str] | None",
skills_by_category: dict[str, list[tuple[str, str]]], category_descriptions: dict[str, str],
compact_categories: "frozenset[str] | None", available_tools: "set[str] | None",
) -> str:
"""Render the ## Skills block; "" when there is nothing to list."""
if not skills_by_category:
@@ -1235,11 +1173,8 @@ def _render_skills_index(
def _build_skills_system_prompt_inner(
skills_dir: "Path",
external_dirs: "list[Path]",
available_tools: "set[str] | None",
available_toolsets: "set[str] | None",
compact_categories: "frozenset[str] | None",
skills_dir: "Path", external_dirs: "list[Path]", available_tools: "set[str] | None",
available_toolsets: "set[str] | None", compact_categories: "frozenset[str] | None",
project_dirs: "list[Path] | None" = None,
) -> str:
# The resolved platform is part of the key: per-platform disabled-skill lists need distinct cache entries.
@@ -1247,14 +1182,10 @@ def _build_skills_system_prompt_inner(
disabled = get_disabled_skill_names(_platform_hint or None)
project_dirs = project_dirs or []
cache_key = (
str(skills_dir),
tuple(str(d) for d in external_dirs),
tuple(str(d) for d in project_dirs),
str(skills_dir), tuple(str(d) for d in external_dirs), tuple(str(d) for d in project_dirs),
tuple(sorted(str(t) for t in (available_tools or set()))),
tuple(sorted(str(ts) for ts in (available_toolsets or set()))),
_platform_hint,
tuple(sorted(disabled)),
tuple(sorted(compact_categories or ())),
_platform_hint, tuple(sorted(disabled)), tuple(sorted(compact_categories or ())),
)
with _SKILLS_PROMPT_CACHE_LOCK:
cached = _SKILLS_PROMPT_CACHE.get(cache_key)
@@ -1264,19 +1195,16 @@ def _build_skills_system_prompt_inner(
def hides(frontmatter_name: str, skill_name: str, conditions: dict) -> bool:
"""Per-build visibility rule shared by every skill source (snapshot, scan, project, external)."""
if frontmatter_name in disabled or skill_name in disabled:
return True
return not _skill_should_show(conditions, available_tools, available_toolsets, _platform_hint or None)
return (frontmatter_name in disabled or skill_name in disabled
or not _skill_should_show(conditions, available_tools, available_toolsets, _platform_hint or None))
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:
candidates = [
(entry, skill_matches_platform_list(entry.get("platforms") or []))
for entry in snapshot.get("skills", []) if isinstance(entry, dict)
]
candidates = [(entry, skill_matches_platform_list(entry.get("platforms") or []))
for entry in snapshot.get("skills", []) if isinstance(entry, dict)]
category_descriptions = {str(k): str(v) for k, v in (snapshot.get("category_descriptions") or {}).items()}
else:
candidates = []
@@ -1292,13 +1220,9 @@ def _build_skills_system_prompt_inner(
project_names: set[str] = set()
if project_dirs:
from agent.skill_utils import iter_project_skill_files
for proj_dir in project_dirs:
if proj_dir.exists():
_collect_extra_skills(
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",
)
for proj_dir in (d for d in project_dirs if d.exists()):
_collect_extra_skills(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")
# 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)
@@ -1314,10 +1238,8 @@ def _build_skills_system_prompt_inner(
# 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 (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",
)
_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")
for cat, cat_desc in _read_category_descriptions(ext_dir, "Could not read external skill description %s: %s").items():
category_descriptions.setdefault(cat, cat_desc)
@@ -1330,14 +1252,8 @@ def _build_skills_system_prompt_inner(
return result
# Context files (SOUL.md, AGENTS.md, .cursorrules)
def _truncate_content(
content: str,
filename: str,
max_chars: Optional[int] = None,
context_length: Optional[int] = None,
content: str, filename: str, max_chars: Optional[int] = None, context_length: Optional[int] = None,
read_path: Optional[str] = None,
) -> str:
"""Head/tail truncation with a marker in the middle; ``read_path`` (default ``filename``) is what the
@@ -1351,7 +1267,9 @@ def _truncate_content(
f"trim the file, pin a larger context_file_max_chars, or use a larger-context model!"
)
logger.warning(msg)
_record_truncation_warning(msg)
if (warnings := _truncation_warnings.get()) is None:
_truncation_warnings.set(warnings := [])
warnings.append(msg)
head_chars = int(max_chars * CONTEXT_TRUNCATE_HEAD_RATIO)
tail_chars = int(max_chars * CONTEXT_TRUNCATE_TAIL_RATIO)
marker = (
@@ -1380,14 +1298,17 @@ 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
def _read_context_file(path: Path) -> str:
"""Stripped text of *path*; "" when empty or unreadable (logged at debug)."""
"""Stripped text of *path*; "" when missing, empty or unreadable (logged at debug)."""
if not path.exists():
return ""
try:
return path.read_text(encoding="utf-8").strip()
except Exception as e:
@@ -1395,18 +1316,8 @@ def _read_context_file(path: Path) -> str:
return ""
def _context_section(
content: str,
label: str,
warn_name: str,
path: Path,
context_length: Optional[int],
*,
strip_frontmatter: bool = False,
) -> str:
def _context_section(content: str, label: str, warn_name: str, path: Path, context_length: Optional[int]) -> str:
"""Threat-scan *content*, render it as ``## <label>``, cap it to the budget (*warn_name* labels warnings)."""
if strip_frontmatter:
content = _strip_yaml_frontmatter(content)
body = f"## {label}\n\n{_scan_context_content(content, label)}"
return _truncate_content(body, warn_name, context_length=context_length, read_path=str(path))
@@ -1418,7 +1329,7 @@ def _load_hermes_md(cwd_path: Path, context_length: Optional[int] = None) -> str
if not content:
return ""
label = str(hermes_md_path.relative_to(cwd_path)) if hermes_md_path.is_relative_to(cwd_path) else hermes_md_path.name
return _context_section(content, label, ".hermes.md", hermes_md_path, context_length, strip_frontmatter=True)
return _context_section(_strip_yaml_frontmatter(content), label, ".hermes.md", hermes_md_path, context_length)
def _agents_md_directory_chain(cwd_path: Path) -> list[Path]:
@@ -1443,27 +1354,25 @@ def _load_agents_md(cwd_path: Path, context_length: Optional[int] = None) -> str
for directory in _agents_md_directory_chain(cwd_resolved):
for name in ("AGENTS.override.md", "AGENTS.md", "agents.md"):
candidate = directory / name
if not candidate.exists():
continue
content = _read_context_file(candidate)
if not content:
continue
if content in seen_content:
break # identical copy along the chain
seen_content.add(content)
label = name if directory == cwd_resolved else os.path.relpath(candidate, cwd_resolved)
sections.append(_context_section(content, label, label, candidate, context_length))
if content not in seen_content: # else: identical copy along the chain
seen_content.add(content)
label = name if directory == cwd_resolved else os.path.relpath(candidate, cwd_resolved)
sections.append(_context_section(content, label, label, candidate, context_length))
break # first name match wins per directory
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"):
content = _read_context_file(cwd_path / name) if (cwd_path / name).exists() else ""
content = _read_context_file(cwd_path / name)
if content:
return _context_section(content, name, "CLAUDE.md", cwd_path / name, context_length)
return ""
@@ -1475,22 +1384,19 @@ def _load_cursorrules(cwd_path: Path, context_length: Optional[int] = None) -> s
cursor_rules_dir = cwd_path / ".cursor" / "rules"
if cursor_rules_dir.is_dir():
candidates += [(f, f".cursor/rules/{f.name}") for f in sorted(cursor_rules_dir.glob("*.mdc"))]
cursorrules_content = ""
for path, label in candidates:
content = _read_context_file(path) if path.exists() else ""
if content:
cursorrules_content += f"## {label}\n\n{_scan_context_content(content, label)}\n\n"
cursorrules_content = "".join(
f"## {label}\n\n{_scan_context_content(content, label)}\n\n"
for path, label in candidates if (content := _read_context_file(path))
)
if not cursorrules_content:
return ""
return _truncate_content(cursorrules_content, ".cursorrules", context_length=context_length, read_path=str(cwd_path / ".cursorrules"))
return _truncate_content(cursorrules_content, ".cursorrules", context_length=context_length,
read_path=str(cwd_path / ".cursorrules"))
def build_context_files_prompt(
cwd: Optional[str] = None,
skip_soul: bool = False,
context_length: Optional[int] = None,
allow_install_tree_fallback: bool = False,
home_override: "Path | None" = None,
cwd: Optional[str] = None, skip_soul: bool = False, context_length: Optional[int] = None,
allow_install_tree_fallback: bool = False, home_override: "Path | None" = None,
) -> str:
"""Discover and load context files for the system prompt (each capped, see ``_get_context_file_max_chars``).
@@ -1502,24 +1408,19 @@ def build_context_files_prompt(
# 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
if cwd is None and not allow_install_tree_fallback and _is_install_tree(cwd_path):
logger.warning(
"skipping project-context discovery: working-directory resolution fell back to the Hermes "
"install tree (%s) — set terminal.cwd to your project directory",
cwd_path,
"install tree (%s) — set terminal.cwd to your project directory", cwd_path,
)
sections = []
else:
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)
]
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]
if not sections:
return ""
return "# Project Context\n\nThe following project context files have been loaded and should be followed:\n\n" + "\n".join(sections)
return ("# Project Context\n\nThe following project context files have been loaded and should be followed:\n\n"
+ "\n".join(sections))