refactor(cli): one foreign-log walker for the CLI picker and the desktop browser

hermes_cli.foreign_sessions._walk owns discovery (glob per source, per-entry
OSError tolerance, symlink-escape and regular-file checks, newest first); the
CLI listing and the desktop browser's candidate rows both consume it instead of
carrying two scans of the same directories.
This commit is contained in:
Teknium
2026-09-06 03:56:54 -07:00
parent f3400ce745
commit 0168bb27ad
2 changed files with 28 additions and 30 deletions

View File

@@ -12,6 +12,7 @@ import uuid
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from stat import S_ISREG
from typing import Any, Dict, List, Optional, Tuple
# User-message texts that are really injected context wrappers, not typed input.
@@ -169,20 +170,33 @@ _SOURCES = {
}
def _list_sessions(source: str, root: Optional[Path]) -> List[ForeignSession]:
default_root, pattern, recursive, parse = _SOURCES[source]
root = Path(root) if root else Path.home().joinpath(*default_root)
results: List[ForeignSession] = []
for jsonl in sorted((root.rglob(pattern) if recursive else root.glob(pattern)) if root.is_dir() else ()):
def _walk(source: str, root: Optional[Path] = None) -> List[Tuple[Path, os.stat_result]]:
"""Regular log files of *source* under *root* (default ``~/<tool dir>``) as ``(path, stat)``,
newest first. Symlinks escaping the root and unreadable/rotated entries are skipped, so one
bad file never hides the rest. Shared by the CLI picker and the desktop browser."""
default_root, pattern, recursive, _ = _SOURCES[source]
root = (Path(root) if root else Path.home().joinpath(*default_root)).resolve()
found: List[Tuple[Path, os.stat_result]] = []
for path in (root.rglob(pattern) if recursive else root.glob(pattern)) if root.is_dir() else ():
try:
mtime = jsonl.stat().st_mtime
resolved = path.resolve()
st = resolved.stat()
except OSError:
continue
parsed = parse(jsonl)
if resolved.is_relative_to(root) and S_ISREG(st.st_mode):
found.append((resolved, st))
found.sort(key=lambda item: item[1].st_mtime, reverse=True)
return found
def _list_sessions(source: str, root: Optional[Path]) -> List[ForeignSession]:
parse = _SOURCES[source][3]
results: List[ForeignSession] = []
for path, st in _walk(source, root):
parsed = parse(path)
if parsed["turns"]:
results.append(ForeignSession(source, jsonl, mtime, parsed["cwd"], parsed["title_guess"],
results.append(ForeignSession(source, path, st.st_mtime, parsed["cwd"], parsed["title_guess"],
len(parsed["turns"]), parsed["session_id"]))
results.sort(key=lambda s: s.mtime, reverse=True)
return results

View File

@@ -4,9 +4,8 @@ import hashlib
import re
import socket
from pathlib import Path
from stat import S_ISREG
from hermes_cli.foreign_sessions import _SOURCES, _SOURCE_LABELS, _SOURCE_DB_NAMES
from hermes_cli.foreign_sessions import _SOURCE_DB_NAMES, _SOURCE_LABELS, _SOURCES, _walk
MAX_LOG_BYTES = 32 * 1024 * 1024
@@ -21,27 +20,12 @@ def _display_title(parsed, source):
def _candidates(source=None):
"""``(mtime, handle, source, path, size)`` rows across sources, newest first. The handle is
the only identifier handed to the client; a request can never name a path."""
if source is not None and source not in _SOURCES:
raise ValueError("Unknown session source")
rows = []
for name, (parts, pattern, recursive, _) in _SOURCES.items():
if source and source != name:
continue
root = Path.home().joinpath(*parts).resolve()
if not root.is_dir():
continue
for path in root.rglob(pattern) if recursive else root.glob(pattern):
try:
resolved = path.resolve()
if not resolved.is_relative_to(root):
continue
stat = resolved.stat()
except OSError:
continue # One inaccessible or rotated log must not hide the rest.
if not S_ISREG(stat.st_mode):
continue
handle = hashlib.sha256(f"{name}:{resolved}".encode()).hexdigest()
rows.append((stat.st_mtime, handle, name, resolved, stat.st_size))
rows = [(st.st_mtime, hashlib.sha256(f"{name}:{path}".encode()).hexdigest(), name, path, st.st_size)
for name in _SOURCES if source in (None, name) for path, st in _walk(name)]
return sorted(rows, key=lambda row: (row[0], row[1]), reverse=True)