feat(cli): cluv estimate and cluv history subcommands

cluv estimate <cluster> <job.sh> [-- program-args...] reproduces the
spec_key and dry-runs the estimator, printing the historical records,
P95, growth slope, confidence, and the SBATCH_MEM it would set.

cluv history list|backfill|clear lets users inspect and manage the
local cache that the estimator and the retry loop both feed.

Author: wietzesuijker

cluv/__main__.py

@@ -19,6 +19,8 @@
 import rich_argparse
 import simple_parsing
 
+from .cli.estimate import estimate
+from .cli.history import history
 from .cli.init import init
 from .cli.login import login
 from .cli.run import run
@@ -37,15 +39,22 @@ def main(argv: list[str] | None = None) -> None:
         argv = sys.argv[1:]
 
     # argparse consumes '--' before REMAINDER sees it, so we extract program
-    # args (everything after the first '--' following 'submit') before parsing.
+    # args (everything after the first '--' following 'submit' or 'estimate')
+    # before parsing.
     submit_program_args: list[str] = []
-    try:
-        sub_idx = argv.index("submit")
-        sep_idx = argv.index("--", sub_idx + 1)
-        submit_program_args = list(argv[sep_idx + 1 :])
-        argv = list(argv[:sep_idx])
-    except ValueError:
-        pass
+    estimate_program_args: list[str] = []
+    for cmd_name, sink in (("submit", "submit"), ("estimate", "estimate")):
+        try:
+            sub_idx = argv.index(cmd_name)
+            sep_idx = argv.index("--", sub_idx + 1)
+            extracted = list(argv[sep_idx + 1 :])
+            argv = list(argv[:sep_idx])
+            if sink == "submit":
+                submit_program_args = extracted
+            else:
+                estimate_program_args = extracted
+        except ValueError:
+            continue
 
     parser = simple_parsing.ArgumentParser(
         description=__doc__,
@@ -75,6 +84,12 @@ def main(argv: list[str] | None = None) -> None:
     status_parser = add_status_args(subparsers)
     _add_v_arg(status_parser)
 
+    estimate_parser = add_estimate_args(subparsers)
+    _add_v_arg(estimate_parser)
+
+    history_parser = add_history_args(subparsers)
+    _add_v_arg(history_parser)
+
     args = parser.parse_args(argv)
     args_dict = vars(args)
 
@@ -85,6 +100,8 @@ def main(argv: list[str] | None = None) -> None:
 
     if subcommand == "submit":
         args_dict["program_args"] = submit_program_args
+    elif subcommand == "estimate":
+        args_dict["program_args"] = estimate_program_args
 
     try:
         if inspect.iscoroutinefunction(function):
@@ -136,6 +153,66 @@ def add_submit_args(
     return submit_parser
 
 
+def add_estimate_args(subparsers: Subparsers) -> argparse.ArgumentParser:
+    estimate_parser = subparsers.add_parser(
+        "estimate",
+        help="Dry-run the memory estimator for a job script on a cluster.",
+        formatter_class=rich_argparse.RichHelpFormatter,
+        usage="cluv estimate <cluster> <job.sh> [-- program-args...]",
+    )
+    estimate_parser.add_argument(
+        "cluster",
+        metavar="<cluster>",
+        help="The cluster whose history cache to read.",
+    )
+    estimate_parser.add_argument(
+        "job_script",
+        metavar="<job.sh>",
+        help="Path to the sbatch job script (relative to project root).",
+    )
+    estimate_parser.add_argument(
+        "--no-backfill",
+        dest="backfill",
+        action="store_false",
+        help="Skip the sacct backfill on a cold cache.",
+    )
+    estimate_parser.set_defaults(func=estimate, backfill=True)
+    return estimate_parser
+
+
+def add_history_args(subparsers: Subparsers) -> argparse.ArgumentParser:
+    history_parser = subparsers.add_parser(
+        "history",
+        help="Inspect or manage the local sacct memory history cache.",
+        formatter_class=rich_argparse.RichHelpFormatter,
+    )
+    history_parser.add_argument(
+        "action",
+        choices=("list", "backfill", "clear"),
+        help="What to do with the cache.",
+    )
+    history_parser.add_argument(
+        "cluster",
+        nargs="?",
+        default=None,
+        metavar="<cluster>",
+        help="Cluster name. Required for `backfill`. Optional filter for `list` and `clear`.",
+    )
+    history_parser.add_argument(
+        "--key",
+        default=None,
+        help="Spec key to target with `clear` (requires <cluster>).",
+    )
+    history_parser.add_argument(
+        "--since-days",
+        type=int,
+        default=60,
+        help="How many days of sacct to pull for `backfill`. Default: 60.",
+    )
+    history_parser.set_defaults(func=history)
+    return history_parser
+
+
 def add_status_args(subparsers: Subparsers) -> argparse.ArgumentParser:
     status_parser = subparsers.add_parser(
         "status",

cluv/cli/estimate.py

@@ -0,0 +1,89 @@
+"""`cluv estimate`: dry-run the memory estimator for the current spec."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from rich.table import Table
+
+from cluv import history as history_module
+from cluv.cli.login import get_remote_without_2fa_prompt
+from cluv.cli.submit import ensure_clean_git_state
+from cluv.config import get_config
+from cluv.utils import console
+
+logger = logging.getLogger(__name__)
+
+__all__ = ["estimate"]
+
+
+async def estimate(
+    cluster: str,
+    job_script: Path,
+    program_args: list[str],
+    backfill: bool,
+) -> None:
+    """Show what `cluv submit` would predict for memory, without submitting anything."""
+    from salvo.history import estimate_mem, spec_key
+
+    git_commit = ensure_clean_git_state()
+    key = spec_key(str(job_script), git_commit, tuple(program_args))
+
+    cfg = get_config().estimate
+    safety = cfg.safety if cfg else 1.2
+    window = cfg.window if cfg else 20
+    min_samples = cfg.min_samples if cfg else 3
+
+    console.print(f"spec key: [bold]{key}[/bold]")
+    console.print(f"cluster:  {cluster}")
+    console.print(f"safety={safety}  window={window}  min_samples={min_samples}")
+
+    records = history_module.load(cluster, key)
+    if not records and backfill:
+        remote = await get_remote_without_2fa_prompt(cluster)
+        if remote is None:
+            console.print(
+                f"[yellow]no active connection to {cluster}; skipping sacct backfill.[/yellow] "
+                f"Run `cluv login {cluster}` first to enable backfill."
+            )
+        else:
+            try:
+                n = await history_module.backfill_from_sacct(remote, cluster)
+                console.print(f"backfilled {n} record(s) from sacct on {cluster}")
+            except Exception as err:
+                console.print(f"[yellow]backfill failed: {err}[/yellow]")
+            records = history_module.load(cluster, key)
+
+    if not records:
+        console.print("[yellow]no records for this key; estimator would skip override.[/yellow]")
+        return
+
+    table = Table(title=f"history ({len(records)} record(s), newest first)")
+    table.add_column("job id")
+    table.add_column("state")
+    table.add_column("mem_mb", justify="right")
+    table.add_column("max_rss_mb", justify="right")
+    table.add_column("submitted_at")
+    for r in records[:window]:
+        table.add_row(
+            r.job_id,
+            r.state,
+            str(r.mem_mb),
+            "-" if r.max_rss_mb is None else str(r.max_rss_mb),
+            r.submitted_at.isoformat(timespec="minutes"),
+        )
+    console.print(table)
+
+    est = estimate_mem(records, safety=safety, window=window, min_samples=min_samples)
+    console.print(f"\n[bold]estimate:[/bold] {est.rationale}")
+    console.print(f"  confidence: {est.confidence}")
+    console.print(f"  n_samples:  {est.n_samples}")
+    if est.p95_mb is not None:
+        console.print(f"  p95_mb:     {est.p95_mb}")
+    if est.growth_slope_mb_per_run is not None:
+        console.print(f"  growth_slope_mb_per_run: {est.growth_slope_mb_per_run:.1f}")
+    if est.mem_mb is None:
+        console.print("[yellow]→ SBATCH_MEM would be left untouched.[/yellow]")
+    else:
+        console.print(f"[green]→ SBATCH_MEM would be set to {est.mem_mb}M.[/green]")

cluv/cli/history.py

@@ -0,0 +1,78 @@
+"""`cluv history`: inspect and manage the local sacct-derived memory cache."""
+
+from __future__ import annotations
+
+import logging
+
+from rich.table import Table
+
+from cluv import history as history_module
+from cluv.cli.login import get_remote_without_2fa_prompt
+from cluv.utils import console
+
+logger = logging.getLogger(__name__)
+
+__all__ = ["history"]
+
+
+async def history(
+    action: str,
+    cluster: str | None,
+    key: str | None,
+    since_days: int,
+) -> None:
+    """Dispatch to `list`, `backfill`, or `clear` based on `action`."""
+    if action == "list":
+        _list(cluster)
+    elif action == "backfill":
+        if not cluster:
+            console.print("[red]`cluv history backfill` requires a <cluster> argument.[/red]")
+            return
+        await _backfill(cluster, since_days)
+    elif action == "clear":
+        _clear(cluster, key)
+    else:
+        console.print(f"[red]unknown history action: {action}[/red]")
+
+
+def _list(cluster: str | None) -> None:
+    rows = history_module.list_keys(cluster)
+    if not rows:
+        scope = f" for {cluster}" if cluster else ""
+        console.print(f"[yellow]no records cached{scope}.[/yellow]")
+        console.print(f"cache dir: {history_module.cache_dir()}")
+        return
+    table = Table(title=f"history cache ({history_module.cache_dir()})")
+    table.add_column("cluster")
+    table.add_column("key")
+    table.add_column("records", justify="right")
+    for c, k, n in rows:
+        table.add_row(c, k, str(n))
+    console.print(table)
+
+
+async def _backfill(cluster: str, since_days: int) -> None:
+    remote = await get_remote_without_2fa_prompt(cluster)
+    if remote is None:
+        console.print(
+            f"[red]no active SSH connection to {cluster}.[/red] "
+            f"Run `cluv login {cluster}` first."
+        )
+        return
+    n = await history_module.backfill_from_sacct(remote, cluster, since_days=since_days)
+    console.print(f"backfilled {n} record(s) from sacct on {cluster} (last {since_days} days).")
+
+
+def _clear(cluster: str | None, key: str | None) -> None:
+    if key and not cluster:
+        console.print("[red]`--key` requires `<cluster>`.[/red]")
+        return
+    deleted = history_module.clear(cluster, key)
+    scope = (
+        f"({cluster}/{key})"
+        if cluster and key
+        else f"({cluster})"
+        if cluster
+        else "(all clusters)"
+    )
+    console.print(f"deleted {deleted} cache file(s) {scope}.")