refactor(tools): squeeze banner blanks, fold manifest read, compact module docstrings

tools/skills_sync_client.py

@@ -123,7 +123,6 @@ def sync_default_opt_in() -> bool:
 
 
 # Local skill eligibility + the personal opt-in flag
-
 def _skills_dir() -> Path:
     from hermes_constants import get_hermes_home
     return get_hermes_home() / "skills"
@@ -227,7 +226,6 @@ def _adopt_manifest_opt_ins(remote_manifest: Optional[Dict[str, bool]]) -> List[
 
 
 # Device label (commit ``author.device``; advisory, never an auth input)
-
 def _default_device_label() -> str:
     """Short hostname + random suffix (two machines can share a hostname); bare uuid if unusable."""
     import socket
@@ -275,7 +273,6 @@ def set_device_name(name: str) -> str:
 
 # Local sync STATE: last HEAD pushed/pulled + its root tree (FULL-digest namespace). Distinct from
 # the bundled manifest (skills_sync.py) and the plane's `sync-manifest`. ~/.hermes/skills/.sync_state.
-
 _EMPTY_STATE: Dict[str, Any] = {"head": None, "skills": {}}
 
 
@@ -324,7 +321,6 @@ def _record_head(state: Dict[str, Any], head: str, root: str) -> None:
 
 
 # Profile snapshot -- the root tree mirrors each skill's relative path (categories = intermediate trees).
-
 def snapshot_profile(skill_names: List[str], *, max_object_bytes: int = DEFAULT_MAX_OBJECT_BYTES,
                      ) -> Tuple[ObjectSet, str, Dict[str, str]]:
     """All objects for *skill_names* + profile root -> ``(objects, root_hash, {name: tree_hash})``.
@@ -350,7 +346,6 @@ def snapshot_profile(skill_names: List[str], *, max_object_bytes: int = DEFAULT_
 
 
 # Personal refs, push, pull
-
 def user_head_ref(owner: str) -> str:
     return f"refs/user/{owner}/HEAD"
 
@@ -500,7 +495,6 @@ def pull_skills(client: Optional[SyncClient] = None, *, identity: Optional[Dict[
 
 
 # Gated public entrypoints (gate-and-swallow, like maybe_run_curator): never raise; dict or None.
-
 def _gate_and_swallow(op: str, run: Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]):
     """Run *run(identity)* only if all gates hold (Nous admin, feature on, base URL); None if inert/error."""
     try:

tools/skills_sync_client_org.py

@@ -76,7 +76,6 @@ def _read_org_head(client: SyncClient, org_id: str) -> Optional[str]:
 
 
 # Local mirror sidecars
-
 def _mirror_root(org_id: str) -> Path:
     return _ssc()._org_dir() / org_id
 
@@ -189,7 +188,6 @@ def list_org_skill_names() -> List[str]:
 
 
 # Pull / propose
-
 def pull_org_skills(client: Optional[SyncClient] = None, *, identity: Optional[Dict[str, Any]] = None,
                     ) -> Dict[str, Any]:
     """Pull the org canonical set into the mirror (fast-forward only, no client merge). A skill with

tools/skills_sync_client_wire.py

@@ -44,7 +44,6 @@ SYNC_MANIFEST_VERSION = 1
 
 # Content addressing: the wire uses the FULL 64-hex sha256 -- a different namespace from the
 # truncated 16-hex local `content_hash` (skills_guard.py).
-
 def wire_address(data: bytes) -> str:
     """``sha256:<64-hex>`` -- the wire address of ``data``."""
     return "sha256:" + hashlib.sha256(data).hexdigest()
@@ -84,7 +83,6 @@ def parse_sync_manifest(data: bytes) -> Optional[Dict[str, bool]]:
 
 
 # Object building
-
 class ObjectSet:
     """Objects to push, ``hash -> (kind, bytes)``, deduped by content address."""
 
@@ -178,7 +176,6 @@ def assemble_root_from_skill_trees(skill_trees: Dict[str, str], objects: ObjectS
 
 
 # HTTP client (routes under /v1/sync/)
-
 class SyncError(RuntimeError):
     """A non-recoverable wire error (4xx the client can't retry)."""
 
@@ -288,7 +285,6 @@ class SyncClient:
 
 
 # Reading remote trees
-
 def read_ref_hash(client: SyncClient, ref: str, *, org_scope: bool = False) -> Optional[str]:
     """Hash of *ref* (queried with itself as prefix), or None if absent."""
     refs = client.get_refs(ref, org_scope=org_scope)
@@ -318,18 +314,11 @@ def skill_trees_of_root(client: SyncClient, root_tree_hash: str, *, org_scope: b
 def read_manifest_of_root(client: SyncClient, root_tree_hash: str) -> Optional[Dict[str, bool]]:
     """``{name: enabled}`` from the root ``sync-manifest`` blob (how a device learns another's opt-ins)."""
     try:
-        tree = client.get_tree_json(root_tree_hash)
+        for e in client.get_tree_json(root_tree_hash).get("entries", []):
+            if e.get("name") == SYNC_MANIFEST_ENTRY_NAME and e.get("kind") == KIND_BLOB:
+                return parse_sync_manifest(client.get_object(e["hash"])[1])
     except Exception as e:
-        logger.debug("skills_sync_client: manifest root read failed: %s", e)
-        return None
-    for e in tree.get("entries", []):
-        if e.get("name") == SYNC_MANIFEST_ENTRY_NAME and e.get("kind") == KIND_BLOB:
-            try:
-                _kind, data = client.get_object(e["hash"])
-            except Exception as ex:
-                logger.debug("skills_sync_client: manifest blob fetch failed: %s", ex)
-                return None
-            return parse_sync_manifest(data)
+        logger.debug("skills_sync_client: manifest read failed: %s", e)
     return None
 
 

tools/threat_patterns.py

@@ -1,14 +1,11 @@
-"""Shared threat-pattern library for context window security scanning.
-
-Single source of truth for prompt-injection / promptware / exfiltration patterns
-(``agent/prompt_builder.py``, ``tools/memory_tool.py``, ``agent/tool_dispatch_helpers.py``).
-Each pattern is ``(regex, pattern_id, scope)``. Scope is cumulative: ``"all"`` everywhere;
+"""Shared threat-pattern library (prompt injection / promptware / exfiltration) for
+``agent/prompt_builder.py``, ``tools/memory_tool.py`` and ``agent/tool_dispatch_helpers.py``.
+Each pattern is ``(regex, pattern_id, scope)``; scope is cumulative: ``"all"`` everywhere,
 ``"context"`` adds promptware / C2 / role hijack for context files, memory and tool results
-(warn-level: tool results carry content the user did not author); ``"strict"`` adds aggressive
-checks only for user-mediated writes (memory, skill installs) where a block is resolvable.
-New patterns must anchor on C2-specific vocabulary or unambiguous attack behavior, NOT bossy
-English ("you must" is common in legitimate AGENTS.md); filler is the bounded ``_FILLER``.
-"""
+(warn-level: that content is not user-authored), ``"strict"`` adds aggressive checks only for
+user-mediated writes (memory, skill installs) where a block is resolvable. New patterns must
+anchor on C2 vocabulary or unambiguous attack behavior, NOT bossy English ("you must" is common
+in legitimate AGENTS.md); filler between tokens is the bounded ``_FILLER``."""
 
 from __future__ import annotations
 

tools/tirith_security.py

@@ -1,13 +1,10 @@
-"""Tirith pre-exec security scanning wrapper.
-
-Runs the tirith binary as a subprocess to scan commands for content-level threats
-(homograph URLs, pipe-to-interpreter, terminal injection, ...). The exit code is the
-verdict source of truth (0 allow, 1 block, 2 warn); JSON stdout only enriches findings.
-Operational failures (spawn error, timeout, unknown exit) respect ``fail_open``;
-programming errors propagate. Auto-install: if tirith is not on PATH / the configured
-path it is downloaded from GitHub releases to $HERMES_HOME/bin/tirith in a background
-thread. SHA-256 is always verified; cosign provenance when cosign is on PATH.
-"""
+"""Tirith pre-exec security scanning wrapper: runs the tirith binary as a subprocess to scan
+commands for content-level threats (homograph URLs, pipe-to-interpreter, terminal injection).
+The exit code is the verdict source of truth (0 allow, 1 block, 2 warn); JSON stdout only
+enriches findings. Operational failures (spawn error, timeout, unknown exit) respect
+``fail_open``; programming errors propagate. Auto-install: a missing tirith is downloaded from
+GitHub releases to $HERMES_HOME/bin/tirith in a background thread -- SHA-256 always verified,
+cosign provenance when cosign is on PATH."""
 
 import hashlib
 import json
@@ -34,7 +31,6 @@ _COSIGN_IDENTITY_REGEXP = f"^https://github.com/{_REPO}/\\.github/workflows/rele
 _COSIGN_ISSUER = "https://token.actions.githubusercontent.com"
 
 # --- Config helpers ---
-
 def _env_bool(key: str, default: bool) -> bool:
     val = os.getenv(key)
     return default if val is None else val.lower() in {"1", "true", "yes"}
@@ -122,7 +118,6 @@ def _set_failed(reason: str) -> None:
 
 
 # --- Disk failure marker ---
-
 def _failure_marker_path() -> str:
     return os.path.join(str(get_hermes_home()), ".tirith-install-failed")
 
@@ -177,7 +172,6 @@ def _disk_marker_blocks_install() -> bool:
 
 
 # --- Auto-install ---
-
 def _hermes_bin_dir() -> str:
     """$HERMES_HOME/bin, created if needed."""
     os.makedirs(d := os.path.join(str(get_hermes_home()), "bin"), exist_ok=True)
@@ -212,8 +206,7 @@ def _download_file(url: str, dest: str, timeout: int = 10):
 
 
 def _verify_cosign(checksums_path: str, sig_path: str, cert_path: str) -> bool | None:
-    """Verify cosign provenance on checksums.txt: True verified, False rejected,
-    None when cosign is not on PATH / failed to execute."""
+    """Cosign provenance of checksums.txt: True verified, False rejected, None if cosign absent/failed."""
     if not (cosign := shutil.which("cosign")):
         logger.info("cosign not found on PATH")
         return None
@@ -349,7 +342,6 @@ def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]:
 
 
 # --- Path resolution ---
-
 def _is_executable(path: str) -> bool:
     return os.path.isfile(path) and os.access(path, os.X_OK)
 
@@ -361,10 +353,9 @@ def _find_local_tirith() -> str | None:
 
 
 def _resolve_locally(configured_path: str, *, warn_missing: bool) -> tuple[str | None, bool]:
-    """Network-free resolution shared by _resolve_tirith_path and ensure_installed ->
-    ``(path, may_install)``. ``path`` set = resolved (module state updated). Otherwise
-    ``may_install`` is False when the miss is terminal (explicit path missing, cached
-    non-retryable failure) and True when the disk marker / install step may proceed."""
+    """Network-free resolution -> ``(path, may_install)``: ``path`` set = resolved (module state
+    updated); else ``may_install`` False = terminal miss (explicit path missing, cached non-retryable
+    failure), True = the disk marker / install step may proceed."""
     global _resolved_path, _install_failure_reason
     expanded = os.path.expanduser(configured_path)
     # An explicit (non-"tirith") path is authoritative: never auto-download a replacement.
@@ -404,10 +395,9 @@ def _record_install_result(installed: str | None, reason: str) -> None:
 
 
 def _resolve_tirith_path(configured_path: str) -> str:
-    """Resolve the tirith binary path, auto-installing synchronously if necessary.
-    Default "tirith": PATH → $HERMES_HOME/bin/tirith → auto-install; failed installs are
-    cached for the process (and on disk for 24h). On any miss the expanded configured path
-    is returned so the spawn fails open via the dedupe'd OSError handler."""
+    """Resolve the tirith path, auto-installing synchronously if needed (default "tirith": PATH →
+    $HERMES_HOME/bin/tirith → install; failures cached in-process and on disk for 24h). On a miss
+    the expanded configured path is returned so the spawn fails open via the dedupe'd OSError."""
     if cached := _cached_path():
         return cached
     expanded = os.path.expanduser(configured_path)
@@ -440,9 +430,8 @@ def _background_install(*, log_failures: bool = True):
 
 
 def ensure_installed(*, log_failures: bool = True):
-    """Ensure tirith is available, downloading in a daemon thread if needed. Local checks
-    are synchronous; the download never blocks startup. Returns the resolved path if
-    available now, else None. Safe to call repeatedly."""
+    """Resolved path if available now, else None after kicking off a daemon-thread download (local
+    checks are synchronous; the download never blocks startup). Safe to call repeatedly."""
     global _install_thread
     cfg = _load_security_config()
     if not cfg["tirith_enabled"]:
@@ -465,7 +454,6 @@ def ensure_installed(*, log_failures: bool = True):
 
 
 # --- Main API ---
-
 _MAX_FINDINGS = 50
 _MAX_SUMMARY_LEN = 500
 _EXIT_ACTIONS = {0: "allow", 1: "block", 2: "warn"}