Merge branch 'simp/r3-33-B' into simp/integration3

tools/path_security.py

@@ -1,17 +1,11 @@
 """Shared path validation helpers for tool implementations (skills, cron, credential files)."""
 
-import logging
 from pathlib import Path
 from typing import Optional
 
-logger = logging.getLogger(__name__)
-
 
 def validate_within_dir(path: Path, root: Path) -> Optional[str]:
-    """Return an error message if *path* does not resolve inside *root*, else None.
-
-    ``Path.resolve()`` follows symlinks and normalises ``..`` before the check.
-    """
+    """Error message if *path* does not resolve inside *root* (symlinks and ``..`` followed)."""
     try:
         path.resolve().relative_to(root.resolve())
     except (ValueError, OSError) as exc:

tools/plugin_guard.py

@@ -1,20 +1,12 @@
 #!/usr/bin/env python3
 """Plugin Guard — security scanner for externally-installed plugins.
 
-Reuses the ``tools/skills_guard.py`` static-analysis engine for
-``hermes plugins install`` / ``update``, which otherwise clone and execute
-arbitrary Git repositories unscanned.
-
-Plugins run Python in-process (more dangerous than skills) but are *expected*
-to read their own API keys from env vars, call provider HTTP APIs and spawn
-subprocesses, so the raw skill patterns would flag every legitimate provider
-plugin. Hence: full pattern set on docs/config files (where prompt-injection
-lives); the "reads own env secret" / "HTTP call with key" family is exempt on
-*code* files while genuinely malicious signals stay; plugin-sized structural
-limits; VCS/venv noise skipped.
-
-Verdict → install policy: ``safe`` installs; ``caution`` requires explicit
-confirmation (prompt, ``--force``, or caller callback); ``dangerous`` is
+Reuses the ``tools/skills_guard.py`` engine for ``hermes plugins install``/``update``.
+Plugins run Python in-process but are *expected* to read their own env keys, call
+provider HTTP APIs and spawn subprocesses, so: full pattern set on docs/config files
+(where prompt-injection lives); the "reads own secret"/"HTTP call with key" family is
+exempt on *code* files; plugin-sized structural limits; VCS/venv noise skipped.
+Verdict policy: ``safe`` installs; ``caution`` needs confirmation; ``dangerous`` is
 blocked and ``--force`` does NOT override.
 """
 
@@ -25,11 +17,7 @@ from pathlib import Path
 from typing import Iterator, List, Optional, Tuple
 
 from tools.skills_guard import (
-    Finding,
-    ScanResult,
-    SUSPICIOUS_BINARY_EXTENSIONS,
-    _determine_verdict,
-    format_scan_report,
+    Finding, ScanResult, SUSPICIOUS_BINARY_EXTENSIONS, _determine_verdict, format_scan_report,
     scan_file,
 )
 
@@ -41,14 +29,12 @@ EXCLUDED_DIRS = {
     ".mypy_cache", ".pytest_cache", ".ruff_cache", ".tox",
 }
 
-# Code files, where "reads an env secret" / "HTTP call with a key variable"
-# is the NORMAL, documented plugin pattern (requires_env).
-CODE_FILE_EXTENSIONS = {
-    ".py", ".js", ".ts", ".sh", ".bash", ".rb", ".pl", ".php",
-}
+# Code files, where "reads an env secret" / "HTTP call with a key" is the normal
+# documented plugin pattern (requires_env).
+CODE_FILE_EXTENSIONS = {".py", ".js", ".ts", ".sh", ".bash", ".rb", ".pl", ".php"}
 
-# skills_guard pattern ids exempt on code files (every legitimate provider
-# plugin exhibits them); they still apply in full to docs/config files.
+# skills_guard pattern ids exempt on code files (every legitimate provider plugin
+# exhibits them); they still apply in full to docs/config files.
 CODE_EXEMPT_PATTERN_IDS = {
     "python_environ_get_secret",
     "python_getenv_secret",
@@ -60,28 +46,24 @@ CODE_EXEMPT_PATTERN_IDS = {
     "env_exfil_fetch",
     "env_exfil_curl",
     "env_exfil_wget",
-    # Agent-facing instruction patterns are meaningless inside code
-    # (docstrings/comments about prompts trip them constantly).
+    # Agent-facing instruction patterns are meaningless inside code (docstrings
+    # about prompts trip them constantly).
     "context_exfil",
     "send_to_url",
     "fake_policy",
-    # Plugins legitimately write their own settings into config.yaml during
-    # post_setup, and encode credentials (e.g. HTTP Basic auth) with base64.
+    # Plugins legitimately write their settings into config.yaml during post_setup
+    # and base64-encode credentials (HTTP Basic auth).
     "agent_config_mod",
     "agent_config_contract",
     "encoded_exfil",
 }
 
-# Severity remaps for plugins. A bundled binary is warn-tier (plugin repos
-# occasionally vendor one legitimately; skills never should). A mere
-# ``~/.hermes/.env`` reference is the DOCUMENTED way plugin READMEs tell users
-# where keys go — informational; actually READING it still trips
-# ``read_secrets_file`` (critical). ``curl | sh`` install instructions are
-# common in READMEs: caution, not an unoverridable block.
+# Plugin severity remaps: a bundled binary is warn-tier (repos occasionally vendor
+# one legitimately); a mere ``~/.hermes/.env`` mention is how READMEs tell users where
+# keys go (READING it still trips ``read_secrets_file``, critical); ``curl | sh``
+# install instructions are common in READMEs — caution, not an unoverridable block.
 SEVERITY_REMAP = {
-    "binary_file": "high",
-    "hermes_env_access": "medium",
-    "curl_pipe_shell": "high",
+    "binary_file": "high", "hermes_env_access": "medium", "curl_pipe_shell": "high"
 }
 
 # Structural limits — plugins are real codebases, far larger than skills.
@@ -169,11 +151,7 @@ def _check_plugin_structure(plugin_dir: Path) -> List[Finding]:
 
 
 def scan_plugin(plugin_dir: Path, source: str = "") -> ScanResult:
-    """Scan a plugin directory (typically the temp clone) for security threats.
-
-    Returns a ScanResult with verdict ``safe`` | ``caution`` | ``dangerous``;
-    every externally installed plugin is ``community`` trust.
-    """
+    """Scan a plugin directory (typically the temp clone); every external plugin is ``community`` trust."""
     all_findings: List[Finding] = []
 
     if plugin_dir.is_dir():
@@ -189,31 +167,20 @@ def scan_plugin(plugin_dir: Path, source: str = "") -> ScanResult:
     else:
         summary = f"{plugin_dir.name}: clean scan, no threats detected"
     result = ScanResult(
-        skill_name=plugin_dir.name,
-        source=source or plugin_dir.name,
-        trust_level="community",
-        verdict=verdict,
-        findings=all_findings,
-        scanned_at=datetime.now(timezone.utc).isoformat(),
+        skill_name=plugin_dir.name, source=source or plugin_dir.name, trust_level="community",
+        verdict=verdict, findings=all_findings, scanned_at=datetime.now(timezone.utc).isoformat(),
         summary=summary,
     )
     result.scan_provenance = {
-        "scanner_version": PLUGIN_SCANNER_VERSION,
-        "verdict": verdict,
-        "source": result.source,
+        "scanner_version": PLUGIN_SCANNER_VERSION, "verdict": verdict, "source": result.source
     }
     return result
 
 
 def should_allow_plugin_install(
-    result: ScanResult,
-    force: bool = False,
+    result: ScanResult, force: bool = False,
 ) -> Tuple[Optional[bool], str]:
-    """Map a plugin scan verdict to ``(allowed, reason)``.
-
-    ``True`` installs, ``None`` needs explicit confirmation (caution), ``False``
-    is blocked — ``force`` never overrides ``dangerous``.
-    """
+    """Map a verdict to ``(allowed, reason)``: True installs, None asks to confirm, False blocks."""
     n = len(result.findings)
     if result.verdict == "safe":
         return True, "Allowed (clean scan)"
@@ -228,8 +195,5 @@ def should_allow_plugin_install(
 
 
 __all__ = [
-    "scan_plugin",
-    "should_allow_plugin_install",
-    "format_scan_report",
-    "PLUGIN_SCANNER_VERSION",
+    "scan_plugin", "should_allow_plugin_install", "format_scan_report", "PLUGIN_SCANNER_VERSION"
 ]

tools/project_tools.py

@@ -1,15 +1,11 @@
 #!/usr/bin/env python3
 """Project tools — the agent's INTENTIONAL handle on first-class Projects.
 
-Projects (per-profile ``projects.db``) are the named workspaces the desktop
-sidebar groups sessions into. Creating / switching a project is a deliberate act
-expressed as explicit tools — never a side effect of a terminal ``cd``.
-
-Exposed only on GUI sessions: the tools live in the `project` toolset (kept off
-``_HERMES_CORE_TOOLS``) which the desktop/TUI gateway folds into its resolved
-toolsets, so no CLI/messaging/cron schema carries them. The GUI also wires
-``set_project_workspace_callback`` so a create/switch re-anchors the live
-session's cwd and the sidebar follows the move; the DB write is the durable part.
+Projects (per-profile ``projects.db``) are the named workspaces the desktop sidebar
+groups sessions into; creating/switching one is an explicit tool call, never a side
+effect of a terminal ``cd``. GUI-only: the `project` toolset stays off
+``_HERMES_CORE_TOOLS`` and is folded in by the desktop/TUI gateway, which also wires
+``set_project_workspace_callback`` so the live session's cwd and sidebar follow.
 """
 
 import json
@@ -18,10 +14,9 @@ from typing import Callable, Optional
 
 from tools.registry import registry
 
-# Set by the GUI gateway (tui_gateway) at session wiring. Receives
-# ``(task_id, primary_path, project_name)`` and re-anchors that session's
-# workspace + refreshes the sidebar. ``None`` in CLI / messaging contexts — the
-# DB write still happens; there's just no live GUI session to move.
+# Set by the GUI gateway at session wiring: ``(task_id, primary_path, project_name)``
+# re-anchors that session's workspace. ``None`` in CLI/messaging contexts — the DB
+# write still happens; there is just no live GUI session to move.
 _workspace_callback: Optional[Callable[[str, str, str], None]] = None
 
 
@@ -66,6 +61,14 @@ def _resolve(conn, token: str):
     return None
 
 
+def _activated(proj, task_id: Optional[str]) -> str:
+    primary = _primary_path(proj)
+    _apply_workspace(task_id, primary, proj.name)
+    return json.dumps({
+        "success": True, "id": proj.id, "slug": proj.slug, "name": proj.name, "primary_path": primary,
+    })
+
+
 def project_list(task_id: Optional[str] = None) -> str:
     from hermes_cli import projects_db as pdb
 
@@ -77,11 +80,8 @@ def project_list(task_id: Optional[str] = None) -> str:
         "active_id": active,
         "projects": [
             {
-                "id": p.id,
-                "slug": p.slug,
-                "name": p.name,
-                "primary_path": _primary_path(p),
-                "active": p.id == active,
+                "id": p.id, "slug": p.slug, "name": p.name,
+                "primary_path": _primary_path(p), "active": p.id == active,
             }
             for p in projects
         ],
@@ -103,9 +103,8 @@ def project_create(name: str, path: Optional[str] = None, task_id: Optional[str]
         with pdb.connect_closing() as conn:
             existing = pdb.find_by_primary_path(conn, folder) if folder else None
             if existing is not None:
-                # Idempotent create: the folder already belongs to a project.
-                # Re-activating it beats minting a duplicate — duplicated
-                # projects render N identical sidebar subtrees (#75820).
+                # Idempotent create: re-activating the folder's project beats minting a
+                # duplicate (duplicates render N identical sidebar subtrees).
                 pdb.set_active(conn, existing.id)
                 proj = existing
             else:
@@ -117,11 +116,7 @@ def project_create(name: str, path: Optional[str] = None, task_id: Optional[str]
 
     if proj is None:
         return json.dumps({"success": False, "error": "project vanished after create"})
-
-    primary = _primary_path(proj)
-    _apply_workspace(task_id, primary, proj.name)
-
-    return json.dumps({"success": True, "id": proj.id, "slug": proj.slug, "name": proj.name, "primary_path": primary})
+    return _activated(proj, task_id)
 
 
 def project_switch(project: str, task_id: Optional[str] = None) -> str:
@@ -132,28 +127,26 @@ def project_switch(project: str, task_id: Optional[str] = None) -> str:
         if proj is None:
             return json.dumps({"success": False, "error": f"no project matching '{project}'"})
         pdb.set_active(conn, proj.id)
+    return _activated(proj, task_id)
 
-    primary = _primary_path(proj)
-    _apply_workspace(task_id, primary, proj.name)
 
-    return json.dumps({"success": True, "id": proj.id, "slug": proj.slug, "name": proj.name, "primary_path": primary})
+_ACTIONS = {
+    "list": lambda args, tid: project_list(task_id=tid),
+    "create": lambda args, tid: project_create(
+        name=args.get("name", ""), path=args.get("path"), task_id=tid,
+    ),
+    "switch": lambda args, tid: project_switch(project=args.get("name", ""), task_id=tid),
+}
 
 
 def _handle_project(args, **kw):
-    action = (args.get("action") or "").strip()
-    tid = kw.get("task_id")
-    if action == "list":
-        return project_list(task_id=tid)
-    if action == "create":
-        return project_create(name=args.get("name", ""), path=args.get("path"), task_id=tid)
-    if action == "switch":
-        return project_switch(project=args.get("name", ""), task_id=tid)
-    return json.dumps({"success": False, "error": "action must be one of: create, switch, list."})
+    action = _ACTIONS.get((args.get("action") or "").strip())
+    if action is None:
+        return json.dumps({"success": False, "error": "action must be one of: create, switch, list."})
+    return action(args, kw.get("task_id"))
 
 
-# Consolidated (#95681, maintainer-directed): project_list/create/switch each
-# re-taught "desktop Projects (named workspaces)"; one action enum says it
-# once (244 -> ~145 tok).
+# One action enum instead of three tools: each re-taught "desktop Projects" (244 -> ~145 tok).
 registry.register(
     name="desktop_project",
     toolset="project",

tools/schema_sanitizer.py

@@ -1,21 +1,11 @@
 """Sanitize tool JSON schemas for broad LLM-backend compatibility.
 
-Some backends are strict about JSON Schema shapes that OpenAI/Anthropic/most
-cloud providers silently accept — llama.cpp's ``json-schema-to-grammar`` fails
-the whole request (``Unrecognized schema: "object"``), Anthropic rejects
-nullable ``anyOf`` at the top of ``input_schema``, Fireworks rejects ``default``
-beside ``$ref``, OpenAI's Codex backend rejects top-level combinators. Known
-hostile constructs:
-
-* ``{"type": "object"}`` with no ``properties``.
-* A bare string (``"object"``) where a schema dict belongs (malformed MCP output).
-* ``"type": ["string", "null"]`` array types.
-* ``anyOf``/``oneOf`` unions whose only purpose is to permit ``null``.
-* ``default`` (etc.) alongside ``$ref`` — e.g. ``{"$ref": "#/$defs/Foo", "default": null}``.
-
-This module walks the final tool schema tree (after MCP normalization and any
-per-tool dynamic rebuilds) and fixes those in place on a deep copy. It is
-deliberately conservative: it only modifies shapes the backend couldn't use.
+Strict backends reject shapes OpenAI/Anthropic silently accept: llama.cpp's
+``json-schema-to-grammar`` fails on ``{"type": "object"}`` without ``properties``,
+bare-string schemas and ``type`` arrays; Anthropic rejects nullable ``anyOf`` at the
+top of ``input_schema``; Fireworks rejects ``default`` beside ``$ref``; OpenAI's
+Codex backend rejects top-level combinators. This module walks the final tool
+schema tree on a deep copy and fixes only those shapes.
 """
 
 from __future__ import annotations
@@ -28,9 +18,9 @@ from typing import Any, Callable
 logger = logging.getLogger(__name__)
 
 
-# Anthropic (and Bedrock/Vertex/Azure fronting it) reject tool input schemas
-# whose property keys don't match this pattern; one bad key anywhere in the
-# tools array 400s the entire request (Cloudflare's MCP ships 61 such keys).
+# Anthropic (and Bedrock/Vertex/Azure fronting it) reject tool input schemas whose
+# property keys don't match this; one bad key anywhere in the tools array 400s the
+# entire request (Cloudflare's MCP ships 61 such keys).
 _PROP_KEY_RE = re.compile(r"^[a-zA-Z0-9_.-]{1,64}$")
 _PROP_KEY_BAD_CHARS = re.compile(r"[^a-zA-Z0-9_.-]")
 
@@ -49,11 +39,10 @@ def sanitize_property_key(key: str) -> str:
 
 
 def _rename_property_keys(props: dict, path: str) -> dict[str, str]:
-    """Return {original_key: conforming_key} for one properties dict.
+    """Return {original_key: conforming_key} for one properties dict (identity entries omitted).
 
-    Identity entries are omitted. Deterministic (insertion order, numeric
-    suffixes on collision) so the model-visible schema and the dispatch-time
-    reverse map computed from the registry's original schema always agree.
+    Deterministic (insertion order, numeric suffixes on collision) so the model-visible
+    schema and the dispatch-time reverse map from the registry's original schema agree.
     """
     renames: dict[str, str] = {}
     taken = {k for k in props if _PROP_KEY_RE.match(k)}
@@ -70,16 +59,15 @@ def _rename_property_keys(props: dict, path: str) -> dict[str, str]:
         renames[key] = candidate
         logger.debug(
             "schema_sanitizer[%s]: renamed property key %r -> %r "
-            "(provider key-pattern compat)", path, key, candidate,
-        )
+            "(provider key-pattern compat)", path, key, candidate)
     return renames
 
 
 def unrename_tool_args(params_schema: Any, args: Any) -> Any:
     """Map sanitized property keys in model-emitted args back to wire names.
 
-    ``params_schema`` is the ORIGINAL (unsanitized) registry schema. Recurses
-    into object values and array items; unknown keys pass through untouched.
+    ``params_schema`` is the ORIGINAL (unsanitized) registry schema. Recurses into
+    object values and array items; unknown keys pass through untouched.
     """
     if not isinstance(params_schema, dict) or not isinstance(args, dict):
         return args
@@ -96,24 +84,20 @@ def unrename_tool_args(params_schema: Any, args: Any) -> Any:
                 value = unrename_tool_args(subschema, value)
             elif isinstance(value, list) and isinstance(subschema.get("items"), dict):
                 value = [
-                    unrename_tool_args(subschema["items"], item)
-                    if isinstance(item, dict) else item
-                    for item in value
-                ]
+                    unrename_tool_args(subschema["items"], item) if isinstance(item, dict) else item
+                    for item in value]
         out[orig] = value
     return out
 
 
 def sanitize_tool_schemas(tools: list[dict]) -> list[dict]:
-    """Return a deep-copied ``tools`` list (OpenAI format) with each tool's
-    parameter schema sanitized; callers may mutate the result freely."""
+    """Deep-copied ``tools`` (OpenAI format) with sanitized parameter schemas; callers may mutate."""
     if not tools:
         return tools
     return [_sanitize_single_tool(tool) for tool in tools]
 
 
 def _sanitize_single_tool(tool: dict) -> dict:
-    """Deep-copy and sanitize a single OpenAI-format tool entry."""
     out = copy.deepcopy(tool)
     fn = out.get("function") if isinstance(out, dict) else None
     if not isinstance(fn, dict):
@@ -134,10 +118,9 @@ def _sanitize_single_tool(tool: dict) -> dict:
             top["type"] = "object"
         if not isinstance(top.get("properties"), dict):
             top["properties"] = {}
-    # Collapse nullable unions the recursive pass leaves intact (it only
-    # handles the array-form ``type: [X, "null"]``); keep ``nullable: true`` so
-    # runtime coercion (``model_tools._schema_allows_null``) still maps a
-    # model-emitted ``"null"`` string to Python ``None``.
+    # Collapse nullable unions the recursive pass leaves intact (it only handles the
+    # array-form ``type: [X, "null"]``); keep ``nullable: true`` so runtime coercion
+    # (``model_tools._schema_allows_null``) still maps a model-emitted ``"null"`` to None.
     top = strip_nullable_unions(top, keep_nullable_hint=True)
     top = _strip_top_level_combinators(top, path=name)
     fn["parameters"] = _strip_ref_siblings(top)
@@ -149,8 +132,7 @@ _REF_FORBIDDEN_SIBLINGS = frozenset({"default"})
 
 
 def _strip_ref_siblings(node: Any) -> Any:
-    """Recursively drop forbidden sibling keywords from nodes carrying ``$ref``
-    (Fireworks: ``keyword(s) ['default'] not allowed at the same level as $ref``)."""
+    """Recursively drop forbidden siblings of ``$ref`` (Fireworks rejects ``default`` there)."""
     if isinstance(node, list):
         return [_strip_ref_siblings(item) for item in node]
     if not isinstance(node, dict):
@@ -166,12 +148,10 @@ _TOP_LEVEL_FORBIDDEN_KEYS = ("allOf", "anyOf", "oneOf", "enum", "not")
 
 
 def _strip_top_level_combinators(params: dict, *, path: str = "<tool>") -> dict:
-    """Drop combinator keywords from the TOP level of a parameters schema only.
+    """Drop combinators from the TOP level only (Codex rejects them there).
 
-    OpenAI's Codex backend rejects ``oneOf/anyOf/allOf/enum/not`` at the top
-    level. They are usually conditional-required hints; dropping them does not
-    change which argument values are valid (handlers re-validate). Nested
-    combinators are preserved.
+    They are usually conditional-required hints; dropping them does not change which
+    argument values are valid (handlers re-validate). Nested combinators are preserved.
     """
     if not isinstance(params, dict):
         return params
@@ -181,8 +161,7 @@ def _strip_top_level_combinators(params: dict, *, path: str = "<tool>") -> dict:
             logger.debug(
                 "schema_sanitizer[%s]: stripped top-level %r combinator "
                 "from tool parameters (strict-backend compat)",
-                path, key,
-            )
+                path, key)
             out.pop(key, None)
     return out
 
@@ -201,23 +180,19 @@ def _carry_union_meta(outer: dict, replacement: dict, *, skip_default_on_ref: bo
             replacement[meta_key] = outer[meta_key]
 
 
-def strip_nullable_unions(
-    schema: Any,
-    *,
-    keep_nullable_hint: bool = True,
-) -> Any:
+def strip_nullable_unions(schema: Any, *, keep_nullable_hint: bool = True) -> Any:
     """Collapse ``anyOf``/``oneOf`` nullable unions to the single non-null branch.
 
-    MCP/Pydantic optional fields arrive as
-    ``{"anyOf": [{"type": "string"}, {"type": "null"}], "default": null}``;
-    Anthropic rejects the null branch, and optionality is already expressed by
-    the parent's ``required``. Only collapses when a null branch was dropped
-    AND exactly one non-null branch survives. Outer metadata is carried over.
-    ``keep_nullable_hint`` sets ``nullable: true`` on the replacement for
-    downstream consumers (runtime ``"null"`` → ``None`` coercion).
+    MCP/Pydantic optional fields arrive as ``{"anyOf": [{"type": "string"}, {"type":
+    "null"}], "default": null}``; Anthropic rejects the null branch, and optionality is
+    already expressed by the parent's ``required``. Only collapses when a null branch was
+    dropped AND exactly one non-null branch survives; outer metadata is carried over.
+    ``keep_nullable_hint`` sets ``nullable: true`` for runtime ``"null"`` → ``None`` coercion.
     """
     if isinstance(schema, list):
-        return [strip_nullable_unions(item, keep_nullable_hint=keep_nullable_hint) for item in schema]
+        return [
+            strip_nullable_unions(item, keep_nullable_hint=keep_nullable_hint) for item in schema
+        ]
     if not isinstance(schema, dict):
         return schema
 
@@ -240,27 +215,22 @@ def strip_nullable_unions(
 
 
 _CONST_PRIMITIVE_TYPES: dict[type, str] = {
-    bool: "boolean",
-    int: "integer",
-    float: "number",
-    str: "string",
+    bool: "boolean", int: "integer", float: "number", str: "string",
 }
 
 
 def _const_branch_type(branch: Any) -> str | None:
     """JSON-Schema primitive type of a pure ``const`` branch, else None.
 
-    Qualifies when the dict carries a primitive ``const`` and any declared
-    ``type`` matches it; ``title``/``description`` are allowed, any other
-    constraining keyword disqualifies.
+    Qualifies when the dict carries a primitive ``const`` and any declared ``type``
+    matches it; ``title``/``description`` are allowed, any other keyword disqualifies.
     """
     if not isinstance(branch, dict) or "const" not in branch:
         return None
     if set(branch) - {"const", "type", "title", "description"}:
         return None
-    value = branch["const"]
     # ``type(value) is`` (not isinstance): bool is a subclass of int.
-    json_type = _CONST_PRIMITIVE_TYPES.get(type(value))
+    json_type = _CONST_PRIMITIVE_TYPES.get(type(branch["const"]))
     if json_type is None:
         return None
     declared = branch.get("type")
@@ -272,17 +242,14 @@ def _const_branch_type(branch: Any) -> str | None:
 def collapse_const_unions(schema: Any) -> Any:
     """Collapse ``anyOf``/``oneOf`` unions of same-typed consts to ``enum``.
 
-    Ported from block/goose ``tool_schema_normalize.rs`` (Apache-2.0). MCP
-    servers generated from Rust/TS union types emit
-    ``{"anyOf": [{"const": "red"}, {"const": "green"}]}``; strict backends
-    mishandle these while ``{"type": "string", "enum": [...]}`` is universal.
-
-    Applies only when EVERY non-null branch is a pure ``const`` of one
-    primitive type (``bool`` never merges with ``integer``). One
-    ``{"type": "null"}`` branch is tolerated and recorded as ``nullable: true``
-    (``strip_nullable_unions`` only handles single-non-null unions, so
-    null+multi-const unions land here). Enum order preserves branch order;
-    outer metadata is carried over; input is never mutated.
+    Ported from block/goose ``tool_schema_normalize.rs`` (Apache-2.0). Rust/TS-generated
+    MCP servers emit ``{"anyOf": [{"const": "red"}, {"const": "green"}]}``; strict
+    backends mishandle these while ``{"type": "string", "enum": [...]}`` is universal.
+    Applies only when EVERY non-null branch is a pure ``const`` of one primitive type
+    (``bool`` never merges with ``integer``). One ``{"type": "null"}`` branch is
+    tolerated and recorded as ``nullable: true`` (null+multi-const unions land here,
+    not in ``strip_nullable_unions``). Enum order preserves branch order; outer
+    metadata is carried over; input is never mutated.
     """
     if isinstance(schema, list):
         return [collapse_const_unions(item) for item in schema]
@@ -294,9 +261,7 @@ def collapse_const_unions(schema: Any) -> Any:
         variants = out.get(key)
         if not isinstance(variants, list) or not variants:
             continue
-        null_branches = [
-            item for item in variants if _is_null_branch(item) and "const" not in item
-        ]
+        null_branches = [item for item in variants if _is_null_branch(item) and "const" not in item]
         const_branches = [item for item in variants if item not in null_branches]
         if len(null_branches) > 1 or not const_branches:
             continue
@@ -315,20 +280,17 @@ def collapse_const_unions(schema: Any) -> Any:
 
 
 _BARE_TYPE_NAMES = frozenset({"object", "string", "number", "integer", "boolean", "array", "null"})
-# Sibling keywords whose values are NOT schemas: recursing would mistake literal
-# strings like "path" for bare-string schemas. Passed through unchanged
-# (``required`` remapped through property renames).
+# Sibling keywords whose values are NOT schemas: recursing would mistake literal strings
+# like "path" for bare-string schemas. Passed through (``required`` follows property renames).
 _NON_SCHEMA_LIST_KEYS = frozenset({"required", "enum", "examples", "dependentRequired"})
 
 
 def _normalize_type_array(value: list, out: dict) -> None:
-    """Normalize a ``type: [...]`` array into *out*.
+    """Normalize a ``type: [...]`` array into *out* (llama.cpp and Gemini-via-OpenAI reject arrays).
 
-    Several backends reject array types (llama.cpp's grammar generator; Gemini
-    via OpenAI-compatible transports 400s). Per the AI-SDK behavior: one
-    non-null type → ``type: X`` (+ ``nullable`` if ``null`` present); several →
-    ``anyOf`` of single-type schemas so EVERY branch survives; none → ``null``
-    or the object fallback. Ported from anomalyco/opencode#31877.
+    Per the AI-SDK behavior: one non-null type → ``type: X`` (+ ``nullable`` if ``null``
+    present); several → ``anyOf`` of single-type schemas so EVERY branch survives; none →
+    ``null`` or the object fallback. Ported from anomalyco/opencode#31877.
     """
     has_null = "null" in value
     non_null = [t for t in value if isinstance(t, str) and t != "null"]
@@ -346,29 +308,22 @@ def _normalize_type_array(value: list, out: dict) -> None:
 def _sanitize_node(node: Any, path: str) -> Any:
     """Recursively sanitize a JSON-Schema fragment.
 
-    - Bare-string schema values become ``{"type": <value>}`` (unknown strings
-      become a permissive object schema rather than something backends reject).
-    - Object-typed nodes gain ``properties: {}`` (llama.cpp can't constrain a
-      free-form object).
-    - ``type`` arrays are normalized (see ``_normalize_type_array``).
-    - Recurses into ``properties``, ``items``, ``additionalProperties``,
-      ``anyOf``/``oneOf``/``allOf`` and ``$defs``/``definitions``; property
-      keys are renamed to the provider-safe pattern and ``required`` follows.
-    - ``required`` entries that don't exist in ``properties`` are pruned
-      (malformed MCP schemas; built-in/plugin tools skip the MCP-level check).
+    Bare-string schemas become ``{"type": <value>}`` (unknown strings → permissive
+    object); object nodes gain ``properties: {}``; ``type`` arrays are normalized;
+    recursion covers ``properties``/``items``/``additionalProperties``/combinators/
+    ``$defs``/``definitions``; property keys are renamed to the provider-safe pattern
+    and ``required`` follows, with entries missing from ``properties`` pruned.
     """
     if isinstance(node, str):
         if node in _BARE_TYPE_NAMES:
             logger.debug(
                 "schema_sanitizer[%s]: replacing bare-string schema %r "
                 "with {'type': %r}",
-                path, node, node,
-            )
+                path, node, node)
             return _empty_object() if node == "object" else {"type": node}
         logger.debug(
             "schema_sanitizer[%s]: replacing non-schema string %r "
-            "with empty object schema", path, node,
-        )
+            "with empty object schema", path, node)
         return _empty_object()
 
     if isinstance(node, list):
@@ -391,8 +346,7 @@ def _sanitize_node(node: Any, path: str) -> Any:
             renames = prop_renames if key == "properties" else {}
             out[key] = {
                 renames.get(sub_k, sub_k): _sanitize_node(sub_v, f"{path}.{key}.{renames.get(sub_k, sub_k)}")
-                for sub_k, sub_v in value.items()
-            }
+                for sub_k, sub_v in value.items()}
         elif key in {"items", "additionalProperties"}:
             # Bool ``additionalProperties`` is valid and widely accepted;
             # ``items: true/false`` is non-standard but preserved rather than dropped.
@@ -420,19 +374,21 @@ def _sanitize_node(node: Any, path: str) -> Any:
     return out
 
 
-# =============================================================================
+# ---------------------------------------------------------------------------
 # Reactive strips — only invoked after a backend rejects a schema
-# =============================================================================
+# ---------------------------------------------------------------------------
 
 _STRIP_ON_RECOVERY_KEYS = frozenset({"pattern", "format"})
 
 
-def _reactive_strip(tools: list[dict], strip_node: Callable[[dict], int], log_msg: str) -> tuple[list[dict], int]:
-    """Walk every tool's parameters in place, applying *strip_node* to each dict
-    node (it returns how many keywords it removed). Handles OpenAI format
-    (``{"function": {"parameters": ...}}``) and Responses format
-    (``{"name": ..., "parameters": ...}`` — codex_responses mode, xAI, etc.).
-    Returns ``(tools, stripped_count)`` — the same list reference."""
+def _reactive_strip(
+    tools: list[dict], strip_node: Callable[[dict], int], log_msg: str,
+) -> tuple[list[dict], int]:
+    """Apply *strip_node* (returns keywords removed) to every dict node of each tool's parameters.
+
+    Handles OpenAI format (``{"function": {"parameters": ...}}``) and Responses format
+    (``{"name": ..., "parameters": ...}``). Returns ``(tools, stripped_count)`` — same list.
+    """
     if not tools:
         return tools, 0
     stripped = 0
@@ -453,8 +409,7 @@ def _reactive_strip(tools: list[dict], strip_node: Callable[[dict], int], log_ms
         fn = tool.get("function")
         if isinstance(fn, dict) and isinstance(fn.get("parameters"), dict):
             _walk(fn["parameters"])
-            continue
-        if isinstance(tool.get("parameters"), dict):
+        elif isinstance(tool.get("parameters"), dict):
             _walk(tool["parameters"])
 
     if stripped:
@@ -465,14 +420,11 @@ def _reactive_strip(tools: list[dict], strip_node: Callable[[dict], int], log_ms
 def strip_pattern_and_format(tools: list[dict]) -> tuple[list[dict], int]:
     """Strip ``pattern``/``format`` keywords from tool schemas, in place.
 
-    Reactive: invoked only after llama.cpp's grammar converter rejected a
-    schema with HTTP 400. Its regex engine supports a small ECMAScript subset
-    (no ``\\d``/``\\w``/``\\s``) and most ``format`` values; cloud providers rely
-    on these as prompting hints, so they stay in the default schema.
-
-    Only strips as a sibling of ``type``/combinators (i.e. on schema nodes), so
-    a property literally *named* ``pattern`` (``search_files``) is untouched —
-    property names live inside ``properties``, not beside ``type``.
+    Reactive: only after llama.cpp's grammar converter rejected a schema (HTTP 400); its
+    regex engine supports a small ECMAScript subset and few ``format`` values, while
+    cloud providers use these as prompting hints, so they stay in the default schema.
+    Only strips beside ``type``/combinators (schema nodes), so a property literally
+    *named* ``pattern`` (``search_files``) inside ``properties`` is untouched.
     """
     def _strip(node: dict) -> int:
         if not ("type" in node or "anyOf" in node or "oneOf" in node or "allOf" in node):
@@ -485,17 +437,15 @@ def strip_pattern_and_format(tools: list[dict]) -> tuple[list[dict], int]:
     return _reactive_strip(
         tools, _strip,
         "schema_sanitizer: stripped %d pattern/format keyword(s) from "
-        "tool schemas (llama.cpp grammar-parse recovery)",
-    )
+        "tool schemas (llama.cpp grammar-parse recovery)")
 
 
 def strip_slash_enum(tools: list[dict]) -> tuple[list[dict], int]:
     """Strip ``enum`` keywords whose string values contain ``/``, in place.
 
-    xAI's ``/v1/responses`` and ``/v1/chat/completions`` compile schemas to a
-    grammar that rejects ``/`` in enum values (HTTP 400 before any token) —
-    typically MCP enums of HuggingFace model IDs or owner/name env IDs. The
-    constraint is a prompting hint only; the model still sees the description.
+    xAI's Responses/chat endpoints compile schemas to a grammar that rejects ``/`` in
+    enum values (HTTP 400 before any token) — typically MCP enums of HuggingFace model
+    IDs. The constraint is a prompting hint only; the model still sees the description.
     """
     def _strip(node: dict) -> int:
         enum_val = node.get("enum")
@@ -507,5 +457,4 @@ def strip_slash_enum(tools: list[dict]) -> tuple[list[dict], int]:
     return _reactive_strip(
         tools, _strip,
         "schema_sanitizer: stripped %d enum keyword(s) containing '/' "
-        "from tool schemas (xAI Responses grammar-compile recovery)",
-    )
+        "from tool schemas (xAI Responses grammar-compile recovery)")

tools/self_repo_guard.py

@@ -14,16 +14,13 @@ from tools.approval import (
     _bash_exec_payload,
     _deobfuscate_shell_word_for_detection,
     _iter_shell_command_starts,
-    _read_shell_word,
-)
-
+    _read_shell_word)
 
 # bisect is included: it drives repeated checkouts of the running root — the
 # exact module-version-skew hazard this guard exists for.
 _WORKTREE_MUTATIONS = frozenset({
     "checkout", "switch", "rebase", "merge", "pull", "restore", "clean",
-    "cherry-pick", "revert", "bisect",
-})
+    "cherry-pick", "revert", "bisect"})
 _WORKTREE_TARGET_ACTIONS = frozenset({"move", "remove"})
 _STASH_SAFE_ACTIONS = frozenset({"list", "show", "create", "store", "drop", "clear"})
 _RESET_WORKTREE_MODES = frozenset({"--hard", "--merge", "--keep"})
@@ -38,29 +35,23 @@ _KNOWN_GIT_BUILTINS = frozenset({
     "maintenance", "merge-base", "mv", "notes", "push", "range-diff", "reflog",
     "remote", "repack", "replace", "reset", "restore", "rev-list", "rev-parse",
     "rm", "shortlog", "show", "show-ref", "stash", "status", "submodule", "tag",
-    "worktree",
-})
+    "worktree"})
 _SHELL_EXECUTABLES = frozenset({"bash", "dash", "ksh", "sh", "zsh"})
 _ASSIGNMENT_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=(.*)", re.DOTALL)
 _RESET_HARD_RE = re.compile(r"--h(?:a(?:r(?:d)?)?)?\Z")
 _NO_OPTIONS: frozenset[str] = frozenset()
-# Wrapper executables that are skipped to reach the real command, mapped to
-# the options that consume a following argument.
+# Wrapper executables skipped to reach the real command -> options that consume an argument.
 _WRAPPER_OPTIONS_WITH_ARG: dict[str, frozenset[str]] = {
     "sudo": frozenset({
         "-C", "--chdir", "-c", "--close-from", "-g", "--group", "-h", "--host",
-        "-p", "--prompt", "-R", "--chroot", "-T", "--command-timeout", "-u", "--user",
-    }),
-    "env": frozenset({
-        "-a", "--argv0", "-C", "--chdir", "-S", "--split-string", "-u", "--unset",
-    }),
+        "-p", "--prompt", "-R", "--chroot", "-T", "--command-timeout", "-u", "--user"}),
+    "env": frozenset({"-a", "--argv0", "-C", "--chdir", "-S", "--split-string", "-u", "--unset"}),
     "command": _NO_OPTIONS,
     "builtin": _NO_OPTIONS,
     "exec": frozenset({"-a"}),
     "nohup": _NO_OPTIONS,
     "setsid": _NO_OPTIONS,
-    "time": frozenset({"-f", "--format", "-o", "--output"}),
-}
+    "time": frozenset({"-f", "--format", "-o", "--output"})}
 _MAX_RECURSION = 4
 # git global options that consume the next argument (-C/--work-tree/-c are acted on).
 _GIT_GLOBAL_OPTIONS_WITH_ARG = frozenset({
@@ -118,9 +109,7 @@ def _shell_words_at(command: str, start: int) -> list[str]:
     cursor = start
     for _ in range(64):
         word_start, word_end, raw_word = _read_shell_word(command, cursor)
-        if word_start == word_end:
-            break
-        if words and "\n" in command[cursor:word_start]:
+        if word_start == word_end or (words and "\n" in command[cursor:word_start]):
             break
         words.append(_deobfuscate_shell_word_for_detection(raw_word))
         cursor = word_end
@@ -128,9 +117,7 @@ def _shell_words_at(command: str, start: int) -> list[str]:
 
 
 def _consume_options(
-    words: list[str],
-    start: int,
-    options_with_arg: frozenset[str] = _NO_OPTIONS,
+    words: list[str], start: int, options_with_arg: frozenset[str] = _NO_OPTIONS,
 ) -> int:
     """Index of the first positional at/after ``start`` (``--`` ends options)."""
     index = start
@@ -140,11 +127,7 @@ def _consume_options(
             return index + 1
         if not option.startswith("-") or option == "-":
             break
-        option_name = option.split("=", 1)[0]
-        if "=" not in option and option_name in options_with_arg:
-            index += 2
-        else:
-            index += 1
+        index += 2 if "=" not in option and option in options_with_arg else 1
     return index
 
 
@@ -152,14 +135,12 @@ def _command_parts(words: list[str]) -> tuple[dict[str, str], str | None, list[s
     """Split leading VAR=value assignments and wrappers off -> (env, executable, args)."""
     env: dict[str, str] = {}
     index = 0
-
     while index < len(words):
         if _ASSIGNMENT_RE.fullmatch(words[index]):
             name, value = words[index].split("=", 1)
             env[name] = value
             index += 1
             continue
-
         executable = _executable_name(words[index])
         wrapper_options = _WRAPPER_OPTIONS_WITH_ARG.get(executable)
         if wrapper_options is None:
@@ -168,7 +149,6 @@ def _command_parts(words: list[str]) -> tuple[dict[str, str], str | None, list[s
         if executable == "command" and words[index + 1 : index + 2] in (["-v"], ["-V"]):
             return env, None, []
         index = _consume_options(words, index + 1, wrapper_options)
-
     return env, None, []
 
 
@@ -233,12 +213,11 @@ def _operator_before(command: str, start: int) -> str | None:
     while index >= 0 and command[index].isspace():
         saw_newline = saw_newline or command[index] == "\n"
         index -= 1
-    if index < 0:
-        return "\n" if saw_newline else None
-    if index > 0 and command[index - 1 : index + 1] in {"&&", "||"}:
-        return command[index - 1 : index + 1]
-    if command[index] in {";", "|", "&", "(", "{"}:
-        return command[index]
+    if index >= 0:
+        if index > 0 and command[index - 1 : index + 1] in {"&&", "||"}:
+            return command[index - 1 : index + 1]
+        if command[index] in {";", "|", "&", "(", "{"}:
+            return command[index]
     return "\n" if saw_newline else None
 
 
@@ -256,21 +235,19 @@ def _shell_script_arg(args: list[str]) -> str | None:
     """Return the script string owned by a shell's ``-c``, if present.
 
     approval.py's ``_bash_exec_payload`` parses bash's real option grammar
-    (``-o pipefail -c '<script>'`` hides ``-c`` behind an operand). When it
-    finds no ``-c``, fall back to a permissive positional scan because
-    zsh/dash/ksh option letters (``zsh -yc``) fall outside bash's alphabet and
-    would otherwise make this block-guard fail open.
+    (``-o pipefail -c '<script>'`` hides ``-c`` behind an operand). When it finds
+    no ``-c``, fall back to a permissive positional scan: zsh/dash/ksh option
+    letters (``zsh -yc``) fall outside bash's alphabet and would otherwise make
+    this block-guard fail open.
     """
     has_c, payload = _bash_exec_payload(args)
     if has_c:
         return payload
     for index, arg in enumerate(args):
-        if arg == "--":
+        if arg == "--" or not arg.startswith("-"):
             break
-        if arg.startswith("-") and "c" in arg[1:]:
+        if "c" in arg[1:]:
             return args[index + 1] if index + 1 < len(args) else None
-        if not arg.startswith("-"):
-            break
     return None
 
 
@@ -318,9 +295,7 @@ def _heredoc_specs(line: str) -> list[_Heredoc]:
             index = end + 1
         else:
             end = index
-            while (
-                end < len(line) and not line[end].isspace() and line[end] not in ";|&<>"
-            ):
+            while end < len(line) and not line[end].isspace() and line[end] not in ";|&<>":
                 end += 1
             delimiter = line[index:end]
             index = end
@@ -335,17 +310,12 @@ def _heredoc_specs(line: str) -> list[_Heredoc]:
             executable
             and _executable_name(executable) in _SHELL_EXECUTABLES
             and _shell_script_arg(args) is None
-            and not any(arg and not arg.startswith("-") for arg in args)
-        )
+            and not any(arg and not arg.startswith("-") for arg in args))
         specs.append(_Heredoc(delimiter, strip_tabs, execute_as_shell))
 
     return specs
 
 
-def _masked_line(line: str) -> str:
-    return "".join(char if char in {"\r", "\n"} else " " for char in line)
-
-
 def _mask_heredocs(command: str) -> tuple[str, list[str]]:
     """Blank heredoc bodies; return (masked command, bodies a bare shell would execute).
 
@@ -365,9 +335,8 @@ def _mask_heredocs(command: str) -> tuple[str, list[str]]:
                 finished.append(pending.pop(0))
             else:
                 current.body.append(line)
-            output.append(_masked_line(line))
+            output.append("".join(char if char in {"\r", "\n"} else " " for char in line))
             continue
-
         output.append(line)
         pending.extend(_heredoc_specs(line))
 
@@ -383,9 +352,7 @@ def _record_alias(config: str, aliases: dict[str, str]) -> None:
 
 
 def _git_target_and_subcommand(
-    args: list[str],
-    current_dir: Path,
-    env: dict[str, str],
+    args: list[str], current_dir: Path, env: dict[str, str],
 ) -> tuple[Path, str | None, list[str], dict[str, str]]:
     """Parse git's global options -> (target dir, subcommand, sub args, inline aliases)."""
     target = current_dir
@@ -457,15 +424,12 @@ _CONDITIONAL_MUTATIONS: dict[str, Callable[[list[str]], bool]] = {
     "reset": _reset_mutates,
     "stash": _stash_mutates,
     "clean": _clean_mutates,
-    "restore": _restore_mutates,
-}
+    "restore": _restore_mutates}
 
 
 def _mutates_worktree(subcommand: str, args: list[str]) -> bool:
     check = _CONDITIONAL_MUTATIONS.get(subcommand)
-    if check is not None:
-        return check(args)
-    return subcommand in _WORKTREE_MUTATIONS
+    return check(args) if check is not None else subcommand in _WORKTREE_MUTATIONS
 
 
 def _inspect_git_worktree(args: list[str], cwd: Path, root: Path) -> str | None:
@@ -477,9 +441,7 @@ def _inspect_git_worktree(args: list[str], cwd: Path, root: Path) -> str | None:
     if action not in _WORKTREE_TARGET_ACTIONS:
         return None
     target_index = _consume_options(args, action_index + 1)
-    if target_index >= len(args):
-        return None
-    if _resolve(args[target_index], cwd) == root:
+    if target_index < len(args) and _resolve(args[target_index], cwd) == root:
         return f"git worktree {action}"
     return None
 
@@ -488,11 +450,7 @@ def _read_git_alias(executable: str, target: Path, alias: str) -> str | None:
     try:
         result = subprocess.run(
             [executable, "-C", str(target), "config", "--get", f"alias.{alias}"],
-            capture_output=True,
-            text=True,
-            timeout=1,
-            check=False,
-        )
+            capture_output=True, text=True, timeout=1, check=False)
     except (OSError, subprocess.SubprocessError):
         return None
     value = result.stdout.strip()
@@ -500,15 +458,11 @@ def _read_git_alias(executable: str, target: Path, alias: str) -> str | None:
 
 
 def _inspect_git(
-    executable: str,
-    args: list[str],
-    current_dir: Path,
-    env: dict[str, str],
-    root: Path,
+    executable: str, args: list[str], current_dir: Path, env: dict[str, str], root: Path,
     depth: int,
 ) -> str | None:
     target, subcommand, sub_args, inline_aliases = _git_target_and_subcommand(
-        args, current_dir, env
+        args, current_dir, env,
     )
     if subcommand is None:
         return None
@@ -533,45 +487,27 @@ def _inspect_git(
         alias_args = shlex.split(alias, posix=True)
     except ValueError:
         return None
-    return _inspect_git(
-        executable,
-        [*alias_args, *sub_args],
-        target,
-        {},
-        root,
-        depth + 1,
-    )
+    return _inspect_git(executable, [*alias_args, *sub_args], target, {}, root, depth + 1)
 
 
 def _inspect_github_cli(
-    executable: str,
-    args: list[str],
-    current_dir: Path,
-    env: dict[str, str],
-    root: Path,
+    executable: str, args: list[str], current_dir: Path, env: dict[str, str], root: Path,
     depth: int,
 ) -> str | None:
     if not _is_within(current_dir, root):
         return None
-    name = _executable_name(executable)
     index = _consume_options(args, 0, frozenset({"-R", "--repo", "--hostname"}))
     if args[index : index + 2] == ["pr", "checkout"]:
-        return f"{name} pr checkout"
+        return f"{_executable_name(executable)} pr checkout"
     return None
 
 
 def _inspect_shell(
-    executable: str,
-    args: list[str],
-    current_dir: Path,
-    env: dict[str, str],
-    root: Path,
+    executable: str, args: list[str], current_dir: Path, env: dict[str, str], root: Path,
     depth: int,
 ) -> str | None:
     script = _shell_script_arg(args)
-    if script:
-        return _find_mutation(script, current_dir, root, depth + 1)
-    return None
+    return _find_mutation(script, current_dir, root, depth + 1) if script else None
 
 
 # executable name -> inspector(executable, args, current_dir, env, root, depth)
@@ -579,8 +515,7 @@ _INSPECTORS: dict[str, Callable[..., str | None]] = {
     "git": _inspect_git,
     "gh": _inspect_github_cli,
     "hub": _inspect_github_cli,
-    **{shell: _inspect_shell for shell in _SHELL_EXECUTABLES},
-}
+    **{shell: _inspect_shell for shell in _SHELL_EXECUTABLES}}
 
 
 def _find_mutation(command: str, cwd: Path, root: Path, depth: int = 0) -> str | None:
@@ -611,8 +546,7 @@ def _find_mutation(command: str, cwd: Path, root: Path, depth: int = 0) -> str |
         if pending is not None and operator in {"&&", ";", "\n"}:
             cwd_by_scope[scope] = pending
 
-        words = _shell_words_at(masked_command, start)
-        env, executable, args = _command_parts(words)
+        env, executable, args = _command_parts(_shell_words_at(masked_command, start))
         if executable is None:
             continue
 
@@ -634,20 +568,16 @@ def _find_mutation(command: str, cwd: Path, root: Path, depth: int = 0) -> str |
 def guard_active() -> bool:
     """Whether the self-repo git guard applies on this platform.
 
-    Windows-only: NTFS locks loaded .py/.pyd files, so an in-place overwrite of
-    the live checkout can corrupt the running process. On POSIX, open handles
-    keep the old inode alive, so already-imported modules keep running; the
-    mixed-module hazard is limited to later lazy imports — not worth blocking
+    Windows-only: NTFS locks loaded .py/.pyd files, so overwriting the live checkout
+    can corrupt the running process. On POSIX, open handles keep the old inode alive;
+    the mixed-module hazard is limited to later lazy imports — not worth blocking
     every git workflow for.
     """
     return os.name == "nt"
 
 
 def detect_self_repo_git_mutation(
-    command: str,
-    cwd: str | None,
-    source_root: Path | None = None,
-) -> tuple[bool, str | None]:
+    command: str, cwd: str | None, source_root: Path | None = None) -> tuple[bool, str | None]:
     """Return whether a command would rewrite the live source checkout."""
     root = source_root if source_root is not None else get_running_source_root()
     if root is None or not command:
@@ -664,7 +594,8 @@ def detect_self_repo_git_mutation(
 def _block_message(operation: str, root: Path) -> str:
     # Suggest a disk-backed scratch dir: /tmp is usually tmpfs (see message).
     hermes_home = os.environ.get("HERMES_HOME", "").strip()
-    scratch = (Path(hermes_home).expanduser() if hermes_home else Path.home() / ".hermes") / "scratch"
+    home = Path(hermes_home).expanduser() if hermes_home else Path.home() / ".hermes"
+    scratch = home / "scratch"
     return (
         f"Blocked: `{operation}` would rewrite Hermes's live source checkout "
         f"({root}) and can mix module versions in this running process. "
@@ -674,5 +605,4 @@ def _block_message(operation: str, root: Path) -> str:
         "tmpfs and a few dependency installs can fill it and ENOSPC other "
         "work. Delete the clone when the branch is pushed. To change this "
         "checkout, stop Hermes, run the command externally, then restart "
-        "Hermes."
-    )
+        "Hermes.")