Stripping ANSI codes from /journey and slash-command output

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #56641 fix(journey): crash on non-dict skill metadata + ANSI leaks in CLI/desktop

  merged to main

  files: agent/learning_graph.py, cli.py, hermes_cli/cli_commands_mixin.py, hermes_cli/journey.py, tests/agent/test_learning_graph.py, tests/hermes_cli/test_journey_render.py, tests/tui_gateway/test_slash_worker_ansi.py, tui_gateway/slash_worker.py

  Show the diff

  • agent/learning_graph.py

    @@ -48,8 +48,16 @@ def _frontmatter(text: str) -> dict[str, Any]:
             return {}
     
     
    +def _hermes_meta(fm: dict[str, Any]) -> dict[str, Any]:
    +    """``metadata.hermes`` as a dict, tolerant of the string-valued frontmatter
    +    that ``parse_frontmatter``'s malformed-YAML fallback produces."""
    +    meta = fm.get("metadata")
    +    hermes = meta.get("hermes") if isinstance(meta, dict) else None
    +    return hermes if isinstance(hermes, dict) else {}
    +
    +
     def _related(fm: dict[str, Any]) -> list[str]:
    -    raw = fm.get("related_skills") or (fm.get("metadata", {}).get("hermes", {}) or {}).get("related_skills")
    +    raw = fm.get("related_skills") or _hermes_meta(fm).get("related_skills")
         if isinstance(raw, list):
             return [str(r).strip() for r in raw if str(r).strip()]
         if isinstance(raw, str):
    @@ -58,7 +66,7 @@ def _related(fm: dict[str, Any]) -> list[str]:
     
     
     def _category(fm: dict[str, Any], skill_md: Path) -> str:
    -    cat = fm.get("category") or (fm.get("metadata", {}).get("hermes", {}) or {}).get("category")
    +    cat = fm.get("category") or _hermes_meta(fm).get("category")
         if cat:
             return str(cat)
         # …/skills/<category>/<skill>/SKILL.md

  • cli.py

    @@ -8640,21 +8640,7 @@ def process_command(self, command: str) -> bool:
             elif canonical == "agents":
                 self._handle_agents_command()
             elif canonical == "journey":
    -            try:
    -                import argparse
    -                import shlex
    -
    -                from hermes_cli.journey import register_cli as _register_journey_cli
    -
    -                parser = argparse.ArgumentParser(prog="/journey", add_help=False)
    -                _register_journey_cli(parser)
    -                argv = shlex.split(cmd_original.split(None, 1)[1]) if len(cmd_original.split(None, 1)) > 1 else []
    -                args = parser.parse_args(argv)
    -                args.func(args)
    -            except SystemExit:
    -                pass
    -            except Exception as exc:
    -                _cprint(f"  /journey failed: {exc}")
    +            self._handle_journey_command(cmd_original)
             elif canonical == "background":
                 self._handle_background_command(cmd_original)
             elif canonical == "queue":

  • hermes_cli/cli_commands_mixin.py

    @@ -294,6 +294,43 @@ def _handle_agents_command(self):
             agent_running = getattr(self, "_agent_running", False)
             _cprint(f"  Agent: {'running' if agent_running else 'idle'}")
     
    +    def _handle_journey_command(self, cmd_original: str) -> None:
    +        """Handle /journey — the learning timeline (see `hermes journey`).
    +
    +        The read-only views (default + ``list``) render Rich color, which
    +        patch_stdout would swallow as raw escapes; capture with forced ANSI and
    +        re-emit through ``_cprint``. ``delete``/``edit`` are interactive
    +        (confirm prompt / ``$EDITOR``) so they keep the real stdio.
    +        """
    +        import argparse
    +        import io
    +        import shlex
    +        from contextlib import redirect_stdout
    +
    +        from cli import _cprint
    +        from hermes_cli.journey import register_cli
    +
    +        parser = argparse.ArgumentParser(prog="/journey", add_help=False)
    +        register_cli(parser)
    +        rest = cmd_original.split(None, 1)
    +        try:
    +            args = parser.parse_args(shlex.split(rest[1]) if len(rest) > 1 else [])
    +        except SystemExit:
    +            return
    +
    +        interactive = getattr(args, "journey_action", None) in ("delete", "edit")
    +        try:
    +            if interactive:
    +                args.func(args)
    +                return
    +            args.force_color = True
    +            buf = io.StringIO()
    +            with redirect_stdout(buf):
    +                args.func(args)
    +            _cprint(buf.getvalue().rstrip("\n"))
    +        except Exception as exc:
    +            _cprint(f"  /journey failed: {exc}")
    +
         def _handle_paste_command(self):
             """Handle /paste — explicitly check clipboard for an image.
     

  • hermes_cli/journey.py

    @@ -171,6 +171,17 @@ def label_row(item) -> Text:
         return Group(*parts)
     
     
    +def _console(*, color: bool, width: Optional[int] = None, force: bool = False):
    +    """A Rich console. ``force`` emits truecolor ANSI even into a captured
    +    stream — the interactive CLI grabs that output and re-renders it through
    +    prompt_toolkit (raw escapes to a real terminal would otherwise be
    +    swallowed). Mirrors the ``ChatConsole`` idiom in ``cli.py``."""
    +    from rich.console import Console
    +
    +    extra = {"force_terminal": True, "color_system": "truecolor"} if force else {}
    +    return Console(no_color=not color, width=width, **extra)
    +
    +
     def _cmd_show(args: argparse.Namespace) -> int:
         from rich.console import Console
     
    @@ -183,7 +194,7 @@ def _cmd_show(args: argparse.Namespace) -> int:
         payload = _build_payload()
         color = not bool(getattr(args, "no_color", False))
         cols, rows = _term_size(getattr(args, "width", None), getattr(args, "height", None))
    -    console = Console(no_color=not color, width=cols)
    +    console = _console(color=color, width=cols, force=bool(getattr(args, "force_color", False)))
     
         if not payload.get("nodes"):
             console.print(
    @@ -226,11 +237,9 @@ def _clamp(v: float, lo: float, hi: float) -> float:
     
     
     def _cmd_list(args: argparse.Namespace) -> int:
    -    from rich.console import Console
    -
         from agent.learning_graph_render import format_date
     
    -    console = Console(no_color=bool(getattr(args, "no_color", False)))
    +    console = _console(color=not bool(getattr(args, "no_color", False)), force=bool(getattr(args, "force_color", False)))
         nodes = sorted(_build_payload().get("nodes", []), key=lambda n: n.get("timestamp") or 0)
         if not nodes:
             console.print("[grey62]No learning yet.[/grey62]")
    @@ -315,6 +324,8 @@ def register_cli(parent: argparse.ArgumentParser) -> None:
         parent.add_argument("--width", type=int, default=None, help="Override render width in columns.")
         parent.add_argument("--height", type=int, default=None, help="Override render height in rows.")
         parent.add_argument("--no-color", action="store_true", help="Disable color output.")
    +    # Force ANSI even when stdout is captured — the interactive CLI re-renders it.
    +    parent.add_argument("--force-color", action="store_true", help=argparse.SUPPRESS)
         parent.add_argument("--json", action="store_true", help="Print the raw graph payload as JSON and exit.")
         parent.set_defaults(func=_cmd_show)
     
    @@ -322,6 +333,7 @@ def register_cli(parent: argparse.ArgumentParser) -> None:
     
         p_list = sub.add_parser("list", help="List node ids (for delete/edit).")
         p_list.add_argument("--no-color", action="store_true")
    +    p_list.add_argument("--force-color", action="store_true", help=argparse.SUPPRESS)
         p_list.set_defaults(func=_cmd_list)
     
         p_del = sub.add_parser("delete", help="Delete a learned skill (archived) or memory by node id.")

  • tests/agent/test_learning_graph.py

    @@ -88,6 +88,30 @@ def test_memory_is_cards_split_on_separator(tmp_path):
         assert any(n["kind"] == "memory" for n in graph["nodes"])
     
     
    +def test_malformed_frontmatter_metadata_does_not_crash(tmp_path):
    +    """``parse_frontmatter``'s malformed-YAML fallback stores every value as a
    +    string, so ``metadata`` can be a str. The graph must tolerate that instead
    +    of crashing on chained ``.get()`` (the /journey base-CLI crash)."""
    +    skill_dir = tmp_path / "skills" / "misc" / "bad-skill"
    +    skill_dir.mkdir(parents=True)
    +    # The unterminated quote makes yaml_load raise → fallback → metadata is a str.
    +    skill_dir.joinpath("SKILL.md").write_text(
    +        '---\nname: bad-skill\nmetadata: not-a-dict\ndescription: "oops\n---\n# Bad\n',
    +        encoding="utf-8",
    +    )
    +
    +    node = learning_graph.build_skill_nodes([("profile", tmp_path / "skills")])["bad-skill"]
    +
    +    assert node.category == "misc"  # directory fallback, not a crash
    +    assert node.related == []
    +
    +
    +def test_hermes_meta_tolerates_non_dict():
    +    assert learning_graph._hermes_meta({"metadata": "junk"}) == {}
    +    assert learning_graph._hermes_meta({"metadata": {"hermes": "junk"}}) == {}
    +    assert learning_graph._hermes_meta({"metadata": {"hermes": {"category": "x"}}}) == {"category": "x"}
    +
    +
     def test_full_payload_shape_and_edge_integrity(tmp_path):
         home = tmp_path / ".hermes"
         home.mkdir()

  • tests/hermes_cli/test_journey_render.py

    @@ -0,0 +1,38 @@
    +"""Behavior contracts for /journey output routing.
    +
    +The interactive CLI captures Rich output and re-renders it through
    +prompt_toolkit, so it needs forced ANSI (``--force-color``); chat surfaces
    +render plain text, so the default captured path must stay escape-free.
    +"""
    +
    +from __future__ import annotations
    +
    +import argparse
    +import contextlib
    +import io
    +
    +
    +def _capture(argv: list[str], *, force: bool) -> str:
    +    from hermes_cli.journey import register_cli
    +
    +    parser = argparse.ArgumentParser(add_help=False)
    +    register_cli(parser)
    +    args = parser.parse_args(argv)
    +    if force:
    +        args.force_color = True
    +
    +    buf = io.StringIO()
    +    with contextlib.redirect_stdout(buf):
    +        args.func(args)
    +    return buf.getvalue()
    +
    +
    +def test_force_color_emits_ansi_for_reemission():
    +    assert "\x1b[" in _capture([], force=True)
    +    assert "\x1b[" in _capture(["list"], force=True)
    +
    +
    +def test_default_capture_is_plain_for_chat_bubbles():
    +    # Rich auto-detects the StringIO as non-tty → no color, no raw escapes.
    +    assert "\x1b[" not in _capture([], force=False)
    +    assert "\x1b[" not in _capture(["list"], force=False)

  • tests/tui_gateway/test_slash_worker_ansi.py

    @@ -0,0 +1,23 @@
    +"""The slash worker feeds desktop chat bubbles, which render plain text — so
    +any ANSI a worker-routed command emits (e.g. /journey's own Rich Console) must
    +be stripped from the worker's return value."""
    +
    +from __future__ import annotations
    +
    +
    +class _FakeCLI:
    +    console = None
    +
    +    def process_command(self, cmd: str) -> None:
    +        import sys
    +
    +        sys.stdout.write("\x1b[38;2;1;2;3mcolored\x1b[0m plain")
    +
    +
    +def test_run_strips_ansi_from_output():
    +    from tui_gateway import slash_worker
    +
    +    out = slash_worker._run(_FakeCLI(), "/anything")
    +
    +    assert "\x1b[" not in out
    +    assert out == "colored plain"

  • tui_gateway/slash_worker.py

    @@ -102,7 +102,14 @@ def _run(cli: HermesCLI, command: str) -> str:
             if old is not None:
                 cli_mod._cprint = old
     
    -    return buf.getvalue().rstrip()
    +    # Desktop chat bubbles render plain text, not ANSI. A worker-routed command
    +    # that emits Rich color (e.g. /journey building its own Console, which picks
    +    # up truecolor from the gateway's inherited COLORTERM) would otherwise leak
    +    # raw escapes; strip them at the single choke point. (The TUI opens /journey
    +    # as an overlay, so it never travels this path.)
    +    from tools.ansi_strip import strip_ansi
    +
    +    return strip_ansi(buf.getvalue().rstrip())
     
     
     def main():