diff --git a/tools/neutts_synth.py b/tools/neutts_synth.py
index 56ecee434d..9d51153199 100644
--- a/tools/neutts_synth.py
+++ b/tools/neutts_synth.py
@@ -1,15 +1,10 @@
 #!/usr/bin/env python3
 """Standalone NeuTTS synthesis helper.
 
-Called by tts_tool.py via subprocess to keep the TTS model (~500MB)
-in a separate process that exits after synthesis — no lingering memory.
-
-Usage:
-    python -m tools.neutts_synth --text "Hello" --out output.wav \
-        --ref-audio samples/jo.wav --ref-text samples/jo.txt
-
-Requires: python -m pip install -U neutts[all]
-System:   apt install espeak-ng  (or brew install espeak-ng)
+Called by tts_tool via subprocess so the ~500MB TTS model lives in a process that exits
+after synthesis. Usage:
+    python -m tools.neutts_synth --text "Hello" --out out.wav --ref-audio jo.wav --ref-text jo.txt
+Requires ``pip install -U neutts[all]`` and espeak-ng (apt/brew).
 """
 
 import argparse
@@ -24,28 +19,12 @@ def _write_wav(path: str, samples, sample_rate: int = 24000) -> None:
 
     if not isinstance(samples, np.ndarray):
         samples = np.array(samples, dtype=np.float32)
-    samples = samples.flatten()
-
-    # Clamp and convert to int16
-    samples = np.clip(samples, -1.0, 1.0)
-    pcm = (samples * 32767).astype(np.int16)
-
-    num_channels = 1
-    bits_per_sample = 16
-    byte_rate = sample_rate * num_channels * (bits_per_sample // 8)
-    block_align = num_channels * (bits_per_sample // 8)
-    data_size = len(pcm) * (bits_per_sample // 8)
-
+    pcm = (np.clip(samples.flatten(), -1.0, 1.0) * 32767).astype(np.int16)
+    data_size = len(pcm) * 2  # 16-bit mono
     with open(path, "wb") as f:
-        f.write(b"RIFF")
-        f.write(struct.pack("<I", 36 + data_size))
-        f.write(b"WAVE")
-        f.write(b"fmt ")
-        f.write(struct.pack("<IHHIIHH", 16, 1, num_channels, sample_rate,
-                            byte_rate, block_align, bits_per_sample))
-        f.write(b"data")
-        f.write(struct.pack("<I", data_size))
-        f.write(pcm.tobytes())
+        f.write(b"RIFF" + struct.pack("<I", 36 + data_size) + b"WAVEfmt ")
+        f.write(struct.pack("<IHHIIHH", 16, 1, 1, sample_rate, sample_rate * 2, 2, 16))
+        f.write(b"data" + struct.pack("<I", data_size) + pcm.tobytes())
 
 
 def main():
@@ -54,18 +33,10 @@ 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()
 
-    # llama_cpp (backbone) offloads to GPU only for the literal string "gpu";
-    # torch (codec) only accepts "cuda". A single --device value can't satisfy
-    # both — "cuda" silently no-ops on the backbone, leaving it on CPU.
-    backbone_device = "gpu" if args.device == "cuda" else args.device
-    codec_device = args.device
-
-    # Validate inputs
     ref_audio = Path(args.ref_audio).expanduser()
     ref_text_path = Path(args.ref_text).expanduser()
     if not ref_audio.exists():
@@ -74,35 +45,32 @@ def main():
     if not ref_text_path.exists():
         print(f"Error: reference text not found: {ref_text_path}", file=sys.stderr)
         sys.exit(1)
-
     ref_text = ref_text_path.read_text(encoding="utf-8").strip()
 
-    # Import and run NeuTTS
     try:
         from neutts import NeuTTS
     except ImportError:
         print("Error: neutts not installed. Run: python -m pip install -U neutts[all]", file=sys.stderr)
         sys.exit(1)
 
+    # llama_cpp (backbone) offloads to GPU only for the literal string "gpu";
+    # torch (codec) only accepts "cuda". A single --device value can't satisfy
+    # both — "cuda" silently no-ops on the backbone, leaving it on CPU.
     tts = NeuTTS(
         backbone_repo=args.model,
-        backbone_device=backbone_device,
+        backbone_device="gpu" if args.device == "cuda" else args.device,
         codec_repo="neuphonic/neucodec",
-        codec_device=codec_device,
+        codec_device=args.device,
     )
-    ref_codes = tts.encode_reference(str(ref_audio))
-    wav = tts.infer(args.text, ref_codes, ref_text)
+    wav = tts.infer(args.text, tts.encode_reference(str(ref_audio)), ref_text)
 
-    # Write output
     out_path = Path(args.out)
     out_path.parent.mkdir(parents=True, exist_ok=True)
-
     try:
         import soundfile as sf
         sf.write(str(out_path), wav, 24000)
     except ImportError:
         _write_wav(str(out_path), wav, 24000)
-
     print(f"OK: {out_path}", file=sys.stderr)
 
 
diff --git a/tools/open_preview_tool.py b/tools/open_preview_tool.py
index 03f3a44d5a..f800653f97 100644
--- a/tools/open_preview_tool.py
+++ b/tools/open_preview_tool.py
@@ -1,10 +1,9 @@
 #!/usr/bin/env python3
 """Open a URL, dev server, or file in the Hermes desktop GUI's preview pane.
 
-Registration moved into the `desktop_preview` tool; this module keeps the normalizer +
-open action for ``tools.preview_tool``. Emits ``preview.open`` via ``desktop_ui``: the
-renderer opens the pane for the window that asked and never steals focus for a
-background session. The desktop_ui toolset reaches desktop clients on any backend.
+Registration lives in the `desktop_preview` tool (``tools.preview_tool``); this module keeps
+the normalizer + open action. Emits ``preview.open`` via ``desktop_ui``: the renderer opens
+the pane for the window that asked and never steals focus for a background session.
 """
 
 import re
@@ -37,7 +36,6 @@ def open_preview_tool(url: str, label: str = "") -> str:
             "url is required — a web URL (https://…), a localhost dev server, or a "
             "file path to show in the preview pane."
         )
-
     label = (label or "").strip()
     return desktop_ui.emit_or_error(
         "preview.open",
diff --git a/tools/osv_check.py b/tools/osv_check.py
index 69036d1a45..7947b182a5 100644
--- a/tools/osv_check.py
+++ b/tools/osv_check.py
@@ -1,12 +1,8 @@
 """OSV malware check for MCP extension packages.
 
-Before launching an MCP server via npx/uvx, queries the OSV (Open Source
-Vulnerabilities) API to check if the package has any known malware advisories
-(MAL-* IDs).  Regular CVEs are ignored — only confirmed malware is blocked.
-
-The API is free, public, and maintained by Google.  Typical latency is ~300ms.
-Fail-open: network errors allow the package to proceed.
-
+Before launching an MCP server via npx/uvx, queries Google's free public OSV API for
+known malware advisories (MAL-* IDs). Regular CVEs are ignored — only confirmed malware
+is blocked. Fail-open: network errors allow the package to proceed (~300ms typical).
 Inspired by Block/goose's extension malware check.
 """
 
@@ -60,17 +56,11 @@ def _cache_put(key, result: Optional[str]) -> None:
         _cache[key] = (time.monotonic() + _CACHE_TTL_S, result)
 
 
-def check_package_for_malware(
-    command: str, args: list
-) -> Optional[str]:
-    """Check if an MCP server package has known malware advisories.
+def check_package_for_malware(command: str, args: list) -> Optional[str]:
+    """Check an MCP server package (inferred from ``command``/``args``) for MAL-* advisories.
 
-    Inspects the *command* (e.g. ``npx``, ``uvx``) and *args* to infer the
-    package name and ecosystem.  Queries the OSV API for MAL-* advisories.
-
-    Returns:
-        An error message string if malware is found, or None if clean/unknown.
-        Returns None (allow) on network errors or unrecognized commands.
+    Returns a BLOCKED message when malware is found, else None — including on network
+    errors and unrecognized commands (fail-open).
     """
     ecosystem = _infer_ecosystem(command)
     if not ecosystem:
@@ -88,22 +78,15 @@ def check_package_for_malware(
     try:
         malware = _query_osv(package, ecosystem, version)
     except Exception as exc:
-        # Fail-open: network errors, timeouts, parse failures → allow.
-        # Deliberately NOT cached — see _CACHE_TTL_S comment.
+        # 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 "
-            f"advisories: {ids}. Details: {summaries}"
-        )
-    else:
-        result = None
+        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}"
     _cache_put(cache_key, result)
     return result
 
@@ -115,26 +98,17 @@ _ECOSYSTEM_BY_COMMAND = {
 
 
 def _infer_ecosystem(command: str) -> Optional[str]:
-    """Infer package ecosystem from the command name."""
     return _ECOSYSTEM_BY_COMMAND.get(os.path.basename(command).lower())
 
 
-def _parse_package_from_args(
-    args: list, ecosystem: str
-) -> Tuple[Optional[str], Optional[str]]:
-    """Extract package name and optional version from command args.
-
-    Returns (package_name, version) or (None, None) if not parseable.
-    """
-    if not args:
-        return None, None
-
+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.
     package_token = None
     take_next = False
-    for arg in args:
+    for arg in args or ():
         if not isinstance(arg, str):
             continue
         if take_next:
@@ -160,54 +134,34 @@ def _parse_package_from_args(
 def _parse_npm_package(token: str) -> Tuple[Optional[str], Optional[str]]:
     """Parse npm package: @scope/name@version or name@version."""
     if token.startswith("@"):
-        # Scoped: @scope/name@version
         match = re.match(r"^(@[^/]+/[^@]+)(?:@(.+))?$", token)
-        if match:
-            return match.group(1), match.group(2)
-        return token, None
-    # Unscoped: name@version
+        return (match.group(1), match.group(2)) if match else (token, None)
     if "@" in token:
-        parts = token.rsplit("@", 1)
-        name = parts[0]
-        version = parts[1] if len(parts) > 1 and parts[1] != "latest" else None
-        return name, version
+        name, version = token.rsplit("@", 1)
+        return name, version if version != "latest" else None
     return token, None
 
 
 def _parse_pypi_package(token: str) -> Tuple[Optional[str], Optional[str]]:
     """Parse PyPI package: name==version or name[extras]==version."""
-    # Strip extras: name[extra1,extra2]==version
     match = re.match(r"^([a-zA-Z0-9._-]+)(?:\[[^\]]*\])?(?:==(.+))?$", token)
-    if match:
-        return match.group(1), match.group(2)
-    return token, None
+    return (match.group(1), match.group(2)) if match else (token, None)
 
 
 _PACKAGE_PARSERS = {"npm": _parse_npm_package, "PyPI": _parse_pypi_package}
 
 
-def _query_osv(
-    package: str, ecosystem: str, version: Optional[str] = None
-) -> list:
-    """Query the OSV API for MAL-* advisories. Returns list of malware vulns."""
+def _query_osv(package: str, ecosystem: str, version: Optional[str] = None) -> list:
+    """Query the OSV API; return only MAL-* advisories (regular CVEs ignored)."""
     payload = {"package": {"name": package, "ecosystem": ecosystem}}
     if version:
         payload["version"] = version
-
-    data = json.dumps(payload).encode("utf-8")
     req = urllib.request.Request(
         _OSV_ENDPOINT,
-        data=data,
-        headers={
-            "Content-Type": "application/json",
-            "User-Agent": "hermes-agent-osv-check/1.0",
-        },
+        data=json.dumps(payload).encode("utf-8"),
+        headers={"Content-Type": "application/json", "User-Agent": "hermes-agent-osv-check/1.0"},
         method="POST",
     )
-
     with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
         result = json.loads(resp.read())
-
-    vulns = result.get("vulns", [])
-    # Only malware advisories — ignore regular CVEs
-    return [v for v in vulns if v.get("id", "").startswith("MAL-")]
+    return [v for v in result.get("vulns", []) if v.get("id", "").startswith("MAL-")]
diff --git a/tools/patch_parser.py b/tools/patch_parser.py
index e48e1b84e5..66d9feb379 100644
--- a/tools/patch_parser.py
+++ b/tools/patch_parser.py
@@ -34,26 +34,23 @@ class OperationType(Enum):
 
 @dataclass
 class HunkLine:
-    """A single line in a patch hunk."""
     prefix: str  # ' ', '-', or '+'
     content: str
 
 
 @dataclass
 class Hunk:
-    """A group of changes within a file."""
     context_hint: Optional[str] = None
     lines: List[HunkLine] = field(default_factory=list)
 
 
 @dataclass
 class PatchOperation:
-    """A single operation in a V4A patch."""
     operation: OperationType
     file_path: str
-    new_path: Optional[str] = None  # For move operations
+    new_path: Optional[str] = None  # MOVE only
     hunks: List[Hunk] = field(default_factory=list)
-    content: Optional[str] = None  # For add file operations
+    content: Optional[str] = None  # ADD only
 
 
 # Markers must occupy the whole line at column 0 so content lines that merely
@@ -92,16 +89,17 @@ def parse_v4a_patch(patch_content: str) -> Tuple[List[PatchOperation], Optional[
     current_op: Optional[PatchOperation] = None
     current_hunk: Optional[Hunk] = None
 
+    def _flush_hunk() -> None:
+        if current_op and current_hunk and current_hunk.lines:
+            current_op.hunks.append(current_hunk)
+
     def _flush() -> None:
         if current_op:
-            if current_hunk and current_hunk.lines:
-                current_op.hunks.append(current_hunk)
+            _flush_hunk()
             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()
@@ -118,8 +116,7 @@ def parse_v4a_patch(patch_content: str) -> Tuple[List[PatchOperation], Optional[
                 current_op = None
         elif line.startswith('@@'):
             if current_op:
-                if current_hunk and current_hunk.lines:
-                    current_op.hunks.append(current_hunk)
+                _flush_hunk()
                 hint_match = _HINT_RE.match(line)
                 current_hunk = Hunk(context_hint=hint_match.group(1) if hint_match else None)
         elif current_op and line:
@@ -146,15 +143,7 @@ def parse_v4a_patch(patch_content: str) -> Tuple[List[PatchOperation], Optional[
 
 def _count_occurrences(text: str, pattern: str) -> int:
     """Count occurrences of *pattern* in *text*, advancing one char per hit (overlaps count)."""
-    count = 0
-    start = 0
-    while True:
-        pos = text.find(pattern, start)
-        if pos == -1:
-            break
-        count += 1
-        start = pos + 1
-    return count
+    return sum(1 for i in range(len(text) + 1) if text.startswith(pattern, i))
 
 
 def _split_hunk(hunk: Hunk) -> Tuple[List[str], List[str]]:
@@ -173,10 +162,15 @@ def _no_match_hint(error: Optional[str], search_pattern: str, content: str) -> s
         return ""
 
 
-def _validate_operations(
-    operations: List[PatchOperation],
-    file_ops: Any,
-) -> List[str]:
+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, ""
+
+
+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
@@ -198,78 +192,65 @@ def _validate_operations(
         if path in pending_content:
             return pending_content[path], None
         r = file_ops.read_file_raw(path)
-        if r.error:
-            return None, r.error
-        return r.content, None
+        return (None, r.error) if r.error else (r.content, None)
+
+    def _validate_update(op: PatchOperation) -> None:
+        nonlocal real_change_count
+        content, read_err = _read(op.file_path)
+        if read_err:
+            errors.append(f"{op.file_path}: {read_err}")
+            return
+        simulated = content
+        for hunk_index, hunk in enumerate(op.hunks, start=1):
+            search_lines, replace_lines = _split_hunk(hunk)
+            if not any(l.prefix in '-+' for l in hunk.lines):
+                # Inert anchor hunk (context only) — models emit these
+                # between real changes; ignore without failing the patch.
+                continue
+            real_change_count += 1
+            if not search_lines:
+                # Addition-only hunk: the context hint must be unique.
+                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")
+                    elif ambiguous:
+                        errors.append(f"{op.file_path}: addition-only hunk {ambiguous}")
+                continue
+            search_pattern = '\n'.join(search_lines)
+            replacement = '\n'.join(replace_lines)
+            if search_lines == replace_lines:
+                # Identical -/+ lines: apply skips it as a no-op, so
+                # 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
+            )
+            if count:
+                simulated = new_simulated
+            elif not is_already_applied(simulated or "", search_pattern, replacement):
+                # Already-applied hunks (edit landed in a prior call) are no-ops so
+                # multi-hunk patches don't fail wholesale; apply performs the same skip.
+                label = f"'{hunk.context_hint}'" if hunk.context_hint else "(no hint)"
+                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)
+                )
+        pending_content[op.file_path] = simulated
 
     for op in operations:
-        if op.operation != OperationType.UPDATE:
-            real_change_count += 1
         if op.operation == OperationType.UPDATE:
-            content, read_err = _read(op.file_path)
-            if read_err:
-                errors.append(f"{op.file_path}: {read_err}")
-                continue
-
-            simulated = content
-            for hunk_index, hunk in enumerate(op.hunks, start=1):
-                search_lines, replace_lines = _split_hunk(hunk)
-                if not any(l.prefix in '-+' for l in hunk.lines):
-                    # Inert anchor hunk (context only) — models emit these
-                    # between real changes; ignore without failing the patch.
-                    continue
-                real_change_count += 1
-                if not search_lines:
-                    # Addition-only hunk: the context hint must be unique.
-                    if hunk.context_hint:
-                        occurrences = _count_occurrences(simulated, hunk.context_hint)
-                        if occurrences == 0:
-                            errors.append(
-                                f"{op.file_path}: addition-only hunk context hint "
-                                f"'{hunk.context_hint}' not found"
-                            )
-                        elif occurrences > 1:
-                            errors.append(
-                                f"{op.file_path}: addition-only hunk context hint "
-                                f"'{hunk.context_hint}' is ambiguous "
-                                f"({occurrences} occurrences)"
-                            )
-                    continue
-
-                search_pattern = '\n'.join(search_lines)
-                replacement = '\n'.join(replace_lines)
-                if search_lines == replace_lines:
-                    # Identical -/+ lines: apply skips it as a no-op, so
-                    # 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
-                )
-                if count == 0:
-                    # Already-applied hunk (edit landed in a prior call): treat
-                    # as a no-op so multi-hunk patches don't fail wholesale.
-                    # The apply phase performs the same skip.
-                    if is_already_applied(simulated or "", search_pattern, replacement):
-                        continue
-                    label = f"'{hunk.context_hint}'" if hunk.context_hint else "(no hint)"
-                    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)
-                    )
-                else:
-                    simulated = new_simulated
-            pending_content[op.file_path] = simulated
-
-        elif op.operation == OperationType.DELETE:
+            _validate_update(op)
+            continue
+        real_change_count += 1
+        if op.operation == OperationType.DELETE:
             _content, read_err = _read(op.file_path)
             if read_err:
                 errors.append(f"{op.file_path}: file not found for deletion")
             else:
                 removed_paths.add(op.file_path)
                 pending_content.pop(op.file_path, None)
-
         elif op.operation == OperationType.MOVE:
             if not op.new_path:
                 errors.append(f"{op.file_path}: MOVE operation missing destination path")
@@ -279,15 +260,12 @@ def _validate_operations(
                 errors.append(f"{op.file_path}: source file not found for move")
             _dst, dst_err = _read(op.new_path)
             if not dst_err:
-                errors.append(
-                    f"{op.new_path}: destination already exists — move would overwrite"
-                )
+                errors.append(f"{op.new_path}: destination already exists — move would overwrite")
             # Only a cleanly-validated move updates the overlay.
             if not src_err and dst_err:
                 pending_content[op.new_path] = src_content if src_content is not None else ""
                 pending_content.pop(op.file_path, None)
                 removed_paths.add(op.file_path)
-
         # ADD: parent directory creation handled by write_file; no pre-check needed.
 
     if not errors and real_change_count == 0:
@@ -299,8 +277,7 @@ def _validate_operations(
 ApplyResult = Tuple[bool, str, Optional[str], Optional[dict]]
 
 
-def apply_v4a_operations(operations: List[PatchOperation],
-                         file_ops: Any) -> 'PatchResult':
+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
@@ -341,10 +318,7 @@ def apply_v4a_operations(operations: List[PatchOperation],
             if not ok:
                 errors.append(f"Failed to {verb} {op.file_path}: {payload}")
                 continue
-            label = op.file_path
-            if op.operation is OperationType.MOVE:
-                label = f"{op.file_path} -> {op.new_path}"
-            bucket.append(label)
+            bucket.append(f"{op.file_path} -> {op.new_path}" if op.operation is OperationType.MOVE else op.file_path)
             all_diffs.append(payload)
             if lsp:
                 lsp_blocks.append(lsp)
@@ -376,18 +350,15 @@ def apply_v4a_operations(operations: List[PatchOperation],
 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
-    basic two-argument form.
+    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.
     """
     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:
@@ -426,15 +397,11 @@ 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 = _count_occurrences(new_content, hunk.context_hint)
-        if occurrences > 1:
-            return None, (
-                f"Addition-only hunk: context hint '{hunk.context_hint}' is ambiguous "
-                f"({occurrences} occurrences) — 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:
-            hint_pos = new_content.find(hunk.context_hint)
-            eol = new_content.find('\n', hint_pos)
+            eol = new_content.find('\n', new_content.find(hunk.context_hint))
             if eol != -1:
                 return new_content[:eol + 1] + insert_text + '\n' + new_content[eol + 1:], None
             return new_content + '\n' + insert_text, None
@@ -472,17 +439,16 @@ def _apply_update(op: PatchOperation, file_ops: Any) -> ApplyResult:
             continue
 
         # Retry inside a window around the context hint, if any.
-        if hunk.context_hint:
-            hint_pos = new_content.find(hunk.context_hint)
-            if hint_pos != -1:
-                window_start = max(0, hint_pos - 500)
-                window_end = min(len(new_content), hint_pos + 2000)
-                window_new, count, _strategy, error = fuzzy_find_and_replace(
-                    new_content[window_start:window_end], search_pattern, replacement, replace_all=False
-                )
-                if count > 0:
-                    new_content = new_content[:window_start] + window_new + new_content[window_end:]
-                    error = None
+        hint_pos = new_content.find(hunk.context_hint) if hunk.context_hint else -1
+        if hint_pos != -1:
+            window_start = max(0, hint_pos - 500)
+            window_end = min(len(new_content), hint_pos + 2000)
+            window_new, count, _strategy, error = fuzzy_find_and_replace(
+                new_content[window_start:window_end], search_pattern, replacement, replace_all=False
+            )
+            if count > 0:
+                new_content = new_content[:window_start] + window_new + new_content[window_end:]
+                error = None
         if error:
             # Mirror the validation-phase already-applied skip, or the two
             # phases disagree and the whole patch fails here.
@@ -499,9 +465,7 @@ def _apply_update(op: PatchOperation, file_ops: Any) -> ApplyResult:
         return False, write_result.error, None, None
 
     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}",
+        current_content.splitlines(keepends=True), new_content.splitlines(keepends=True),
+        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)
diff --git a/tools/preview_tool.py b/tools/preview_tool.py
index 8bc7f3804d..6856118f17 100644
--- a/tools/preview_tool.py
+++ b/tools/preview_tool.py
@@ -12,10 +12,6 @@ from tools.open_preview_tool import _normalize_target, open_preview_tool
 from tools.registry import registry, tool_error
 
 
-def preview_open(url: str, label: str = "") -> str:
-    return open_preview_tool(url=url, label=label)
-
-
 def preview_close(url: str = "") -> str:
     target = _normalize_target(url or "")
     return desktop_ui.emit_or_error(
@@ -28,19 +24,16 @@ def preview_close(url: str = "") -> str:
 
 
 _ACTIONS = {
-    "open": lambda args: preview_open(url=args.get("url", ""), label=args.get("label", "")),
+    "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)."),
 }
 
 
 def _handle_preview(args, **kw):
     """Non-read actions only: action=read is dispatched at the agent level."""
-    action = (args.get("action") or "").strip()
-    fn = _ACTIONS.get(action)
+    fn = _ACTIONS.get((args.get("action") or "").strip())
     if fn is None:
         return tool_error("action must be one of: open, close, read.")
     return fn(args)
diff --git a/tools/read_extract.py b/tools/read_extract.py
index 5c8876f8e0..aeed3c6ae3 100644
--- a/tools/read_extract.py
+++ b/tools/read_extract.py
@@ -1,12 +1,10 @@
 """Stdlib document-to-text extraction for ``read_file``.
 
-Supports Jupyter notebooks, DOCX, and XLSX without hard dependencies. When the
-optional ``firecrawl-anydoc`` package is installed (imports as ``anydoc``),
-coverage widens to legacy Office (.doc/.ppt/.xls), OpenDocument, RTF, EPUB, and
-PDF via its Rust core. The stdlib extractors stay authoritative for their three
-formats so behavior is identical whether or not anydoc is present. Malformed
-documents raise :class:`ExtractionError`; callers then fall back to normal
-text/binary handling.
+Jupyter, DOCX and XLSX need no dependencies. The optional ``firecrawl-anydoc`` package
+(imports as ``anydoc``) widens coverage to legacy Office, OpenDocument, RTF, EPUB and PDF.
+The stdlib extractors stay authoritative for their three formats so behavior is identical
+with or without anydoc. Malformed documents raise :class:`ExtractionError`; callers then
+fall back to normal text/binary handling.
 """
 
 from __future__ import annotations
@@ -38,11 +36,8 @@ __all__ = [
 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",
+    ".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.
@@ -80,9 +75,8 @@ _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, so one transient failure
-    (network blip, pip race) does not stick in long-lived workers.
+    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).
     """
     global _anydoc_module, _anydoc_failed_at
     if _anydoc_module is not _ANYDOC_UNSET:
@@ -90,22 +84,15 @@ def _anydoc() -> Optional[Any]:
     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)
-        except Exception:
-            _anydoc_failed_at = time.monotonic()
-            return None
-        try:
             _anydoc_module = importlib.import_module("anydoc")
-        except Exception:  # ImportError or a broken native binding
+        except Exception:  # install failure, ImportError or a broken native binding
             _anydoc_failed_at = time.monotonic()
             return None
         _anydoc_failed_at = None
@@ -116,10 +103,6 @@ def is_extractable_document(path: str) -> bool:
     return bool(_extension(path))
 
 
-def _unsupported(path: str) -> ExtractionError:
-    return ExtractionError(f"Unsupported document type: {path!r}")
-
-
 def _check_size(size: int, limit: int) -> None:
     if size > limit:
         raise ExtractionError(f"Document too large to convert ({size:,} bytes, limit is {limit:,})")
@@ -136,10 +119,8 @@ def _temp_copy(data: bytes, suffix: str) -> Iterator[str]:
         yield temp_path
     finally:
         if temp_path:
-            try:
+            with contextlib.suppress(OSError):
                 os.unlink(temp_path)
-            except OSError:
-                pass
 
 
 def extract_document_text(path: str) -> str:
@@ -149,7 +130,7 @@ def extract_document_text(path: str) -> str:
         return extractor(path)
     if ext in ANYDOC_EXTENSIONS:
         return _extract_anydoc(path)
-    raise _unsupported(path)
+    raise ExtractionError(f"Unsupported document type: {path!r}")
 
 
 def extract_document_bytes(data: bytes, path: str) -> str:
@@ -159,18 +140,15 @@ def extract_document_bytes(data: bytes, path: str) -> str:
     if ext in ANYDOC_EXTENSIONS:
         return _extract_anydoc_bytes(data, path)
     if ext not in EXTRACTABLE_EXTENSIONS:
-        raise _unsupported(path)
+        raise ExtractionError(f"Unsupported document type: {path!r}")
     # The stdlib extractors are path-oriented.
     with _temp_copy(data, ext) as temp_path:
         return extract_document_text(temp_path)
 
 
 def _anydoc_missing_error(path: str) -> str:
-    """Teaching error for anydoc-gated formats when the converter is absent.
-
-    The schema deliberately omits these formats and this caveat; the explanation
-    (and the fix) is paid for only by sessions that actually hit one.
-    """
+    """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)."""
     return (
         f"Cannot convert {path!r}: this format needs the optional anydoc "
         "converter, which is not installed (install blocked or first "
@@ -184,33 +162,27 @@ def _anydoc_missing_error(path: str) -> str:
 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 while scrape/search
-    worked (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.
+    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.
     """
     api_key = os.environ.get("FIRECRAWL_API_KEY") or None
     enabled = api_key is not None
-    try:
+    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
         if isinstance(section, dict) and section.get("hosted_ocr") is False:
             enabled = False
-    except Exception:  # noqa: BLE001
-        pass
     return enabled, api_key, None
 
 
 def hosted_ocr_available() -> bool:
-    """Public probe for read_file's schema line: is hosted OCR unlocked?
-
-    Same single 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 (same gate as :func:`_hosted_ocr_config`);
+    a key that fails at conversion time lands in the NEEDS-OCR warning instead."""
     return _hosted_ocr_config()[0]
 
 
@@ -239,10 +211,8 @@ def _needs_ocr_warning(path: str, pages, hosted_error: str = "") -> str:
 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 the extraction: a footer on a long
-    document would sit on a page the model may never fetch. The note covers
-    PARTIAL gaps (text layer plus some scanned pages) that convert without
-    raising NeedsOcrError.
+    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.
     """
     if not isinstance(text, str) or not text.strip():
         raise ExtractionError("Document contains no extractable text")
@@ -261,16 +231,11 @@ def _ocr_scanned_pdf(mod: Any, path: str, exc: BaseException) -> str:
     hosted_error = ""
     if enabled:
         try:
-            kwargs = {"ocr": "hosted"}
-            if api_key:
-                kwargs["api_key"] = api_key
-            if api_url:
-                kwargs["api_url"] = api_url
+            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"
         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 — nothing to
-    # extract, so the warning IS the result.
+    # No route / disabled / hosted failed: whole doc is scans — the warning IS the result.
     return _needs_ocr_warning(path, pages, hosted_error)
 
 
@@ -296,9 +261,8 @@ def _extract_anydoc(path: str) -> str:
         needs_ocr = getattr(mod, "NeedsOcrError", None)
         if needs_ocr is not None and isinstance(exc, needs_ocr):
             return _ocr_scanned_pdf(mod, path, exc)
-        # anydoc raises one ConvertError subclass per failure mode (Unsupported,
-        # Malformed, Encrypted, ResourceLimit, MissingPart); all mean "no
-        # meaningful text", so read_file falls back to path/binary handling.
+        # anydoc raises one ConvertError subclass per failure mode (Unsupported, Malformed,
+        # Encrypted, ResourceLimit, MissingPart); all mean "no meaningful text".
         raise ExtractionError(f"{type(exc).__name__}: {exc}") from exc
     return _finalize_anydoc_text(text, path, lambda: _pdf_coverage_note(path))
 
@@ -314,13 +278,10 @@ def _extract_anydoc_bytes(data: bytes, path: str) -> str:
 
 
 # ── Scanned-PDF coverage detection ──────────────────────────────────
-# Text-layer extractors return nothing for scanned pages and emit no
-# placeholders, 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 separators) and warn when many pages are empty.
-
-# A page with fewer extracted characters than this is considered empty.
-PDF_EMPTY_PAGE_CHARS = 20
+# 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.
+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
 PDF_COVERAGE_MIN_RATIO = 0.2
@@ -336,11 +297,7 @@ 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:
@@ -379,8 +336,7 @@ def _gap_map(counts: list[int], texts: list[str], empty: list[int]) -> str:
         lines.append(f"  {span} ({n} page{'s' if n != 1 else ''}){label}")
     if len(ranges) > PDF_GAP_MAP_MAX_ENTRIES:
         rest = ranges[PDF_GAP_MAP_MAX_ENTRIES:]
-        rest_pages = sum(b - a + 1 for a, b in rest)
-        lines.append(f"  … {len(rest)} more gaps ({rest_pages} pages)")
+        lines.append(f"  … {len(rest)} more gaps ({sum(b - a + 1 for a, b in rest)} pages)")
     return "\n".join(lines)
 
 
@@ -398,10 +354,7 @@ def _pdf_coverage_note(path: str, display_path: Optional[str] = None) -> str:
     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) / total < PDF_COVERAGE_MIN_RATIO and len(empty) < PDF_COVERAGE_ABSOLUTE_EMPTY:
         return ""
     shown = display_path or path
     return (
@@ -422,11 +375,8 @@ def _pdf_coverage_note(path: str, display_path: Optional[str] = None) -> str:
 
 
 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`` — the path the agent's terminal backend can see.
-    """
+    """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)."""
     try:
         with _temp_copy(data, ".pdf") as temp_path:
             return _pdf_coverage_note(temp_path, display_path=display_path)
@@ -454,16 +404,12 @@ 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`` of each line.
-    """
+    """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``."""
     from tools.ansi_strip import strip_ansi
 
-    cleaned = strip_ansi(text).replace("\r\n", "\n")
     lines = []
-    for line in cleaned.split("\n"):
+    for line in strip_ansi(text).replace("\r\n", "\n").split("\n"):
         frames = [frame for frame in line.split("\r") if frame]
         lines.append(frames[-1] if frames else "")
     return "\n".join(lines)
@@ -480,61 +426,49 @@ _V3_MIME_KEYS = (("png", "image/png"), ("jpeg", "image/jpeg"), ("svg", "image/sv
 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.
+    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.
     """
     if not isinstance(output, dict):
         return ""
     otype = output.get("output_type")
-
     if otype == "stream":
         body = _clean_stream_text(_source_text(output.get("text", "")))
         return body if body.strip() else ""
-
     if otype in {"error", "pyerr"}:
         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 ""
 
-    if otype in {"execute_result", "display_data", "pyout"}:
-        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 "application/vnd.jupyter.widget-view+json" in data:
-            return "[interactive widget — omitted]"
-
-        # Prefer readable text: models consume text/plain far better than markup.
-        for mime in ("text/plain", "text/markdown"):
-            if mime in data:
-                body = _clean_stream_text(_source_text(data[mime]))
-                if body.strip():
-                    return body
-
-        for mime, value in data.items():
-            if isinstance(mime, str) and mime.startswith("image/"):
-                size = _base64_bytes(_source_text(value))
-                return f"[{mime} output — {_human_size(size)}, omitted]"
-
-        if "text/html" in data:
-            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]"
-
-    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 "application/vnd.jupyter.widget-view+json" in data:
+        return "[interactive widget — omitted]"
+    # Prefer readable text: models consume text/plain far better than markup.
+    for mime in ("text/plain", "text/markdown"):
+        if mime in data:
+            body = _clean_stream_text(_source_text(data[mime]))
+            if body.strip():
+                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]"
+    if "text/html" in data:
+        return f"[text/html output — {len(_source_text(data['text/html'])):,} chars, omitted]"
+    mimes = ", ".join(str(m) for m in data) or "unknown"
+    return f"[{mimes} output — omitted]"
 
 
 def _notebook_outputs(cell: dict, jq_pointer: str = "", filename: str = "") -> str:
@@ -570,8 +504,7 @@ def _extract_notebook(path: str) -> str:
     else:
         cells = [
             (f".worksheets[{wi}].cells[{ci}].outputs", cell)
-            for wi, ws in enumerate(nb.get("worksheets", []))
-            if isinstance(ws, dict)
+            for wi, ws in enumerate(nb.get("worksheets", [])) if isinstance(ws, dict)
             for ci, cell in enumerate(ws.get("cells", []))
         ]
     if not cells:
@@ -598,33 +531,35 @@ def _extract_notebook(path: str) -> str:
     return "\n".join(out).rstrip("\n") + "\n"
 
 
-def _zip_xml(zf: zipfile.ZipFile, name: str) -> ET.Element:
+@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."""
+    try:
+        with zipfile.ZipFile(path) as zf:
+            yield zf
+    except zipfile.BadZipFile as exc:
+        raise ExtractionError(f"Not a valid {kind}: {exc}") from exc
+    except OSError as exc:
+        raise ExtractionError(str(exc)) from exc
+
+
+def _zip_xml(zf: zipfile.ZipFile, name: str, optional: bool = False) -> Any:
+    """Parse a package part; ``optional`` parts yield None when absent or malformed."""
     try:
         return ET.fromstring(zf.read(name))
     except KeyError as exc:
+        if optional:
+            return None
         raise ExtractionError(f"Missing {name}") from exc
     except ET.ParseError as exc:
+        if optional:
+            return None
         raise ExtractionError(f"Malformed XML in {name}: {exc}") from exc
 
 
-def _optional_zip_xml(zf: zipfile.ZipFile, names: set[str], name: str) -> Optional[ET.Element]:
-    """Parse an optional package part; None when absent or malformed."""
-    if name not in names:
-        return None
-    try:
-        return ET.fromstring(zf.read(name))
-    except ET.ParseError:
-        return None
-
-
 def _extract_docx(path: str) -> str:
-    try:
-        with zipfile.ZipFile(path) as zf:
-            root = _zip_xml(zf, "word/document.xml")
-    except zipfile.BadZipFile as exc:
-        raise ExtractionError(f"Not a valid DOCX: {exc}") from exc
-    except OSError as exc:
-        raise ExtractionError(str(exc)) from exc
+    with _open_zip(path, "DOCX") as zf:
+        root = _zip_xml(zf, "word/document.xml")
 
     w = f"{{{_NS_W}}}"
     lines: list[str] = []
@@ -644,40 +579,35 @@ def _extract_docx(path: str) -> str:
 
 
 def _extract_xlsx(path: str) -> str:
-    try:
-        with zipfile.ZipFile(path) as zf:
-            names = set(zf.namelist())
-            shared = _shared_strings(zf, names)
-            sheets = _workbook_sheets(zf)
-            rels = _workbook_rels(zf, names)
-            out: list[str] = []
-            for name, state, rid in sheets:
-                if state in {"hidden", "veryHidden"}:
-                    continue
-                part = _sheet_part(rels.get(rid, ""))
-                if part not in names:
-                    continue
-                try:
-                    rows = _sheet_rows(zf.read(part), shared)
-                except ET.ParseError:
-                    continue
-                out.append(f"# ── Sheet: {name} ──")
-                out.extend("\t".join(row) for row in rows)
-                if not rows:
-                    out.append("(empty)")
-                out.append("")
-    except zipfile.BadZipFile as exc:
-        raise ExtractionError(f"Not a valid XLSX: {exc}") from exc
-    except OSError as exc:
-        raise ExtractionError(str(exc)) from exc
+    with _open_zip(path, "XLSX") as zf:
+        names = set(zf.namelist())
+        shared = _shared_strings(zf)
+        rels = _workbook_rels(zf)
+        out: list[str] = []
+        for name, state, rid in _workbook_sheets(zf):
+            if state in {"hidden", "veryHidden"}:
+                continue
+            target = rels.get(rid, "").lstrip("/")
+            part = posixpath.normpath(target if target.startswith("xl/") else f"xl/{target}")
+            if part not in names:
+                continue
+            try:
+                rows = _sheet_rows(zf.read(part), shared)
+            except ET.ParseError:
+                continue
+            out.append(f"# ── Sheet: {name} ──")
+            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, names: set[str]) -> list[str]:
-    root = _optional_zip_xml(zf, names, "xl/sharedStrings.xml")
+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}}}"
@@ -693,19 +623,14 @@ def _workbook_sheets(zf: zipfile.ZipFile) -> list[tuple[str, str, str]]:
     ]
 
 
-def _workbook_rels(zf: zipfile.ZipFile, names: set[str]) -> dict[str, str]:
-    root = _optional_zip_xml(zf, names, "xl/_rels/workbook.xml.rels")
+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")}
 
 
-def _sheet_part(target: str) -> str:
-    target = target.lstrip("/")
-    return posixpath.normpath(target if target.startswith("xl/") else f"xl/{target}")
-
-
 def _col_index(ref: str) -> int:
     idx = 0
     for ch in ref:
diff --git a/tools/read_preview_tool.py b/tools/read_preview_tool.py
index 66f4cf490b..57b52db34a 100644
--- a/tools/read_preview_tool.py
+++ b/tools/read_preview_tool.py
@@ -3,14 +3,13 @@
 
 The preview's content lives in the renderer (a sandboxed ``<webview>``), so this
 round-trips through the gateway's blocking-prompt bridge like ``read_terminal``
-(``preview.read.request`` -> ``preview.read.respond``). Registration moved into
-`desktop_preview`; the agent dispatches action=read here with the injected callback.
+(``preview.read.request`` -> ``preview.read.respond``). Registered as action=read of
+`desktop_preview`; the agent dispatches here with the injected callback.
 """
 
 from typing import Callable, Optional
 
-from tools.desktop_ui import passthrough_json
-from tools.registry import tool_error
+from tools.read_terminal_tool import read_pane
 
 
 def read_preview_tool(
@@ -19,23 +18,9 @@ def read_preview_tool(
     callback: Optional[Callable] = None,
 ) -> str:
     """Return the active preview tab's contents (+ metadata) as a JSON string."""
-    if callback is None:
-        return tool_error("read_preview is only available in the Hermes desktop app.")
-
-    try:
-        window = {
-            key: max(floor, int(val))
-            for key, val, floor in (("start", start, 0), ("count", count, 1))
-            if val is not None
-        }
-    except (TypeError, ValueError):
-        return tool_error("start and count must be integers.")
-
-    try:
-        raw = callback(**window)
-    except Exception as exc:
-        return tool_error(f"Failed to read the preview pane: {exc}")
-
-    if not raw:
-        return tool_error("No preview tab is open, or the read timed out.")
-    return passthrough_json(raw)
+    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.",
+    ))
diff --git a/tools/read_terminal_tool.py b/tools/read_terminal_tool.py
index 5fad29f0aa..945bedf474 100644
--- a/tools/read_terminal_tool.py
+++ b/tools/read_terminal_tool.py
@@ -1,49 +1,51 @@
 #!/usr/bin/env python3
 """Read the in-app terminal pane in the Hermes desktop GUI.
 
-The buffer lives in the desktop renderer (xterm.js), so this round-trips through
-the gateway's blocking-prompt bridge (as `clarify` does): tui_gateway emits
-``terminal.read.request``, the renderer answers ``terminal.read.respond``.  Lives
-in the ``desktop_ui`` toolset, enabled only for desktop-sourced sessions.
+The buffer lives in the desktop renderer (xterm.js), so this round-trips through the
+gateway's blocking-prompt bridge (as `clarify` does): tui_gateway emits
+``terminal.read.request``, the renderer answers ``terminal.read.respond``. Lives in the
+``desktop_ui`` toolset, enabled only for desktop-sourced sessions.
 """
 
-import json
 from typing import Callable, Optional
 
+from tools.desktop_ui import passthrough_json
 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).
+    """
+    if callback is None:
+        return tool_error(errors[0])
+    try:
+        kwargs = {key: max(floor, int(val)) for key, val, floor in window if val is not None}
+    except (TypeError, ValueError):
+        return tool_error(errors[1])
+    try:
+        raw = callback(**kwargs)
+    except Exception as exc:
+        return tool_error(f"{errors[2]}{exc}")
+    if not raw:
+        return tool_error(errors[3])
+    return passthrough_json(raw)
+
+
 def read_terminal_tool(
     start_line: Optional[int] = None,
     count: Optional[int] = None,
     callback: Optional[Callable] = None,
 ) -> str:
     """Return the in-app terminal's contents (+ line metadata) as a JSON string."""
-    if callback is None:
-        return tool_error("read_terminal is only available in the Hermes desktop app.")
-
-    try:
-        window = {
-            key: max(floor, int(val))
-            for key, val, floor in (("start", start_line, 0), ("count", count, 1))
-            if val is not None
-        }
-    except (TypeError, ValueError):
-        return tool_error("start_line and count must be integers.")
-
-    try:
-        raw = callback(**window)
-    except Exception as exc:
-        return tool_error(f"Failed to read terminal: {exc}")
-
-    if not raw:
-        return tool_error("No in-app terminal is open, or the read timed out.")
-
-    # Desktop answers with a JSON object; pass it through, else wrap the raw text.
-    try:
-        return json.dumps(json.loads(raw), ensure_ascii=False)
-    except (TypeError, ValueError):
-        return json.dumps({"text": str(raw)}, ensure_ascii=False)
+    return read_pane(callback, (("start", start_line, 0), ("count", count, 1)), (
+        "read_terminal is only available in the Hermes desktop app.",
+        "start_line and count must be integers.",
+        "Failed to read terminal: ",
+        "No in-app terminal is open, or the read timed out.",
+    ))
 
 
 READ_TERMINAL_SCHEMA = {
diff --git a/tools/read_window_tool.py b/tools/read_window_tool.py
index c772110eb8..ba0c5cf45c 100644
--- a/tools/read_window_tool.py
+++ b/tools/read_window_tool.py
@@ -8,28 +8,19 @@ main process (native window enumeration) -> ``window.read.respond``.
 
 from typing import Callable, Optional
 
-from tools.desktop_ui import passthrough_json
-from tools.registry import registry, tool_error
+from tools.read_terminal_tool import read_pane
+from tools.registry import registry
 
 
 def read_window_below_tool(callback: Optional[Callable] = None) -> str:
     """Return the window underneath the Hermes window as a JSON string."""
-    if callback is None:
-        return tool_error(
-            "read_window_below is only available in the Hermes desktop app."
-        )
-
-    try:
-        raw = callback()
-    except Exception as exc:
-        return tool_error(f"Failed to read the window below: {exc}")
-
-    if not raw:
-        return tool_error(
-            "Could not determine the window underneath (the desktop app did "
-            "not answer, or window enumeration is unavailable on this system)."
-        )
-    return passthrough_json(raw)
+    return read_pane(callback, (), (
+        "read_window_below is only available in the Hermes desktop app.",
+        "",
+        "Failed to read the window below: ",
+        "Could not determine the window underneath (the desktop app did "
+        "not answer, or window enumeration is unavailable on this system).",
+    ))
 
 
 READ_WINDOW_BELOW_SCHEMA = {
diff --git a/tools/shell_heredoc.py b/tools/shell_heredoc.py
index 9e64d6944c..c8fa8bb0c9 100644
--- a/tools/shell_heredoc.py
+++ b/tools/shell_heredoc.py
@@ -1,22 +1,17 @@
 """Conservative heredoc masking for shell-command scanners.
 
-Guards that scan raw command text (the foreground 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).
+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
-(commit 69c7663c6de6b6cb05bf99203fa39673efe01ccf).
+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.
 """
 
 from __future__ import annotations
@@ -62,8 +57,7 @@ def _mask_simple_quotes(command: str) -> str:
                 break
             result.append("''")
             cursor = closing + 1
-            continue
-        if char == '"':
+        elif char == '"':
             end = _span_end(command, cursor, '"')
             if not command[cursor:end].endswith('"'):
                 result.append(command[cursor:])
@@ -71,14 +65,13 @@ def _mask_simple_quotes(command: str) -> str:
             segment = command[cursor:end]
             result.append(segment if "$(" in segment or "`" in segment else '""')
             cursor = end
-            continue
-        if char == "`":
+        elif char == "`":
             end = _span_end(command, cursor, "`")
             result.append(command[cursor:end])
             cursor = end
-            continue
-        result.append(char)
-        cursor += 1
+        else:
+            result.append(char)
+            cursor += 1
     return "".join(result)
 
 
@@ -88,9 +81,8 @@ def _parse_heredoc_operator(command: str, index: int):
         return None
 
     cursor = index + 2
-    strip_tabs = False
-    if cursor < len(command) and command[cursor] == "-":
-        strip_tabs = True
+    strip_tabs = cursor < len(command) and command[cursor] == "-"
+    if strip_tabs:
         cursor += 1
     while cursor < len(command) and command[cursor] in " \t":
         cursor += 1
@@ -161,7 +153,6 @@ def _scan_heredoc_command_unit(command: str, start: int):
                 return cursor, specs, unknown_operator, has_list_operator
             cursor += 1
             continue
-
         if quote is not None:
             if quote in {'"', "`"} and char == "\\" and cursor + 1 < len(command):
                 cursor += 2
@@ -170,7 +161,6 @@ def _scan_heredoc_command_unit(command: str, start: int):
                 quote = None
             cursor += 1
             continue
-
         if char == "\\" and cursor + 1 < len(command):
             # Includes line continuations: the logical command keeps going.
             cursor += 2
@@ -206,12 +196,7 @@ 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:
@@ -237,9 +222,7 @@ 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: