feat: close sessions from Bach + Started column

src/bach/cli/sessions.py

@@ -16,6 +16,9 @@
 
 import json
 import logging
+import os
+import signal
+from datetime import datetime
 from pathlib import Path
 from typing import Annotated
 
@@ -27,13 +30,19 @@
 from bach.config.settings import load_config
 from bach.config.themes import THEMES, ThemePalette, get_theme
 from bach.domain.models import DiscoveredSession
-from bach.runtimes.session_discovery import default_process_probe, discover_sessions
+from bach.runtimes.session_discovery import (
+    default_cwd_probe,
+    default_process_probe,
+    discover_sessions,
+    find_session_pids,
+)
 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,
+    close_session,
     find_artifact_for_ref,
     list_sessions,
     resume_session,
@@ -139,6 +148,9 @@ def _output_sessions_json(rows: list[SessionRow]) -> None:
                 "runtime": row.session.runtime.value,
                 "liveness": row.session.liveness.value,
                 "project_dir": str(row.session.project_dir) if row.session.project_dir else None,
+                "started_at": (
+                    row.session.started_at.isoformat() if row.session.started_at else None
+                ),
                 "last_activity": (
                     row.session.last_activity.isoformat() if row.session.last_activity else None
                 ),
@@ -150,6 +162,16 @@ def _output_sessions_json(rows: list[SessionRow]) -> None:
     _console.print(json.dumps(output, indent=2))
 
 
+def _format_started(started_at: datetime | None) -> str:
+    """Compact local start time: today → HH:MM, else MM-DD HH:MM."""
+    if started_at is None:
+        return "—"
+    local = started_at.astimezone()
+    if local.date() == datetime.now().astimezone().date():
+        return local.strftime("%H:%M")
+    return local.strftime("%m-%d %H:%M")
+
+
 def _render_sessions_table(rows: list[SessionRow], t: ThemePalette) -> None:
     """Render discovered sessions as a Rich table."""
     if not rows:
@@ -162,6 +184,7 @@ def _render_sessions_table(rows: list[SessionRow], t: ThemePalette) -> None:
     table.add_column("State")
     table.add_column("Task")
     table.add_column("Project")
+    table.add_column("Started", style="dim")
     table.add_column("Preview", max_width=40)
 
     # Color mapping for liveness states — makes at-a-glance scanning fast.
@@ -179,13 +202,15 @@ def _render_sessions_table(rows: list[SessionRow], t: ThemePalette) -> None:
         project_name = s.project_dir.name if s.project_dir else "—"
         task_display = row.task_id or "—"
         preview_display = (row.preview or "")[:40] if row.preview else "—"
+        started_display = _format_started(s.started_at)
 
         table.add_row(
             sid_short,
             s.runtime.value,
             f"[{color}]{liveness_val}[/{color}]",
             task_display,
             project_name,
+            started_display,
             preview_display,
         )
 
@@ -338,3 +363,75 @@ def sessions_watch(
         judge_fn=judge_session,
     )
     raise typer.Exit(code=exit_code)
+
+
+# ---------------------------------------------------------------------------
+# `bach sessions close`
+# ---------------------------------------------------------------------------
+
+
+@sessions_app.command("close")
+def sessions_close(
+    session_ref: Annotated[str, typer.Argument(help="Session ID or unique prefix.")],
+    pid: Annotated[
+        int | None,
+        typer.Option("--pid", help="Disambiguate: kill exactly this candidate PID."),
+    ] = None,
+    yes: Annotated[
+        bool,
+        typer.Option("--yes", "-y", help="Skip the confirmation prompt."),
+    ] = False,
+) -> None:
+    """Terminate the process behind a running or idle session (SIGTERM).
+
+    Process attribution is heuristic (runtime binary + working directory),
+    so when several sessions share a project the command lists candidate
+    PIDs instead of guessing — rerun with --pid to pick one.
+    """
+    t = _theme()
+
+    if not yes and not typer.confirm(f"Send SIGTERM to the process behind {session_ref}?"):
+        raise typer.Exit(code=0)
+
+    def _find_pids(session: DiscoveredSession) -> list[int]:
+        return find_session_pids(
+            session,
+            process_probe=default_process_probe,
+            cwd_probe=default_cwd_probe,
+        )
+
+    try:
+        result = close_session(
+            session_ref,
+            discover_fn=_default_discover_fn,
+            find_pids_fn=_find_pids,
+            kill_fn=lambda p: os.kill(p, signal.SIGTERM),
+            pid_override=pid,
+        )
+    except ValueError as exc:
+        _console.print(f"[{t.error}]error[/{t.error}] {exc}")
+        raise typer.Exit(code=1) from exc
+
+    if result.reason == "terminated":
+        _console.print(
+            f"[{t.success}]closed[/{t.success}] session {result.session_id[:12]} "
+            f"(pid {result.killed_pid})"
+        )
+    elif result.reason == "not_running":
+        _console.print(f"[{t.muted}]session {result.session_id[:12]} already ended[/{t.muted}]")
+    elif result.reason == "no_process_found":
+        _console.print(
+            f"[{t.warning}]no live process attributable to "
+            f"{result.session_id[:12]}[/{t.warning}] — it may have exited already, "
+            "or its project has no unambiguous working directory"
+        )
+        raise typer.Exit(code=1)
+    else:  # ambiguous
+        _console.print(
+            f"[{t.warning}]multiple candidate processes[/{t.warning}] for "
+            f"{result.session_id[:12]} (same runtime + project):"
+        )
+        for cand in result.candidates:
+            _console.print(f"  pid {cand}")
+        _console.print(f"Pick one: bach sessions close {session_ref} --pid <PID>")
+        raise typer.Exit(code=1)

src/bach/domain/models.py

@@ -309,6 +309,9 @@ class DiscoveredSession:
     project_dir: Path | None  # decoded working dir if derivable, else None
     liveness: SessionLiveness
     last_activity: datetime | None  # transcript mtime; None on stat failure
+    # Transcript birth time (macOS st_birthtime, mtime fallback) — when the
+    # session started. Default None keeps older constructor sites valid.
+    started_at: datetime | None = None
 
 
 @dataclass(frozen=True)

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 [project] [running|idle|ended] [claude|codex] [all]",
-        args="free-form tokens: project name, state, runtime, `all` for full history",
+        syntax="/sessions [project] [state] [claude|codex] [all] | /sessions close <id>",
+        args="filter tokens in any order; `close <id-prefix>` ends a session (confirms first)",
         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

@@ -17,6 +17,10 @@
 from __future__ import annotations
 
 import logging
+import os
+import signal
+from collections.abc import Callable
+from datetime import datetime
 from pathlib import Path
 from typing import Any
 
@@ -59,6 +63,7 @@ def render_sessions_table(
     table.add_column("State", max_width=10)
     table.add_column("Task", max_width=8)
     table.add_column("Project", max_width=20)
+    table.add_column("Started", style="dim", max_width=12)
     table.add_column("Preview", max_width=44)
 
     for row in rows:
@@ -72,20 +77,32 @@ def render_sessions_table(
         project_name = s.project_dir.name if s.project_dir else "—"
         task_display = row.task_id or "—"
         preview_text = row.preview[:44] if row.preview else "—"
+        started_text = _format_started(s.started_at)
 
         table.add_row(
             sid_short,
             s.runtime.value,
             f"[{color}]{liveness_val}[/{color}]",
             task_display,
             project_name,
+            started_text,
             preview_text,
         )
 
     console.print(table)
     logger.debug("event=sessions_table_rendered rows=%d", len(rows))
 
 
+def _format_started(started_at: datetime | None) -> str:
+    """Compact local start time: today → HH:MM, else MM-DD HH:MM."""
+    if started_at is None:
+        return "—"
+    local = started_at.astimezone()
+    if local.date() == datetime.now().astimezone().date():
+        return local.strftime("%H:%M")
+    return local.strftime("%m-%d %H:%M")
+
+
 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).
 
@@ -150,6 +167,17 @@ def _discover() -> list[DiscoveredSession]:
             idle_threshold_s=cfg.liveness_idle_minutes * 60,
         )
 
+    tokens = args.split()
+    if tokens and tokens[0].lower() == "close":
+        ref = tokens[1] if len(tokens) > 1 else ""
+        if not ref:
+            console.print(
+                f"[{theme.error}]usage:[/{theme.error}] /sessions close <session-id-prefix>"
+            )
+            return
+        _close_from_repl(ref, console=console, theme=theme, discover_fn=_discover)
+        return
+
     project, runtime, state, show_all = _parse_sessions_args(args)
 
     try:
@@ -169,3 +197,75 @@ def _discover() -> list[DiscoveredSession]:
         return
 
     render_sessions_table(rows, console=console, theme=theme)
+
+
+def _close_from_repl(
+    ref: str,
+    *,
+    console: Console,
+    theme: ThemePalette,
+    discover_fn: Callable[[], list[DiscoveredSession]],
+    confirm_fn: Callable[[str], bool] | None = None,
+    kill_fn: Callable[[int], None] | None = None,
+) -> None:
+    """Terminate a session's process from inside the REPL.
+
+    Killing a process is the one destructive action in the sessions view,
+    so it always confirms first (plain input() — safe between prompt_toolkit
+    prompts). Ambiguity is never resolved by guessing: candidates are listed
+    and the user reruns with an exact PID via the CLI.
+    """
+    from bach.runtimes.session_discovery import (
+        default_cwd_probe,
+        default_process_probe,
+        find_session_pids,
+    )
+    from bach.services.sessions_service import close_session
+
+    def _default_confirm(prompt: str) -> bool:
+        try:
+            return input(prompt).strip().lower() in ("y", "yes")
+        except (EOFError, KeyboardInterrupt):
+            return False
+
+    confirm = confirm_fn or _default_confirm
+    if not confirm(f"Send SIGTERM to the process behind {ref}? [y/N] "):
+        console.print(f"[{theme.muted}]aborted[/{theme.muted}]")
+        return
+
+    def _find_pids(session: DiscoveredSession) -> list[int]:
+        return find_session_pids(
+            session,
+            process_probe=default_process_probe,
+            cwd_probe=default_cwd_probe,
+        )
+
+    try:
+        result = close_session(
+            ref,
+            discover_fn=discover_fn,
+            find_pids_fn=_find_pids,
+            kill_fn=kill_fn or (lambda p: os.kill(p, signal.SIGTERM)),
+        )
+    except ValueError as exc:
+        console.print(f"[{theme.error}]close error:[/{theme.error}] {exc}")
+        return
+
+    if result.reason == "terminated":
+        console.print(
+            f"[{theme.success}]closed[/{theme.success}] {result.session_id[:12]} "
+            f"(pid {result.killed_pid})"
+        )
+    elif result.reason == "not_running":
+        console.print(f"[{theme.muted}]{result.session_id[:12]} already ended[/{theme.muted}]")
+    elif result.reason == "no_process_found":
+        console.print(
+            f"[{theme.warning}]no live process attributable to "
+            f"{result.session_id[:12]}[/{theme.warning}]"
+        )
+    else:  # ambiguous
+        console.print(
+            f"[{theme.warning}]multiple candidates[/{theme.warning}] for {result.session_id[:12]}: "
+            f"{', '.join(str(c) for c in result.candidates)} — "
+            f"pick one with: bach sessions close {ref} --pid <PID>"
+        )