From a9011bec27559d53108074176482f06b80c025e4 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:01:41 -0700
Subject: [PATCH 1/4] =?UTF-8?q?refactor(agent):=20prompt=5Fbuilder=20pass-?=
 =?UTF-8?q?2=20=E2=80=94=20pack=20signatures,=20fold=20context-file=20guar?=
 =?UTF-8?q?ds=20into=20=5Fread=5Fcontext=5Ffile,=20inline=20truncation=20r?=
 =?UTF-8?q?ecorder?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 agent/prompt_builder.py | 279 ++++++++++++++--------------------------
 1 file changed, 97 insertions(+), 182 deletions(-)

diff --git a/agent/prompt_builder.py b/agent/prompt_builder.py
index 8de4b28382..1d6892213c 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]:
@@ -73,12 +70,8 @@ def _find_hermes_md(cwd: Path) -> Optional[Path]:
 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 = (
@@ -371,6 +364,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 +670,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 +677,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()
 
 
@@ -750,17 +741,12 @@ def _run_backend_probe(env_type: str, terminal_tool) -> str:
     # 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(image_key, "") if image_key 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 +769,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,17 +800,14 @@ 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:
@@ -856,7 +838,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 +847,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 +884,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 +894,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 +919,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,18 +945,15 @@ 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
 
 
@@ -1000,11 +966,8 @@ def _load_skills_snapshot(skills_dir: Path) -> Optional[dict]:
         snapshot = json.loads(snapshot_path.read_text(encoding="utf-8"))
     except Exception:
         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
 
@@ -1023,11 +986,8 @@ def _build_snapshot_entry(skill_file: Path, skills_dir: Path, frontmatter: dict,
     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 +1014,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 +1047,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 +1070,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 +1096,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 +1131,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 +1184,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 +1193,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 +1206,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 +1231,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 +1249,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)
 
@@ -1334,10 +1267,7 @@ def _build_skills_system_prompt_inner(
 
 
 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 +1281,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 +1312,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 +1330,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 +1343,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 +1368,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 +1398,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 +1422,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))

From 2c3816d1ed85f32c68e4f844a0723d0855c8e063 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:10:47 -0700
Subject: [PATCH 2/4] =?UTF-8?q?refactor(agent):=20plugin=5Fllm=20pass-2=20?=
 =?UTF-8?q?=E2=80=94=20collapse=20policy-load=20ladder,=20pack=20call=20kw?=
 =?UTF-8?q?args,=20drop=20section=20banners?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 agent/plugin_llm.py | 151 ++++++++++++--------------------------------
 1 file changed, 40 insertions(+), 111 deletions(-)

diff --git a/agent/plugin_llm.py b/agent/plugin_llm.py
index 46fd27145b..a67d221414 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,10 +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",
-)
+_OVERRIDE_FLAGS = ("allow_provider_override", "allow_model_override", "allow_agent_id_override",
+                   "allow_profile_override", "allow_task_override")
 
 
 def _normalize_ref(raw: str) -> str:
@@ -135,19 +127,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 +151,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:
@@ -174,11 +160,8 @@ def _gate_ref_override(policy: _TrustPolicy, kind: str, requested: str) -> str:
         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."
-        )
+        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 +182,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 +196,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 +219,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 +236,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 +251,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 +269,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 +283,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 +304,6 @@ def _build_structured_messages(
     return messages
 
 
-# -- JSON parsing / response extraction --------------------------------------
-
-
 _FENCE_RE = re.compile(r"```(?:json)?\s*(.+?)```", re.DOTALL | re.IGNORECASE)
 
 
@@ -352,20 +313,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 +331,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 +353,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 +368,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 +385,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 +400,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 +420,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 +443,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 +495,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 +508,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 +543,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 +552,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 +580,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)

From a5858a9e98b381c08f40f486cc1b6cc73c58d3af Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:21:06 -0700
Subject: [PATCH 3/4] =?UTF-8?q?refactor(agent):=20process=5Fbootstrap/mark?=
 =?UTF-8?q?down=5Ftables/onboarding=20pass-2=20=E2=80=94=20inline=20=5Fpad?=
 =?UTF-8?q?=5Fto=5Fwidth,=20fold=20proxy=20bypass=20and=20pool=20guard,=20?=
 =?UTF-8?q?pack=20signatures?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 agent/markdown_tables.py   | 34 ++++++---------------
 agent/onboarding.py        | 13 +++------
 agent/process_bootstrap.py | 60 ++++++++++----------------------------
 3 files changed, 29 insertions(+), 78 deletions(-)

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/process_bootstrap.py b/agent/process_bootstrap.py
index 8b56ddbd87..64ceaeaee8 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
 
 
@@ -303,14 +280,10 @@ 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 +297,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

From 123d962dc7544144d0f1d3b1b9ce9e1a47c39db2 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:32:12 -0700
Subject: [PATCH 4/4] =?UTF-8?q?refactor(agent):=20prompt=5Fbuilder/plugin?=
 =?UTF-8?q?=5Fllm/process=5Fbootstrap=20pass-2=20=E2=80=94=20inline=20=5Fn?=
 =?UTF-8?q?ormalize=5Fref,=20fold=20snapshot=20read,=20proxy-env=20scan?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 agent/plugin_llm.py        |  8 ++------
 agent/process_bootstrap.py |  7 ++-----
 agent/prompt_builder.py    | 34 ++++++++++------------------------
 3 files changed, 14 insertions(+), 35 deletions(-)

diff --git a/agent/plugin_llm.py b/agent/plugin_llm.py
index a67d221414..d084d465e2 100644
--- a/agent/plugin_llm.py
+++ b/agent/plugin_llm.py
@@ -109,16 +109,12 @@ _OVERRIDE_FLAGS = ("allow_provider_override", "allow_model_override", "allow_age
                    "allow_profile_override", "allow_task_override")
 
 
-def _normalize_ref(raw: str) -> str:
-    return (raw or "").strip().lower()
-
-
 def _coerce_allowlist(raw: Any) -> tuple[Optional[frozenset], bool]:
     """YAML list → ``(frozenset_or_None, allow_any)``. A ``"*"`` entry sets
     ``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
 
 
@@ -159,7 +155,7 @@ 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:
+    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()
diff --git a/agent/process_bootstrap.py b/agent/process_bootstrap.py
index 64ceaeaee8..3f0fae516f 100644
--- a/agent/process_bootstrap.py
+++ b/agent/process_bootstrap.py
@@ -269,11 +269,8 @@ 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]:
diff --git a/agent/prompt_builder.py b/agent/prompt_builder.py
index 1d6892213c..0889204a47 100644
--- a/agent/prompt_builder.py
+++ b/agent/prompt_builder.py
@@ -59,11 +59,9 @@ 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
 
 
@@ -247,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
@@ -739,9 +736,8 @@ 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}
@@ -812,10 +808,8 @@ def _local_host_hints() -> list[str]:
 
 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 + (
@@ -959,12 +953,9 @@ def _build_skills_manifest(skills_dir: Path) -> dict[str, list[int]]:
 
 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)):
@@ -979,10 +970,8 @@ 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 = {
@@ -1263,9 +1252,6 @@ 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,
     read_path: Optional[str] = None,