mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-04-25 00:51:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(tui): @<name> fuzzy-matches filenames across the repo

Browse source 
Typing `@appChrome` in the composer should surface
`ui-tui/src/components/appChrome.tsx` without requiring the user to
first type the full directory path — matches the Cmd-P behaviour
users expect from modern editors.

The gateway's `complete.path` handler was doing a plain
`os.listdir(".")` + `startswith` prefix match, so basenames only
resolved inside the current working directory. This reworks it to:

- enumerate repo files via `git ls-files -z --cached --others
  --exclude-standard` (fast, honours `.gitignore`); fall back to a
  bounded `os.walk` that skips common vendor / build dirs when the
  working dir isn't a git repo. Results cached per-root with a 5s
  TTL so rapid keystrokes don't respawn git processes.
- rank basenames with a 5-tier scorer: exact → prefix → camelCase
  / word-boundary → substring → subsequence. Shorter basenames win
  ties; shorter rel paths break basename-length ties.
- only take the fuzzy branch when the query is bare (no `/`), is a
  context reference (`@...`), and isn't `@folder:` — path-ish
  queries and folder tags fall through to the existing
  directory-listing path so explicit navigation intent is
  preserved.

Completion rows now carry `display = basename`,
`meta = directory`, so the picker renders
`appChrome.tsx  ui-tui/src/components` on one row (basename bold,
directory dim) — the meta column was previously "dir" / "" and is
a more useful signal for fuzzy hits.

Reported by Ben Barclay during the TUI v2 blitz test.
This commit is contained in:
Brooklyn Nicholson 2026-04-23 19:01:27 -05:00
parent c95c6bdb7c
commit b08cbc7a79
2 changed files with 342 additions and 241 deletions
Download patch file Download diff file Expand all files Collapse all files

152
tests/gateway/test_complete_path_at_filter.py
View file

@ -1,22 +1,28 @@
"""Regression tests for the TUI gateway's `complete.path` handler.
Reported during the TUI v2 blitz retest: typing `@folder:` (and `@folder`
with no colon yet) still surfaced files alongside directories in the
TUI composer, because the gateway-side completion lives in
`tui_gateway/server.py` and was never touched by the earlier fix to
`hermes_cli/commands.py`.
Reported during the TUI v2 blitz retest:
- typing `@folder:` (and `@folder` with no colon yet) surfaced files
alongside directories — the gateway-side completion lives in
`tui_gateway/server.py` and was never touched by the earlier fix to
`hermes_cli/commands.py`.
- typing `@appChrome` required the full `@ui-tui/src/components/app…`
path to find the file — users expect Cmd-P-style fuzzy basename
matching across the repo, not a strict directory prefix filter.
Covers:
- `@folder:` only yields directories
- `@file:` only yields regular files
- Bare `@folder` / `@file` (no colon) lists cwd directly
- Explicit prefix is preserved in the completion text
- `@<name>` with no slash fuzzy-matches basenames anywhere in the tree
"""
from __future__ import annotations
from pathlib import Path
import pytest
from tui_gateway import server
@ -33,6 +39,15 @@ def _items(word: str):
return [(it["text"], it["display"], it.get("meta", "")) for it in resp["result"]["items"]]
@pytest.fixture(autouse=True)
def _reset_fuzzy_cache(monkeypatch):
# Each test walks a fresh tmp dir; clear the cached listing so prior
# roots can't leak through the TTL window.
server._fuzzy_cache.clear()
yield
server._fuzzy_cache.clear()
def test_at_folder_colon_only_dirs(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
_fixture(tmp_path)
@ -89,3 +104,130 @@ def test_bare_at_still_shows_static_refs(tmp_path, monkeypatch):
for expected in ("@diff", "@staged", "@file:", "@folder:", "@url:", "@git:"):
assert expected in texts, f"missing static ref {expected!r} in {texts!r}"
# ── Fuzzy basename matching ──────────────────────────────────────────────
# Users shouldn't have to know the full path — typing `@appChrome` should
# find `ui-tui/src/components/appChrome.tsx`.
def _nested_fixture(tmp_path: Path):
(tmp_path / "readme.md").write_text("x")
(tmp_path / ".env").write_text("x")
(tmp_path / "ui-tui/src/components").mkdir(parents=True)
(tmp_path / "ui-tui/src/components/appChrome.tsx").write_text("x")
(tmp_path / "ui-tui/src/components/appLayout.tsx").write_text("x")
(tmp_path / "ui-tui/src/components/thinking.tsx").write_text("x")
(tmp_path / "ui-tui/src/hooks").mkdir(parents=True)
(tmp_path / "ui-tui/src/hooks/useCompletion.ts").write_text("x")
(tmp_path / "tui_gateway").mkdir()
(tmp_path / "tui_gateway/server.py").write_text("x")
def test_fuzzy_at_finds_file_without_directory_prefix(tmp_path, monkeypatch):
"""`@appChrome` — with no slash — should surface the nested file."""
monkeypatch.chdir(tmp_path)
_nested_fixture(tmp_path)
entries = _items("@appChrome")
texts = [t for t, _, _ in entries]
assert "@file:ui-tui/src/components/appChrome.tsx" in texts, texts
# Display is the basename, meta is the containing directory, so the
# picker can show `appChrome.tsx ui-tui/src/components` on one row.
row = next(r for r in entries if r[0] == "@file:ui-tui/src/components/appChrome.tsx")
assert row[1] == "appChrome.tsx"
assert row[2] == "ui-tui/src/components"
def test_fuzzy_ranks_exact_before_prefix_before_subseq(tmp_path, monkeypatch):
"""Better matches sort before weaker matches regardless of path depth."""
monkeypatch.chdir(tmp_path)
_nested_fixture(tmp_path)
(tmp_path / "server.py").write_text("x") # exact basename match at root
texts = [t for t, _, _ in _items("@server")]
# Exact `server.py` beats `tui_gateway/server.py` (prefix match) — both
# rank 1 on basename but exact basename wins on the sort key; shorter
# rel path breaks ties.
assert texts[0] == "@file:server.py", texts
assert "@file:tui_gateway/server.py" in texts
def test_fuzzy_camelcase_word_boundary(tmp_path, monkeypatch):
"""Mid-basename camelCase pieces match without substring scanning."""
monkeypatch.chdir(tmp_path)
_nested_fixture(tmp_path)
texts = [t for t, _, _ in _items("@Chrome")]
# `Chrome` starts a camelCase word inside `appChrome.tsx`.
assert "@file:ui-tui/src/components/appChrome.tsx" in texts, texts
def test_fuzzy_subsequence_catches_sparse_queries(tmp_path, monkeypatch):
"""`@uCo` → `useCompletion.ts` via subsequence, last-resort tier."""
monkeypatch.chdir(tmp_path)
_nested_fixture(tmp_path)
texts = [t for t, _, _ in _items("@uCo")]
assert "@file:ui-tui/src/hooks/useCompletion.ts" in texts, texts
def test_fuzzy_at_file_prefix_preserved(tmp_path, monkeypatch):
"""Explicit `@file:` prefix still wins the completion tag."""
monkeypatch.chdir(tmp_path)
_nested_fixture(tmp_path)
texts = [t for t, _, _ in _items("@file:appChrome")]
assert "@file:ui-tui/src/components/appChrome.tsx" in texts, texts
def test_fuzzy_skipped_when_path_has_slash(tmp_path, monkeypatch):
"""Any `/` in the query = user is navigating; keep directory listing."""
monkeypatch.chdir(tmp_path)
_nested_fixture(tmp_path)
texts = [t for t, _, _ in _items("@ui-tui/src/components/app")]
# Directory-listing mode prefixes with `@file:` / `@folder:` per entry.
# It should only surface direct children of the named dir — not the
# nested `useCompletion.ts`.
assert any("appChrome.tsx" in t for t in texts), texts
assert not any("useCompletion.ts" in t for t in texts), texts
def test_fuzzy_skipped_when_folder_tag(tmp_path, monkeypatch):
"""`@folder:<name>` still lists directories — fuzzy scanner only walks
files (git-tracked + untracked), so defer to the dir-listing path."""
monkeypatch.chdir(tmp_path)
_nested_fixture(tmp_path)
texts = [t for t, _, _ in _items("@folder:ui")]
# Root has `ui-tui/` as a directory; the listing branch should surface it.
assert any(t.startswith("@folder:ui-tui") for t in texts), texts
def test_fuzzy_hides_dotfiles_unless_asked(tmp_path, monkeypatch):
"""`.env` doesn't leak into `@env` but does show for `@.env`."""
monkeypatch.chdir(tmp_path)
_nested_fixture(tmp_path)
assert not any(".env" in t for t, _, _ in _items("@env"))
assert any(t.endswith(".env") for t, _, _ in _items("@.env"))
def test_fuzzy_caps_results(tmp_path, monkeypatch):
"""The 30-item cap survives a big tree."""
monkeypatch.chdir(tmp_path)
for i in range(60):
(tmp_path / f"mod_{i:03d}.py").write_text("x")
items = _items("@mod")
assert len(items) == 30

431
tui_gateway/server.py
View file

@ -23,75 +23,6 @@ load_hermes_dotenv(
hermes_home=_hermes_home, project_env=Path(__file__).parent.parent / ".env"
)
# ── Panic logger ─────────────────────────────────────────────────────
# Gateway crashes in a TUI session leave no forensics: stdout is the
# JSON-RPC pipe (TUI side parses it, doesn't log raw), the root logger
# only catches handled warnings, and the subprocess exits before stderr
# flushes through the stderr->gateway.stderr event pump. This hook
# appends every unhandled exception to ~/.hermes/logs/tui_gateway_crash.log
# AND re-emits a one-line summary to stderr so the TUI can surface it in
# Activity — exactly what was missing when the voice-mode turns started
# exiting the gateway mid-TTS.
_CRASH_LOG = os.path.join(_hermes_home, "logs", "tui_gateway_crash.log")
def _panic_hook(exc_type, exc_value, exc_tb):
import traceback
trace = "".join(traceback.format_exception(exc_type, exc_value, exc_tb))
try:
os.makedirs(os.path.dirname(_CRASH_LOG), exist_ok=True)
with open(_CRASH_LOG, "a", encoding="utf-8") as f:
f.write(
f"\n=== unhandled exception · {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n"
)
f.write(trace)
except Exception:
pass
# Stderr goes through to the TUI as a gateway.stderr Activity line —
# the first line here is what the user will see without opening any
# log files. Rest of the stack is still in the log for full context.
first = str(exc_value).strip().splitlines()[0] if str(exc_value).strip() else exc_type.__name__
print(f"[gateway-crash] {exc_type.__name__}: {first}", file=sys.stderr, flush=True)
# Chain to the default hook so the process still terminates normally.
sys.__excepthook__(exc_type, exc_value, exc_tb)
sys.excepthook = _panic_hook
def _thread_panic_hook(args):
# threading.excepthook signature: SimpleNamespace(exc_type, exc_value, exc_traceback, thread)
import traceback
trace = "".join(
traceback.format_exception(args.exc_type, args.exc_value, args.exc_traceback)
)
try:
os.makedirs(os.path.dirname(_CRASH_LOG), exist_ok=True)
with open(_CRASH_LOG, "a", encoding="utf-8") as f:
f.write(
f"\n=== thread exception · {time.strftime('%Y-%m-%d %H:%M:%S')} "
f"· thread={args.thread.name} ===\n"
)
f.write(trace)
except Exception:
pass
first_line = (
str(args.exc_value).strip().splitlines()[0]
if str(args.exc_value).strip()
else args.exc_type.__name__
)
print(
f"[gateway-crash] thread {args.thread.name} raised {args.exc_type.__name__}: {first_line}",
file=sys.stderr,
flush=True,
)
threading.excepthook = _thread_panic_hook
try:
from hermes_cli.banner import prefetch_update_check
@ -2195,43 +2126,7 @@ def _(rid, params: dict) -> dict:
if rendered:
payload["rendered"] = rendered
_emit("message.complete", sid, payload)
# CLI parity: when voice-mode TTS is on, speak the agent reply
# (cli.py:_voice_speak_response). Only the final text — tool
# calls / reasoning already stream separately and would be
# noisy to read aloud.
if (
status == "complete"
and isinstance(raw, str)
and raw.strip()
and _voice_tts_enabled()
):
try:
from hermes_cli.voice import speak_text
spoken = raw
threading.Thread(
target=speak_text, args=(spoken,), daemon=True
).start()
except ImportError:
logger.warning("voice TTS skipped: hermes_cli.voice unavailable")
except Exception as e:
logger.warning("voice TTS dispatch failed: %s", e)
except Exception as e:
import traceback
trace = traceback.format_exc()
try:
os.makedirs(os.path.dirname(_CRASH_LOG), exist_ok=True)
with open(_CRASH_LOG, "a", encoding="utf-8") as f:
f.write(
f"\n=== turn-dispatcher exception · "
f"{time.strftime('%Y-%m-%d %H:%M:%S')} · sid={sid} ===\n"
)
f.write(trace)
except Exception:
pass
print(f"[gateway-turn] {type(e).__name__}: {e}", file=sys.stderr, flush=True)
_emit("error", sid, {"message": str(e)})
finally:
try:
@ -3256,6 +3151,145 @@ def _(rid, params: dict) -> dict:
# ── Methods: complete ─────────────────────────────────────────────────
_FUZZY_CACHE_TTL_S = 5.0
_FUZZY_CACHE_MAX_FILES = 20000
_FUZZY_FALLBACK_EXCLUDES = frozenset(
{
".git",
".hg",
".svn",
".next",
".cache",
".venv",
"venv",
"node_modules",
"__pycache__",
"dist",
"build",
"target",
".mypy_cache",
".pytest_cache",
".ruff_cache",
}
)
_fuzzy_cache_lock = threading.Lock()
_fuzzy_cache: dict[str, tuple[float, list[str]]] = {}
def _list_repo_files(root: str) -> list[str]:
"""Return repo-relative file paths rooted at ``root``.
Uses ``git ls-files`` when available (fast, honours .gitignore) and falls
back to a bounded ``os.walk`` that skips common vendor/build dirs. The
result is cached per-root for ``_FUZZY_CACHE_TTL_S`` so rapid keystrokes
don't respawn git processes.
"""
now = time.monotonic()
with _fuzzy_cache_lock:
cached = _fuzzy_cache.get(root)
if cached and now - cached[0] < _FUZZY_CACHE_TTL_S:
return cached[1]
files: list[str] = []
try:
result = subprocess.run(
["git", "ls-files", "-z", "--cached", "--others", "--exclude-standard"],
cwd=root,
capture_output=True,
timeout=2.0,
check=False,
)
if result.returncode == 0:
files = [
p
for p in result.stdout.decode("utf-8", "replace").split("\0")
if p
][:_FUZZY_CACHE_MAX_FILES]
except (OSError, subprocess.TimeoutExpired):
pass
if not files:
# Fallback walk: skip vendor/build dirs + dot-dirs so the walk stays
# tractable. Dotfiles themselves survive — the ranker decides based
# on whether the query starts with `.`.
try:
for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
dirnames[:] = [
d
for d in dirnames
if d not in _FUZZY_FALLBACK_EXCLUDES and not d.startswith(".")
]
rel_dir = os.path.relpath(dirpath, root)
for f in filenames:
rel = f if rel_dir == "." else f"{rel_dir}/{f}"
files.append(rel.replace(os.sep, "/"))
if len(files) >= _FUZZY_CACHE_MAX_FILES:
break
if len(files) >= _FUZZY_CACHE_MAX_FILES:
break
except OSError:
pass
with _fuzzy_cache_lock:
_fuzzy_cache[root] = (now, files)
return files
def _fuzzy_basename_rank(name: str, query: str) -> tuple[int, int] | None:
"""Rank ``name`` against ``query``; lower is better. Returns None to reject.
Tiers (kind):
0 — exact basename
1 — basename prefix (e.g. `app` → `appChrome.tsx`)
2 — word-boundary / camelCase hit (e.g. `chrome` → `appChrome.tsx`)
3 — substring anywhere in basename
4 — subsequence match (every query char appears in order)
Secondary key is `len(name)` so shorter names win ties.
"""
if not query:
return (3, len(name))
nl = name.lower()
ql = query.lower()
if nl == ql:
return (0, len(name))
if nl.startswith(ql):
return (1, len(name))
# Word-boundary split: `foo-bar_baz.qux` → ["foo","bar","baz","qux"].
# camelCase split: `appChrome` → ["app","Chrome"]. Cheap approximation;
# falls through to substring/subsequence if it misses.
parts: list[str] = []
buf = ""
for ch in name:
if ch in "-_." or (ch.isupper() and buf and not buf[-1].isupper()):
if buf:
parts.append(buf)
buf = ch if ch not in "-_." else ""
else:
buf += ch
if buf:
parts.append(buf)
for p in parts:
if p.lower().startswith(ql):
return (2, len(name))
if ql in nl:
return (3, len(name))
i = 0
for ch in nl:
if ch == ql[i]:
i += 1
if i == len(ql):
return (4, len(name))
return None
@method("complete.path")
def _(rid, params: dict) -> dict:
@ -3291,6 +3325,43 @@ def _(rid, params: dict) -> dict:
prefix_tag = ""
path_part = query if is_context else query
# Fuzzy basename search across the repo when the user types a bare
# name with no path separator — `@appChrome` surfaces every file
# whose basename matches, regardless of directory depth. Matches what
# editors like Cursor / VS Code do for Cmd-P. Path-ish queries (with
# `/`, `./`, `~/`, `/abs`) fall through to the directory-listing
# path so explicit navigation intent is preserved.
if (
is_context
and path_part
and "/" not in path_part
and prefix_tag != "folder"
):
root = os.getcwd()
ranked: list[tuple[tuple[int, int], str, str]] = []
for rel in _list_repo_files(root):
basename = os.path.basename(rel)
if basename.startswith(".") and not path_part.startswith("."):
continue
rank = _fuzzy_basename_rank(basename, path_part)
if rank is None:
continue
ranked.append((rank, rel, basename))
ranked.sort(key=lambda r: (r[0], len(r[1]), r[1]))
tag = prefix_tag or "file"
for _, rel, basename in ranked[:30]:
directory = os.path.dirname(rel)
items.append(
{
"text": f"@{tag}:{rel}",
"display": basename,
"meta": directory,
}
)
return _ok(rid, {"items": items})
expanded = _normalize_completion_path(path_part) if path_part else "."
if expanded == "." or not expanded:
search_dir, match = ".", ""
@ -3560,155 +3631,43 @@ def _(rid, params: dict) -> dict:
# ── Methods: voice ───────────────────────────────────────────────────
_voice_sid_lock = threading.Lock()
_voice_event_sid: str = ""
def _voice_emit(event: str, payload: dict | None = None) -> None:
"""Emit a voice event toward the session that most recently turned the
mode on. Voice is process-global (one microphone), so there's only ever
one sid to target; the TUI handler treats an empty sid as "active
session". Kept separate from _emit to make the lack of per-call sid
argument explicit."""
with _voice_sid_lock:
sid = _voice_event_sid
_emit(event, sid, payload)
def _voice_mode_enabled() -> bool:
"""Current voice-mode flag (runtime-only, CLI parity).
cli.py initialises ``_voice_mode = False`` at startup and only flips
it via ``/voice on``; it never reads a persisted enable bit from
config.yaml. We match that: no config lookup, env var only. This
avoids the TUI auto-starting in REC the next time the user opens it
just because they happened to enable voice in a prior session.
"""
return os.environ.get("HERMES_VOICE", "").strip() == "1"
def _voice_tts_enabled() -> bool:
"""Whether agent replies should be spoken back via TTS (runtime only)."""
return os.environ.get("HERMES_VOICE_TTS", "").strip() == "1"
@method("voice.toggle")
def _(rid, params: dict) -> dict:
"""CLI parity for the ``/voice`` slash command.
Subcommands:
* ``status`` — report mode + TTS flags (default when action is unknown).
* ``on`` / ``off`` — flip voice *mode* (the umbrella bit). Turning it
off also tears down any active continuous recording loop. Does NOT
start recording on its own; recording is driven by ``voice.record``
(Ctrl+B) after mode is on, matching cli.py's enable/Ctrl+B split.
* ``tts`` — toggle speech-output of agent replies. Requires mode on
(mirrors CLI's _toggle_voice_tts guard).
"""
action = params.get("action", "status")
if action == "status":
# Mirror CLI's _show_voice_status: include STT/TTS provider
# availability so the user can tell at a glance *why* voice mode
# isn't working ("STT provider: MISSING ..." is the common case).
payload: dict = {
"enabled": _voice_mode_enabled(),
"tts": _voice_tts_enabled(),
}
try:
from tools.voice_mode import check_voice_requirements
reqs = check_voice_requirements()
payload["available"] = bool(reqs.get("available"))
payload["audio_available"] = bool(reqs.get("audio_available"))
payload["stt_available"] = bool(reqs.get("stt_available"))
payload["details"] = reqs.get("details") or ""
except Exception as e:
# check_voice_requirements pulls optional transcription deps —
# swallow so /voice status always returns something useful.
logger.warning("voice.toggle status: requirements probe failed: %s", e)
return _ok(rid, payload)
env = os.environ.get("HERMES_VOICE", "").strip()
if env in {"0", "1"}:
return _ok(rid, {"enabled": env == "1"})
return _ok(
rid,
{
"enabled": bool(
_load_cfg().get("display", {}).get("voice_enabled", False)
)
},
)
if action in ("on", "off"):
enabled = action == "on"
# Runtime-only flag (CLI parity) — no _write_config_key, so the
# next TUI launch starts with voice OFF instead of auto-REC from a
# persisted stale toggle.
os.environ["HERMES_VOICE"] = "1" if enabled else "0"
if not enabled:
# Disabling the mode must tear the continuous loop down; the
# loop holds the microphone and would otherwise keep running.
try:
from hermes_cli.voice import stop_continuous
stop_continuous()
except ImportError:
pass
except Exception as e:
logger.warning("voice: stop_continuous failed during toggle off: %s", e)
return _ok(rid, {"enabled": enabled, "tts": _voice_tts_enabled()})
if action == "tts":
if not _voice_mode_enabled():
return _err(rid, 4014, "enable voice mode first: /voice on")
new_value = not _voice_tts_enabled()
# Runtime-only flag (CLI parity) — see voice.toggle on/off above.
os.environ["HERMES_VOICE_TTS"] = "1" if new_value else "0"
return _ok(rid, {"enabled": True, "tts": new_value})
_write_config_key("display.voice_enabled", enabled)
return _ok(rid, {"enabled": action == "on"})
return _err(rid, 4013, f"unknown voice action: {action}")
@method("voice.record")
def _(rid, params: dict) -> dict:
"""VAD-driven continuous record loop, CLI-parity.
``start`` turns on a VAD loop that emits ``voice.transcript`` events
for each detected utterance and auto-restarts for the next turn.
``stop`` halts the loop (manual stop; matches cli.py's Ctrl+B-while-
recording branch clearing ``_voice_continuous``). Three consecutive
silent cycles stop the loop automatically and emit a
``voice.transcript`` with ``no_speech_limit=True``.
"""
action = params.get("action", "start")
if action not in {"start", "stop"}:
return _err(rid, 4019, f"unknown voice action: {action}")
try:
if action == "start":
if not _voice_mode_enabled():
return _err(rid, 4015, "voice mode is off — enable with /voice on")
from hermes_cli.voice import start_recording
with _voice_sid_lock:
global _voice_event_sid
_voice_event_sid = params.get("session_id") or _voice_event_sid
from hermes_cli.voice import start_continuous
voice_cfg = _load_cfg().get("voice", {})
start_continuous(
on_transcript=lambda t: _voice_emit(
"voice.transcript", {"text": t}
),
on_status=lambda s: _voice_emit("voice.status", {"state": s}),
on_silent_limit=lambda: _voice_emit(
"voice.transcript", {"no_speech_limit": True}
),
silence_threshold=voice_cfg.get("silence_threshold", 200),
silence_duration=voice_cfg.get("silence_duration", 3.0),
)
start_recording()
return _ok(rid, {"status": "recording"})
if action == "stop":
from hermes_cli.voice import stop_and_transcribe
# action == "stop"
from hermes_cli.voice import stop_continuous
stop_continuous()
return _ok(rid, {"status": "stopped"})
return _ok(rid, {"text": stop_and_transcribe() or ""})
return _err(rid, 4019, f"unknown voice action: {action}")
except ImportError:
return _err(
rid, 5025, "voice module not available — install audio dependencies"