diff --git a/agent/markdown_tables.py b/agent/markdown_tables.py index 20e752eddd..a3158f8cef 100644 --- a/agent/markdown_tables.py +++ b/agent/markdown_tables.py @@ -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) diff --git a/agent/onboarding.py b/agent/onboarding.py index 4898bf4eed..e8bb0ded65 100644 --- a/agent/onboarding.py +++ b/agent/onboarding.py @@ -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(): diff --git a/agent/plugin_llm.py b/agent/plugin_llm.py index 46fd27145b..d084d465e2 100644 --- a/agent/plugin_llm.py +++ b/agent/plugin_llm.py @@ -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) diff --git a/agent/process_bootstrap.py b/agent/process_bootstrap.py index 8b56ddbd87..3f0fae516f 100644 --- a/agent/process_bootstrap.py +++ b/agent/process_bootstrap.py @@ -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 diff --git a/agent/prompt_builder.py b/agent/prompt_builder.py index 8de4b28382..0889204a47 100644 --- a/agent/prompt_builder.py +++ b/agent/prompt_builder.py @@ -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\\\\, 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\\\\, 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 ``##