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

tools/neutts_synth.py

@@ -16,7 +16,6 @@ from pathlib import Path
 def _write_wav(path: str, samples, sample_rate: int = 24000) -> None:
     """Write a WAV file from float32 samples (no soundfile dependency)."""
     import numpy as np
-
     if not isinstance(samples, np.ndarray):
         samples = np.array(samples, dtype=np.float32)
     pcm = (np.clip(samples.flatten(), -1.0, 1.0) * 32767).astype(np.int16)
@@ -33,18 +32,17 @@ def main():
     parser.add_argument("--out", required=True, help="Output WAV path")
     parser.add_argument("--ref-audio", required=True, help="Reference voice audio path")
     parser.add_argument("--ref-text", required=True, help="Reference voice transcript path")
-    parser.add_argument("--model", default="neuphonic/neutts-air-q4-gguf", help="HuggingFace backbone model repo")
+    parser.add_argument("--model", default="neuphonic/neutts-air-q4-gguf",
+                        help="HuggingFace backbone model repo")
     parser.add_argument("--device", default="cpu", help="Device (cpu/cuda/mps)")
     args = parser.parse_args()
 
     ref_audio = Path(args.ref_audio).expanduser()
     ref_text_path = Path(args.ref_text).expanduser()
-    if not ref_audio.exists():
-        print(f"Error: reference audio not found: {ref_audio}", file=sys.stderr)
-        sys.exit(1)
-    if not ref_text_path.exists():
-        print(f"Error: reference text not found: {ref_text_path}", file=sys.stderr)
-        sys.exit(1)
+    for label, p in (("audio", ref_audio), ("text", ref_text_path)):
+        if not p.exists():
+            print(f"Error: reference {label} not found: {p}", file=sys.stderr)
+            sys.exit(1)
     ref_text = ref_text_path.read_text(encoding="utf-8").strip()
 
     try:
@@ -60,8 +58,7 @@ def main():
         backbone_repo=args.model,
         backbone_device="gpu" if args.device == "cuda" else args.device,
         codec_repo="neuphonic/neucodec",
-        codec_device=args.device,
-    )
+        codec_device=args.device)
     wav = tts.infer(args.text, tts.encode_reference(str(ref_audio)), ref_text)
 
     out_path = Path(args.out)

tools/open_preview_tool.py

@@ -34,13 +34,11 @@ def open_preview_tool(url: str, label: str = "") -> str:
     if not target:
         return tool_error(
             "url is required — a web URL (https://…), a localhost dev server, or a "
-            "file path to show in the preview pane."
-        )
+            "file path to show in the preview pane.")
     label = (label or "").strip()
     return desktop_ui.emit_or_error(
         "preview.open",
         {"url": target, "label": label},
         "Failed to open the preview pane: ",
         "The preview pane is only available in the Hermes desktop app.",
-        {"success": True, "url": target, "label": label},
-    )
+        {"success": True, "url": target, "label": label})

tools/openrouter_client.py

@@ -11,7 +11,6 @@ def check_api_key() -> bool:
     """
     try:
         from agent.secret_scope import UnscopedSecretError, get_secret
-
         try:
             return bool(get_secret("OPENROUTER_API_KEY"))
         except UnscopedSecretError:

tools/osv_check.py

@@ -20,12 +20,11 @@ logger = logging.getLogger(__name__)
 _OSV_ENDPOINT = os.getenv("OSV_ENDPOINT", "https://api.osv.dev/v1/query")
 _TIMEOUT = 10  # seconds
 
-# Result cache: (ecosystem, package, version) -> (expiry_monotonic, result).
-# MCP reconnect ladders and parked-server self-probes re-run the preflight for
-# the SAME package on every spawn; uncached, a flapping server becomes a
-# sustained OSV/DNS query stream. Advisories don't flip on second timescales,
-# so clean AND blocked verdicts are reusable. Network failures are NOT cached:
-# fail-open covers them and caching one could mask a real advisory later.
+# Result cache: (ecosystem, package, version) -> (expiry_monotonic, result). Reconnect
+# ladders and parked-server self-probes re-run the preflight for the SAME package on every
+# spawn; uncached, a flapping server becomes a sustained OSV/DNS query stream. Clean AND
+# blocked verdicts are reusable; network failures are NOT cached (fail-open covers them and
+# caching one could mask a real advisory later).
 _CACHE_TTL_S = float(os.getenv("OSV_CHECK_CACHE_TTL", "3600"))
 _CACHE_MAX_ENTRIES = 256
 _cache: dict = {}
@@ -36,13 +35,10 @@ def _cache_get(key) -> Tuple[bool, Optional[str]]:
     """Return (hit, result) for a fresh cache entry."""
     with _cache_lock:
         entry = _cache.get(key)
-        if entry is None:
-            return False, None
-        expiry, result = entry
-        if time.monotonic() >= expiry:
-            del _cache[key]
-            return False, None
-        return True, result
+        if entry is not None and time.monotonic() < entry[0]:
+            return True, entry[1]
+        _cache.pop(key, None)  # absent or expired
+        return False, None
 
 
 def _cache_put(key, result: Optional[str]) -> None:
@@ -58,43 +54,35 @@ def _cache_put(key, result: Optional[str]) -> None:
 
 def check_package_for_malware(command: str, args: list) -> Optional[str]:
     """Check an MCP server package (inferred from ``command``/``args``) for MAL-* advisories.
-
-    Returns a BLOCKED message when malware is found, else None — including on network
-    errors and unrecognized commands (fail-open).
-    """
+    Returns a BLOCKED message, else None — also on network errors/unknown commands (fail-open)."""
     ecosystem = _infer_ecosystem(command)
     if not ecosystem:
         return None  # not npx/uvx — skip
-
     package, version = _parse_package_from_args(args, ecosystem)
     if not package:
         return None
-
     cache_key = (ecosystem, package, version)
     hit, cached = _cache_get(cache_key)
     if hit:
         return cached
-
     try:
         malware = _query_osv(package, ecosystem, version)
     except Exception as exc:
         # Fail-open; deliberately NOT cached — see _CACHE_TTL_S comment.
         logger.debug("OSV check failed for %s/%s (allowing): %s", ecosystem, package, exc)
         return None
-
     result = None
     if malware:
         ids = ", ".join(m["id"] for m in malware[:3])
         summaries = "; ".join(m.get("summary", m["id"])[:100] for m in malware[:3])
-        result = f"BLOCKED: Package '{package}' ({ecosystem}) has known malware advisories: {ids}. Details: {summaries}"
+        result = (f"BLOCKED: Package '{package}' ({ecosystem}) has known malware "
+                  f"advisories: {ids}. Details: {summaries}")
     _cache_put(cache_key, result)
     return result
 
 
 _ECOSYSTEM_BY_COMMAND = {
-    "npx": "npm", "npx.cmd": "npm",
-    "uvx": "PyPI", "uvx.cmd": "PyPI", "pipx": "PyPI",
-}
+    "npx": "npm", "npx.cmd": "npm", "uvx": "PyPI", "uvx.cmd": "PyPI", "pipx": "PyPI"}
 
 
 def _infer_ecosystem(command: str) -> Optional[str]:
@@ -103,9 +91,8 @@ def _infer_ecosystem(command: str) -> Optional[str]:
 
 def _parse_package_from_args(args: list, ecosystem: str) -> Tuple[Optional[str], Optional[str]]:
     """Extract (package_name, version) from command args, or (None, None) if not parseable."""
-    # Skip flags to find the package token. Honor npx's explicit install target
-    # (--package=NAME / --package NAME / -p NAME), which names a package distinct
-    # from the executed binary; otherwise the first bare positional is used.
+    # Skip flags to find the package token. npx's explicit install target (--package=NAME /
+    # --package NAME / -p NAME) names a package distinct from the executed binary.
     package_token = None
     take_next = False
     for arg in args or ():
@@ -124,7 +111,6 @@ def _parse_package_from_args(args: list, ecosystem: str) -> Tuple[Optional[str],
             continue
         package_token = arg
         break
-
     if not package_token:
         return None, None
     parser = _PACKAGE_PARSERS.get(ecosystem)
@@ -160,8 +146,7 @@ def _query_osv(package: str, ecosystem: str, version: Optional[str] = None) -> l
         _OSV_ENDPOINT,
         data=json.dumps(payload).encode("utf-8"),
         headers={"Content-Type": "application/json", "User-Agent": "hermes-agent-osv-check/1.0"},
-        method="POST",
-    )
+        method="POST")
     with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
         result = json.loads(resp.read())
     return [v for v in result.get("vulns", []) if v.get("id", "").startswith("MAL-")]

tools/patch_parser.py

@@ -1,17 +1,9 @@
 #!/usr/bin/env python3
 """V4A patch format parser and applier (format used by codex, cline, etc.).
 
-    *** Begin Patch
-    *** Update File: path/to/file.py
-    @@ optional context hint @@
-     context line (space prefix)
-    -removed line
-    +added line
-    *** Add File: path/to/new.py
-    +new file content
-    *** Delete File: path/to/old.py
-    *** Move File: old/path.py -> new/path.py
-    *** End Patch
+    *** Begin Patch / *** End Patch wrap the operations:
+    *** Update File: p.py   then hunks: ``@@ hint @@``, `` ctx``, ``-old``, ``+new``
+    *** Add File: n.py      then ``+`` lines;  *** Delete File: o.py;  *** Move File: a -> b
 
     operations, error = parse_v4a_patch(patch_content)
     result = apply_v4a_operations(operations, file_ops)
@@ -61,17 +53,13 @@ _OP_MARKERS: List[Tuple[OperationType, re.Pattern]] = [
     (OperationType.UPDATE, re.compile(r'\*\*\*\s*Update\s+File:\s*(.+)')),
     (OperationType.ADD, re.compile(r'\*\*\*\s*Add\s+File:\s*(.+)')),
     (OperationType.DELETE, re.compile(r'\*\*\*\s*Delete\s+File:\s*(.+)')),
-    (OperationType.MOVE, re.compile(r'\*\*\*\s*Move\s+File:\s*(.+?)\s*->\s*(.+)')),
-]
+    (OperationType.MOVE, re.compile(r'\*\*\*\s*Move\s+File:\s*(.+?)\s*->\s*(.+)'))]
 _HINT_RE = re.compile(r'@@\s*(.+?)\s*@@')
 
 
 def parse_v4a_patch(patch_content: str) -> Tuple[List[PatchOperation], Optional[str]]:
-    """Parse a V4A patch into operations.
-
-    Returns ``(operations, None)`` — ``[]`` for an empty patch is not an
-    error — or ``([], "Parse error: ...")`` for malformed operations.
-    """
+    """Parse a V4A patch -> ``(operations, None)`` (``[]`` for an empty patch is not an
+    error) or ``([], "Parse error: ...")`` for malformed operations."""
     # Tolerate CRLF bodies: a stray ``\r`` would otherwise end up in every
     # HunkLine.content and defeat the anchored Begin/End markers.
     lines = [ln[:-1] if ln.endswith('\r') else ln for ln in patch_content.split('\n')]
@@ -99,15 +87,15 @@ def parse_v4a_patch(patch_content: str) -> Tuple[List[PatchOperation], Optional[
             operations.append(current_op)
 
     for line in lines[start_idx + 1:end_idx]:
-        op_match = next(((kind, m) for kind, rx in _OP_MARKERS if (m := rx.match(line))), None)
+        op_match = next(
+            ((kind, m) for kind, rx in _OP_MARKERS if (m := rx.match(line))), None)
         if op_match:
             kind, m = op_match
             _flush()
             current_op = PatchOperation(
                 operation=kind,
                 file_path=m.group(1).strip(),
-                new_path=m.group(2).strip() if kind is OperationType.MOVE else None,
-            )
+                new_path=m.group(2).strip() if kind is OperationType.MOVE else None)
             # UPDATE hunks start lazily (at '@@' or the first hunk line); ADD
             # collects all '+' lines into one hunk; DELETE/MOVE are complete.
             current_hunk = Hunk() if kind is OperationType.ADD else None
@@ -166,25 +154,23 @@ def _hint_ambiguity(content: str, hint: str, tail: str = "") -> Tuple[int, str]:
     """(occurrences, error) for an addition-only hunk's context hint; error is '' when unique."""
     occurrences = _count_occurrences(content, hint)
     if occurrences > 1:
-        return occurrences, f"context hint '{hint}' is ambiguous ({occurrences} occurrences){tail}"
+        return occurrences, (f"context hint '{hint}' is ambiguous "
+                             f"({occurrences} occurrences){tail}")
     return occurrences, ""
 
 
 def _validate_operations(operations: List[PatchOperation], file_ops: Any) -> List[str]:
-    """Dry-run every operation; return error strings (empty list = safe to apply).
-
-    UPDATE hunks are simulated in order so later hunks validate against
-    post-earlier-hunk content, exactly as the apply phase will see it.
-    """
+    """Dry-run every operation; return error strings (empty list = safe to apply). UPDATE
+    hunks are simulated in order so later hunks see post-earlier-hunk content, as apply will."""
     from tools.fuzzy_match import fuzzy_find_and_replace, is_already_applied
 
     errors: List[str] = []
     real_change_count = 0
 
-    # Virtual overlay so inter-op state validates (e.g. a MOVE creating the
-    # destination a later UPDATE targets). UPDATE/MOVE reads consult it first.
-    pending_content: dict = {}   # path -> content produced by an earlier op
-    removed_paths: set = set()   # paths a MOVE/DELETE has taken away
+    # Virtual overlay so inter-op state validates (e.g. a MOVE creating the destination
+    # a later UPDATE targets): path -> content from an earlier op; paths MOVE/DELETE removed.
+    pending_content: dict = {}
+    removed_paths: set = set()
 
     def _read(path: str):
         if path in removed_paths and path not in pending_content:
@@ -213,7 +199,8 @@ def _validate_operations(operations: List[PatchOperation], file_ops: Any) -> Lis
                 if hunk.context_hint:
                     occurrences, ambiguous = _hint_ambiguity(simulated, hunk.context_hint)
                     if occurrences == 0:
-                        errors.append(f"{op.file_path}: addition-only hunk context hint '{hunk.context_hint}' not found")
+                        errors.append(f"{op.file_path}: addition-only hunk context hint "
+                                      f"'{hunk.context_hint}' not found")
                     elif ambiguous:
                         errors.append(f"{op.file_path}: addition-only hunk {ambiguous}")
                 continue
@@ -224,8 +211,7 @@ def _validate_operations(operations: List[PatchOperation], file_ops: Any) -> Lis
                 # validation must not reject it with the identical-strings error.
                 continue
             new_simulated, count, _strategy, match_error = fuzzy_find_and_replace(
-                simulated, search_pattern, replacement, replace_all=False
-            )
+                simulated, search_pattern, replacement, replace_all=False)
             if count:
                 simulated = new_simulated
             elif not is_already_applied(simulated or "", search_pattern, replacement):
@@ -235,8 +221,7 @@ def _validate_operations(operations: List[PatchOperation], file_ops: Any) -> Lis
                 errors.append(
                     f"{op.file_path}: hunk {hunk_index} {label} not found"
                     + (f" — {match_error}" if match_error else "")
-                    + _no_match_hint(match_error, search_pattern, simulated)
-                )
+                    + _no_match_hint(match_error, search_pattern, simulated))
         pending_content[op.file_path] = simulated
 
     for op in operations:
@@ -279,11 +264,8 @@ ApplyResult = Tuple[bool, str, Optional[str], Optional[dict]]
 
 def apply_v4a_operations(operations: List[PatchOperation], file_ops: Any) -> 'PatchResult':
     """Validate all operations, then apply them (two-phase, atomic on validation failure).
-
-    A phase-2 failure (e.g. a race between validation and apply) is reported
-    with a note to run ``git diff`` since state may be inconsistent.
-    ``file_ops`` needs ``read_file_raw``, ``write_file``, ``delete_file``, ``move_file``.
-    """
+    A phase-2 failure (validate/apply race) is reported with a ``git diff`` note since state
+    may be inconsistent. ``file_ops`` needs read_file_raw/write_file/delete_file/move_file."""
     from tools.file_operations import PatchResult  # avoid circular import
 
     validation_errors = _validate_operations(operations, file_ops)
@@ -291,12 +273,9 @@ def apply_v4a_operations(operations: List[PatchOperation], file_ops: Any) -> 'Pa
         return PatchResult(
             success=False,
             error="Patch validation failed (no files were modified):\n"
-                  + "\n".join(f"  • {e}" for e in validation_errors),
-        )
+                  + "\n".join(f"  • {e}" for e in validation_errors))
 
-    files_modified: List[str] = []
-    files_created: List[str] = []
-    files_deleted: List[str] = []
+    files: Dict[str, List[str]] = {"created": [], "deleted": [], "modified": []}
     all_diffs: List[str] = []
     # V4A bypasses the WriteResult/PatchResult plumbing that write_file uses,
     # so LSP diagnostics and lint must be propagated explicitly per file.
@@ -304,21 +283,17 @@ def apply_v4a_operations(operations: List[PatchOperation], file_ops: Any) -> 'Pa
     errors: List[str] = []
     lint_results: Dict[str, dict] = {}
 
-    dispatch: Dict[OperationType, Tuple[Callable[[PatchOperation, Any], ApplyResult], List[str], str]] = {
-        OperationType.ADD: (_apply_add, files_created, "add"),
-        OperationType.DELETE: (_apply_delete, files_deleted, "delete"),
-        OperationType.MOVE: (_apply_move, files_modified, "move"),
-        OperationType.UPDATE: (_apply_update, files_modified, "update"),
-    }
-
     for op in operations:
         try:
-            handler, bucket, verb = dispatch[op.operation]
+            handler, verb, bucket = _APPLY_DISPATCH[op.operation]
             ok, payload, lsp, lint = handler(op, file_ops)
             if not ok:
                 errors.append(f"Failed to {verb} {op.file_path}: {payload}")
                 continue
-            bucket.append(f"{op.file_path} -> {op.new_path}" if op.operation is OperationType.MOVE else op.file_path)
+            label = op.file_path
+            if op.operation is OperationType.MOVE:
+                label = f"{op.file_path} -> {op.new_path}"
+            files[bucket].append(label)
             all_diffs.append(payload)
             if lsp:
                 lsp_blocks.append(lsp)
@@ -331,34 +306,28 @@ def apply_v4a_operations(operations: List[PatchOperation], file_ops: Any) -> 'Pa
     # concatenation keeps per-file attribution.
     result_kwargs = dict(
         diff='\n'.join(all_diffs),
-        files_modified=files_modified,
-        files_created=files_created,
-        files_deleted=files_deleted,
+        files_modified=files["modified"], files_created=files["created"], files_deleted=files["deleted"],
         lint=lint_results if lint_results else None,
-        lsp_diagnostics="\n\n".join(lsp_blocks) if lsp_blocks else None,
-    )
+        lsp_diagnostics="\n\n".join(lsp_blocks) if lsp_blocks else None)
     if errors:
         return PatchResult(
             success=False,
             error="Apply phase failed (state may be inconsistent — run `git diff` to assess):\n"
                   + "\n".join(f"  • {e}" for e in errors),
-            **result_kwargs,
-        )
+            **result_kwargs)
     return PatchResult(success=True, **result_kwargs)
 
 
 def _write_file_accepts_pre_content(file_ops: Any) -> bool:
-    """True when ``file_ops.write_file`` accepts a ``pre_content`` kwarg.
-
-    Decided from the signature rather than catching TypeError around the call, so a
-    TypeError raised *inside* a capable write_file propagates instead of triggering a
-    second, duplicate write. Unintrospectable callables get the two-argument form.
-    """
+    """True when ``file_ops.write_file`` accepts ``pre_content``. Decided from the signature,
+    not by catching TypeError around the call, so a TypeError raised *inside* a capable
+    write_file propagates instead of triggering a duplicate write."""
     try:
         params = inspect.signature(file_ops.write_file).parameters
     except (TypeError, ValueError):
         return False
-    return "pre_content" in params or any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values())
+    return "pre_content" in params or any(
+        p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values())
 
 
 def _apply_add(op: PatchOperation, file_ops: Any) -> ApplyResult:
@@ -382,8 +351,7 @@ def _apply_delete(op: PatchOperation, file_ops: Any) -> ApplyResult:
         return False, result.error, None, None
     diff = ''.join(difflib.unified_diff(
         read_result.content.splitlines(keepends=True), [],
-        fromfile=f"a/{op.file_path}", tofile="/dev/null",
-    ))
+        fromfile=f"a/{op.file_path}", tofile="/dev/null"))
     return True, diff or f"# Deleted: {op.file_path}", None, None
 
 
@@ -397,7 +365,8 @@ def _apply_move(op: PatchOperation, file_ops: Any) -> ApplyResult:
 def _insert_addition_only(new_content: str, hunk: Hunk, insert_text: str) -> Tuple[Optional[str], Optional[str]]:
     """Place an addition-only hunk after its context hint (or at EOF). Returns (content, error)."""
     if hunk.context_hint:
-        occurrences, ambiguous = _hint_ambiguity(new_content, hunk.context_hint, " — provide a more unique hint")
+        occurrences, ambiguous = _hint_ambiguity(
+            new_content, hunk.context_hint, " — provide a more unique hint")
         if ambiguous:
             return None, f"Addition-only hunk: {ambiguous}"
         if occurrences == 1:
@@ -433,8 +402,7 @@ def _apply_update(op: PatchOperation, file_ops: Any) -> ApplyResult:
         search_pattern = '\n'.join(search_lines)
         replacement = '\n'.join(replace_lines)
         new_content, count, _strategy, error = fuzzy_find_and_replace(
-            new_content, search_pattern, replacement, replace_all=False
-        )
+            new_content, search_pattern, replacement, replace_all=False)
         if not (error and count == 0):
             continue
 
@@ -466,6 +434,14 @@ def _apply_update(op: PatchOperation, file_ops: Any) -> ApplyResult:
 
     diff = ''.join(difflib.unified_diff(
         current_content.splitlines(keepends=True), new_content.splitlines(keepends=True),
-        fromfile=f"a/{op.file_path}", tofile=f"b/{op.file_path}",
-    ))
+        fromfile=f"a/{op.file_path}", tofile=f"b/{op.file_path}"))
     return True, diff, getattr(write_result, "lsp_diagnostics", None), getattr(write_result, "lint", None)
+
+
+# operation -> (handler, verb for error text, files_* bucket)
+_APPLY_DISPATCH: Dict[OperationType, Tuple[Callable[[PatchOperation, Any], ApplyResult], str, str]] = {
+    OperationType.ADD: (_apply_add, "add", "created"),
+    OperationType.DELETE: (_apply_delete, "delete", "deleted"),
+    OperationType.MOVE: (_apply_move, "move", "modified"),
+    OperationType.UPDATE: (_apply_update, "update", "modified"),
+}

tools/preview_tool.py

@@ -27,7 +27,8 @@ _ACTIONS = {
     "open": lambda args: open_preview_tool(url=args.get("url", ""), label=args.get("label", "")),
     "close": lambda args: preview_close(url=args.get("url", "")),
     # read needs the GUI callback and is dispatched at the agent level.
-    "read": lambda args: tool_error("preview read must run inside a desktop session (no GUI callback here)."),
+    "read": lambda args: tool_error(
+        "preview read must run inside a desktop session (no GUI callback here)."),
 }
 
 

tools/read_extract.py

@@ -25,22 +25,16 @@ from pathlib import Path
 from typing import Any, Callable, Iterator, Optional
 from xml.etree import ElementTree as ET
 
-__all__ = [
-    "EXTRACTABLE_EXTENSIONS",
-    "ExtractionError",
-    "extract_document_bytes",
-    "extract_document_text",
-    "is_extractable_document",
-]
+__all__ = ["EXTRACTABLE_EXTENSIONS", "ExtractionError", "extract_document_bytes",
+           "extract_document_text", "is_extractable_document"]
 
 EXTRACTABLE_EXTENSIONS = frozenset({".ipynb", ".docx", ".xlsx"})
 # Formats handled only when the optional anydoc converter is installed.
 ANYDOC_EXTENSIONS = frozenset({
     ".doc", ".docm", ".ppt", ".pps", ".pot", ".pptx", ".pptm", ".ppsx", ".ppsm",
-    ".xls", ".xlsm", ".xlsb", ".odt", ".ods", ".odp", ".rtf", ".epub", ".pdf",
-})
-# anydoc loads the whole file through its Rust core with no streaming, and the
-# read_file char budget only applies after conversion — cap the input size.
+    ".xls", ".xlsm", ".xlsb", ".odt", ".ods", ".odp", ".rtf", ".epub", ".pdf"})
+# anydoc loads whole files with no streaming and the read_file char budget only applies
+# after conversion — cap the input size.
 MAX_ANYDOC_BYTES = 50 * 1024 * 1024
 MAX_DOCUMENT_BYTES = 50 * 1024 * 1024
 _MAX_XLSX_ROWS_PER_SHEET = 5000
@@ -58,9 +52,8 @@ class ExtractionError(Exception):
 
 def _extension(path: str) -> str:
     ext = Path(path).suffix.lower()
-    if ext in EXTRACTABLE_EXTENSIONS or (ext in ANYDOC_EXTENSIONS and _anydoc() is not None):
-        return ext
-    return ""
+    known = ext in EXTRACTABLE_EXTENSIONS or (ext in ANYDOC_EXTENSIONS and _anydoc() is not None)
+    return ext if known else ""
 
 
 _ANYDOC_UNSET = object()
@@ -73,22 +66,19 @@ _anydoc_failed_at: Optional[float] = None
 
 
 def _anydoc() -> Optional[Any]:
-    """Lazily import the optional anydoc converter; None when unavailable.
-
-    A failed load is retried after :data:`ANYDOC_RETRY_SECONDS` rather than disabling
-    extraction for the rest of the process (one transient pip/network blip must not stick).
-    """
+    """Lazily import the optional anydoc converter; None when unavailable. A failed load is
+    retried after ANYDOC_RETRY_SECONDS so one transient pip/network blip does not stick."""
     global _anydoc_module, _anydoc_failed_at
     if _anydoc_module is not _ANYDOC_UNSET:
         return _anydoc_module
     with _anydoc_lock:
         if _anydoc_module is not _ANYDOC_UNSET:
             return _anydoc_module
-        if _anydoc_failed_at is not None and time.monotonic() - _anydoc_failed_at < ANYDOC_RETRY_SECONDS:
+        if (_anydoc_failed_at is not None
+                and time.monotonic() - _anydoc_failed_at < ANYDOC_RETRY_SECONDS):
             return None
         try:
             from tools.lazy_deps import ensure as _lazy_ensure
-
             # prompt=False: read_file must never block on an install prompt.
             _lazy_ensure("tool.doc_extract", prompt=False)
             _anydoc_module = importlib.import_module("anydoc")
@@ -147,92 +137,73 @@ def extract_document_bytes(data: bytes, path: str) -> str:
 
 
 def _anydoc_missing_error(path: str) -> str:
-    """Teaching error for anydoc-gated formats; the schema deliberately omits this caveat
-    so only sessions that hit one pay for the explanation (and the fix)."""
+    """Teaching error for anydoc-gated formats (deliberately absent from the schema so only
+    sessions that hit one pay for it)."""
     return (
         f"Cannot convert {path!r}: this format needs the optional anydoc "
         "converter, which is not installed (install blocked or first "
         "attempt failed; retried every 5 minutes). Fix: `pip install "
         "firecrawl-anydoc` in Hermes's environment, or convert the file "
         "yourself via terminal (e.g. libreoffice --headless --convert-to "
-        "txt)."
-    )
+        "txt).")
 
 
 def _hosted_ocr_config() -> tuple:
-    """Resolve hosted-OCR settings: (enabled, api_key, api_url). Never raises.
-
-    Maintainer decision: the ONLY route is a direct ``FIRECRAWL_API_KEY`` (anydoc defaults
-    api_url to https://api.firecrawl.dev); the Nous managed gateway is NOT used — its Parse
-    proxy live-probed broken (revisit when it grows Parse support). ``file_tools.hosted_ocr:
-    false`` disables even with a key; true/unset → enabled iff the key is present. Env probe
-    only, no network at schema-build time.
-    """
+    """Resolve hosted-OCR settings: (enabled, api_key, api_url). Never raises; no network.
+    Maintainer decision: the ONLY route is a direct ``FIRECRAWL_API_KEY`` (anydoc defaults the
+    api_url); the Nous gateway is NOT used — its Parse proxy live-probed broken (revisit when
+    it grows Parse support). ``file_tools.hosted_ocr: false`` disables even with a key."""
     api_key = os.environ.get("FIRECRAWL_API_KEY") or None
     enabled = api_key is not None
     with contextlib.suppress(Exception):
         from hermes_cli.config import load_config_readonly
-
-        cfg = load_config_readonly()
-        section = cfg.get("file_tools") if isinstance(cfg, dict) else None
+        section = load_config_readonly().get("file_tools")
         if isinstance(section, dict) and section.get("hosted_ocr") is False:
             enabled = False
     return enabled, api_key, None
 
 
 def hosted_ocr_available() -> bool:
-    """Public probe for read_file's schema line (same gate as :func:`_hosted_ocr_config`);
-    a key that fails at conversion time lands in the NEEDS-OCR warning instead."""
+    """Public probe for read_file's schema line; a key failing at conversion time lands in
+    the NEEDS-OCR warning instead."""
     return _hosted_ocr_config()[0]
 
 
 def _needs_ocr_warning(path: str, pages, hosted_error: str = "") -> str:
-    """Result text when anydoc raises NeedsOcrError and hosted OCR is off/failed.
-
-    Hints at CHECKING for an OCR skill (never names one — none is guaranteed to
-    exist) and never advertises the hosted_ocr config knob.
-    """
+    """Result text when anydoc raises NeedsOcrError and hosted OCR is off/failed. Hints at
+    CHECKING for an OCR skill (never names one) and never advertises the hosted_ocr knob."""
     page_list = ", ".join(str(p) for p in pages) if pages else "unknown"
     msg = (
         f"[NEEDS OCR: pages {page_list} of this PDF are scanned images "
-        "with no text layer — their content is MISSING below. "
-    )
+        "with no text layer — their content is MISSING below. ")
     if hosted_error:
         msg += f"Hosted OCR was attempted and failed ({hosted_error}). "
     msg += (
         "If the missing pages matter: render just those pages with "
         f"`pdftoppm -jpeg -r 150 -f <first> -l <last> '{path}' /tmp/page` "
         "and inspect via vision_analyze, or check whether an OCR skill is "
-        "available (skills_list)."
-    )
+        "available (skills_list).")
     return msg + "]\n"
 
 
 def _finalize_anydoc_text(text: Any, path: str, pdf_note: Callable[[], str]) -> str:
-    """Normalize converter output and, for PDFs, PREPEND the coverage note.
-
-    Prepended because read_file paginates: a footer on a long document would sit on a page
-    the model may never fetch. Covers PARTIAL gaps that convert without NeedsOcrError.
-    """
+    """Normalize converter output and, for PDFs, PREPEND the coverage note (read_file
+    paginates: a footer may never be fetched). Covers PARTIAL gaps without NeedsOcrError."""
     if not isinstance(text, str) or not text.strip():
         raise ExtractionError("Document contains no extractable text")
-    text = text.rstrip("\n") + "\n"
-    if Path(path).suffix.lower() == ".pdf":
-        note = pdf_note()
-        if note:
-            text = note + text
-    return text
+    note = pdf_note() if Path(path).suffix.lower() == ".pdf" else ""
+    return (note or "") + text.rstrip("\n") + "\n"
 
 
 def _ocr_scanned_pdf(mod: Any, path: str, exc: BaseException) -> str:
-    """Typed scanned-pages signal (anydoc >= 0.2): try hosted OCR when a Firecrawl route exists, else teach recovery."""
+    """anydoc >= 0.2 scanned-pages signal: hosted OCR when a route exists, else teach recovery."""
     pages = list(getattr(exc, "pages", []) or [])
     enabled, api_key, api_url = _hosted_ocr_config()
     hosted_error = ""
     if enabled:
         try:
-            kwargs = {"ocr": "hosted", **{k: v for k, v in (("api_key", api_key), ("api_url", api_url)) if v}}
-            return mod.to_markdown(path, **kwargs).rstrip("\n") + "\n"
+            extra = {k: v for k, v in (("api_key", api_key), ("api_url", api_url)) if v}
+            return mod.to_markdown(path, ocr="hosted", **extra).rstrip("\n") + "\n"
         except Exception as hosted_exc:  # noqa: BLE001
             hosted_error = f"{type(hosted_exc).__name__}: {hosted_exc}"
     # No route / disabled / hosted failed: whole doc is scans — the warning IS the result.
@@ -277,10 +248,9 @@ def _extract_anydoc_bytes(data: bytes, path: str) -> str:
     return _finalize_anydoc_text(text, path, lambda: _pdf_coverage_note_from_bytes(data, path))
 
 
-# ── Scanned-PDF coverage detection ──────────────────────────────────
-# Text-layer extractors return nothing for scanned pages, so a mostly-scanned PDF
-# converts "successfully" into headers with empty bodies — silent data loss the model
-# cannot detect. Count per-page text via pdftotext (form-feed separated) and warn.
+# ── Scanned-PDF coverage detection: text-layer extractors return nothing for scanned
+# pages, so a mostly-scanned PDF converts "successfully" into headers with empty bodies —
+# silent data loss. Count per-page text via pdftotext (form-feed separated) and warn.
 PDF_EMPTY_PAGE_CHARS = 20  # fewer extracted chars than this = empty page
 # Warn when empty pages reach both MIN_EMPTY and MIN_RATIO, or ABSOLUTE_EMPTY alone.
 PDF_COVERAGE_MIN_EMPTY = 2
@@ -297,32 +267,26 @@ def _pdf_page_texts(path: str) -> Optional[list[str]]:
     if shutil.which("pdftotext") is None:
         return None
     try:
-        proc = subprocess.run(["pdftotext", path, "-"], capture_output=True, timeout=PDF_PAGE_SCAN_TIMEOUT)
+        proc = subprocess.run(
+            ["pdftotext", path, "-"], capture_output=True, timeout=PDF_PAGE_SCAN_TIMEOUT)
     except (OSError, subprocess.SubprocessError):
         return None
-    if proc.returncode != 0:
-        return None
-    pages = proc.stdout.decode("utf-8", errors="replace").split("\f")
+    out = proc.stdout.decode("utf-8", errors="replace") if proc.returncode == 0 else ""
+    pages = out.split("\f") if out else []
     if pages and not pages[-1].strip():
         pages.pop()  # trailing form-feed artifact
     return pages or None
 
 
-def _group_ranges(pages: list[int]) -> list[list[int]]:
-    """Group sorted 1-based page numbers into [start, end] runs."""
-    ranges: list[list[int]] = []
-    for p in pages:
+def _gap_map(counts: list[int], texts: list[str], empty: list[int]) -> str:
+    """Per-gap breakdown, each empty range labeled with the last text seen before
+    it (usually a section header), so the agent can pick WHICH gaps to OCR."""
+    ranges: list[list[int]] = []  # sorted 1-based page numbers -> [start, end] runs
+    for p in empty:
         if ranges and p == ranges[-1][1] + 1:
             ranges[-1][1] = p
         else:
             ranges.append([p, p])
-    return ranges
-
-
-def _gap_map(counts: list[int], texts: list[str], empty: list[int]) -> str:
-    """Per-gap breakdown, each empty range labeled with the last text seen before
-    it (usually a section header), so the agent can pick WHICH gaps to OCR."""
-    ranges = _group_ranges(empty)
     lines: list[str] = []
     for a, b in ranges[:PDF_GAP_MAP_MAX_ENTRIES]:
         label = ""
@@ -341,20 +305,17 @@ def _gap_map(counts: list[int], texts: list[str], empty: list[int]) -> str:
 
 
 def _pdf_coverage_note(path: str, display_path: Optional[str] = None) -> str:
-    """Warning header when many PDF pages produced no text, else ''.
-
-    ``path`` is scanned with pdftotext (may be a host temp file); ``display_path``
-    is what the recovery command shows — the path the agent's terminal can see.
-    """
+    """Warning header when many PDF pages produced no text, else ''. ``path`` is scanned
+    (may be a host temp file); ``display_path`` is what the recovery command shows."""
     texts = _pdf_page_texts(path)
     if not texts or len(texts) < 2:
         return ""
     counts = [len(page.strip()) for page in texts]
     empty = [i + 1 for i, n in enumerate(counts) if n < PDF_EMPTY_PAGE_CHARS]
     total = len(counts)
-    if len(empty) < PDF_COVERAGE_MIN_EMPTY:
-        return ""
-    if len(empty) / total < PDF_COVERAGE_MIN_RATIO and len(empty) < PDF_COVERAGE_ABSOLUTE_EMPTY:
+    if len(empty) < PDF_COVERAGE_MIN_EMPTY or (
+        len(empty) / total < PDF_COVERAGE_MIN_RATIO and len(empty) < PDF_COVERAGE_ABSOLUTE_EMPTY
+    ):
         return ""
     shown = display_path or path
     return (
@@ -370,13 +331,12 @@ def _pdf_coverage_note(path: str, display_path: Optional[str] = None) -> str:
         f"`pdftoppm -jpeg -r 150 -f <first> -l <last> '{shown}' /tmp/page` "
         "and inspect each image with the vision_analyze tool, or use the "
         "ocr-and-documents skill (marker-pdf) for bulk OCR of large "
-        "ranges.]\n"
-    )
+        "ranges.]\n")
 
 
 def _pdf_coverage_note_from_bytes(data: bytes, display_path: str) -> str:
-    """Coverage note for backend-transferred PDF bytes: pdftotext is path-oriented, so scan a
-    host temp copy; the recovery command still names ``display_path`` (visible to the agent)."""
+    """Coverage note for backend-transferred PDF bytes via a host temp copy (pdftotext is
+    path-oriented); the recovery command still names ``display_path``."""
     try:
         with _temp_copy(data, ".pdf") as temp_path:
             return _pdf_coverage_note(temp_path, display_path=display_path)
@@ -384,6 +344,13 @@ def _pdf_coverage_note_from_bytes(data: bytes, display_path: str) -> str:
         return ""
 
 
+def _joined(lines: list[str], empty_error: str) -> str:
+    """Join extracted lines with a single trailing newline; raise when nothing non-blank."""
+    if not any(line.strip() for line in lines):
+        raise ExtractionError(empty_error)
+    return "\n".join(lines).rstrip("\n") + "\n"
+
+
 def _source_text(source) -> str:
     if isinstance(source, str):
         return source
@@ -404,10 +371,8 @@ def _base64_bytes(payload: str) -> int:
 
 
 def _clean_stream_text(text: str) -> str:
-    """Strip ANSI escapes and collapse ``\\r`` progress-bar rewrites: Jupyter renders only
-    the final frame of a ``\\r``-redrawn line (tqdm), so keep the text after the last ``\\r``."""
+    """Strip ANSI escapes; keep only the final ``\\r`` frame of each line (tqdm redraws)."""
     from tools.ansi_strip import strip_ansi
-
     lines = []
     for line in strip_ansi(text).replace("\r\n", "\n").split("\n"):
         frames = [frame for frame in line.split("\r") if frame]
@@ -415,21 +380,16 @@ def _clean_stream_text(text: str) -> str:
     return "\n".join(lines)
 
 
-# Notebook outputs longer than this are tail-truncated per output block so a
-# single runaway training log cannot flood the extracted text.
+# Per-output-block truncation so one runaway training log cannot flood the extraction.
 _MAX_OUTPUT_CHARS = 20_000
-
 # nbformat v3 stores mime data flat on the output dict under these keys.
 _V3_MIME_KEYS = (("png", "image/png"), ("jpeg", "image/jpeg"), ("svg", "image/svg+xml"), ("html", "text/html"))
 
 
 def _notebook_output_text(output: Any) -> str:
-    """Render one notebook output as compact text.
-
-    Keeps stream text, tracebacks, and textual results; replaces token-heavy payloads
-    (base64 images, HTML, widget state) with short sized placeholders. Handles nbformat
-    v4 and legacy v3 (``pyout``/``pyerr``) shapes.
-    """
+    """Render one notebook output as compact text: stream text, tracebacks and textual
+    results kept; token-heavy payloads (images, HTML, widgets) become sized placeholders.
+    Handles nbformat v4 and legacy v3 (``pyout``/``pyerr``) shapes."""
     if not isinstance(output, dict):
         return ""
     otype = output.get("output_type")
@@ -440,20 +400,17 @@ def _notebook_output_text(output: Any) -> str:
         traceback = output.get("traceback")
         tb_text = ""
         if isinstance(traceback, list):
-            tb_text = _clean_stream_text("\n".join(line for line in traceback if isinstance(line, str)))
+            tb_text = _clean_stream_text(
+                "\n".join(line for line in traceback if isinstance(line, str)))
         header = f"Error: {output.get('ename', '')}: {output.get('evalue', '')}".rstrip(": ")
         return f"{header}\n{tb_text}".rstrip()
     if otype not in {"execute_result", "display_data", "pyout"}:
         return ""
 
     data = output.get("data")
-    if not isinstance(data, dict):
-        data = {}
-        if isinstance(output.get("text"), (str, list)):
-            data["text/plain"] = output["text"]
-        for v3_key, mime in _V3_MIME_KEYS:
-            if v3_key in output:
-                data[mime] = output[v3_key]
+    if not isinstance(data, dict):  # legacy v3: mime payloads sit flat on the output dict
+        data = {"text/plain": output["text"]} if isinstance(output.get("text"), (str, list)) else {}
+        data.update((mime, output[k]) for k, mime in _V3_MIME_KEYS if k in output)
     if "application/vnd.jupyter.widget-view+json" in data:
         return "[interactive widget — omitted]"
     # Prefer readable text: models consume text/plain far better than markup.
@@ -464,9 +421,11 @@ def _notebook_output_text(output: Any) -> str:
                 return body
     for mime, value in data.items():
         if isinstance(mime, str) and mime.startswith("image/"):
-            return f"[{mime} output — {_human_size(_base64_bytes(_source_text(value)))}, omitted]"
+            size = _base64_bytes(_source_text(value))
+            return f"[{mime} output — {_human_size(size)}, omitted]"
     if "text/html" in data:
-        return f"[text/html output — {len(_source_text(data['text/html'])):,} chars, omitted]"
+        html = _source_text(data["text/html"])
+        return f"[text/html output — {len(html):,} chars, omitted]"
     mimes = ", ".join(str(m) for m in data) or "unknown"
     return f"[{mimes} output — omitted]"
 
@@ -497,7 +456,6 @@ def _extract_notebook(path: str) -> str:
         raise ExtractionError(f"Not a valid notebook: {exc}") from exc
     if not isinstance(nb, dict):
         raise ExtractionError("Notebook root is not an object")
-
     raw_cells = nb.get("cells")
     if isinstance(raw_cells, list):
         cells = [(f".cells[{i}].outputs", cell) for i, cell in enumerate(raw_cells)]
@@ -505,11 +463,9 @@ def _extract_notebook(path: str) -> str:
         cells = [
             (f".worksheets[{wi}].cells[{ci}].outputs", cell)
             for wi, ws in enumerate(nb.get("worksheets", [])) if isinstance(ws, dict)
-            for ci, cell in enumerate(ws.get("cells", []))
-        ]
+            for ci, cell in enumerate(ws.get("cells", []))]
     if not cells:
         raise ExtractionError("Notebook contains no cells")
-
     nb_name = os.path.basename(path)
     counts = dict.fromkeys(_CELL_LABELS, 0)
     out: list[str] = []
@@ -526,14 +482,12 @@ def _extract_notebook(path: str) -> str:
             rendered = _notebook_outputs(cell, jq_pointer, nb_name)
             if rendered:
                 out.extend((f"# ── Output (cell {counts[typ]}) ──", rendered.rstrip("\n"), ""))
-    if not out:
-        raise ExtractionError("Notebook contains no readable cells")
-    return "\n".join(out).rstrip("\n") + "\n"
+    return _joined(out, "Notebook contains no readable cells")
 
 
 @contextlib.contextmanager
 def _open_zip(path: str, kind: str) -> Iterator[zipfile.ZipFile]:
-    """Open an OOXML package, mapping bad-zip/OS failures (also from the body) to ExtractionError."""
+    """Open an OOXML package; bad-zip/OS failures (body included) become ExtractionError."""
     try:
         with zipfile.ZipFile(path) as zf:
             yield zf
@@ -560,34 +514,32 @@ def _zip_xml(zf: zipfile.ZipFile, name: str, optional: bool = False) -> Any:
 def _extract_docx(path: str) -> str:
     with _open_zip(path, "DOCX") as zf:
         root = _zip_xml(zf, "word/document.xml")
-
     w = f"{{{_NS_W}}}"
+    breaks = {f"{w}tab": "\t", f"{w}br": "\n", f"{w}cr": "\n"}
     lines: list[str] = []
     for para in root.iter(f"{w}p"):
-        buf: list[str] = []
-        for node in para.iter():
-            if node.tag == f"{w}t":
-                buf.append(node.text or "")
-            elif node.tag == f"{w}tab":
-                buf.append("\t")
-            elif node.tag in {f"{w}br", f"{w}cr"}:
-                buf.append("\n")
+        buf = [(node.text or "") if node.tag == f"{w}t" else breaks.get(node.tag, "")
+               for node in para.iter()]
         lines.extend("".join(buf).split("\n"))
-    if not any(line.strip() for line in lines):
-        raise ExtractionError("DOCX contains no extractable text")
-    return "\n".join(lines).rstrip("\n") + "\n"
+    return _joined(lines, "DOCX contains no extractable text")
 
 
 def _extract_xlsx(path: str) -> str:
+    s, r, pr = f"{{{_NS_S}}}", f"{{{_NS_REL}}}", f"{{{_NS_PKG_REL}}}"
     with _open_zip(path, "XLSX") as zf:
         names = set(zf.namelist())
-        shared = _shared_strings(zf)
-        rels = _workbook_rels(zf)
+        sst = _zip_xml(zf, "xl/sharedStrings.xml", optional=True)
+        shared = [] if sst is None else [
+            "".join(t.text or "" for t in item.iter(f"{s}t")) for item in sst.iter(f"{s}si")]
+        rels_root = _zip_xml(zf, "xl/_rels/workbook.xml.rels", optional=True)
+        rels = {} if rels_root is None else {
+            rel.get("Id", ""): rel.get("Target", "")
+            for rel in rels_root.iter(f"{pr}Relationship") if rel.get("Id")}
         out: list[str] = []
-        for name, state, rid in _workbook_sheets(zf):
-            if state in {"hidden", "veryHidden"}:
+        for sheet in _zip_xml(zf, "xl/workbook.xml").iter(f"{s}sheet"):
+            if sheet.get("state", "visible") in {"hidden", "veryHidden"}:
                 continue
-            target = rels.get(rid, "").lstrip("/")
+            target = rels.get(sheet.get(f"{r}id", ""), "").lstrip("/")
             part = posixpath.normpath(target if target.startswith("xl/") else f"xl/{target}")
             if part not in names:
                 continue
@@ -595,40 +547,12 @@ def _extract_xlsx(path: str) -> str:
                 rows = _sheet_rows(zf.read(part), shared)
             except ET.ParseError:
                 continue
-            out.append(f"# ── Sheet: {name} ──")
+            out.append(f"# ── Sheet: {sheet.get('name', 'Sheet')} ──")
             out.extend("\t".join(row) for row in rows)
             if not rows:
                 out.append("(empty)")
             out.append("")
-
-    if not out:
-        raise ExtractionError("XLSX has no visible sheets with content")
-    return "\n".join(out).rstrip("\n") + "\n"
-
-
-def _shared_strings(zf: zipfile.ZipFile) -> list[str]:
-    root = _zip_xml(zf, "xl/sharedStrings.xml", optional=True)
-    if root is None:
-        return []
-    s = f"{{{_NS_S}}}"
-    return ["".join(t.text or "" for t in item.iter(f"{s}t")) for item in root.iter(f"{s}si")]
-
-
-def _workbook_sheets(zf: zipfile.ZipFile) -> list[tuple[str, str, str]]:
-    root = _zip_xml(zf, "xl/workbook.xml")
-    s, r = f"{{{_NS_S}}}", f"{{{_NS_REL}}}"
-    return [
-        (sheet.get("name", "Sheet"), sheet.get("state", "visible"), sheet.get(f"{r}id", ""))
-        for sheet in root.iter(f"{s}sheet")
-    ]
-
-
-def _workbook_rels(zf: zipfile.ZipFile) -> dict[str, str]:
-    root = _zip_xml(zf, "xl/_rels/workbook.xml.rels", optional=True)
-    if root is None:
-        return {}
-    rel_tag = f"{{{_NS_PKG_REL}}}Relationship"
-    return {rel.get("Id", ""): rel.get("Target", "") for rel in root.iter(rel_tag) if rel.get("Id")}
+    return _joined(out, "XLSX has no visible sheets with content")
 
 
 def _col_index(ref: str) -> int:
@@ -674,14 +598,9 @@ def _cell_value(cell: ET.Element, shared: list[str], s: str) -> str:
         return "" if inline is None else "".join(t.text or "" for t in inline.iter(f"{s}t"))
     if typ == "b":
         return "TRUE" if value.strip() in {"1", "true", "TRUE"} else "FALSE"
-    if typ == "e":
-        return value or "#ERROR"
-    return value
+    return (value or "#ERROR") if typ == "e" else value
 
 
 # Extension -> stdlib extractor; anydoc formats fall through in extract_document_text.
 _STDLIB_EXTRACTORS: dict[str, Callable[[str], str]] = {
-    ".ipynb": _extract_notebook,
-    ".docx": _extract_docx,
-    ".xlsx": _extract_xlsx,
-}
+    ".ipynb": _extract_notebook, ".docx": _extract_docx, ".xlsx": _extract_xlsx}

tools/read_preview_tool.py

@@ -13,14 +13,11 @@ from tools.read_terminal_tool import read_pane
 
 
 def read_preview_tool(
-    start: Optional[int] = None,
-    count: Optional[int] = None,
-    callback: Optional[Callable] = None,
+    start: Optional[int] = None, count: Optional[int] = None, callback: Optional[Callable] = None
 ) -> str:
     """Return the active preview tab's contents (+ metadata) as a JSON string."""
     return read_pane(callback, (("start", start, 0), ("count", count, 1)), (
         "read_preview is only available in the Hermes desktop app.",
         "start and count must be integers.",
         "Failed to read the preview pane: ",
-        "No preview tab is open, or the read timed out.",
-    ))
+        "No preview tab is open, or the read timed out."))

tools/read_terminal_tool.py

@@ -14,11 +14,9 @@ from tools.registry import registry, tool_error
 
 
 def read_pane(callback: Optional[Callable], window, errors: tuple) -> str:
-    """Shared body of the read_terminal / read_preview bridges.
-
-    ``window`` is ``((key, value, floor), ...)``; None values are omitted, others are
-    int-coerced and floored. ``errors`` = (not_desktop, not_integers, fail_prefix, empty).
-    """
+    """Shared body of the read_terminal / read_preview / read_window bridges. ``window`` is
+    ``((key, value, floor), ...)`` (None omitted, else int-coerced and floored); ``errors`` =
+    (not_desktop, not_integers, fail_prefix, empty)."""
     if callback is None:
         return tool_error(errors[0])
     try:
@@ -77,9 +75,7 @@ registry.register(
     toolset="desktop_ui",
     schema=READ_TERMINAL_SCHEMA,
     handler=lambda args, **kw: read_terminal_tool(
-        start_line=args.get("start_line"),
-        count=args.get("count"),
-        callback=kw.get("callback"),
+        start_line=args.get("start_line"), count=args.get("count"), callback=kw.get("callback")
     ),
     emoji="🖥️",
 )

tools/read_window_tool.py

@@ -34,8 +34,7 @@ READ_WINDOW_BELOW_SCHEMA = {
         "retrying. Metadata only; never captures pixels."
     ),
     "parameters": {
-        "type": "object",
-        "properties": {},
+        "type": "object", "properties": {}
     },
 }
 

tools/shell_heredoc.py

@@ -1,18 +1,11 @@
-"""Conservative heredoc masking for shell-command scanners.
-
-Guards that scan raw command text (the background-'&' guard in ``tools/terminal_tool.py``,
-blocked-command checks, ``cron/lifecycle_guard``) false-positive on heredoc *bodies*, which
-are usually inline data. Naively stripping every body is unsafe the other way (fake ``<<``
-in quotes can swallow a real operator; unquoted bodies expand; ``bash <<'EOF'`` executes).
-
-A body is masked ONLY when: every delimiter on the opener is quoted (no expansion); every
-heredoc is terminated by an exact delimiter line; the opener is a single command (no
-``;``/``|``/``&`` and no ``$(...)``, backtick or process substitution); and the consumer is
-an allowlisted non-shell interpreter (``_INERT_HEREDOC_CONSUMER_RE``). Otherwise the command
-is returned untouched: a false positive is acceptable, hiding real shell syntax from a guard
-is not. Masked bodies become an equal number of newlines so ``re.MULTILINE`` scanning keeps
-its line structure. Adapted from Wolfram Ravenwolf's security-hardened rework of PR #63788.
-"""
+"""Conservative heredoc masking for shell-command scanners (terminal '&' guard, blocked-command
+checks, cron lifecycle_guard) that false-positive on heredoc *bodies*. Stripping every body is
+unsafe the other way (a fake ``<<`` in quotes can swallow a real operator; unquoted bodies
+expand; ``bash <<'EOF'`` executes), so a body is masked ONLY when every delimiter is quoted,
+every heredoc has an exact terminator line, the opener is a single command (no ``;|&``,
+``$(...)``, backticks or process substitution) and the consumer is an allowlisted non-shell
+interpreter. Otherwise the command is returned untouched: a false positive is acceptable,
+hiding shell syntax from a guard is not. Masked bodies keep their newline count (re.MULTILINE)."""
 
 from __future__ import annotations
 
@@ -22,13 +15,9 @@ import re
 # THAT interpreter. Optional VAR=... assignments, ``env`` and a path prefix are
 # allowed. Deliberately narrow: anything unmatched keeps its body visible.
 _INERT_HEREDOC_CONSUMER_RE = re.compile(
-    r"^\s*"
-    r"(?:[A-Z_][A-Z0-9_]*=\S+\s+)*"
-    r"(?:env\s+)?"
-    r"(?:[A-Za-z0-9_./-]+/)?"
+    r"^\s*(?:[A-Z_][A-Z0-9_]*=\S+\s+)*(?:env\s+)?(?:[A-Za-z0-9_./-]+/)?"
     r"(?:python(?:3(?:\.\d+)*)?|osascript|cat)(?=\s|$)",
-    re.IGNORECASE,
-)
+    re.IGNORECASE)
 
 
 def _span_end(command: str, cursor: int, closer: str) -> int:
@@ -134,11 +123,8 @@ def _parse_heredoc_operator(command: str, index: int):
 
 
 def _scan_heredoc_command_unit(command: str, start: int):
-    """Scan one logical command -> ``(end, specs, unknown_operator, has_list_operator)``.
-
-    ``unknown_operator``: an unparseable ``<<`` (caller must fail closed).
-    ``has_list_operator``: unquoted ``;``/``|``/``&`` on the opener.
-    """
+    """Scan one logical command -> ``(end, specs, unknown_operator, has_list_operator)``:
+    an unparseable ``<<`` (caller must fail closed) / unquoted ``;|&`` on the opener."""
     cursor = start
     quote = None
     comment = False
@@ -196,7 +182,8 @@ def _scan_heredoc_command_unit(command: str, start: int):
     return len(command), specs, unknown_operator, has_list_operator
 
 
-def _find_heredoc_close(command: str, body_start: int, delimiter: str, strip_tabs: bool) -> int | None:
+def _find_heredoc_close(
+        command: str, body_start: int, delimiter: str, strip_tabs: bool) -> int | None:
     """Return the position after an exact shell heredoc terminator line."""
     cursor = body_start
     while True:
@@ -222,7 +209,8 @@ def strip_inert_heredoc_bodies(command: str) -> str:
     command_start = 0
 
     while command_start <= last_opener_index:
-        command_end, specs, unknown_operator, has_list_operator = _scan_heredoc_command_unit(command, command_start)
+        command_end, specs, unknown_operator, has_list_operator = (
+            _scan_heredoc_command_unit(command, command_start))
         if unknown_operator:
             return command
         if not specs: