refactor(hermes_cli): session-export/projects pass 2 — fold export renderers, projects_db row/meta helpers, projects_cmd str-return handlers

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 <action> [options]\nRun 'hermes project --help' for the full list.",
-                file=sys.stderr,
-            )
+            print("usage: hermes project <action> [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 <name>`.")
-        return 0
+        return "No projects yet. Create one with `hermes project create <name>`."
     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.<op>(conn, proj.id)`` followed by ``"<verb> <slug>"``."""
-
-    @_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 = {

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 ``<project-slug>/<task-id>[-<title-slug>]`` branch name for a project-linked kanban
     task (stable and human-meaningful, replacing the random ``wt/<task-id>`` 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

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")

hermes_cli/session_export_html.py

@@ -642,6 +642,7 @@ HTML_TEMPLATE = """<!DOCTYPE html>
 </html>
 """
 
+
 def _escape_html(text: Any) -> str:
     return (
         str(text).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
@@ -715,34 +716,31 @@ def _generate_messages_html(messages: List[Dict[str, Any]]) -> str:
             )
         if content:
             escaped = _escape_html(content)
-            body = f"<pre><code>{escaped}</code></pre>" if role == "tool" else escaped
-            html += f'  <div class="content">{body}</div>'
+            html += f'  <div class="content">{f"<pre><code>{escaped}</code></pre>" if role == "tool" else escaped}</div>'
         if reasoning := msg.get("reasoning") or msg.get("reasoning_content"):
             html += _collapsible(
-                "reasoning", ICON_SPARKLES, "Reasoning",
-                f'<div class="content">{_escape_html(reasoning)}</div>', " " * 12,
+                "reasoning", ICON_SPARKLES, "Reasoning", f'<div class="content">{_escape_html(reasoning)}</div>', " " * 12,
             )
         html_list.append(html + "  </div></div>")
     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"""
             <a class="session-item" data-id="{_escape_html(sid)}" href="#{quote(sid, safe='')}">
                 <div class="session-item-title">{_escape_html(title)}</div>
                 <div class="session-item-meta">
                     <span>{_escape_html(sid[:8])}</span>
-                    <span>{date}</span>
+                    <span>{_format_timestamp(s.get("started_at", 0)).split(" ")[0]}</span>
                 </div>
             </a>
-            """)
+            """
+
+
+def _sidebar_html(sessions: List[Dict[str, Any]]) -> str:
     return f"""
         <aside class="sidebar">
             <div class="sidebar-header">
@@ -755,7 +753,7 @@ def _sidebar_html(sessions: List[Dict[str, Any]]) -> str:
                 </div>
             </div>
             <div class="session-list">
-                {"".join(items)}
+                {"".join(_sidebar_item_html(s) for s in sessions)}
             </div>
         </aside>
         """
@@ -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'<div class="content">{_escape_html(system_prompt)}</div>', " " * 12,
-            outer_class="system-prompt-section active",
-        )
+    system_html = _collapsible(
+        "system-prompt", ICON_SHIELD, "System Prompt (Persona)",
+        f'<div class="content">{_escape_html(s["system_prompt"])}</div>', " " * 12,
+        outer_class="system-prompt-section active",
+    ) if s.get("system_prompt") else ""
     return f"""
         <div class="{"session-view" if is_multi else "session-view active"}" id="view-{escaped_sid}">
             <header class="fade-in">

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:

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)"

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 <id-or-title>` 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 <session id>` or `/resume <number>` from `/resume`.")
-    lines.append("More: `/sessions all`, `/sessions full`, `/sessions search <query>`.")
-    return "\n".join(lines)
+    return "\n".join([
+        *lines, "", *([notice] if notice else []),
+        "Resume: `/resume <session id>` or `/resume <number>` from `/resume`.",
+        "More: `/sessions all`, `/sessions full`, `/sessions search <query>`.",
+    ])