diff --git a/hermes_cli/foreign_sessions.py b/hermes_cli/foreign_sessions.py index 4c1983028c..9ace1d7d97 100644 --- a/hermes_cli/foreign_sessions.py +++ b/hermes_cli/foreign_sessions.py @@ -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 ``~/``) 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 diff --git a/hermes_cli/foreign_sessions_browser.py b/hermes_cli/foreign_sessions_browser.py index 627917e7e9..c49ac73865 100644 --- a/hermes_cli/foreign_sessions_browser.py +++ b/hermes_cli/foreign_sessions_browser.py @@ -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)