hermes-agent/hermes_cli/journey.py

"""``hermes journey`` — what Hermes has learned, on a timeline."""

from __future__ import annotations

import argparse
import shutil
import sys
import time
from functools import lru_cache
from typing import Any, Optional

_TITLE_COLOR = "#E8C463"
_CHARTED_SIGNAL_MIN_CONTRAST = 4.5


def _build_payload() -> dict[str, Any]:
    from agent.learning_graph import build_learning_graph

    return build_learning_graph()


@lru_cache(maxsize=1)
def _primary_hex() -> str:
    """The active skin's primary color (mirrors the TUI theme primary)."""
    try:
        from hermes_cli.skin_engine import get_active_skin

        skin = get_active_skin()
        return skin.get_color("ui_primary", "") or skin.get_color("banner_title", "#FFD700")
    except Exception:
        return "#FFD700"


@lru_cache(maxsize=1)
def _palette() -> dict[str, str]:
    from agent.learning_graph_render import derive_palette

    return derive_palette(_primary_hex(), dark=True)


def _fade(base: Optional[str], alpha: float) -> Optional[str]:
    from agent.learning_graph_render import hex_to_rgb, mix_rgb, rgb_to_hex

    if not base:
        return None
    if alpha >= 0.999:
        return base
    return rgb_to_hex(mix_rgb(hex_to_rgb(_palette()["bg"]), hex_to_rgb(base), alpha))


def _resolve(style: str, alpha: float) -> Optional[str]:
    """Fade the style's base ink toward the background by ``alpha`` (rgba-over-bg)."""
    return _fade(_palette().get(style), alpha)


def _relative_luminance(rgb: tuple[int, int, int]) -> float:
    def channel(value: int) -> float:
        n = value / 255
        return n / 12.92 if n <= 0.03928 else ((n + 0.055) / 1.055) ** 2.4

    red, green, blue = (channel(value) for value in rgb)
    return 0.2126 * red + 0.7152 * green + 0.0722 * blue


def _contrast_ratio(foreground: tuple[int, int, int], background: tuple[int, int, int]) -> float:
    high, low = sorted((_relative_luminance(foreground), _relative_luminance(background)), reverse=True)
    return (high + 0.05) / (low + 0.05)


def _ensure_contrast(color: Optional[str], background: str, minimum: float) -> Optional[str]:
    """Lift a foreground toward the readable pole until it clears ``minimum``."""
    if not color:
        return None
    from agent.learning_graph_render import hex_to_rgb, mix_rgb, rgb_to_hex

    foreground_rgb = hex_to_rgb(color)
    background_rgb = hex_to_rgb(background)
    if _contrast_ratio(foreground_rgb, background_rgb) >= minimum:
        return color
    pole = (0, 0, 0) if _relative_luminance(background_rgb) > 0.5 else (255, 255, 255)
    for step in range(1, 21):
        candidate = mix_rgb(foreground_rgb, pole, step * 0.05)
        if _contrast_ratio(candidate, background_rgb) >= minimum:
            return rgb_to_hex(candidate)
    return rgb_to_hex(pole)


def _resolve_charted_signal(style: str, alpha: float) -> Optional[str]:
    """Keep age tinting without allowing explanatory labels to disappear."""
    return _ensure_contrast(_resolve(style, alpha), _palette()["bg"], _CHARTED_SIGNAL_MIN_CONTRAST)


def _row_to_text(row: list, color: bool):
    from rich.text import Text

    text = Text()
    for run in row:
        chunk, style = run[0], run[1]
        alpha = run[2] if len(run) > 2 else 1.0
        override = run[3] if len(run) > 3 else None
        if not color:
            text.append(chunk)
        elif override:
            text.append(chunk, style=_fade(override, alpha))
        else:
            text.append(chunk, style=_resolve(style, alpha))
    return text


def _term_size(width: Optional[int], height: Optional[int]) -> tuple[int, int]:
    size = shutil.get_terminal_size((90, 30))
    return max(40, width or size.columns), max(10, height or size.lines)


def _frame_renderable(payload, *, cols, rows, reveal, color):
    from rich.console import Group
    from rich.text import Text

    from agent import learning_graph_render as render

    legend = render.build_legend(payload)
    categories = render.category_legend(payload)
    summary = render.build_summary(payload)
    axis = render.axis_labels(payload)
    # Lines are pad_left(2), so content must fit in cols-2.
    inner = max(24, cols - 2)
    # Reserve rows for title/legend/blank/axis/footer/labels + summary; field gets rest.
    field_rows = max(6, rows - 10 - len(summary))
    frame = render.render_graph(payload, cols=inner, rows=field_rows, reveal=reveal)
    count = len(payload.get("nodes", []))

    def st(style: Optional[str]) -> Optional[str]:
        return style if color else None

    title = Text()
    title.append("✦ Journey ", style=st(f"bold {_TITLE_COLOR}"))
    title.append("· learned skills & memories over time", style=st("grey62"))
    parts: list[Any] = [title]

    legend_line = Text("  ")
    for i, item in enumerate(legend):
        if i:
            legend_line.append("   ")
        legend_line.append(item["glyph"] + " ", style=_resolve(item["style"], 1.0) if color else None)
        legend_line.append(item["label"], style=st("grey62"))
    parts.append(legend_line)

    if categories:
        cat_line = Text("  ")
        for i, item in enumerate(categories):
            if i:
                cat_line.append("  ")
            cat_line.append(item["glyph"] + " ", style=_fade(item.get("color"), 1.0) if color else None)
            cat_line.append(item["label"], style=st("grey54"))
        parts.append(cat_line)

    parts.append(Text(""))
    for grow in frame["grid"]:
        line = _row_to_text(grow, color)
        line.pad_left(2)
        parts.append(line)

    # Date axis under the field (oldest → now), with the playhead date centered.
    axis_line = Text("  ")
    axis_line.append(axis["start"], style=st("grey54"))
    axis_line.append(" " * max(1, inner - len(axis["start"]) - len(axis["end"])))
    axis_line.append(axis["end"], style=st("grey54"))
    parts.append(axis_line)

    pct = int(round(reveal * 100))
    foot = Text("  ")
    foot.append("◷ ", style=st("grey54"))
    foot.append(frame["date"] or "—", style=st(_TITLE_COLOR))
    foot.append(f"   {frame['visible']}/{count} revealed · {pct}%", style=st("grey54"))
    parts.append(foot)

    labels = frame.get("labels", [])
    if labels:
        parts.append(Text(""))
        parts.append(Text("  charted signals", style=st("grey62")))
        for item in labels[:6]:
            row = Text("  ")
            row.append(f"{item['key']} ", style=st("grey70"))
            signal_style = (
                _resolve_charted_signal(item["style"], float(item.get("alpha", 1.0))) if color else None
            )
            row.append(f"{item['glyph']} ", style=signal_style)
            row.append(str(item["label"]), style=signal_style)
            meta = str(item["meta"])
            row.append(f"  {meta if len(meta) <= 32 else meta[:29] + '…'}", style=st("grey54"))
            parts.append(row)

    for line_text in summary:
        parts.append(Text("  " + line_text, style=st("grey62")))
    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

    if getattr(args, "json", False):
        import json

        Console(no_color=bool(getattr(args, "no_color", False))).print_json(json.dumps(_build_payload()))
        return 0

    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(color=color, width=cols, force=bool(getattr(args, "force_color", False)))

    if not payload.get("nodes"):
        console.print(
            "[grey62]No learning yet — use Hermes a while and your learned skills and "
            "memories will start mapping out here.[/grey62]"
        )
        return 0
    if getattr(args, "play", False):
        return _play(console, payload, cols=cols, rows=rows, color=color, fps=getattr(args, "fps", 12))
    reveal = _clamp(float(getattr(args, "reveal", 1.0) or 1.0), 0.0, 1.0)
    console.print(_frame_renderable(payload, cols=cols, rows=rows, reveal=reveal, color=color))
    return 0


def _play(console, payload, *, cols, rows, color, fps: int) -> int:
    from rich.live import Live

    frames = 42
    delay = 1.0 / max(1, min(60, fps))
    try:
        with Live(console=console, refresh_per_second=max(1, fps), screen=False) as live:
            for i in range(frames):
                reveal = i / (frames - 1)
                live.update(_frame_renderable(payload, cols=cols, rows=rows, reveal=reveal, color=color))
                time.sleep(delay)
            live.update(_frame_renderable(payload, cols=cols, rows=rows, reveal=1.0, color=color))
    except KeyboardInterrupt:
        console.print("[grey54]interrupted[/grey54]")
        return 130
    return 0


def _clamp(v: float, lo: float, hi: float) -> float:
    return lo if v < lo else hi if v > hi else v


# ── list / delete / edit ─────────────────────────────────────────────────────


def _cmd_list(args: argparse.Namespace) -> int:
    from agent.learning_graph_render import format_date

    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]")
        return 0
    for node in nodes:
        glyph = "◆" if node.get("kind") == "memory" else "●"
        date = format_date(node.get("timestamp"))
        console.print(f"[grey54]{node['id']}[/grey54]  {glyph} {node.get('label', '')}  [grey54]{date}[/grey54]")
    return 0


def _lookup_node(node_id: str) -> Optional[dict]:
    """Resolve a node id via ``node_detail``; print the failure and return None when missing."""
    from agent.learning_mutations import node_detail

    detail = node_detail(node_id)
    if not detail.get("ok"):
        print(f"  {detail.get('message', 'not found')}")
        return None
    return detail


def _cmd_delete(args: argparse.Namespace) -> int:
    from agent.learning_mutations import delete_node

    detail = _lookup_node(args.node)
    if detail is None:
        return 1
    if not getattr(args, "yes", False):
        try:
            if input(f"  Delete {detail['label']!r}? [y/N] ").strip().lower() not in ("y", "yes"):
                print("  aborted")
                return 1
        except (EOFError, KeyboardInterrupt):
            print("\n  aborted")
            return 1
    res = delete_node(args.node)
    print(f"  {res['message']}")
    return 0 if res.get("ok") else 1


def _cmd_edit(args: argparse.Namespace) -> int:
    from agent.learning_mutations import edit_node

    detail = _lookup_node(args.node)
    if detail is None:
        return 1
    edited = _open_in_editor(detail["content"], suffix=".md" if detail["kind"] == "skill" else ".txt")
    if edited is None or edited.strip() == detail["content"].strip():
        print("  no changes")
        return 0
    res = edit_node(args.node, edited)
    print(f"  {res['message']}")
    return 0 if res.get("ok") else 1


def _open_in_editor(initial: str, *, suffix: str) -> Optional[str]:
    import os
    import subprocess
    import tempfile

    editor = os.environ.get("EDITOR") or os.environ.get("VISUAL") or "vi"
    with tempfile.NamedTemporaryFile("w", suffix=suffix, delete=False, encoding="utf-8") as fh:
        fh.write(initial)
        path = fh.name
    try:
        subprocess.call([*editor.split(), path])
        with open(path, encoding="utf-8-sig") as fh:
            return fh.read()
    except OSError as exc:
        print(f"  editor failed: {exc}")
        return None
    finally:
        try:
            os.unlink(path)
        except OSError:
            pass


def register_cli(parent: argparse.ArgumentParser) -> None:
    parent.add_argument(
        "--reveal", type=float, default=1.0, metavar="0..1",
        help="Render the timeline built up to this point (0=oldest, 1=now).",
    )
    parent.add_argument("--play", action="store_true", help="Animate the build-up over time (Ctrl-C to stop).")
    parent.add_argument("--fps", type=int, default=12, help="Animation frames per second for --play (default 12).")
    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)

    sub = parent.add_subparsers(dest="journey_action")

    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.")
    p_del.add_argument("node", help="Node id (skill name or memory:<source>:<index>; see `journey list`).")
    p_del.add_argument("-y", "--yes", action="store_true", help="Skip the confirmation prompt.")
    p_del.set_defaults(func=_cmd_delete)

    p_edit = sub.add_parser("edit", help="Edit a learned skill or memory by node id in $EDITOR.")
    p_edit.add_argument("node", help="Node id (skill name or memory:<source>:<index>; see `journey list`).")
    p_edit.set_defaults(func=_cmd_edit)


if __name__ == "__main__":
    _p = argparse.ArgumentParser(prog="hermes journey")
    register_cli(_p)
    _a = _p.parse_args()
    sys.exit(_a.func(_a))


# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
# Names external plugins imported from this module before the Sep 2026 decomposition.
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
# The whole block is removed by reverting the commit that added it.

def cmd_journey(args: argparse.Namespace) -> int:
    return _cmd_show(args)
# ---- END PLUGIN-COMPAT ----