Merge branch 'simp/r3-17-g2-tui' into simp/integration3

This commit is contained in:
Teknium
2026-09-02 21:59:18 -07:00
5 changed files with 1118 additions and 2069 deletions
+14 -34
View File
@@ -1,9 +1,8 @@
"""Baked-in build metadata for Hermes Agent.
Source installs report their git revision live via ``git rev-parse`` (see ``hermes_cli/dump.py`` and
``hermes_cli/banner.py``). That doesn't work inside the published Docker image because
``.dockerignore`` excludes ``.git``, so those callsites fall back to ``"(unknown)"`` / drop the
banner suffix entirely.
Source installs report their git revision live via ``git rev-parse`` (``hermes_cli/dump.py``,
``hermes_cli/banner.py``). Inside the published Docker image ``.dockerignore`` excludes ``.git``,
so those callsites fall back to the ``.hermes_build_sha`` file written by the Dockerfile.
"""
from __future__ import annotations
@@ -11,11 +10,8 @@ from __future__ import annotations
from pathlib import Path
from typing import Optional
# Path is resolved relative to this module so it works regardless of cwd —
# matches the pattern used by ``banner._resolve_repo_dir``.
# Resolved relative to this module so it works regardless of cwd (as banner._resolve_repo_dir).
_BUILD_SHA_FILE = Path(__file__).parent.parent / ".hermes_build_sha"
_code_identity_cache: Optional[dict] = None
@@ -28,17 +24,16 @@ def _sha_or_none(value: str) -> Optional[str]:
def _resolve_git_head_sha(project_root: Path) -> Optional[str]:
"""Resolve the checkout's HEAD commit sha by reading .git directly.
"""HEAD sha by reading .git directly (worktrees/submodules, loose refs, packed-refs).
Deliberately NOT ``git rev-parse``: this runs in library paths (runtime-status writes, update
receipts) where spawning is slow and hostile to tests that mock ``subprocess.run`` tightly.
Handles worktrees/submodules (``gitdir:`` + ``commondir``), loose refs, and packed-refs.
Returns None on any failure.
"""
try:
git_path = project_root / ".git"
if git_path.is_file():
# Worktree/submodule: ".git" is a pointer file.
# Worktree/submodule: ".git" is a "gitdir: <path>" pointer file.
pointer = _read_stripped(git_path)
if not pointer.startswith("gitdir:"):
return None
@@ -49,24 +44,19 @@ def _resolve_git_head_sha(project_root: Path) -> Optional[str]:
git_dir = git_path
else:
return None
# Refs live in the COMMON git dir for worktrees.
common_dir = git_dir
commondir_file = git_dir / "commondir"
if commondir_file.is_file():
common = Path(_read_stripped(commondir_file))
common_dir = common if common.is_absolute() else (git_dir / common).resolve()
head = _read_stripped(git_dir / "HEAD")
if not head.startswith("ref:"):
# Detached HEAD: the file holds the sha itself.
return _sha_or_none(head)
return _sha_or_none(head) # detached HEAD holds the sha itself
ref_name = head[len("ref:"):].strip()
loose = common_dir / ref_name
if loose.is_file():
return _sha_or_none(_read_stripped(loose))
packed = common_dir / "packed-refs"
if packed.is_file():
for line in _read_stripped(packed).splitlines():
@@ -82,20 +72,16 @@ def _resolve_git_head_sha(project_root: Path) -> Optional[str]:
def get_code_identity(refresh: bool = False) -> dict:
"""Return the running checkout's code identity as a dict.
"""``{sha, short_sha, version, source}`` for the running checkout; never raises.
Resolution order mirrors the banner/dump callsites: live ``git rev-parse`` for source installs,
the baked ``.hermes_build_sha`` for Docker images (no ``.git`` inside the published image), else
unknown.
Cached per process — code identity cannot change while a process is running (an updated checkout
requires a restart to take effect, which is exactly the property the fleet version verification
relies on). Never raises; every field degrades to ``None`` independently.
Resolution mirrors the banner/dump callsites: live git for source installs, the baked
``.hermes_build_sha`` for Docker images, else unknown. Cached per process — code identity
cannot change while a process runs (an updated checkout needs a restart), which is exactly
the property fleet version verification relies on. Each field degrades to None independently.
"""
global _code_identity_cache
if _code_identity_cache is not None and not refresh:
return dict(_code_identity_cache)
project_root = Path(__file__).parent.parent
source = "unknown"
sha = _resolve_git_head_sha(project_root)
@@ -105,17 +91,14 @@ def get_code_identity(refresh: bool = False) -> dict:
sha = get_build_sha(short=0)
if sha:
source = "build-file"
version: Optional[str] = None
try:
import tomllib
with open(project_root / "pyproject.toml", "rb") as fh: # windows-footgun: ok — binary mode, tomllib requires bytes
raw_version = tomllib.load(fh).get("project", {}).get("version")
version = str(raw_version) if raw_version else None
except Exception:
version = None
_code_identity_cache = {
"sha": sha,
"short_sha": sha[:8] if sha else None,
@@ -126,11 +109,8 @@ def get_code_identity(refresh: bool = False) -> dict:
def get_build_sha(short: int = 8) -> Optional[str]:
"""Return the baked-in build SHA, truncated to ``short`` chars, or None.
Reads ``<project_root>/.hermes_build_sha``, written by the Dockerfile's ``HERMES_GIT_SHA``
build-arg (full 40-char hash on one line).
"""
"""Baked-in build SHA (Dockerfile ``HERMES_GIT_SHA`` build-arg, full 40-char hash on one
line), truncated to ``short`` chars (``short<=0`` = full), or None."""
try:
if not _BUILD_SHA_FILE.is_file():
return None
+11 -52
View File
@@ -1,9 +1,4 @@
"""Shared CLI output helpers for Hermes CLI modules.
Extracts the identical ``print_info/success/warning/error`` and ``prompt()``
functions previously duplicated across setup.py, tools_config.py,
mcp_config.py, and memory_setup.py.
"""
"""Shared CLI output helpers (``print_*`` + ``prompt()``) for the setup/config wizards."""
import sys
@@ -11,91 +6,57 @@ from hermes_cli.colors import Colors, color
from hermes_cli.secret_prompt import masked_secret_prompt
# ─── Print Helpers ────────────────────────────────────────────────────────────
def print_info(text: str) -> None:
"""Print a dim informational message."""
print(color(f" {text}", Colors.DIM))
def print_success(text: str) -> None:
"""Print a green success message with ✓ prefix."""
print(color(f"✓ {text}", Colors.GREEN))
def print_warning(text: str) -> None:
"""Print a yellow warning message with ⚠ prefix."""
print(color(f"⚠ {text}", Colors.YELLOW))
def print_error(text: str) -> None:
"""Print a red error message with ✗ prefix."""
print(color(f"✗ {text}", Colors.RED))
def print_header(text: str) -> None:
"""Print a bold yellow header."""
print(color(f"\n {text}", Colors.YELLOW))
# ─── Input Prompts ────────────────────────────────────────────────────────────
def line_input(prompt_text: str) -> str:
"""Read non-secret text with normal cursor-editing keys on a real TTY.
Setup and model-selection commands run outside the interactive chat's
prompt-toolkit application, so they can safely use a short-lived prompt
here. Redirected input and output retain the built-in ``input`` behavior
used by scripts, tests, and numbered fallbacks.
Setup/model-selection commands run outside the chat's prompt_toolkit application, so a
short-lived prompt is safe here. Redirected stdin/stdout keep the built-in ``input`` used by
scripts, tests and numbered fallbacks.
"""
if not (sys.stdin.isatty() and sys.stdout.isatty()):
return input(prompt_text)
try:
from prompt_toolkit import prompt as prompt_toolkit_prompt
from prompt_toolkit.formatted_text import ANSI
except ImportError:
return input(prompt_text)
try:
return prompt_toolkit_prompt(ANSI(prompt_text))
except (KeyboardInterrupt, EOFError):
raise
except Exception:
# Some terminals report isatty() == True yet reject registering stdin
# with the asyncio event-loop selector (observed on macOS, where kqueue
# raises EINVAL / "Invalid argument" for fd 0). prompt_toolkit cannot
# attach its input there, so fall back to the built-in line reader,
# which needs no selector and works in cooked mode. Any prompt_toolkit
# runtime failure (OSError, ValueError, RuntimeError) degrades the same
# way — the wizard proceeds instead of crashing.
# Some terminals report isatty() yet reject registering stdin with the asyncio selector
# (macOS kqueue raises EINVAL for fd 0). Any prompt_toolkit runtime failure degrades to
# the built-in reader, which needs no selector — the wizard proceeds instead of crashing.
return input(prompt_text)
def prompt(
question: str,
default: str | None = None,
password: bool = False,
) -> str:
"""Prompt the user for input with optional default and password masking.
Replaces the four independent ``_prompt()`` / ``prompt()`` implementations
in setup.py, tools_config.py, mcp_config.py, and memory_setup.py.
Returns the user's input (stripped), or *default* if the user presses Enter.
Returns empty string on Ctrl-C or EOF.
"""
def prompt(question: str, default: str | None = None, password: bool = False) -> str:
"""Prompt for input (stripped), or ``default`` on plain Enter; "" on Ctrl-C/EOF."""
suffix = f" [{default}]" if default else ""
display = color(f" {question}{suffix}: ", Colors.YELLOW)
try:
if password:
value = masked_secret_prompt(display)
else:
value = line_input(display)
value = value.strip()
value = (masked_secret_prompt(display) if password else line_input(display)).strip()
return value if value else (default or "")
except (KeyboardInterrupt, EOFError):
print()
@@ -103,9 +64,7 @@ def prompt(
def prompt_yes_no(question: str, default: bool = True) -> bool:
"""Prompt for a yes/no answer. Returns bool."""
hint = "Y/n" if default else "y/N"
answer = prompt(f"{question} ({hint})")
answer = prompt(f"{question} ({'Y/n' if default else 'y/N'})")
if not answer:
return default
return answer.lower().startswith("y")
+1078 -1966
View File
File diff suppressed because it is too large Load Diff
+5 -6
View File
@@ -131,15 +131,14 @@ def test_handle_enter_never_gates_on_command_running():
cli_path = Path(__file__).resolve().parents[2] / "hermes_cli" / "cli_tui_mixin.py"
tree = ast.parse(cli_path.read_text(encoding="utf-8"))
target = None
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef) and node.name == "_tui_handle_enter":
target = node
break
assert target is not None, "_tui_handle_enter not found in cli_tui_mixin.py"
handlers = ("_tui_handle_enter", "_tui_enter_inline_command", "_tui_enter_overlay",
"_tui_enter_clarify_freetext", "_tui_enter_clarify_choice", "_tui_enter_while_busy")
targets = [n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name in handlers]
assert {n.name for n in targets} == set(handlers), "Enter handlers not found in cli_tui_mixin.py"
offenders = [
node.attr
for target in targets
for node in ast.walk(target)
if isinstance(node, ast.Attribute)
and node.attr in {"_command_running", "_command_blocks_input"}
+10 -11
View File
@@ -25,18 +25,17 @@ import ast
from pathlib import Path
def _load_handle_enter_node() -> ast.FunctionDef:
"""Extract the ``handle_enter`` nested function node from cli.py."""
ENTER_HANDLERS = ("_tui_handle_enter", "_tui_enter_inline_command", "_tui_enter_overlay",
"_tui_enter_clarify_freetext", "_tui_enter_clarify_choice")
def _load_handle_enter_node() -> ast.Module:
"""Return a synthetic module holding the Enter handler and its phase helpers."""
cli_path = Path(__file__).resolve().parents[2] / "hermes_cli" / "cli_tui_mixin.py"
tree = ast.parse(cli_path.read_text(encoding="utf-8"))
target = None
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef) and node.name == "_tui_handle_enter":
target = node
break
assert target is not None, "_tui_handle_enter not found in cli_tui_mixin.py"
return target
found = [n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name in ENTER_HANDLERS]
assert {n.name for n in found} == set(ENTER_HANDLERS), "Enter handlers not found in cli_tui_mixin.py"
return ast.Module(body=found, type_ignores=[])
def _is_buffer_reset(node: ast.stmt) -> bool:
@@ -61,7 +60,7 @@ def _is_invalidate(node: ast.stmt) -> bool:
return isinstance(func, ast.Attribute) and func.attr == "invalidate"
def _collect_reset_blocks(func: ast.FunctionDef) -> list[list[ast.stmt]]:
def _collect_reset_blocks(func: ast.AST) -> list[list[ast.stmt]]:
"""Find every statement sequence (a block body/orelse/finalbody) within
``handle_enter`` that contains a ``buffer.reset()`` call."""
blocks: list[list[ast.stmt]] = []