diff --git a/agent/skill_bundles.py b/agent/skill_bundles.py index 6fdf8d1c87..00eeea1fbe 100644 --- a/agent/skill_bundles.py +++ b/agent/skill_bundles.py @@ -32,9 +32,7 @@ _bundles_cache_mtime: Optional[float] = None def _bundles_dir() -> Path: """Bundles directory: ``HERMES_BUNDLES_DIR`` override (tests) or ``/skill-bundles``.""" override = os.environ.get("HERMES_BUNDLES_DIR") - if override: - return Path(override).expanduser() - return get_hermes_home() / "skill-bundles" + return Path(override).expanduser() if override else get_hermes_home() / "skill-bundles" def _iter_bundle_files() -> List[Path]: @@ -63,10 +61,8 @@ def _load_bundle_file(path: Path) -> Optional[Dict[str, Any]]: except yaml.YAMLError as exc: logger.warning("Invalid YAML in bundle %s: %s", path, exc) return None - def _skip(reason: str) -> None: logger.warning("Bundle %s %s; skipping", path, reason) - if not isinstance(data, dict): return _skip("is not a mapping") name = str(data.get("name") or path.stem).strip() @@ -135,10 +131,7 @@ def list_bundles() -> List[Dict[str, Any]]: def build_bundle_invocation_message( - cmd_key: str, - user_instruction: str = "", - task_id: str | None = None, - platform: str | None = None, + cmd_key: str, user_instruction: str = "", task_id: str | None = None, platform: str | None = None, ) -> Optional[Tuple[str, List[str], List[str]]]: """Build the user message for a bundle invocation: ``(message, loaded_skill_names, missing_skill_names)`` or ``None`` if the bundle wasn't @@ -148,10 +141,8 @@ def build_bundle_invocation_message( info = get_skill_bundles().get(cmd_key) if not info: return None - # Late import keeps skill_bundles cheap to import (no tools/* at import time). from agent.skill_commands import _disabled_skill_names, _load_skill_blocks, _load_skill_payload, _scaffold_header - bundle_name = info["name"] loaded_names, missing, disabled, skill_blocks = _load_skill_blocks( [(skill_id or "").strip() for skill_id in info["skills"]], @@ -162,7 +153,6 @@ def build_bundle_invocation_message( ) if not skill_blocks: return None - header = _scaffold_header( f'"{bundle_name}" skill bundle', loaded_names, @@ -186,13 +176,7 @@ def bundle_path_for(name: str) -> Path: return _bundles_dir() / f"{slug}.yaml" -def save_bundle( - name: str, - skills: List[str], - description: str = "", - instruction: str = "", - overwrite: bool = False, -) -> Path: +def save_bundle(name: str, skills: List[str], description: str = "", instruction: str = "", overwrite: bool = False) -> Path: """Write a bundle to disk and refresh the cache. Raises ``FileExistsError`` if the target exists and not ``overwrite``; ``ValueError`` for unusable inputs.""" name = (name or "").strip() @@ -201,17 +185,12 @@ def save_bundle( cleaned_skills = [str(s).strip() for s in skills if str(s).strip()] if not cleaned_skills: raise ValueError("Bundle must reference at least one skill") - path = bundle_path_for(name) if path.exists() and not overwrite: raise FileExistsError(f"Bundle already exists at {path}") - path.parent.mkdir(parents=True, exist_ok=True) payload: Dict[str, Any] = {"name": name, "skills": cleaned_skills} - if description: - payload["description"] = description - if instruction: - payload["instruction"] = instruction + payload.update({k: v for k, v in (("description", description), ("instruction", instruction)) if v}) path.write_text(yaml.safe_dump(payload, sort_keys=False, allow_unicode=True), encoding="utf-8") scan_bundles() return path diff --git a/agent/skill_commands.py b/agent/skill_commands.py index df3d96c7b9..f4704f8235 100644 --- a/agent/skill_commands.py +++ b/agent/skill_commands.py @@ -56,8 +56,7 @@ SKILL_EXCERPT_JOINT = "\x1e" def slugify_skill_name(name: str) -> str: """Normalize a skill/bundle name to a ``/command`` slug (``Foo Bar`` -> ``foo-bar``); strips chars (``+``, ``/``) that would make invalid Telegram command names.""" - cmd = name.lower().replace(" ", "-").replace("_", "-") - cmd = _SKILL_INVALID_CHARS.sub("", cmd) + cmd = _SKILL_INVALID_CHARS.sub("", name.lower().replace(" ", "-").replace("_", "-")) return _SKILL_MULTI_HYPHEN.sub("-", cmd).strip("-") @@ -114,9 +113,7 @@ def _cut_after(message: str, marker: str, stop_marker: str, find) -> Optional[st marker_idx = find(marker) if marker_idx < 0: return None - text = message[marker_idx + len(marker):] - idx = text.find(stop_marker) - return (text[:idx] if idx >= 0 else text).strip() or None + return message[marker_idx + len(marker):].split(stop_marker, 1)[0].strip() or None def _resolve_skill_commands_platform() -> Optional[str]: @@ -125,7 +122,6 @@ def _resolve_skill_commands_platform() -> Optional[str]: own ``skills.platform_disabled`` view.""" try: from gateway.session_context import get_session_env - resolved_platform = os.getenv("HERMES_PLATFORM") or get_session_env("HERMES_SESSION_PLATFORM") except Exception: resolved_platform = os.getenv("HERMES_PLATFORM") @@ -136,7 +132,6 @@ def _resolve_skill_commands_home() -> str: """Effective Hermes home the scan is scoped to (profiles carry their own ``skills.external_dirs``, so a profile switch must invalidate the cache).""" from hermes_constants import get_hermes_home - return str(get_hermes_home()) @@ -148,7 +143,6 @@ def _load_skill_payload(skill_identifier: str, task_id: str | None = None) -> tu try: from tools.skills_tool import _skills_dir, skill_view from agent.skill_utils import normalize_skill_lookup_name - normalized = normalize_skill_lookup_name(raw_identifier) loaded_skill = json.loads(skill_view(normalized, task_id=task_id, preprocess=False)) except Exception: @@ -174,7 +168,6 @@ def _inject_skill_config(loaded_skill: dict[str, Any], parts: list[str]) -> None values so the agent needn't read config.yaml. Any failure leaves the message without it.""" try: from agent.skill_utils import extract_skill_config_vars, parse_frontmatter, resolve_skill_config_values - raw_content = str(loaded_skill.get("raw_content") or loaded_skill.get("content") or "") frontmatter, _ = parse_frontmatter(raw_content) resolved = resolve_skill_config_values(extract_skill_config_vars(frontmatter)) @@ -230,7 +223,6 @@ def _build_skill_message( ) -> str: """Format a loaded skill into a user/system message payload.""" from tools.skills_tool import _skills_dir - # Preprocess first so downstream blocks see the expanded content. content = preprocess_skill_content( str(loaded_skill.get("content") or ""), skill_dir, session_id, skills_cfg=_load_skills_config(), @@ -245,7 +237,6 @@ def _build_skill_message( "`templates/config.yaml`) against that directory, then run them " "with the terminal tool using the absolute path.", ] - _inject_skill_config(loaded_skill, parts) setup_note = _setup_note(loaded_skill) if setup_note: @@ -322,7 +313,6 @@ _SCAN_SKIP_PARTS = {'.git', '.github', '.hub', '.archive'} def _scan_skill_md(skill_md: Path, disabled: set, seen_names: set, commands: Dict[str, Dict[str, Any]], resolve_command) -> None: """Register one SKILL.md in *commands* (no-op when filtered or colliding).""" from tools.skills_tool import _parse_frontmatter, skill_matches_platform, skill_matches_environment - if any(part in _SCAN_SKIP_PARTS for part in skill_md.parts): return frontmatter, body = _parse_frontmatter(skill_md.read_text(encoding='utf-8')) @@ -382,7 +372,6 @@ def scan_skill_commands() -> Dict[str, Dict[str, Any]]: from hermes_cli.commands import resolve_command disabled = _get_disabled_skill_names() seen_names: set = set() - # Precedence: project (through the quarantine chokepoint) > local > external. # Resolve the local dir at call time: import-time SKILLS_DIR is frozen to # the launch home, but a multiplexed profile scope may have changed it. @@ -501,9 +490,7 @@ def split_stacked_skill_commands(rest: str) -> tuple[list[str], str]: stripped = remaining.lstrip() if not stripped.startswith("/"): break - parts = stripped.split(None, 1) - token = parts[0] - tail = parts[1] if len(parts) > 1 else "" + token, tail = (stripped.split(None, 1) + [""])[:2] cmd_key = resolve_skill_command_key(token.lstrip("/")) if cmd_key is None or cmd_key in keys: break diff --git a/agent/skill_preprocessing.py b/agent/skill_preprocessing.py index eadc425aa9..dd15586b46 100644 --- a/agent/skill_preprocessing.py +++ b/agent/skill_preprocessing.py @@ -23,7 +23,6 @@ def load_skills_config() -> dict: """Load the ``skills`` section of config.yaml (best-effort).""" try: from hermes_cli.config import load_config_readonly - skills_cfg = (load_config_readonly() or {}).get("skills") if isinstance(skills_cfg, dict): return skills_cfg @@ -69,7 +68,6 @@ def run_inline_shell(command: str, cwd: Path | None, timeout: int) -> str: if isinstance(exc, RuntimeError) and "live-system guard: blocked os.kill" in str(exc): return f"[inline-shell timeout after {timeout}s: {command}]" return f"[inline-shell error: {exc}]" - output = (completed.stdout or "").rstrip("\n") or (completed.stderr or "").rstrip("\n") if len(output) > _INLINE_SHELL_MAX_OUTPUT: output = output[:_INLINE_SHELL_MAX_OUTPUT] + "...[truncated]" @@ -80,11 +78,9 @@ def expand_inline_shell(content: str, skill_dir: Path | None, timeout: int) -> s """Replace every !`cmd` snippet with its stdout, run with the skill dir as CWD.""" if "!`" not in content: return content - def _replace(match: re.Match) -> str: cmd = match.group(1).strip() return run_inline_shell(cmd, skill_dir, timeout) if cmd else "" - return _INLINE_SHELL_RE.sub(_replace, content) diff --git a/agent/skill_utils.py b/agent/skill_utils.py index 0a5c4e868c..ebd656f68d 100644 --- a/agent/skill_utils.py +++ b/agent/skill_utils.py @@ -93,7 +93,6 @@ def yaml_load(content: str): if _yaml_load_fn is None: import functools import yaml - _yaml_load_fn = functools.partial(yaml.load, Loader=getattr(yaml, "CSafeLoader", None) or yaml.SafeLoader) return _yaml_load_fn(content) @@ -106,7 +105,6 @@ def parse_frontmatter(content: str) -> Tuple[Dict[str, Any], str]: end_match = re.search(r"\n---\s*\n", content[3:]) if content.startswith("---") else None if not end_match: return {}, content - yaml_content = content[3 : end_match.start() + 3] body = content[end_match.end() + 3 :] frontmatter: Dict[str, Any] = {} @@ -126,16 +124,13 @@ def skill_matches_platform_list(platforms: Any) -> bool: """Return True when *platforms* is compatible with the current OS.""" if not platforms: return True - if not isinstance(platforms, list): - platforms = [platforms] - current = sys.platform running_in_termux = is_termux() - for platform in platforms: + for platform in platforms if isinstance(platforms, list) else [platforms]: normalized = str(platform).lower().strip() mapped = PLATFORM_MAP.get(normalized, normalized) # Termux is a Linux userland on Android: accept linux-tagged skills # whether sys.platform is "linux" (pre-3.13) or "android" (3.13+). - if current.startswith(mapped) or (running_in_termux and mapped in ("linux", "termux", "android")): + if sys.platform.startswith(mapped) or (running_in_termux and mapped in ("linux", "termux", "android")): return True return False @@ -159,7 +154,6 @@ def _detect_kanban() -> bool: if os.getenv("HERMES_KANBAN_TASK") or os.getenv("HERMES_KANBAN_BOARD"): try: from agent.delegation_context import is_dispatcher_owned_worker_context - owned = is_dispatcher_owned_worker_context() except Exception: owned = True @@ -167,7 +161,6 @@ def _detect_kanban() -> bool: return True try: from tools.kanban_tools import _profile_has_kanban_toolset - return bool(_profile_has_kanban_toolset()) except Exception: return False @@ -176,7 +169,6 @@ def _detect_kanban() -> bool: def _detect_docker() -> bool: try: from hermes_constants import is_container - return is_container() except Exception: return False @@ -239,7 +231,6 @@ def _load_raw_config() -> Dict[str, Any]: cached = _RAW_CONFIG_CACHE.get(cache_key) if cache_key is not None else None if cached is not None: return cached - try: parsed = yaml_load(config_path.read_text(encoding="utf-8")) except Exception as e: @@ -247,7 +238,6 @@ def _load_raw_config() -> Dict[str, Any]: return {} if not isinstance(parsed, dict): return {} - if cache_key is not None: _RAW_CONFIG_CACHE.clear() _RAW_CONFIG_CACHE[cache_key] = parsed @@ -274,7 +264,6 @@ def _expand_path(entry: str) -> Path: def _home_relative(p: Path) -> Path: """Anchor a relative config path at HERMES_HOME; absolute paths pass through.""" from hermes_constants import get_hermes_home - return p if p.is_absolute() else get_hermes_home() / p @@ -289,7 +278,6 @@ def get_disabled_skill_names(platform: str | None = None) -> Set[str]: skills_cfg = _skills_cfg() if skills_cfg is None: return set() - from gateway.session_context import get_session_env resolved_platform = platform or os.getenv("HERMES_PLATFORM") or get_session_env("HERMES_SESSION_PLATFORM") disabled = _normalize_string_set(skills_cfg.get("disabled")) @@ -354,7 +342,6 @@ def get_external_skills_dirs() -> List[Path]: skills_cfg = _skills_cfg() if skills_cfg is None: return [] - local_skills = get_skills_dir().resolve() result: List[Path] = [] for entry in _config_str_list(skills_cfg.get("external_dirs")): @@ -394,14 +381,12 @@ def display_skill_create_dir() -> str: """User-facing path where new skills are created (``~/`` shorthand when possible); tool schema descriptions and prompts follow ``skills.create_dir``.""" from hermes_constants import display_hermes_home - create_dir = get_skill_create_dir() if create_dir is None: return f"{display_hermes_home()}/skills/" - try: + if create_dir.is_relative_to(Path.home()): return "~/" + create_dir.relative_to(Path.home()).as_posix() + "/" - except ValueError: - return create_dir.as_posix() + "/" + return create_dir.as_posix() + "/" def get_all_skills_dirs() -> List[Path]: @@ -433,7 +418,6 @@ def find_project_root(start: Optional[Path] = None) -> Optional[Path]: try: if start is None: from agent.runtime_cwd import scope_terminal_cwd - env_cwd = scope_terminal_cwd() start = Path(env_cwd) if env_cwd else Path.cwd() cur = Path(start).resolve() @@ -478,8 +462,7 @@ def _candidate_project_skills_dirs(root: Path) -> List[Path]: (HERMES_HOME itself may live inside a git checkout).""" local_skills = get_skills_dir().resolve() dirs: List[Path] = [] - for sub in PROJECT_SKILLS_SUBDIRS: - cand = root / sub + for cand in (root / sub for sub in PROJECT_SKILLS_SUBDIRS): try: if cand.is_dir() and cand.resolve() != local_skills: dirs.append(cand.resolve()) @@ -505,10 +488,8 @@ def get_project_skills_dirs() -> List[Path]: def get_untrusted_project_skills_root() -> Optional[Tuple[Path, int]]: """(root, skill_count) when cwd's project has skills but is NOT trusted, else None.""" root = _current_project_root(trusted=False) - if root is None: - return None count = 0 - for d in _candidate_project_skills_dirs(root): + for d in _candidate_project_skills_dirs(root) if root is not None else (): try: count += sum(1 for _ in iter_skill_index_files(d, "SKILL.md")) except OSError: @@ -540,7 +521,6 @@ def is_quarantined_project_skill(skill_md) -> bool: try: from tools.skills_guard import scan_skill_cached from hermes_constants import get_hermes_home - result, _prov = scan_skill_cached( skill_dir, source=_PROJECT_SCAN_SOURCE, cache_dir=get_hermes_home() / "cache" / "project_skill_scans", ) @@ -567,11 +547,9 @@ def normalize_skill_lookup_name(identifier: str) -> str: raw_identifier = (identifier or "").strip() if not raw_identifier: return raw_identifier - identifier_path = Path(raw_identifier).expanduser() if not identifier_path.is_absolute(): return raw_identifier.lstrip("/") - # Resolve the primary root via tools.skills_tool at CALL time: tests patch # ``tools.skills_tool.SKILLS_DIR`` and skill_view() enforces ``_skills_dir()`` # (which follows the live profile-scoped HERMES_HOME), so normalization @@ -581,14 +559,12 @@ def normalize_skill_lookup_name(identifier: str) -> str: primary_root = _skills_tool._skills_dir() except Exception: primary_root = get_skills_dir() - trusted_roots = [primary_root] for getter in (get_project_skills_dirs, get_external_skills_dirs): try: trusted_roots.extend(getter()) except Exception: pass - # Prefer the lexical path under a trusted root before resolving symlinks: # ~/.hermes/skills/ may be a symlink to a checkout elsewhere, and # resolving first would turn that trusted path into one skill_view rejects. @@ -713,13 +689,12 @@ def resolve_skill_config_values(config_vars: List[Dict[str, Any]]) -> Dict[str, config = _load_raw_config() resolved: Dict[str, Any] = {} for var in config_vars: - logical_key = var["key"] - value = _resolve_dotpath(config, f"{SKILL_CONFIG_PREFIX}.{logical_key}") + value = _resolve_dotpath(config, f"{SKILL_CONFIG_PREFIX}.{var['key']}") if value is None or (isinstance(value, str) and not value.strip()): value = var.get("default", "") if isinstance(value, str) and ("~" in value or "${" in value): value = os.path.expanduser(os.path.expandvars(value)) - resolved[logical_key] = value + resolved[var["key"]] = value return resolved diff --git a/agent/system_prompt.py b/agent/system_prompt.py index 91dbdfc1c4..67ca082905 100644 --- a/agent/system_prompt.py +++ b/agent/system_prompt.py @@ -34,8 +34,7 @@ _PLUGIN_SECTION_FRAME_RE = re.compile( r"\n\n", re.MULTILINE, ) -_GATE_WORDS = {"true": True, "always": True, "yes": True, "on": True, - "false": False, "never": False, "no": False, "off": False} +_GATE_WORDS = {**dict.fromkeys(("true", "always", "yes", "on"), True), **dict.fromkeys(("false", "never", "no", "off"), False)} def _ra(): @@ -70,12 +69,9 @@ def _resolve_platform_hint(agent: Any, platform_key: str, default_hint: str) -> spec = {"append": spec} if not isinstance(spec, dict): return default_hint - replace_text = spec.get("replace") - base = replace_text.strip() if isinstance(replace_text, str) and replace_text.strip() else default_hint - append_text = spec.get("append") - if isinstance(append_text, str) and append_text.strip(): - return f"{base}\n\n{append_text.strip()}".strip() - return base + replace_text, append_text = (v.strip() if isinstance(v, str) else "" for v in (spec.get("replace"), spec.get("append"))) + base = replace_text or default_hint + return f"{base}\n\n{append_text}".strip() if append_text else base _TUI_EMBEDDED_PANE_CLARIFIER = ( @@ -106,7 +102,6 @@ def _plugin_session_info(agent: Any) -> Dict[str, str]: def _ambient_plugin_profile_name() -> str: from hermes_cli.profiles import get_active_profile_name - return str(get_active_profile_name() or "default") @@ -134,7 +129,6 @@ def _frozen_plugin_prompt_sections(agent: Any) -> tuple: else: try: from hermes_cli.plugins import render_system_prompt_sections - rendered = tuple(render_system_prompt_sections(_plugin_session_info(agent))) except Exception as exc: rendered = getattr(agent, "_plugin_system_prompt_sections_previous", None) @@ -155,7 +149,6 @@ def _restore_plugin_prompt_sections(prompt: str) -> tuple: MAX_SYSTEM_PROMPT_SECTION_CHARS, PLUGIN_SECTIONS_END, PLUGIN_SECTIONS_START, RenderedPluginSystemPromptSection, format_system_prompt_sections, ) - start = prompt.rfind(PLUGIN_SECTIONS_START) end = prompt.find(PLUGIN_SECTIONS_END, start + len(PLUGIN_SECTIONS_START)) if start >= 0 else -1 if end < 0: @@ -164,7 +157,6 @@ def _restore_plugin_prompt_sections(prompt: str) -> tuple: if not prompt[after_end:].startswith("\n\nConversation started:"): return () framed = prompt[start:after_end] - restored = [] for match in _PLUGIN_SECTION_FRAME_RE.finditer(framed): content_len = int(match.group("chars")) @@ -184,7 +176,6 @@ def restore_plugin_prompt_sections(agent: Any, prompt: str) -> None: def _plugin_section_blocks(sections: tuple, position: str) -> List[str]: from hermes_cli.plugins import format_system_prompt_sections - block = format_system_prompt_sections([s for s in sections if s.position == position]) return [block] if block else [] @@ -198,7 +189,6 @@ def _session_start_like(agent: Any, now: Any) -> Any: then ``now``. Stamps are box-local wall-clock: attach that zone first, then convert to ``now``'s zone so the date matches the per-turn clock.""" from datetime import datetime - def _to_display_tz(dt: Any) -> Any: if dt.tzinfo is None: try: @@ -211,7 +201,6 @@ def _session_start_like(agent: Any, now: Any) -> Any: except (ValueError, OSError): pass return dt - session_id = getattr(agent, "session_id", None) db = getattr(agent, "_session_db", None) try: @@ -237,7 +226,6 @@ def _agent_home(agent: Any) -> Optional[Path]: ContextVar, where ambient resolution would leak the launch profile.""" try: from hermes_constants import get_hermes_home_override - override = get_hermes_home_override() if override: return Path(override) @@ -253,7 +241,7 @@ def _agent_home(agent: Any) -> Optional[Path]: def _agent_skills_dir(agent: Any) -> Optional[Path]: """The agent's own ``/skills`` dir, or None to use ambient home.""" home = _agent_home(agent) - return (home / "skills") if home is not None else None + return home / "skills" if home is not None else None def _profile_name_for_home(home: Path) -> str: @@ -263,7 +251,6 @@ def _profile_name_for_home(home: Path) -> str: would misreport as "default".""" try: from hermes_constants import get_default_hermes_root - rel = home.resolve().relative_to((get_default_hermes_root() / "profiles").resolve()) return rel.parts[0] if rel.parts else "default" except (ValueError, OSError): @@ -304,7 +291,6 @@ def _skills_prompt(agent: Any, _r: Any) -> str: avail_toolsets = {_r.get_toolset_for_tool(tool_name) for tool_name in agent.valid_tool_names} - {None, ""} try: from agent.coding_context import coding_compact_skill_categories - _compact_cats = coding_compact_skill_categories(platform=agent.platform, cwd=resolve_context_cwd()) except Exception: _compact_cats = frozenset() @@ -323,7 +309,6 @@ def _bot_mode_parts(agent: Any) -> List[str]: parts: List[str] = [] try: from tools.bot_mode_probe import BOT_CHAT_TITLE, epoch_line, get_bot_mode_protocol_section - _title = str(getattr(agent, "_session_title_hint", "") or "").strip() if not _title: _sdb = getattr(agent, "_session_db", None) @@ -343,7 +328,6 @@ def _bot_mode_parts(agent: Any) -> List[str]: def _ambient_file_safety_profile_name() -> str: from agent.file_safety import _resolve_active_profile_name - return _resolve_active_profile_name() @@ -489,24 +473,22 @@ def _identity_parts(agent: Any, _r: Any, ctx_len: Optional[int]) -> Tuple[List[s """SOUL.md (primary identity; cron keeps the persona while skipping cwd instructions, scoped to the agent's OWN home) or the default identity. Returns ``(parts, soul_loaded)``.""" - if agent.load_soul_identity or not agent.skip_context_files: - _soul_content = _r.load_soul_md(ctx_len, home_override=_agent_home(agent)) - if _soul_content: - return [_soul_content], True - return [DEFAULT_AGENT_IDENTITY], False + wants_soul = agent.load_soul_identity or not agent.skip_context_files + _soul_content = _r.load_soul_md(ctx_len, home_override=_agent_home(agent)) if wants_soul else None + return ([_soul_content], True) if _soul_content else ([DEFAULT_AGENT_IDENTITY], False) def _guidance_parts(agent: Any) -> List[str]: """Universal + tool-aware + model-gated guidance blocks, each gated by its config.yaml key.""" parts: List[str] = [] if agent.valid_tool_names: - if getattr(agent, "_task_completion_guidance", True): - parts.append(TASK_COMPLETION_GUIDANCE) - if getattr(agent, "_parallel_tool_call_guidance", True): - parts.append(PARALLEL_TOOL_CALL_GUIDANCE) - _tool_block = _tool_guidance_block(agent) - if _tool_block: - parts.append(_tool_block) + parts += [ + text for flag, text in ( + ("_task_completion_guidance", TASK_COMPLETION_GUIDANCE), + ("_parallel_tool_call_guidance", PARALLEL_TOOL_CALL_GUIDANCE), + ) if getattr(agent, flag, True) + ] + parts.append(_tool_guidance_block(agent)) # None/empty entries are dropped by _join_tier if not agent.valid_tool_names: return parts # Steering only lands inside tool results, so only reachable with tools. @@ -517,8 +499,7 @@ def _guidance_parts(agent: Any) -> List[str]: # DeepSeek/Kimi/Qwen-class models get it even with enforcement off. if _model_gate(agent._tool_use_enforcement, agent.model, TOOL_USE_ENFORCEMENT_MODELS): parts.append(TOOL_USE_ENFORCEMENT_GUIDANCE) - _model_lower = (agent.model or "").lower() - if "gemini" in _model_lower or "gemma" in _model_lower: + if any(g in (agent.model or "").lower() for g in ("gemini", "gemma")): parts.append(GOOGLE_MODEL_OPERATIONAL_GUIDANCE) if _model_gate(getattr(agent, "_execution_guidance", "auto"), agent.model, EXECUTION_GUIDANCE_MODELS): from agent.prompt_builder import execution_guidance_text @@ -531,7 +512,7 @@ def _alibaba_identity_part(agent: Any) -> List[str]: the real identity so the agent can answer correctly.""" if agent.provider != "alibaba": return [] - _model_short = agent.model.split("/")[-1] if "/" in agent.model else agent.model + _model_short = agent.model.rsplit("/", 1)[-1] return [ f"You are powered by the model named {_model_short}. " f"The exact model ID is {agent.model}. " @@ -545,7 +526,6 @@ def _coding_parts(agent: Any) -> Tuple[List[str], List[str], List[str]]: without tools or when probing fails (it must never block prompt build).""" try: from agent.coding_context import coding_system_prompt_parts - if agent.valid_tool_names: return coding_system_prompt_parts( platform=agent.platform, cwd=resolve_context_cwd(), @@ -564,15 +544,12 @@ def _post_workspace_parts(agent: Any) -> List[str]: if getattr(agent, "_environment_probe", True): try: from tools.env_probe import get_environment_probe_line - _probe_line = get_environment_probe_line() - if _probe_line: - parts.append(_probe_line) + parts.append(get_environment_probe_line()) except Exception: pass # Probe failure must never block prompt build. if getattr(agent, "_bot_mode_protocol", True): parts.extend(_bot_mode_parts(agent)) - parts.append(_active_profile_line(agent)) - parts.append(_platform_hint(agent)) + parts += [_active_profile_line(agent), _platform_hint(agent)] return parts @@ -585,15 +562,13 @@ def _context_files_part(agent: Any, _r: Any, ctx_len: Optional[int], soul_loaded if agent.skip_context_files: return [] launch_artifact = getattr(agent, "_context_cwd_is_launch_artifact", False) - context_files_prompt = _r.build_context_files_prompt( - cwd=None if launch_artifact else resolve_context_cwd(), skip_soul=soul_loaded, - context_length=ctx_len, - allow_install_tree_fallback=agent.platform in ("cli", "tui"), - home_override=_agent_home(agent)) - return [context_files_prompt] if context_files_prompt else [] + return [_r.build_context_files_prompt( + cwd=None if launch_artifact else resolve_context_cwd(), skip_soul=soul_loaded, context_length=ctx_len, + allow_install_tree_fallback=agent.platform in ("cli", "tui"), home_override=_agent_home(agent))] -def _join_tier(parts: List[str]) -> str: +def _join_tier(parts: List[Optional[str]]) -> str: + """Join non-empty parts; None/blank entries are dropped.""" return "\n\n".join(p.strip() for p in parts if p and p.strip()) @@ -607,7 +582,6 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None) # Model context window scales the context-file caps; stable per conversation. _cc_len = getattr(getattr(agent, "context_compressor", None), "context_length", None) _ctx_len = _cc_len if isinstance(_cc_len, int) and _cc_len > 0 else None - # ── Stable tier ──────────────────────────────────────────────── stable_parts, _soul_loaded = _identity_parts(agent, _r, _ctx_len) # The skill_view() pointer dangles without skill tools OR without the @@ -616,25 +590,19 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None) _help_guidance_slot = len(stable_parts) stable_parts.append(HERMES_AGENT_HELP_GUIDANCE_NO_SKILLS) stable_parts.extend(_guidance_parts(agent)) - skills_prompt = _skills_prompt(agent, _r) # Skill-pointer variant requires BOTH skill_view AND the hermes-agent skill # in the rendered index (pure string check — inherits the index's stability). if "skill_view" in (agent.valid_tool_names or set()) and "- hermes-agent:" in skills_prompt: stable_parts[_help_guidance_slot] = HERMES_AGENT_HELP_GUIDANCE - stable_parts.extend(_alibaba_identity_part(agent)) - _env_hints = _r.build_environment_hints() - if _env_hints: - stable_parts.append(_env_hints) - + stable_parts.append(_r.build_environment_hints()) # Coding posture: operating brief stays in the stable prefix; the live # git/workspace snapshot sits behind its own cache boundary, and the blocks # below it must keep their historical post-snapshot position. coding_prefix_parts, coding_workspace_parts, coding_trailing_parts = _coding_parts(agent) stable_parts.extend(coding_prefix_parts) post_workspace_parts = _post_workspace_parts(agent) - # ── Context tier (cwd-dependent, may change between sessions) ─ context_parts: List[str] = [] (context_parts if coding_workspace_parts else stable_parts).extend( @@ -644,17 +612,14 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None) if system_message is not None: context_parts.append(system_message) context_parts.extend(_context_files_part(agent, _r, _ctx_len, _soul_loaded)) - # ── Volatile tier (most likely to differ on a rebuild; kept last so the stable prefix stays reusable) ── # Skills are runtime-mutable, so the index leads the volatile band: on a longest-prefix # backend an unchanged index stays inside the reused prefix; a changed one re-prefills from here. - volatile_parts: List[str] = [skills_prompt] if skills_prompt else [] - volatile_parts.extend(_memory_parts(agent)) + volatile_parts: List[str] = [skills_prompt, *_memory_parts(agent)] # Plugin sections are confined to one coarse anchor in the volatile tail so # a resumed process can reconstruct the stable prefix without re-running plugins. volatile_parts.extend(_plugin_section_blocks(_frozen_plugin_prompt_sections(agent), "after_memory")) volatile_parts.append(_timestamp_line(agent)) - return { "stable": _join_tier(stable_parts), "context": _join_tier(context_parts), @@ -667,12 +632,11 @@ def build_system_prompt(agent: Any, system_message: Optional[str] = None) -> str only rebuilt after compression. Tiers are ordered stable -> context -> volatile so implicit longest-prefix caches keep the unchanged scaffold.""" parts = build_system_prompt_parts(agent, system_message=system_message) - joined = "\n\n".join(p for p in (parts["stable"], parts["context"], parts["volatile"]) if p) agent._cached_system_prompt_static = parts["stable"] # Surface context-file truncation warnings in chat, not only in logs. for warning in drain_truncation_warnings(): agent._emit_status(warning) - return joined + return "\n\n".join(p for p in (parts["stable"], parts["context"], parts["volatile"]) if p) def invalidate_system_prompt(agent: Any) -> None: