refactor(hermes_cli): group A — body blank-line squeeze, slash tap/diff-line helpers

This commit is contained in:
Teknium
2026-09-02 21:34:25 -07:00
parent 2bc919015b
commit b69d922d78
7 changed files with 15 additions and 59 deletions
-2
View File
@@ -89,7 +89,6 @@ def _select_platform() -> Optional[str]:
def _toggle_by_category(skills: List[dict], disabled: Set[str]) -> Set[str]:
"""Toggle all skills in a category at once."""
from hermes_cli.curses_ui import curses_checklist
categories = _get_categories(skills)
cat_skills = [{s["name"] for s in skills if (s["category"] or "uncategorized") == cat}
for cat in categories]
@@ -111,7 +110,6 @@ def _toggle_by_category(skills: List[dict], disabled: Set[str]) -> Set[str]:
def skills_command(args=None):
"""Entry point for `hermes skills`."""
from hermes_cli.curses_ui import curses_checklist
config = load_config()
skills = _list_all_skills()
if not skills:
+15 -30
View File
@@ -362,12 +362,10 @@ def _fetch_browse_results(c: Console, source: str):
# parallel_search_sources invokes the callback from the collecting thread as each
# source completes; the page itself is rendered once over the final, fully sorted set.
_done: List[str] = []
def _on_source_done(sid: str, count: int) -> None:
_done.append(f"{sid} ({count})")
status.update(
f"[bold]Fetching skills from registries...[/] [dim]done: {', '.join(_done)}[/]")
return parallel_search_sources(
_sources(), query="", per_source_limits=_BROWSE_LIMITS, source_filter=source,
overall_timeout=30, on_source_done=_on_source_done)
@@ -467,7 +465,6 @@ def do_inspect(identifier: str, console: Optional[Console] = None) -> None:
if not meta:
_print_error(c, f"Could not find '{identifier}' in any source.")
return
c.print()
info_lines = [
f"[bold]Name:[/] {meta.name}",
@@ -487,11 +484,9 @@ def do_inspect(identifier: str, console: Optional[Console] = None) -> None:
def inspect_skill(identifier: str) -> Optional[dict]:
"""Skill metadata (+ SKILL.md preview) for programmatic callers."""
class _Q:
def print(self, *a, **k):
pass
ident, meta, bundle, _ = _resolve_identifier(identifier, _sources(), _Q())
if not ident or not meta:
return None
@@ -1002,15 +997,12 @@ def do_list_modified(console: Optional[Console] = None, as_json: bool = False) -
def _print_diff_line(c: Console, line: str) -> None:
"""Unified-diff line with light coloring."""
if line.startswith("+") and not line.startswith("+++"):
c.print(f"[green]{line}[/]")
elif line.startswith("-") and not line.startswith("---"):
c.print(f"[red]{line}[/]")
elif line.startswith("@@"):
c.print(f"[cyan]{line}[/]")
else:
c.print(line, highlight=False)
"""Unified-diff line with light coloring (file headers +++/--- stay plain)."""
for prefix, style in (("+", "green"), ("-", "red"), ("@@", "cyan")):
if line.startswith(prefix) and not line.startswith(prefix * 3):
c.print(f"[{style}]{line}[/]")
return
c.print(line, highlight=False)
def do_diff(name: str, console: Optional[Console] = None) -> None:
@@ -1302,12 +1294,10 @@ def do_snapshot_export(output_path: str, console: Optional[Console] = None) -> N
"hermes_version": "0.1.0",
"exported_at": datetime.now(timezone.utc).isoformat(),
"skills": [
{
"name": entry["name"],
"source": entry.get("source", ""),
"identifier": entry.get("identifier", ""),
"category": str(Path(entry.get("install_path", "")).parent)
if "/" in entry.get("install_path", "") else ""}
{"name": entry["name"], "source": entry.get("source", ""),
"identifier": entry.get("identifier", ""),
"category": str(Path(entry.get("install_path", "")).parent)
if "/" in entry.get("install_path", "") else ""}
for entry in installed],
"taps": tap_list}
payload = json.dumps(snapshot, indent=2, ensure_ascii=False) + "\n"
@@ -1465,22 +1455,16 @@ def _slash_search(args, c):
def _slash_snapshot(args, c):
if args and args[0] == "export" and len(args) > 1:
if len(args) > 1 and args[0] == "export":
do_snapshot_export(args[1], console=c)
elif args and args[0] == "import" and len(args) > 1:
elif len(args) > 1 and args[0] == "import":
do_snapshot_import(args[1], force="--force" in args, console=c)
else:
c.print("[bold red]Usage:[/] /skills snapshot export <file> | /skills snapshot import <file>\n")
def _slash_tap(args, c):
if not args:
do_tap("list", console=c)
return
do_tap(args[0], repo=args[1] if len(args) > 1 else "", console=c)
def _first_positional(args):
"""First argument unless it is a flag (audit's historical parse)."""
return args[0] if args and not args[0].startswith("--") else None
@@ -1517,7 +1501,8 @@ _SLASH_ACTIONS = {
args[0], target=_opt_value(args, "--to", "github", last=True),
repo=_opt_value(args, "--repo", "", last=True), console=c),
"snapshot": _slash_snapshot,
"tap": _slash_tap,
"tap": lambda args, c: (do_tap(args[0], repo=args[1] if len(args) > 1 else "", console=c) if args
else do_tap("list", console=c)),
"help": lambda args, c: _print_skills_help(c),
"--help": lambda args, c: _print_skills_help(c),
"-h": lambda args, c: _print_skills_help(c)}
-6
View File
@@ -23,7 +23,6 @@ def _skins_dir() -> Path:
def _active_skin() -> str:
from hermes_cli.config import load_config
display = (load_config() or {}).get("display") or {}
return str(display.get("skin") or "default")
@@ -31,13 +30,11 @@ def _active_skin() -> str:
def _use(name: str) -> None:
"""Activate a skin (persists display.skin via the shared config writer)."""
from hermes_cli.config import config_command
config_command(argparse.Namespace(config_command="set", key="display.skin", value=name, force=True))
def _skin_set(key: str, value: str, skin: str | None) -> int:
import yaml
if not _HEX_RE.match(value):
print(f"✗ {value!r} is not a #rrggbb hex color", file=sys.stderr)
return 1
@@ -52,7 +49,6 @@ def _skin_set(key: str, value: str, skin: str | None) -> int:
# Built-in (or missing): fork into an editable copy that keeps its full palette, under a
# fresh name so the built-in stays intact for revert.
from hermes_cli.skin_engine import load_skin
resolved = load_skin(name)
target = f"{name}-custom"
path = _skins_dir() / f"{target}.yaml"
@@ -66,7 +62,6 @@ def _skin_set(key: str, value: str, skin: str | None) -> int:
# Atomic write: write_text truncates with no fsync; safe_load("") → None → {} would
# permanently lose the palette on the next set.
from utils import atomic_yaml_write
atomic_yaml_write(path, data, sort_keys=False)
if target != name:
_use(target)
@@ -76,7 +71,6 @@ def _skin_set(key: str, value: str, skin: str | None) -> int:
def _skin_list() -> int:
from hermes_cli.skin_engine import list_skins
active = _active_skin()
for s in list_skins():
mark = "*" if s["name"] == active else " "
-3
View File
@@ -369,7 +369,6 @@ def _build_skin_config(data: Dict[str, Any]) -> SkinConfig:
"""Build a SkinConfig from a raw dict (built-in or loaded from YAML)."""
default = _BUILTIN_SKINS["default"]
skin_name = str(data.get("name", "unknown"))
def section(key: str) -> Dict[str, Any]:
value = data.get(key)
if isinstance(value, dict):
@@ -378,10 +377,8 @@ def _build_skin_config(data: Dict[str, Any]) -> SkinConfig:
logger.warning("Skin '%s' has invalid '%s' section type (%s); ignoring section",
skin_name, key, type(value).__name__)
return {}
def merged(key: str) -> Dict[str, Any]:
return {**default.get(key, {}), **section(key)}
# Paired palettes are NOT merged over the default skin's blocks: an empty block means
# "no hand-tuned variant for that polarity" and consumers (the TUI) fall back to `colors`
# + automatic adaptation, which beats the default's gold light palette under a crimson skin.
-3
View File
@@ -19,7 +19,6 @@ def _build_full_manifest(
UI after pasting.
"""
from hermes_cli.commands import slack_app_manifest
if messaging_experience is None:
messaging_experience = "assistant" if include_assistant else "none"
messaging_experience = str(messaging_experience).strip().lower()
@@ -115,7 +114,6 @@ def slack_manifest_command(args) -> int:
if slashes_only:
from hermes_cli.commands import slack_app_manifest
manifest = slack_app_manifest()["features"]["slash_commands"]
else:
manifest = _build_full_manifest(
@@ -129,7 +127,6 @@ def slack_manifest_command(args) -> int:
return 0
if isinstance(write_target, bool) and write_target: # bare --write → default location
from hermes_constants import get_hermes_home
target = Path(get_hermes_home()) / "slack-manifest.json"
else:
target = Path(write_target).expanduser()
-11
View File
@@ -18,7 +18,6 @@ __all__ = ["CommandContext", "CommandReply", "EXECUTORS", "execute_command", "re
@dataclass(frozen=True)
class CommandContext:
"""Surface-provided inputs for a shared command executor."""
surface: str = "cli" # "cli" | "gateway" | "tui" — decoration only
args: str = "" # raw argument string after the command word
options: Mapping[str, Any] = field(default_factory=dict) # surface params (page_size, ...)
@@ -29,7 +28,6 @@ class CommandContext:
class CommandReply:
"""Canonical result of a shared executor: surface-independent ``text`` plus the structured
``data`` it derived so a surface can re-render with its own decoration."""
text: str
data: Mapping[str, Any] = field(default_factory=dict)
format: str = "plain" # "plain" | "markdown" (hint, not a contract)
@@ -39,14 +37,12 @@ class CommandReply:
def _exec_version(ctx: CommandContext) -> CommandReply:
"""Core /version text — the banner version label."""
from hermes_cli.banner import format_banner_version_label
return CommandReply(format_banner_version_label())
def _exec_egress(ctx: CommandContext) -> CommandReply:
"""Core /egress text — Docker egress proxy status."""
from hermes_cli.proxy_cli import format_status_text
return CommandReply(format_status_text())
@@ -60,17 +56,14 @@ def _exec_profile(ctx: CommandContext) -> CommandReply:
home_display = str(ctx.options.get("home_display") or "").strip()
if not profile_name:
from hermes_cli.profiles import get_active_profile_name
profile_name = get_active_profile_name()
if not home_display:
from hermes_constants import display_hermes_home
home_display = display_hermes_home()
# Presentation-only display name (profile.yaml); `data.profile` stays the canonical id.
label = profile_name
try:
from hermes_cli.profiles import format_profile_label, get_profile_dir, read_profile_meta
display = read_profile_meta(get_profile_dir(profile_name)).get("display_name", "")
label = format_profile_label(profile_name, display)
except Exception:
@@ -107,7 +100,6 @@ def _skill_commands() -> dict:
"""Registered skill commands, or ``{}`` when the skill subsystem is unavailable."""
try:
from agent.skill_commands import get_skill_commands
return get_skill_commands() or {}
except Exception:
return {}
@@ -117,7 +109,6 @@ def _exec_help(ctx: CommandContext) -> CommandReply:
"""Core gateway /help body (pre platform mention decoration)."""
from agent.i18n import t
from hermes_cli.commands import gateway_help_lines
lines = [t("gateway.help.header"), *gateway_help_lines()]
skill_cmds = _skill_commands()
try:
@@ -140,7 +131,6 @@ def _exec_commands(ctx: CommandContext) -> CommandReply:
"""
from agent.i18n import t
from hermes_cli.commands import gateway_help_lines
raw_args = (ctx.args or "").strip()
if raw_args:
try:
@@ -209,7 +199,6 @@ def run_execute(cmd_def: Any, ctx: CommandContext) -> CommandReply | None:
def execute_command(name: str, ctx: CommandContext) -> CommandReply:
"""Run the shared executor for ``name``; ``LookupError`` when unknown or not migrated."""
from hermes_cli.commands import resolve_command
cmd_def = resolve_command(name)
reply = run_execute(cmd_def, ctx) if cmd_def is not None else None
if reply is None:
-4
View File
@@ -32,7 +32,6 @@ def _resolve_origin() -> Optional[Dict[str, Any]]:
accepted job delivers back to the accepting chat; None lets create_job use the home channel."""
try:
from gateway.session_context import get_session_env
platform = get_session_env("HERMES_SESSION_PLATFORM")
chat_id = get_session_env("HERMES_SESSION_CHAT_ID")
if platform and chat_id:
@@ -50,7 +49,6 @@ def _accept(store, rest: str, origin, surface: str) -> str:
if not rest:
return "Usage: /suggestions accept <number|id>"
from cron.scheduler import CronSchedulerRegistrationError
try:
job = store.accept_suggestion(rest, origin=origin)
except CronSchedulerRegistrationError as e:
@@ -75,7 +73,6 @@ def _dismiss(store, rest: str, origin, surface: str) -> str:
def _catalog(store, rest: str, origin, surface: str) -> str:
try:
from cron.suggestion_catalog import seed_catalog_suggestions
created = seed_catalog_suggestions()
except Exception as e:
logger.debug("catalog seed failed: %s", e)
@@ -117,7 +114,6 @@ def handle_suggestions_command(
except Exception as e: # pragma: no cover - import guard
logger.debug("suggestions store import failed: %s", e)
return "Suggestions are unavailable in this build."
parts = (args or "").strip().split()
handler = _SUBCOMMANDS.get(parts[0].lower() if parts else "")
if handler is None: