fix: sessions radar — real codex IDs, codex projects, recency window, REPL filters

src/bach/cli/sessions.py

@@ -31,6 +31,7 @@
 from bach.runtimes.transcript import read_tail_events
 from bach.services.session_tracker import SESSIONS_DIRNAME
 from bach.services.sessions_service import (
+    DEFAULT_ENDED_WINDOW_S,
     SessionRow,
     adopt_session,
     find_artifact_for_ref,
@@ -84,7 +85,7 @@ def _default_artifacts_dir() -> Path:
 def sessions_list(
     project: Annotated[
         str | None,
-        typer.Option("--project", help="Filter by project path (exact)."),
+        typer.Option("--project", help="Filter by project path or name."),
     ] = None,
     runtime: Annotated[
         str | None,
@@ -98,6 +99,10 @@ def sessions_list(
         bool,
         typer.Option("--json", help="Output as JSON array instead of Rich table."),
     ] = False,
+    show_all: Annotated[
+        bool,
+        typer.Option("--all", help="Include ended sessions older than the 7-day window."),
+    ] = False,
 ) -> None:
     """List discovered runtime sessions, optionally filtered and joined with Bach tasks."""
     t = _theme()
@@ -111,6 +116,7 @@ def sessions_list(
             project=project,
             runtime=runtime,
             state=state,
+            max_ended_age_s=None if show_all else DEFAULT_ENDED_WINDOW_S,
         )
     except Exception as exc:
         _console.print(f"[{t.error}]error[/{t.error}] {exc}")

src/bach/repl/help_content.py

@@ -982,8 +982,8 @@ class CommandHelp:
     # =====================================================================
     "sessions": CommandHelp(
         synopsis="Show live and recent agent sessions (runtime-plane radar).",
-        syntax="/sessions",
-        args="(none — use `bach sessions list --state/--runtime/--project` for filters)",
+        syntax="/sessions [project] [running|idle|ended] [claude|codex] [all]",
+        args="free-form tokens: project name, state, runtime, `all` for full history",
         what=(
             "Scans ~/.claude/projects/ and ~/.codex/sessions/ to find all "
             "active and recently-ended Claude Code / Codex sessions. Shows "

src/bach/repl/sessions_view.py

@@ -25,7 +25,7 @@
 
 from bach.config.themes import ThemePalette
 from bach.domain.models import DiscoveredSession
-from bach.services.sessions_service import SessionRow
+from bach.services.sessions_service import DEFAULT_ENDED_WINDOW_S, SessionRow
 
 logger = logging.getLogger(__name__)
 
@@ -86,12 +86,39 @@ def render_sessions_table(
     logger.debug("event=sessions_table_rendered rows=%d", len(rows))
 
 
-def cmd_sessions(app: Any, _args: str) -> None:
+def _parse_sessions_args(args: str) -> tuple[str | None, str | None, str | None, bool]:
+    """Parse free-form /sessions tokens into (project, runtime, state, show_all).
+
+    Token classification by value, not position — `/sessions bach running`
+    and `/sessions running bach` mean the same thing. Unrecognized tokens
+    become the project filter (last one wins).
+    """
+    project: str | None = None
+    runtime: str | None = None
+    state: str | None = None
+    show_all = False
+    for token in args.split():
+        lowered = token.lower().lstrip("-")
+        if lowered in ("running", "idle", "ended"):
+            state = lowered
+        elif lowered in ("claude-code", "claude", "codex"):
+            runtime = "claude-code" if lowered == "claude" else lowered
+        elif lowered == "all":
+            show_all = True
+        elif lowered == "list":
+            continue  # tolerate the CLI-style `/sessions list` spelling
+        else:
+            project = token
+    return project, runtime, state, show_all
+
+
+def cmd_sessions(app: Any, args: str) -> None:
     """/sessions REPL command handler.
 
-    Discovers live sessions and renders them inline. No filters — the REPL
-    command is a quick-look tool; detailed filtering belongs to the CLI
-    `bach sessions list` command.
+    Accepts free-form filter tokens: a project name/path, a state
+    (running/idle/ended), a runtime (claude/codex), and `all` to include
+    ended sessions beyond the default 7-day window — e.g.
+    `/sessions bach running` or `/sessions codex all`.
 
     `app` is typed as object to avoid importing BachRepl here (circular dep).
     The duck-typed attribute access (app.console, app.current_theme) is safe:
@@ -123,12 +150,18 @@ def _discover() -> list[DiscoveredSession]:
             idle_threshold_s=cfg.liveness_idle_minutes * 60,
         )
 
+    project, runtime, state, show_all = _parse_sessions_args(args)
+
     try:
         rows = list_sessions(
             sessions_dir=bach_home() / "sessions",
             artifacts_dir=bach_home(),
             discover_fn=_discover,
             read_events_fn=read_tail_events,
+            project=project,
+            runtime=runtime,
+            state=state,
+            max_ended_age_s=None if show_all else DEFAULT_ENDED_WINDOW_S,
         )
     except Exception as exc:
         logger.warning("event=repl_sessions_error reason=%s", exc)

src/bach/runtimes/session_discovery.py

@@ -32,6 +32,7 @@
 
 import json
 import logging
+import re
 import subprocess
 from collections.abc import Callable
 from datetime import UTC, datetime
@@ -249,7 +250,11 @@ def _scan_codex(
             logger.debug("event=codex_discovery_skip_filename path=%s", transcript)
             continue
 
-        project_dir = index_map.get(session_id)
+        # The live session_index.jsonl carries only (id, thread_name,
+        # updated_at) — no working directory. The authoritative cwd lives in
+        # the rollout's first line (session_meta payload), so fall back to
+        # reading just that one line when the index can't resolve a project.
+        project_dir = index_map.get(session_id) or _read_codex_cwd(transcript)
         last_activity, liveness = _classify_liveness(
             transcript,
             has_runtime_process=has_codex_process,
@@ -271,6 +276,22 @@ def _scan_codex(
     return sessions
 
 
+def _read_codex_cwd(transcript: Path) -> Path | None:
+    """Best-effort cwd from a rollout's first line (session_meta payload).
+
+    Reads ONE line only — rollouts can be tens of MB and this runs per
+    session during discovery, so the read must stay O(1).
+    """
+    try:
+        with transcript.open(encoding="utf-8") as fh:
+            first = fh.readline()
+        record = json.loads(first)
+        cwd = record.get("payload", {}).get("cwd")
+        return Path(cwd) if isinstance(cwd, str) and cwd else None
+    except (OSError, json.JSONDecodeError, AttributeError):
+        return None
+
+
 def _read_codex_index(codex_home: Path) -> dict[str, Path]:
     """Read session_index.jsonl and return {session_id → project_dir}.
 
@@ -305,18 +326,22 @@ def _read_codex_index(codex_home: Path) -> dict[str, Path]:
     return result
 
 
+# Real Codex rollout filenames embed a HYPHENATED timestamp before the UUID
+# (e.g. 'rollout-2026-06-09T00-19-05-<uuid>.jsonl'), so splitting on the first
+# '-' returns a timestamp fragment, not the UUID. Anchoring a UUID regex at the
+# END of the stem is the only split that survives both timestamp formats.
+_UUID_TAIL_RE = re.compile(
+    r"([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})$"
+)
+
+
 def _extract_codex_session_id(filename: str) -> str | None:
     """Extract the session UUID from a Codex rollout filename.
 
-    Expected format: rollout-<ts>-<uuid>.jsonl
-    where <ts> has no hyphens (e.g. '20260115T120000Z') and <uuid>
-    may contain hyphens (e.g. '550e8400-e29b-41d4-a716-446655440000').
-
-    The extraction strategy: strip the 'rollout-' prefix and '.jsonl'
-    suffix, then split on '-' taking the SECOND token as the start of
-    the UUID (the first token is the timestamp). Because the UUID itself
-    contains hyphens, we must join everything after the first '-' separator
-    that follows the 'rollout-' prefix.
+    Expected format: rollout-<ts>-<uuid>.jsonl where <ts> is an ISO-ish
+    timestamp that itself contains hyphens (e.g. '2026-06-09T00-19-05')
+    and <uuid> is a standard hyphenated UUID. The UUID is matched from
+    the END of the stem so the timestamp's hyphens cannot confuse the split.
 
     Returns None if the filename doesn't match the expected pattern.
     """
@@ -325,14 +350,11 @@ def _extract_codex_session_id(filename: str) -> str | None:
     stem = filename[:-6]  # strip '.jsonl'
     if not stem.startswith("rollout-"):
         return None
-    rest = stem[len("rollout-") :]  # remove 'rollout-' prefix
-    # rest is now '<ts>-<uuid-with-hyphens>'
-    # The timestamp has no hyphens, so split on the FIRST '-' gives ts and uuid.
-    dash_pos = rest.find("-")
-    if dash_pos < 0:
-        # No separator between timestamp and UUID — malformed filename.
+    match = _UUID_TAIL_RE.search(stem)
+    if match is None:
+        # No trailing UUID — malformed or unexpected filename.
         return None
-    return rest[dash_pos + 1 :]
+    return match.group(1)
 
 
 # ---------------------------------------------------------------------------

src/bach/services/sessions_service.py

@@ -29,7 +29,7 @@
 import shlex
 from collections.abc import Callable
 from dataclasses import dataclass
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
 
@@ -292,6 +292,18 @@ def find_artifact_for_ref(
     return None
 
 
+# Default visibility window for ended sessions in list output (7 days).
+DEFAULT_ENDED_WINDOW_S = 7 * 86400
+
+
+def _project_matches(project_dir: Path | None, query: str) -> bool:
+    """Match a project filter against full path, dir name, or substring."""
+    if project_dir is None:
+        return False
+    path_str = str(project_dir)
+    return query == path_str or query == project_dir.name or query in path_str
+
+
 def list_sessions(
     *,
     sessions_dir: Path,
@@ -302,12 +314,25 @@ def list_sessions(
     runtime: str | None = None,
     state: str | None = None,
     artifact_paths: list[Path] | None = None,
+    max_ended_age_s: int | None = DEFAULT_ENDED_WINDOW_S,
+    now: datetime | None = None,
 ) -> list[SessionRow]:
     """Discover sessions and join them with Bach task artifacts.
 
     Filters are applied in order (project → runtime → state). All filters
     are AND-ed — a session must match every supplied filter.
 
+    The project filter matches the full path, the directory name, or a
+    path substring — users think in project names ("bach"), not absolute
+    paths.
+
+    Ended sessions older than max_ended_age_s are hidden by default
+    (None disables the window) so the table stays a radar, not an
+    archive — months of dead sessions otherwise drown the live ones.
+    Explicitly filtering state="ended" also disables the window: asking
+    for ended sessions means you want the history.
+    Rows are sorted by last_activity, newest first.
+
     artifact_paths can be injected for testing; when None, the function
     scans artifacts_dir recursively. This allows tests to provide specific
     artifacts without touching real project directories.
@@ -319,13 +344,23 @@ def list_sessions(
 
     # Apply filters before the join to avoid reading events for filtered-out sessions.
     if project is not None:
-        discovered = [
-            s for s in discovered if s.project_dir is not None and str(s.project_dir) == project
-        ]
+        discovered = [s for s in discovered if _project_matches(s.project_dir, project)]
     if runtime is not None:
         discovered = [s for s in discovered if s.runtime.value == runtime]
     if state is not None:
         discovered = [s for s in discovered if s.liveness.value == state]
+    elif max_ended_age_s is not None:
+        cutoff = (now or datetime.now(UTC)) - timedelta(seconds=max_ended_age_s)
+        discovered = [
+            s
+            for s in discovered
+            if s.liveness is not SessionLiveness.ended
+            or (s.last_activity is not None and s.last_activity >= cutoff)
+        ]
+
+    # Newest activity first; sessions with no timestamp sink to the bottom.
+    epoch = datetime.fromtimestamp(0, UTC)
+    discovered.sort(key=lambda s: s.last_activity or epoch, reverse=True)
 
     # Build artifact lookup from injected paths or scanned directory.
     paths = artifact_paths if artifact_paths is not None else _scan_artifact_paths(artifacts_dir)