From c4693838c60470f82754704129ae6892c5fbe603 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 3 Sep 2026 00:06:57 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20session-export/projec?= =?UTF-8?q?ts=20pass=202=20=E2=80=94=20fold=20export=20renderers,=20projec?= =?UTF-8?q?ts=5Fdb=20row/meta=20helpers,=20projects=5Fcmd=20str-return=20h?= =?UTF-8?q?andlers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/projects_cmd.py | 82 +++++--------- hermes_cli/projects_db.py | 178 +++++++++++------------------- hermes_cli/session_export.py | 61 ++++------ hermes_cli/session_export_html.py | 42 ++++--- hermes_cli/session_export_md.py | 41 +++---- hermes_cli/session_filters.py | 31 ++---- hermes_cli/session_listing.py | 46 +++----- 7 files changed, 178 insertions(+), 303 deletions(-) diff --git a/hermes_cli/projects_cmd.py b/hermes_cli/projects_cmd.py index 0b1ff36d13..469de52091 100644 --- a/hermes_cli/projects_cmd.py +++ b/hermes_cli/projects_cmd.py @@ -22,18 +22,15 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu ), ) sub = parser.add_subparsers(dest="project_action") - p_create = sub.add_parser("create", help="Create a new project") p_create.add_argument("name", help="Human name, e.g. 'Hermes Agent'") p_create.add_argument("folders", nargs="*", help="Folder paths to include (first = primary)") p_create.add_argument("--slug", default=None, help="Explicit slug override") p_create.add_argument("--primary", default=None, metavar="PATH", help="Primary repo path") - p_create.add_argument("--description", default=None) - p_create.add_argument("--icon", default=None) - p_create.add_argument("--color", default=None) + for opt in ("--description", "--icon", "--color"): + p_create.add_argument(opt, default=None) p_create.add_argument("--board", default=None, metavar="SLUG", help="Bind a kanban board") p_create.add_argument("--use", action="store_true", help="Set as the active project") - p_list = sub.add_parser("list", aliases=["ls"], help="List projects") p_list.add_argument("--all", action="store_true", dest="include_archived", help="Include archived projects") @@ -43,27 +40,20 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu return sp project_sub("show", "Show a project's details") - p_add = project_sub("add-folder", "Add a folder to a project") p_add.add_argument("path", help="Folder path") p_add.add_argument("--label", default=None) p_add.add_argument("--primary", action="store_true", help="Mark as primary repo") - project_sub("remove-folder", "Remove a folder from a project").add_argument("path", help="Folder path") project_sub("rename", "Rename a project").add_argument("name", help="New name") - project_sub("set-primary", "Set the primary folder").add_argument( - "path", help="Folder path (must already be in project)" - ) - + project_sub("set-primary", "Set the primary folder").add_argument("path", help="Folder path (must already be in project)") p_use = sub.add_parser("use", help="Set the active project") p_use.add_argument("project", nargs="?", default=None, help="Project id or slug (omit to clear)") - project_sub("archive", "Archive a project") project_sub("restore", "Restore an archived project") project_sub("bind-board", "Bind a kanban board to a project").add_argument( "board", nargs="?", default="", help="Board slug (omit to unbind)" ) - parser.set_defaults(_project_parser=parser) return parser @@ -76,10 +66,7 @@ def projects_command(args: argparse.Namespace) -> int: if parser is not None: parser.print_help() else: - print( - "usage: hermes project [options]\nRun 'hermes project --help' for the full list.", - file=sys.stderr, - ) + print("usage: hermes project [options]\nRun 'hermes project --help' for the full list.", file=sys.stderr) return 0 handler = _HANDLERS.get(action) if handler is None: @@ -101,16 +88,21 @@ def _resolve(conn, ident: str): def _db_command(fn): - """Open the DB and run ``fn(args, conn)``; a ``ValueError`` prints ``project: …`` and exits 2.""" + """Open the DB and run ``fn(args, conn)``; a ``str`` result is printed (rc 0), an ``int`` is the rc; + a ``ValueError`` prints ``project: …`` and exits 2.""" @functools.wraps(fn) def wrapper(args: argparse.Namespace) -> int: try: with pdb.connect_closing() as conn: - return fn(args, conn) + out = fn(args, conn) except ValueError as exc: print(f"project: {exc}", file=sys.stderr) return 2 + if isinstance(out, str): + print(out) + return 0 + return out return wrapper @@ -119,7 +111,7 @@ def _with_project(fn): """Like ``_db_command`` but also resolves ``args.project`` into ``fn(args, conn, proj)``.""" @functools.wraps(fn) - def wrapper(args: argparse.Namespace, conn) -> int: + def wrapper(args: argparse.Namespace, conn): proj = _resolve(conn, args.project) return 1 if proj is None else fn(args, conn, proj) @@ -156,12 +148,11 @@ def _cmd_create(args, conn) -> int: @_db_command -def _cmd_list(args, conn) -> int: +def _cmd_list(args, conn): active = pdb.get_active_id(conn) projs = pdb.list_projects(conn, include_archived=getattr(args, "include_archived", False)) if not projs: - print("No projects yet. Create one with `hermes project create `.") - return 0 + return "No projects yet. Create one with `hermes project create `." for p in projs: flags = " (archived)" if p.archived else "" print(f"{'*' if p.id == active else ' '} {p.slug:<24} {p.name}{flags} [{len(p.folders)} folder(s)]") @@ -175,68 +166,53 @@ def _cmd_show(args, conn, proj) -> int: @_with_project -def _cmd_add_folder(args, conn, proj) -> int: +def _cmd_add_folder(args, conn, proj) -> str: path = pdb.add_folder(conn, proj.id, args.path, label=args.label, is_primary=args.primary) - print(f"Added {path} to {proj.slug}") - return 0 + return f"Added {path} to {proj.slug}" @_with_project -def _cmd_remove_folder(args, conn, proj) -> int: +def _cmd_remove_folder(args, conn, proj): if not pdb.remove_folder(conn, proj.id, args.path): return _err(f"folder not in project: {args.path}") - print(f"Removed {args.path} from {proj.slug}") - return 0 + return f"Removed {args.path} from {proj.slug}" @_with_project -def _cmd_rename(args, conn, proj) -> int: +def _cmd_rename(args, conn, proj) -> str: pdb.update_project(conn, proj.id, name=args.name) - print(f"Renamed {proj.slug} -> {args.name}") - return 0 + return f"Renamed {proj.slug} -> {args.name}" @_with_project -def _cmd_set_primary(args, conn, proj) -> int: +def _cmd_set_primary(args, conn, proj): if not pdb.set_primary(conn, proj.id, args.path): return _err(f"'{args.path}' is not a folder of {proj.slug}; add it first with `hermes project add-folder`.") - print(f"Set primary of {proj.slug} -> {args.path}") - return 0 + return f"Set primary of {proj.slug} -> {args.path}" @_db_command -def _cmd_use(args, conn) -> int: +def _cmd_use(args, conn): if not args.project: pdb.set_active(conn, None) - print("Cleared active project") - return 0 + return "Cleared active project" proj = _resolve(conn, args.project) if proj is None: return 1 pdb.set_active(conn, proj.id) - print(f"Active project: {proj.slug}") - return 0 + return f"Active project: {proj.slug}" def _flag_command(op: str, verb: str): """Handler for ``pdb.(conn, proj.id)`` followed by ``" "``.""" - - @_with_project - def handler(args, conn, proj) -> int: - getattr(pdb, op)(conn, proj.id) - print(f"{verb} {proj.slug}") - return 0 - - return handler + return _with_project(lambda args, conn, proj: (getattr(pdb, op)(conn, proj.id), f"{verb} {proj.slug}")[1]) @_with_project -def _cmd_bind_board(args, conn, proj) -> int: +def _cmd_bind_board(args, conn, proj) -> str: pdb.update_project(conn, proj.id, board_slug=args.board) if not args.board.strip(): - print(f"Unbound board from {proj.slug}") - return 0 - print(f"Bound {proj.slug} -> board {args.board}") + return f"Unbound board from {proj.slug}" if proj.primary_path: # best-effort: point the bound board's default_workdir at the primary repo try: from hermes_cli import kanban_db as kb @@ -246,7 +222,7 @@ def _cmd_bind_board(args, conn, proj) -> int: kb.write_board_metadata(slug, default_workdir=proj.primary_path) except Exception: pass - return 0 + return f"Bound {proj.slug} -> board {args.board}" _HANDLERS = { diff --git a/hermes_cli/projects_db.py b/hermes_cli/projects_db.py index a90c761a0e..91019cd07b 100644 --- a/hermes_cli/projects_db.py +++ b/hermes_cli/projects_db.py @@ -66,11 +66,19 @@ CREATE TABLE IF NOT EXISTS discovered_repos ( ); """ -# --- Slug + id helpers ------------------------------------------------------- - # Lowercase alphanumerics, hyphens, underscores; 1-64 chars; no leading separator. Strict enough to # stop traversal/path separators, loose enough for kebab-case. Display formatting lives in ``name``. _SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9\-_]{0,63}$") +# Deterministic branch slug: lowercase, separators collapsed, capped. +_BRANCH_SAFE_RE = re.compile(r"[^a-z0-9._-]+") +_INITIALIZED_PATHS: set[str] = set() +# TEXT columns added to `projects` after v1; re-applied idempotently on every open so a legacy DB +# upgrades in place. +_OPTIONAL_PROJECT_COLUMNS = ("board_slug", "primary_path", "icon", "color") +# Nullable TEXT columns that may be absent from a legacy row. +_OPTIONAL_ROW_FIELDS = ("description", "icon", "color", "board_slug", "primary_path") +_ACTIVE_META_KEY = "active_id" +_DISCOVERY_POLICY_META_KEY = "repo_discovery_policy" def _slugify(name: str) -> str: @@ -81,9 +89,7 @@ def _slugify(name: str) -> str: def normalize_slug(slug: Optional[str]) -> Optional[str]: """Lowercase + strip a slug; validate; return ``None`` for empty.""" - if slug is None: - return None - s = str(slug).strip().lower() + s = str(slug).strip().lower() if slug is not None else "" if not s: return None if not _SLUG_RE.match(s): @@ -105,15 +111,6 @@ def _normalize_path(path: str) -> str: return p.rstrip("/\\") or p -# --- Connection management --------------------------------------------------- - -_INITIALIZED_PATHS: set[str] = set() - -# TEXT columns added to `projects` after v1; re-applied idempotently on every open so a legacy DB -# upgrades in place. -_OPTIONAL_PROJECT_COLUMNS = ("board_slug", "primary_path", "icon", "color") - - def connect(db_path: Optional[Path] = None) -> sqlite3.Connection: """Open (and initialize if needed) the per-profile projects DB. @@ -155,8 +152,6 @@ def connect_closing(db_path: Optional[Path] = None): conn.close() -# --- Dataclasses ------------------------------------------------------------- - @dataclass class ProjectFolder: path: str @@ -183,42 +178,31 @@ class Project: folders: List[ProjectFolder] = field(default_factory=list) def to_dict(self) -> dict: - d = {k: getattr(self, k) for k in ("id", "slug", "name", "description", "icon", "color", "board_slug", "primary_path")} + d = {k: getattr(self, k) for k in ("id", "slug", "name", *_OPTIONAL_ROW_FIELDS)} return {**d, "archived": bool(self.archived), "created_at": self.created_at, "folders": [f.to_dict() for f in self.folders]} -# Nullable TEXT columns that may be absent from a legacy row. -_OPTIONAL_ROW_FIELDS = ("description", "icon", "color", "board_slug", "primary_path") - - def _load_project(conn: sqlite3.Connection, row: sqlite3.Row) -> Project: """Materialize a ``projects`` row together with its folders.""" keys = row.keys() - project = Project( + folders = conn.execute( + "SELECT path, label, is_primary, added_at FROM project_folders WHERE project_id = ? ORDER BY is_primary DESC, added_at ASC", + (row["id"],), + ).fetchall() + return Project( id=row["id"], slug=row["slug"], name=row["name"], created_at=row["created_at"], archived=bool(row["archived"]) if "archived" in keys else False, + folders=[ProjectFolder(r["path"], r["label"], bool(r["is_primary"]), r["added_at"]) for r in folders], **{f: row[f] for f in _OPTIONAL_ROW_FIELDS if f in keys}, ) - project.folders = [ - ProjectFolder(path=r["path"], label=r["label"], is_primary=bool(r["is_primary"]), added_at=r["added_at"]) - for r in conn.execute( - "SELECT path, label, is_primary, added_at FROM project_folders WHERE project_id = ? ORDER BY is_primary DESC, added_at ASC", - (project.id,), - ).fetchall() - ] - return project -# --- CRUD -------------------------------------------------------------------- - def _unique_slug(conn: sqlite3.Connection, candidate: str) -> str: """Return ``candidate`` or ``candidate-2``, ``-3`` ... if taken.""" - n = 1 - slug = candidate + n, slug = 1, candidate while conn.execute("SELECT 1 FROM projects WHERE slug = ?", (slug,)).fetchone() is not None: n += 1 - suffix = f"-{n}" - slug = (candidate[: 64 - len(suffix)]).rstrip("-_") + suffix + slug = candidate[: 64 - len(f"-{n}")].rstrip("-_") + f"-{n}" return slug @@ -231,9 +215,7 @@ def find_by_primary_path(conn: sqlite3.Connection, path: str, *, include_archive """The first (oldest) project whose primary path matches ``path`` (separator/case normalized so equivalent Windows spellings don't slip past the dedup check), else None.""" key = _primary_path_key(path) - if not key: - return None - for proj in list_projects(conn, include_archived=include_archived): + for proj in list_projects(conn, include_archived=include_archived) if key else (): primary = proj.primary_path or next( (f.path for f in proj.folders if f.is_primary), proj.folders[0].path if proj.folders else None ) @@ -252,31 +234,21 @@ def create_project( name = str(name or "").strip() if not name: raise ValueError("project name must not be empty") - slug_candidate = normalize_slug(slug) if slug else _slugify(name) pid = "p_" + secrets.token_hex(4) now = _now() - - folder_paths: List[str] = [] - for f in folders or []: - norm = _normalize_path(f) - if norm and norm not in folder_paths: - folder_paths.append(norm) - + folder_paths = list(dict.fromkeys(p for p in map(_normalize_path, folders or []) if p)) primary = _normalize_path(primary_path) if primary_path else None if primary and primary not in folder_paths: folder_paths.insert(0, primary) if primary is None and folder_paths: primary = folder_paths[0] - - if primary and not allow_duplicate_path: - existing = find_by_primary_path(conn, primary) - if existing is not None: - raise ValueError( - f"folder already belongs to project '{existing.slug}' ({existing.id}); " - "switch to it instead of creating a duplicate" - ) - + existing = find_by_primary_path(conn, primary) if primary and not allow_duplicate_path else None + if existing is not None: + raise ValueError( + f"folder already belongs to project '{existing.slug}' ({existing.id}); " + "switch to it instead of creating a duplicate" + ) with write_txn(conn): conn.execute( "INSERT INTO projects (id, slug, name, description, icon, color, board_slug, primary_path, created_at, archived) " @@ -284,11 +256,10 @@ def create_project( (pid, _unique_slug(conn, slug_candidate), name, description, icon, color, normalize_slug(board_slug) if board_slug else None, primary, now), ) - for path in folder_paths: - conn.execute( - "INSERT INTO project_folders (project_id, path, label, is_primary, added_at) VALUES (?, ?, ?, ?, ?)", - (pid, path, None, 1 if path == primary else 0, now), - ) + conn.executemany( + "INSERT INTO project_folders (project_id, path, label, is_primary, added_at) VALUES (?, ?, ?, ?, ?)", + [(pid, path, None, 1 if path == primary else 0, now) for path in folder_paths], + ) return pid @@ -319,18 +290,16 @@ def update_project( if board_slug is not None: board_slug = normalize_slug(board_slug) if board_slug.strip() else "" # (column, provided value, stored value) — "" clears icon/color/board_slug to NULL. - fields = ( - ("name", name, name), - ("description", description, description), - ("icon", icon, icon or None), - ("color", color, color or None), - ("board_slug", board_slug, board_slug or None), - ) - sets = [f"{col} = ?" for col, given, _ in fields if given is not None] - if not sets: + fields = [ + (col, given, stored) for col, given, stored in ( + ("name", name, name), ("description", description, description), ("icon", icon, icon or None), + ("color", color, color or None), ("board_slug", board_slug, board_slug or None), + ) if given is not None + ] + if not fields: return False - params = [stored for _, given, stored in fields if given is not None] + [project_id] - return _execute_rowcount(conn, f"UPDATE projects SET {', '.join(sets)} WHERE id = ?", params) > 0 + sets = ", ".join(f"{col} = ?" for col, _, _ in fields) + return _execute_rowcount(conn, f"UPDATE projects SET {sets} WHERE id = ?", [f[2] for f in fields] + [project_id]) > 0 def _execute_rowcount(conn: sqlite3.Connection, sql: str, params) -> int: @@ -412,10 +381,6 @@ def delete_project(conn: sqlite3.Connection, project_id: str) -> bool: # --- Active-project pointer + discovery policy (project_meta KV) -------------- -_ACTIVE_META_KEY = "active_id" -_DISCOVERY_POLICY_META_KEY = "repo_discovery_policy" - - def _upsert_meta_locked(conn: sqlite3.Connection, key: str, value: str) -> None: """Upsert a project_meta row (caller already holds a write txn).""" conn.execute( @@ -446,30 +411,31 @@ def get_discovery_policy_key(conn: sqlite3.Connection) -> Optional[str]: return _get_meta(conn, _DISCOVERY_POLICY_META_KEY) +def _clear_repos_locked(conn: sqlite3.Connection, clear: bool, policy_key: Optional[str]) -> None: + """Optionally wipe the scan cache, then record the policy key when given (caller holds a write txn).""" + if clear: + conn.execute("DELETE FROM discovered_repos") + if policy_key is not None: + _upsert_meta_locked(conn, _DISCOVERY_POLICY_META_KEY, policy_key) + + def reconcile_discovered_repos_policy(conn: sqlite3.Connection, policy_key: str, *, preserve_unversioned: bool = False) -> bool: """Clear cached scan rows when their discovery policy changes; pre-policy rows are retained only for the backward-compatible default policy. Returns whether rows were cleared.""" current = get_discovery_policy_key(conn) if current == policy_key: return False - cleared = current is not None or not preserve_unversioned with write_txn(conn): - if cleared: - conn.execute("DELETE FROM discovered_repos") - _upsert_meta_locked(conn, _DISCOVERY_POLICY_META_KEY, policy_key) + _clear_repos_locked(conn, cleared, policy_key) return cleared def clear_discovered_repos(conn: sqlite3.Connection, *, policy_key: Optional[str] = None) -> None: with write_txn(conn): - conn.execute("DELETE FROM discovered_repos") - if policy_key is not None: - _upsert_meta_locked(conn, _DISCOVERY_POLICY_META_KEY, policy_key) + _clear_repos_locked(conn, True, policy_key) -# --- Discovered repos (filesystem scan cache) -------------------------------- - def record_discovered_repos( conn: sqlite3.Connection, repos: Iterable[tuple[str, Optional[str]]], *, replace: bool = False, policy_key: Optional[str] = None, @@ -478,12 +444,10 @@ def record_discovered_repos( return the row count. ``replace`` = authoritative fresh scan: stale rows are deleted first so old eval/worktree noise doesn't live forever.""" now = _now() - rows = [] - for root, label in repos: - norm = _normalize_path(root) - if norm: - rows.append((norm, (label or os.path.basename(norm) or norm), now)) - + rows = [ + (norm, label or os.path.basename(norm) or norm, now) + for norm, label in ((_normalize_path(root), label) for root, label in repos) if norm + ] with write_txn(conn): if replace: conn.execute("DELETE FROM discovered_repos") @@ -493,19 +457,15 @@ def record_discovered_repos( "ON CONFLICT(root) DO UPDATE SET label = excluded.label, last_seen = excluded.last_seen", rows, ) - if policy_key is not None: - _upsert_meta_locked(conn, _DISCOVERY_POLICY_META_KEY, policy_key) + _clear_repos_locked(conn, False, policy_key) return len(rows) def list_discovered_repos(conn: sqlite3.Connection) -> List[dict]: """All cached discovered repo roots, most-recently-seen first.""" - rows = conn.execute("SELECT root, label, last_seen FROM discovered_repos ORDER BY last_seen DESC").fetchall() - return [dict(r) for r in rows] + return [dict(r) for r in conn.execute("SELECT root, label, last_seen FROM discovered_repos ORDER BY last_seen DESC").fetchall()] -# --- Resolution + naming ----------------------------------------------------- - def project_for_path(conn: sqlite3.Connection, path: str, *, include_archived: bool = False) -> Optional[Project]: """Return the project owning ``path``: a folder owns it when equal or an ancestor, and the longest folder wins so nested projects resolve to the innermost one.""" @@ -515,28 +475,18 @@ def project_for_path(conn: sqlite3.Connection, path: str, *, include_archived: b sql = "SELECT pf.project_id AS pid, pf.path AS folder FROM project_folders pf JOIN projects p ON p.id = pf.project_id" if not include_archived: sql += " WHERE p.archived = 0" - best_pid: Optional[str] = None - best_len = -1 - for row in conn.execute(sql).fetchall(): - folder = row["folder"] + + def owns(folder: str) -> bool: stem = folder.rstrip("/\\") - owns = target == folder or target.startswith(stem + os.sep) or target.startswith(stem + "/") - if owns and len(folder) > best_len: - best_len = len(folder) - best_pid = row["pid"] - return None if best_pid is None else get_project(conn, best_pid) + return target == folder or target.startswith(stem + os.sep) or target.startswith(stem + "/") - -# Deterministic branch slug: lowercase, separators collapsed, capped. -_BRANCH_SAFE_RE = re.compile(r"[^a-z0-9._-]+") + owners = [row for row in conn.execute(sql).fetchall() if owns(row["folder"])] + return get_project(conn, max(owners, key=lambda r: len(r["folder"]))["pid"]) if owners else None def branch_name_for(project: Project, task_id: str, *, title: str = "") -> str: """Deterministic ``/[-]`` branch name for a project-linked kanban task (stable and human-meaningful, replacing the random ``wt/`` fallback).""" base = f"{project.slug or _slugify(project.name)}/{task_id}" - if title: - tslug = _BRANCH_SAFE_RE.sub("-", str(title).strip().lower()).strip("-")[:40].strip("-") - if tslug: - base = f"{base}-{tslug}" - return base + tslug = _BRANCH_SAFE_RE.sub("-", str(title).strip().lower()).strip("-")[:40].strip("-") if title else "" + return f"{base}-{tslug}" if tslug else base diff --git a/hermes_cli/session_export.py b/hermes_cli/session_export.py index 93a867129a..0ffdcb3f95 100644 --- a/hermes_cli/session_export.py +++ b/hermes_cli/session_export.py @@ -51,20 +51,27 @@ def render_sessions_export(sessions: Iterable[Dict[str, Any]], *, fmt: str = "js rows = iter_user_prompt_records(session_list) if prompts_only else session_list lines = [json.dumps(row, ensure_ascii=False) for row in rows] return ("\n".join(lines) + "\n") if lines else "" + # One session → its own H1 with body at H2; several → a shared H1, each session H2/H3. if prompts_only: - lines = _render_sessions_markdown( - session_list, "User prompts export", - lambda session: f"User prompts for session {_heading_text(_session_id(session))}", - lambda session: f"Session {_heading_text(_session_id(session))}", - _append_prompt_records, - ) - if not session_list: - lines += ["_No user prompts found._", ""] - return _finish_markdown(lines) - heading = lambda session: f"Session: {_heading_text(_session_title_or_id(session))}" # noqa: E731 - return _finish_markdown(_render_sessions_markdown( - session_list, "Hermes sessions export", heading, heading, _append_session_messages, - )) + multi_title, append_body = "User prompts export", _append_prompt_records + headings = (lambda s: f"User prompts for session {_heading_text(_session_id(s))}", + lambda s: f"Session {_heading_text(_session_id(s))}") + else: + multi_title, append_body = "Hermes sessions export", _append_session_messages + headings = (lambda s: f"Session: {_heading_text(_session_title_or_id(s))}",) * 2 + lines: List[str] = [] + single = len(session_list) == 1 + if not single: + lines += [f"# {multi_title}", ""] + for session in session_list: + level = 1 if single else 2 + lines += [f"{'#' * level} {headings[level - 1](session)}", *_session_metadata_lines(session), ""] + append_body(lines, session, heading_level=level + 1) + if prompts_only and not session_list: + lines += ["_No user prompts found._", ""] + while lines and lines[-1] == "": + lines.pop() + return "\n".join(lines) + "\n" def export_record_count(sessions: Iterable[Dict[str, Any]], *, only: Optional[str] = None) -> Tuple[int, str]: @@ -79,11 +86,8 @@ def iter_user_prompt_records(sessions: Iterable[Dict[str, Any]]) -> Iterator[Dic """Yield one normalized record for each user-authored prompt.""" for session in sessions: session_id = str(session.get("id") or session.get("session_id") or "") - index = 0 - for message in _messages(session): - if message.get("role") != "user": - continue - index += 1 + prompts = [m for m in _messages(session) if m.get("role") == "user"] + for index, message in enumerate(prompts, start=1): record: Dict[str, Any] = { "session_id": session_id, "index": index, @@ -98,21 +102,6 @@ def iter_user_prompt_records(sessions: Iterable[Dict[str, Any]]) -> Iterator[Dic yield record -def _render_sessions_markdown(sessions, multi_title, single_heading, multi_heading, append_body) -> List[str]: - """One session → its own H1 with body at H2; several → a shared H1, each session H2/H3.""" - lines: List[str] = [] - if len(sessions) == 1: - session = sessions[0] - lines += [f"# {single_heading(session)}", *_session_metadata_lines(session), ""] - append_body(lines, session, heading_level=2) - else: - lines += [f"# {multi_title}", ""] - for session in sessions: - lines += [f"## {multi_heading(session)}", *_session_metadata_lines(session), ""] - append_body(lines, session, heading_level=3) - return lines - - def _append_prompt_records(lines: List[str], session: Dict[str, Any], *, heading_level: int) -> None: prompts = list(iter_user_prompt_records([session])) if not prompts: @@ -213,12 +202,6 @@ def _fenced_text(text: str, *, language: str = "text") -> str: return f"{fence}{language}\n{text}\n{fence}" -def _finish_markdown(lines: List[str]) -> str: - while lines and lines[-1] == "": - lines.pop() - return "\n".join(lines) + "\n" - - # --- Current-session save helper (shared by CLI /save and gateway /save) --- SAVE_FORMATS = ("json", "md", "html") diff --git a/hermes_cli/session_export_html.py b/hermes_cli/session_export_html.py index bfd6dee1c4..fbf33e901f 100644 --- a/hermes_cli/session_export_html.py +++ b/hermes_cli/session_export_html.py @@ -642,6 +642,7 @@ HTML_TEMPLATE = """ """ + def _escape_html(text: Any) -> str: return ( str(text).replace("&", "&").replace("<", "<").replace(">", ">") @@ -715,34 +716,31 @@ def _generate_messages_html(messages: List[Dict[str, Any]]) -> str: ) if content: escaped = _escape_html(content) - body = f"
{escaped}
" if role == "tool" else escaped - html += f'
{body}
' + html += f'
{f"
{escaped}
" if role == "tool" else escaped}
' if reasoning := msg.get("reasoning") or msg.get("reasoning_content"): html += _collapsible( - "reasoning", ICON_SPARKLES, "Reasoning", - f'
{_escape_html(reasoning)}
', " " * 12, + "reasoning", ICON_SPARKLES, "Reasoning", f'
{_escape_html(reasoning)}
', " " * 12, ) html_list.append(html + " ") return "\n".join(html_list) -def _sidebar_html(sessions: List[Dict[str, Any]]) -> str: - items = [] - for s in sessions: - sid = str(s.get("id", "N/A")) - title = s.get("title") or s.get("preview") or "Untitled Session" - if len(title) > 50: - title = title[:47] + "..." - date = _format_timestamp(s.get("started_at", 0)).split(" ")[0] - items.append(f""" +def _sidebar_item_html(s: Dict[str, Any]) -> str: + sid = str(s.get("id", "N/A")) + title = s.get("title") or s.get("preview") or "Untitled Session" + title = title[:47] + "..." if len(title) > 50 else title + return f"""
{_escape_html(title)}
{_escape_html(sid[:8])} - {date} + {_format_timestamp(s.get("started_at", 0)).split(" ")[0]}
- """) + """ + + +def _sidebar_html(sessions: List[Dict[str, Any]]) -> str: return f""" """ @@ -763,13 +761,11 @@ def _sidebar_html(sessions: List[Dict[str, Any]]) -> str: def _session_view_html(s: Dict[str, Any], is_multi: bool) -> str: escaped_sid = _escape_html(str(s.get("id", "N/A"))) - system_html = "" - if system_prompt := s.get("system_prompt"): - system_html = _collapsible( - "system-prompt", ICON_SHIELD, "System Prompt (Persona)", - f'
{_escape_html(system_prompt)}
', " " * 12, - outer_class="system-prompt-section active", - ) + system_html = _collapsible( + "system-prompt", ICON_SHIELD, "System Prompt (Persona)", + f'
{_escape_html(s["system_prompt"])}
', " " * 12, + outer_class="system-prompt-section active", + ) if s.get("system_prompt") else "" return f"""
diff --git a/hermes_cli/session_export_md.py b/hermes_cli/session_export_md.py index 1e6544f480..6aa9b7da77 100644 --- a/hermes_cli/session_export_md.py +++ b/hermes_cli/session_export_md.py @@ -64,23 +64,19 @@ def _message_count(session: dict[str, Any]) -> int: def _render_messages(session: dict[str, Any]) -> str: - parts: list[str] = ["## Messages\n"] segments = _segments(session) if _message_count(session) == 0: - parts.append("_No messages in this session._\n") - return "\n".join(parts).rstrip() + "\n" - - multi_segment = len(segments) > 1 + return "## Messages\n\n_No messages in this session._\n" + parts: list[str] = ["## Messages\n"] for segment in segments: - if multi_segment: + if len(segments) > 1: parts.append(f"## Compression segment: {_session_id(segment)}\n") for message in list(segment.get("messages") or []): parts.append(_message_heading(message) + "\n") content = message.get("content") - if content is not None: - rendered = content.rstrip() if isinstance(content, str) else _json_block(content) - if rendered: - parts.append(rendered + "\n") + rendered = "" if content is None else content.rstrip() if isinstance(content, str) else _json_block(content) + if rendered: + parts.append(rendered + "\n") if tool_calls := message.get("tool_calls"): parts.append("\n\n## Tool calls\n\n" + _json_block(tool_calls) + "\n") parts.append("") @@ -91,7 +87,6 @@ def _export_body_without_hash(session: dict[str, Any], *, fmt: str, exported_at: session_id = _session_id(session) exported_iso = _iso_timestamp(exported_at) message_count = _message_count(session) - fields = [ ("session_id", session_id), ("title", session.get("title")), @@ -105,18 +100,14 @@ def _export_body_without_hash(session: dict[str, Any], *, fmt: str, exported_at: ("archived", bool(session.get("archived"))), ("message_count", message_count), ("tool_call_count", session.get("tool_call_count") or 0), + *([("lineage_session_ids", session["lineage_session_ids"])] if session.get("lineage_session_ids") else []), + ("format", fmt), ("exported_at", exported_iso), ("exporter", EXPORTER_VERSION), ] - if session.get("lineage_session_ids"): - fields.append(("lineage_session_ids", session.get("lineage_session_ids"))) - fields += [("format", fmt), ("exported_at", exported_iso), ("exporter", EXPORTER_VERSION)] - frontmatter = ["---", *(_frontmatter_line(k, v) for k, v in fields), "---", ""] - - parts = ["\n".join(frontmatter), f"# {session.get('title') or session_id}\n", f"Session ID: `{session_id}`\n"] - if session.get("source"): - parts.append(f"Source: `{session.get('source')}`\n") - if session.get("cwd"): - parts.append(f"Working directory: `{session.get('cwd')}`\n") - parts += [ + parts = [ + "\n".join(["---", *(_frontmatter_line(k, v) for k, v in fields), "---", ""]), + f"# {session.get('title') or session_id}\n", f"Session ID: `{session_id}`\n", + *([f"Source: `{session.get('source')}`\n"] if session.get("source") else []), + *([f"Working directory: `{session.get('cwd')}`\n"] if session.get("cwd") else []), _render_messages(session), f"{_VERIFICATION_HEADING}\n", f"- Session id: `{session_id}`", @@ -192,11 +183,7 @@ def redact_session_data(session: dict[str, Any]) -> dict[str, Any]: return {k: _clean(v) for k, v in value.items()} return value - redacted = dict(session) - for key in ("messages", "segments"): - if redacted.get(key) is not None: - redacted[key] = _clean(redacted[key]) - return redacted + return {**session, **{k: _clean(session[k]) for k in ("messages", "segments") if session.get(k) is not None}} def _export_dir(output_dir: Path | str) -> Path: diff --git a/hermes_cli/session_filters.py b/hermes_cli/session_filters.py index e23b9f5f9e..11f6212640 100644 --- a/hermes_cli/session_filters.py +++ b/hermes_cli/session_filters.py @@ -23,8 +23,6 @@ def parse_duration_seconds(value: str) -> Optional[float]: """Parse ``5h`` / ``30m`` / ``2d`` / ``1w`` / ``90`` (bare = days, backward compatible with ``--older-than 90``) into seconds. Returns None when the value doesn't look like a duration.""" s = str(value).strip().lower() - if not s: - return None if re.fullmatch(r"\d+(?:\.\d+)?", s): return float(s) * 86400 m = _DURATION_RE.match(s) @@ -99,11 +97,10 @@ def build_prune_filters(args: Any) -> Dict[str, Any]: ``--older-than`` / ``--newer-than`` bound last activity (latest message timestamp, falling back to ``started_at`` for empty sessions); ``--before`` / ``--after`` bound session start time. """ - bounds: Dict[str, Optional[float]] = {} - for key, attr, flag, _ in _TIME_BOUNDS: - raw = getattr(args, attr, None) - bounds[key] = None if raw is None else parse_point_in_time(raw, flag) - + bounds: Dict[str, Optional[float]] = { + key: None if (raw := getattr(args, attr, None)) is None else parse_point_in_time(raw, flag) + for key, attr, flag, _ in _TIME_BOUNDS + } for lo, hi, label, lo_flag, hi_flag in _WINDOWS: if bounds[hi] is not None and bounds[lo] is not None and bounds[lo] >= bounds[hi]: raise ValueError( @@ -114,20 +111,16 @@ def build_prune_filters(args: Any) -> Dict[str, Any]: # older_than_days=None: the epoch bounds are the whole story; otherwise prune_sessions' default # 90-day cutoff would silently cap an --after/--newer-than-only window. - filters: Dict[str, Any] = {"older_than_days": None, **bounds} - for key, attr, _ in _ARG_FILTERS: - filters[key] = getattr(args, attr, None) - return filters + return {"older_than_days": None, **bounds, **{key: getattr(args, attr, None) for key, attr, _ in _ARG_FILTERS}} def describe_filters(filters: Dict[str, Any]) -> str: """Human-readable summary of active filters for confirmation prompts.""" - parts = [] - for key, _, _, template in _TIME_BOUNDS: - if (value := filters.get(key)) is not None: - parts.append(template.format(v=format_epoch(value))) - for key, _, template in _ARG_FILTERS: - value = filters.get(key) - if (value is not None) if key.startswith(("min_", "max_")) else bool(value): - parts.append(template.format(v=value)) + parts = [ + template.format(v=format_epoch(filters[key])) for key, _, _, template in _TIME_BOUNDS + if filters.get(key) is not None + ] + [ + template.format(v=filters[key]) for key, _, template in _ARG_FILTERS + if ((filters.get(key) is not None) if key.startswith(("min_", "max_")) else bool(filters.get(key))) + ] return ", ".join(parts) if parts else "no filters (all ended sessions)" diff --git a/hermes_cli/session_listing.py b/hermes_cli/session_listing.py index a4c677bb12..7c4806a7a0 100644 --- a/hermes_cli/session_listing.py +++ b/hermes_cli/session_listing.py @@ -18,24 +18,20 @@ def parse_session_listing_args(raw_args: str) -> tuple[bool, bool, str, str | No a target so `/sessions ` can delegate to `/resume`. """ parts = shlex.split(raw_args or "") - include_all = False - include_unnamed = False + flags = {"all": False, "full": False} target_parts: list[str] = [] for i, part in enumerate(parts): lower = part.strip().lower() if not target_parts: if lower in _LIST_WORDS: continue - if lower in {"all", "--all"}: - include_all = True - continue - if lower in {"full", "--full"}: - include_unnamed = True + if lower in {"all", "--all", "full", "--full"}: + flags[lower.lstrip("-")] = True continue if lower in _SEARCH_WORDS: - return include_all, include_unnamed, "", " ".join(parts[i + 1:]).strip() + return flags["all"], flags["full"], "", " ".join(parts[i + 1:]).strip() target_parts.append(part) - return include_all, include_unnamed, " ".join(target_parts).strip(), None + return flags["all"], flags["full"], " ".join(target_parts).strip(), None def query_session_listing( @@ -71,13 +67,11 @@ def query_session_listing( result: list[dict[str, Any]] = [] for row in rows: is_current = bool(current_session_id and row.get("id") == current_session_id) - if is_current and not include_current_session: + if (is_current and not include_current_session) or ( + not include_unnamed and not row.get("title") and not search and not is_current + ): continue - if not include_unnamed and not row.get("title") and not search and not is_current: - continue - if is_current: - row = {**row, "is_current_session": True} - result.append(row) + result.append({**row, "is_current_session": True} if is_current else row) if len(result) >= limit: break return result @@ -96,15 +90,12 @@ def format_gateway_session_listing( (``all``) was declined, so the caller isn't left guessing why sessions are missing. """ if not rows: - parts = [ + return "\n".join([ "No sessions found.\n" "Use `/title My Session` to name this chat, or `/sessions full` " - "to include unnamed sessions." - ] - if notice: - parts.append(notice) - return "\n".join(parts) - + "to include unnamed sessions.", + *([notice] if notice else []), + ]) lines = [f"📋 **{title}**", ""] for idx, row in enumerate(rows, start=1): current_part = " (current)" if row.get("is_current_session") else "" @@ -116,9 +107,8 @@ def format_gateway_session_listing( f"{idx}. **{row.get('title') or '—'}**{current_part}{source_part}" f" — `{row.get('id') or ''}`{preview_part}" ) - lines.append("") - if notice: - lines.append(notice) - lines.append("Resume: `/resume ` or `/resume ` from `/resume`.") - lines.append("More: `/sessions all`, `/sessions full`, `/sessions search `.") - return "\n".join(lines) + return "\n".join([ + *lines, "", *([notice] if notice else []), + "Resume: `/resume ` or `/resume ` from `/resume`.", + "More: `/sessions all`, `/sessions full`, `/sessions search `.", + ])