From e954301a092e7c87e71eb54b4a190824abe67911 Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Sun, 4 Oct 2026 16:34:03 +0000 Subject: [PATCH] feat(cli): pick a prior run interactively when --resume has no name strix --resume with no run name now lists the runs in ./strix_runs inline in the terminal (started, target, status, findings, run name), newest first, with arrow-key selection, type-to-search and esc to cancel. Picking a run continues through the same path as --resume . Headless or non-TTY launches error with the run list instead. --- strix/interface/cli_args.py | 85 +++++-- strix/interface/main.py | 40 +++- strix/interface/resume_picker.py | 392 +++++++++++++++++++++++++++++++ strix/report/runs.py | 86 +++++++ tests/test_cli_resume_picker.py | 111 +++++++++ tests/test_main_launch.py | 16 +- tests/test_resume_picker.py | 221 +++++++++++++++++ tests/test_run_summaries.py | 124 ++++++++++ 8 files changed, 1050 insertions(+), 25 deletions(-) create mode 100644 strix/interface/resume_picker.py create mode 100644 strix/report/runs.py create mode 100644 tests/test_cli_resume_picker.py create mode 100644 tests/test_resume_picker.py create mode 100644 tests/test_run_summaries.py diff --git a/strix/interface/cli_args.py b/strix/interface/cli_args.py index cea33099..f9675782 100644 --- a/strix/interface/cli_args.py +++ b/strix/interface/cli_args.py @@ -9,7 +9,7 @@ from pathlib import Path from strix.config import apply_config_override from strix.config.settings import DEFAULT_MAX_TURNS -from strix.core.paths import run_dir_for, runtime_state_dir +from strix.core.paths import RUNS_DIR_NAME, run_dir_for, runtime_state_dir from strix.interface.scan_setup import attach_workspace_mount, build_targets_info from strix.interface.update_check import self_update from strix.interface.utils import ( @@ -18,6 +18,7 @@ from strix.interface.utils import ( resolve_workspace_files, validate_config_file, ) +from strix.report.runs import list_run_summaries # Severities ``--fail-on`` accepts, most severe first. @@ -300,11 +301,14 @@ Strix Cloud: parser.add_argument( "--resume", type=str, + nargs="?", + const="", metavar="RUN_NAME", help=( "Resume a prior scan by its run name (the dir under ./strix_runs/). " "Picks up the root + every non-terminal subagent's full LLM history " - "and agent topology. Skips fresh run-name generation." + "and agent topology. Skips fresh run-name generation. Without a " + "name, opens a picker of the prior runs." ), ) @@ -312,6 +316,7 @@ Strix Cloud: # Startup-resolved state lives alongside the parsed flags. The full schema # is established here so downstream code reads attributes directly. args.needs_setup = False + args.resume_picker = False args.targets_info = [] args.local_sources = [] args.diff_scope = {"active": False} @@ -366,22 +371,28 @@ Strix Cloud: # transcript shows as their opening message. args.user_instruction = args.instruction or None - if args.resume: + if args.resume is not None: if args.target or args.target_list: parser.error( "Cannot combine --resume with --target/--target-list. " "--resume picks up where the prior run left off, including the " "original target list." ) - _load_resume_state(args, parser) - agents_path = runtime_state_dir(run_dir_for(args.resume)) / "agents.json" - if not agents_path.exists(): - parser.error( - f"--resume {args.resume}: missing {agents_path}. The run was " - f"persisted but never reached its first agent snapshot — " - f"there's nothing to resume from. Pick a fresh --run-name " - f"or remove --resume to start over with the same targets." - ) + if not args.resume.strip(): + # A bare --resume: main() opens the inline picker of prior runs + # before anything launches; headless has nobody to pick, so it + # lists them. + args.resume = None + args.resume_picker = True + if args.non_interactive: + parser.error(resume_run_list_message("--resume needs a run name in headless mode.")) + if not list_run_summaries(): + parser.error(f"--resume: no runs in ./{RUNS_DIR_NAME} to resume") + return args + try: + load_resume_state(args) + except ResumeError as exc: + parser.error(str(exc)) else: if not args.target and not args.target_list: if args.non_interactive: @@ -403,21 +414,45 @@ Strix Cloud: return args -def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None: - """Populate ``args.targets_info`` and friends from a prior run's run.json.""" +class ResumeError(ValueError): + """A prior run cannot be resumed as recorded; the message names why.""" + + +def resume_run_list_message(lead: str) -> str: + """``lead`` followed by the runs that ``--resume `` would accept.""" + runs = list_run_summaries() + if not runs: + return f"{lead} There are no runs in ./{RUNS_DIR_NAME}." + name_width = max(len(run.run_name) for run in runs) + status_width = max(len(run.status) for run in runs) + lines = [f"{lead} Runs in ./{RUNS_DIR_NAME}:"] + lines.extend( + f" {run.run_name:<{name_width}} {run.status:<{status_width}} " + f"{run.started_at[:19]:<19} {run.target}".rstrip() + for run in runs + ) + return "\n".join(lines) + + +def load_resume_state(args: argparse.Namespace) -> None: + """Populate ``args.targets_info`` and friends from a prior run's run.json. + + Raises :class:`ResumeError` when the run is missing, unreadable, or its + recorded workspace is gone. + """ from strix.report.writer import read_run_record run_dir = run_dir_for(args.resume) state_path = run_dir / "run.json" if not state_path.exists(): - parser.error( + raise ResumeError( f"--resume {args.resume}: no such run " f"(missing {state_path}; remove --resume for a fresh start)" ) try: state = read_run_record(run_dir) except (RuntimeError, TypeError) as exc: - parser.error(f"--resume {args.resume}: run.json unreadable: {exc}") + raise ResumeError(f"--resume {args.resume}: run.json unreadable: {exc}") from exc args.targets_info = state.get("targets_info") or [] # A target-less run has no targets_info at all. It is driven by its @@ -425,7 +460,7 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser # mount was declined, so either of those is enough to resume it. workspace_mount = state.get("workspace_mount") or None if not args.targets_info and not workspace_mount and not state.get("user_instruction"): - parser.error(f"--resume {args.resume}: run.json has no targets_info") + raise ResumeError(f"--resume {args.resume}: run.json has no targets_info") for target in args.targets_info: if not isinstance(target, dict): @@ -435,7 +470,7 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser try: check_mountable_dir(Path(details["target_path"]).expanduser()) except ValueError as exc: - parser.error(f"--resume {args.resume}: {exc}") + raise ResumeError(f"--resume {args.resume}: {exc}") from exc continue if target.get("type") != "repository": continue @@ -443,7 +478,7 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser if not cloned: continue if not Path(cloned).expanduser().exists(): - parser.error( + raise ResumeError( f"--resume {args.resume}: cloned repo at {cloned} is missing. " f"It was deleted between runs. Pick a fresh --run-name to " f"re-clone, or restore the directory before resuming." @@ -474,10 +509,10 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser try: args.workspace_files = resolve_workspace_files(restored) except ValueError as error: - parser.error(f"--resume {args.resume}: invalid workspace file: {error}") + raise ResumeError(f"--resume {args.resume}: invalid workspace file: {error}") from error if workspace_mount: if not Path(workspace_mount).expanduser().is_dir(): - parser.error( + raise ResumeError( f"--resume {args.resume}: the working directory {workspace_mount} " f"is missing. Restore it before resuming, or start a fresh run." ) @@ -487,3 +522,11 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser persisted_scan_mode = state.get("scan_mode") if persisted_scan_mode and args.scan_mode == "deep": args.scan_mode = persisted_scan_mode + agents_path = runtime_state_dir(run_dir) / "agents.json" + if not agents_path.exists(): + raise ResumeError( + f"--resume {args.resume}: missing {agents_path}. The run was " + f"persisted but never reached its first agent snapshot — " + f"there's nothing to resume from. Pick a fresh --run-name " + f"or remove --resume to start over with the same targets." + ) diff --git a/strix/interface/main.py b/strix/interface/main.py index 1f8bd4d5..fa8da3fa 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -15,8 +15,14 @@ from rich.panel import Panel from rich.text import Text from strix.config import codex, load_settings, persist_current -from strix.core.paths import run_dir_for -from strix.interface.cli_args import FAIL_ON_SEVERITIES, parse_arguments +from strix.core.paths import RUNS_DIR_NAME, run_dir_for +from strix.interface.cli_args import ( + FAIL_ON_SEVERITIES, + ResumeError, + load_resume_state, + parse_arguments, + resume_run_list_message, +) from strix.interface.environment import ( check_docker_installed, pull_docker_image, @@ -391,6 +397,33 @@ def _print_model_connection_error(exc: BaseException, model_name: str) -> None: console.print() +def _print_cli_error(message: str) -> None: + Console(stderr=True, soft_wrap=True).print( + f"strix: error: {message}", markup=False, highlight=False + ) + + +def _pick_run_to_resume(args: argparse.Namespace) -> None: + """A bare --resume: let the user pick a run, then load it like --resume .""" + from strix.interface.resume_picker import PickerUnavailableError, pick_run + from strix.report.runs import list_run_summaries + + try: + chosen = pick_run(list_run_summaries(), runs_dir=RUNS_DIR_NAME) + except PickerUnavailableError as exc: + _print_cli_error(resume_run_list_message(f"{exc}.")) + sys.exit(2) + if chosen is None: + Console().print("No run selected.", style="dim") + sys.exit(0) + args.resume = chosen.run_name + try: + load_resume_state(args) + except ResumeError as exc: + _print_cli_error(str(exc)) + sys.exit(2) + + def _bootstrap_scan(args: argparse.Namespace) -> None: """Warm up the model and prepare the run before the interface starts. @@ -475,6 +508,9 @@ def main() -> None: restart_after_update() sys.exit(0) + if args.resume_picker: + _pick_run_to_resume(args) + check_docker_installed() pull_docker_image() validate_environment() diff --git a/strix/interface/resume_picker.py b/strix/interface/resume_picker.py new file mode 100644 index 00000000..8a4e2d1f --- /dev/null +++ b/strix/interface/resume_picker.py @@ -0,0 +1,392 @@ +"""Inline arrow-key picker for ``strix --resume`` with no run name. + +Draws a short list in the normal terminal (no alternate screen), redraws it in +place on every key, and returns the chosen run. Keys: up/down, page up/down, +home/end, enter, esc, typing filters the rows, backspace edits the filter. +""" + +from __future__ import annotations + +import os +import sys +from datetime import UTC, datetime +from typing import TYPE_CHECKING, TextIO + +from rich.console import Console +from rich.text import Text + + +if TYPE_CHECKING: + from collections.abc import Callable + + from strix.report.runs import RunSummary + + +KEY_UP = "up" +KEY_DOWN = "down" +KEY_PAGE_UP = "pageup" +KEY_PAGE_DOWN = "pagedown" +KEY_HOME = "home" +KEY_END = "end" +KEY_ENTER = "enter" +KEY_ESCAPE = "escape" +KEY_BACKSPACE = "backspace" +KEY_INTERRUPT = "interrupt" + +_CSI_KEYS = { + "A": KEY_UP, + "B": KEY_DOWN, + "H": KEY_HOME, + "F": KEY_END, + "1~": KEY_HOME, + "4~": KEY_END, + "5~": KEY_PAGE_UP, + "6~": KEY_PAGE_DOWN, + "7~": KEY_HOME, + "8~": KEY_END, +} +_WINDOWS_KEYS = { + "H": KEY_UP, + "P": KEY_DOWN, + "I": KEY_PAGE_UP, + "Q": KEY_PAGE_DOWN, + "G": KEY_HOME, + "O": KEY_END, +} +_CONTROL_KEYS = { + "\r": KEY_ENTER, + "\n": KEY_ENTER, + "\x1b": KEY_ESCAPE, + "\x7f": KEY_BACKSPACE, + "\x08": KEY_BACKSPACE, + "\x03": KEY_INTERRUPT, +} + +_CURSOR_HIDE = "\x1b[?25l" +_CURSOR_SHOW = "\x1b[?25h" +_CLEAR_BELOW = "\x1b[J" + +_GREEN = "#22c55e" +_AMBER = "#f59e0b" +_STATUS_STYLES = { + "completed": _GREEN, + "running": _GREEN, + "interrupted": _AMBER, + "stopped": _AMBER, + "failed": "#ef4444", +} +_NO_STATE = "no state" +_CURSOR = " \u276f " +_STARTED_WIDTH = 14 +_FINDINGS_WIDTH = 8 +_MIN_TARGET_WIDTH = 12 +_MAX_RUN_WIDTH = 40 +_CHROME_LINES = 8 + + +class PickerUnavailableError(RuntimeError): + """The picker needs an interactive terminal on stdin and stdout.""" + + +if sys.platform == "win32": + import msvcrt + + def read_raw_key(_stream: TextIO) -> str: + char = str(msvcrt.getwch()) + if char in ("\x00", "\xe0"): + return _WINDOWS_KEYS.get(str(msvcrt.getwch()), "") + return char + +else: + import select + import termios + import tty + + def read_raw_key(stream: TextIO) -> str: + fd = stream.fileno() + saved = termios.tcgetattr(fd) + try: + tty.setcbreak(fd) + char = os.read(fd, 1).decode("utf-8", "replace") + if char != "\x1b": + return char + sequence = "" + while select.select([fd], [], [], 0.05)[0]: + sequence += os.read(fd, 1).decode("utf-8", "replace") + if (sequence.startswith("[") and sequence[-1].isalpha()) or sequence.endswith("~"): + break + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, saved) + if not sequence: + return "\x1b" + return _CSI_KEYS.get(sequence[1:], "") if sequence[0] in "[O" else "" + + +def read_key(stream: TextIO) -> str: + return translate_key(read_raw_key(stream)) + + +def translate_key(raw: str) -> str: + return _CONTROL_KEYS.get(raw, raw) + + +def relative_time(stamp: str, now: datetime | None = None) -> str: + try: + started = datetime.fromisoformat(stamp) + except ValueError: + return stamp[:16] if stamp else "unknown" + if started.tzinfo is None: + started = started.replace(tzinfo=UTC) + now = now or datetime.now(UTC) + seconds = (now - started).total_seconds() + for limit, unit, label in ( + (60, 1, ""), + (3600, 60, "min"), + (86400, 3600, "h"), + (7 * 86400, 86400, "d"), + ): + if seconds < limit: + return "just now" if not label else f"{int(seconds // unit)} {label} ago" + local = started.astimezone() + if local.year == now.astimezone().year: + return f"{local:%b} {local.day}, {local:%H:%M}" + return f"{local:%b} {local.day}, {local.year}" + + +def filter_runs(runs: list[RunSummary], needle: str) -> list[RunSummary]: + needle = needle.strip().lower() + if not needle: + return list(runs) + return [ + run + for run in runs + if needle in run.run_name.lower() + or needle in run.target.lower() + or needle in run.status.lower() + ] + + +def _fit(text: str, width: int) -> str: + if len(text) <= width: + return text.ljust(width) + return text[: width - 1] + "\u2026" + + +def _status_text(run: RunSummary) -> str: + return run.status if run.resumable else f"{run.status} \u00b7 {_NO_STATE}" + + +class ResumePicker: + def __init__( + self, + runs: list[RunSummary], + *, + console: Console, + runs_dir: str, + now: datetime | None = None, + ) -> None: + self.runs = runs + self.console = console + self.runs_dir = runs_dir + self.now = now + self.filter = "" + self.cursor = 0 + self.offset = 0 + self.notice = "" + self._drawn = 0 + + @property + def rows(self) -> list[RunSummary]: + return filter_runs(self.runs, self.filter) + + def _visible(self) -> int: + return max(3, min(len(self.rows), self.console.height - _CHROME_LINES)) + + def _columns(self) -> tuple[int, int, int]: + width = max(40, self.console.width - 1) + status_width = max(len("status"), *(len(_status_text(run)) for run in self.runs)) + run_width = min(_MAX_RUN_WIDTH, max(len("run"), *(len(run.run_name) for run in self.runs))) + fixed = len(_CURSOR) + _STARTED_WIDTH + _FINDINGS_WIDTH + status_width + 4 * 2 + target_width = width - fixed - run_width + if target_width < _MIN_TARGET_WIDTH: + run_width = max(8, run_width + target_width - _MIN_TARGET_WIDTH) + target_width = width - fixed - run_width + return max(_MIN_TARGET_WIDTH, target_width), status_width, run_width + + def _scroll(self) -> range: + rows = self.rows + visible = self._visible() + self.cursor = max(0, min(self.cursor, len(rows) - 1)) + if self.cursor < self.offset: + self.offset = self.cursor + elif self.cursor >= self.offset + visible: + self.offset = self.cursor - visible + 1 + self.offset = max(0, min(self.offset, max(0, len(rows) - visible))) + return range(self.offset, min(len(rows), self.offset + visible)) + + def render(self) -> list[Text]: + rows = self.rows + widths = self._columns() + window = self._scroll() + + title = Text() + title.append(" Resume a run", style="bold") + title.append(f" {len(self.runs)} runs in ./{self.runs_dir}", style="dim") + if self.filter: + title.append(" search: ", style="dim") + title.append(self.filter) + header = Text( + " " * len(_CURSOR) + + self._cells("started", "target", "status", "findings", "run", widths), + style="dim", + ) + lines = [title, Text(), header] + if not rows: + lines.append(Text(f" no runs match {self.filter!r}", style="dim")) + if window.start: + lines.append(Text(f" \u2026 {window.start} more above", style="dim")) + lines.extend(self._row(rows[index], index == self.cursor, widths) for index in window) + if window.stop < len(rows): + lines.append(Text(f" \u2026 {len(rows) - window.stop} more below", style="dim")) + footer = Text(" ") + if self.notice: + footer.append(self.notice, style=_AMBER) + else: + footer.append( + "\u2191\u2193 move enter resume type to search esc cancel", style="dim" + ) + lines.extend([Text(), footer]) + return lines + + @staticmethod + def _cells( + started: str, + target: str, + status: str, + findings: str, + run: str, + widths: tuple[int, int, int], + ) -> str: + target_width, status_width, run_width = widths + return " ".join( + [ + _fit(started, _STARTED_WIDTH), + _fit(target, target_width), + _fit(status, status_width), + _fit(findings, _FINDINGS_WIDTH), + _fit(run, run_width), + ] + ) + + def _row(self, run: RunSummary, selected: bool, widths: tuple[int, int, int]) -> Text: + target_width, status_width, run_width = widths + primary = "bold" if selected else "" + muted = "" if selected else "dim" + status_style = _STATUS_STYLES.get(run.status, "") + if not run.resumable: + primary = muted = status_style = "dim" + line = Text() + line.append(_CURSOR if selected else " " * len(_CURSOR), style=_GREEN) + line.append(_fit(relative_time(run.started_at, self.now), _STARTED_WIDTH), style=muted) + line.append(" ") + line.append(_fit(run.target, target_width), style=primary) + line.append(" ") + line.append(_fit(_status_text(run), status_width), style=status_style) + line.append(" ") + line.append(_fit(str(run.findings), _FINDINGS_WIDTH), style=muted) + line.append(" ") + line.append(_fit(run.run_name, run_width), style=muted) + return line + + def draw(self) -> None: + self.clear() + lines = self.render() + for line in lines: + self.console.print(line, soft_wrap=True, overflow="crop", end="\n") + self.console.file.flush() + self._drawn = len(lines) + + def clear(self) -> None: + if self._drawn: + self.console.file.write(f"\x1b[{self._drawn}A\r{_CLEAR_BELOW}") + self.console.file.flush() + self._drawn = 0 + + def handle(self, key: str) -> tuple[bool, RunSummary | None]: + """Apply one key: (done, run), where done with no run means cancelled.""" + self.notice = "" + if key == KEY_INTERRUPT: + raise KeyboardInterrupt + if key == KEY_ESCAPE: + if not self.filter: + return True, None + self.filter = "" + self.cursor = 0 + return False, None + if key == KEY_ENTER: + return self._choose() + if key == KEY_BACKSPACE: + self.filter = self.filter[:-1] + self.cursor = 0 + elif len(key) == 1 and key.isprintable(): + self.filter += key + self.cursor = 0 + else: + self._move(key) + self.cursor = max(0, min(self.cursor, max(0, len(self.rows) - 1))) + return False, None + + def _choose(self) -> tuple[bool, RunSummary | None]: + rows = self.rows + if not rows: + return False, None + run = rows[self.cursor] + if not run.resumable: + self.notice = f"{run.run_name} has no saved agent state to resume from" + return False, None + return True, run + + def _move(self, key: str) -> None: + steps = { + KEY_UP: -1, + KEY_DOWN: 1, + KEY_PAGE_UP: -self._visible(), + KEY_PAGE_DOWN: self._visible(), + } + if key in steps: + self.cursor += steps[key] + elif key == KEY_HOME: + self.cursor = 0 + elif key == KEY_END: + self.cursor = len(self.rows) - 1 + + def run(self, next_key: Callable[[], str]) -> RunSummary | None: + self.console.file.write(_CURSOR_HIDE) + try: + self.draw() + while True: + done, chosen = self.handle(next_key()) + if done: + return chosen + self.draw() + finally: + self.clear() + self.console.file.write(_CURSOR_SHOW) + self.console.file.flush() + + +def pick_run( + runs: list[RunSummary], + *, + runs_dir: str, + console: Console | None = None, + stdin: TextIO | None = None, +) -> RunSummary | None: + stdin = stdin or sys.stdin + console = console or Console() + if not (hasattr(stdin, "isatty") and stdin.isatty() and console.is_terminal): + raise PickerUnavailableError( + "--resume needs a run name when there is no interactive terminal" + ) + picker = ResumePicker(runs, console=console, runs_dir=runs_dir) + return picker.run(lambda: read_key(stdin)) diff --git a/strix/report/runs.py b/strix/report/runs.py new file mode 100644 index 00000000..c2e82046 --- /dev/null +++ b/strix/report/runs.py @@ -0,0 +1,86 @@ +"""Summaries of the runs under ./strix_runs, for pickers and listings.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from strix.core.paths import run_record_path, runs_base_dir, runtime_state_dir + + +if TYPE_CHECKING: + from pathlib import Path + + +@dataclass(frozen=True) +class RunSummary: + run_name: str + target: str + started_at: str + ended_at: str + status: str + findings: int + resumable: bool + + +def list_run_summaries(*, cwd: Path | None = None) -> list[RunSummary]: + """Every run with a run.json, newest activity first (run.json mtime).""" + base = runs_base_dir(cwd=cwd) + if not base.is_dir(): + return [] + rows: list[tuple[float, RunSummary]] = [] + for child in base.iterdir(): + record_path = run_record_path(child) + try: + if not record_path.is_file(): + continue + modified = record_path.stat().st_mtime + except OSError: + continue + rows.append((modified, _summarize(child, _load_json(record_path, default={})))) + rows.sort(key=lambda row: (row[0], row[1].run_name), reverse=True) + return [summary for _, summary in rows] + + +def _summarize(run_dir: Path, record: Any) -> RunSummary: + if not isinstance(record, dict): + record = {} + findings = _load_json(run_dir / "vulnerabilities.json", default=[]) + return RunSummary( + run_name=run_dir.name, + target=_describe_target(record), + started_at=str(record.get("start_time") or ""), + ended_at=str(record.get("end_time") or ""), + status=str(record.get("status") or "unknown"), + findings=len(findings) if isinstance(findings, list) else 0, + resumable=(runtime_state_dir(run_dir) / "agents.json").is_file(), + ) + + +def _describe_target(record: dict[str, Any]) -> str: + targets = record.get("targets_info") + originals = [ + str(entry["original"]) + for entry in targets or [] + if isinstance(entry, dict) and entry.get("original") + ] + if originals: + if len(originals) == 1: + return originals[0] + return f"{originals[0]} +{len(originals) - 1} more" + mount = record.get("workspace_mount") + if isinstance(mount, str) and mount: + return f"{mount} (workspace)" + instruction = record.get("user_instruction") or record.get("instruction") + return str(instruction or "").strip().splitlines()[0] if instruction else "" + + +def _load_json(path: Path, *, default: Any) -> Any: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return default + + +__all__ = ["RunSummary", "list_run_summaries"] diff --git a/tests/test_cli_resume_picker.py b/tests/test_cli_resume_picker.py new file mode 100644 index 00000000..56096187 --- /dev/null +++ b/tests/test_cli_resume_picker.py @@ -0,0 +1,111 @@ +"""Tests for `strix --resume` without a run name.""" + +from __future__ import annotations + +import importlib +import json +import sys +from typing import TYPE_CHECKING, Any + +import pytest + + +if TYPE_CHECKING: + from pathlib import Path + + +cli_main: Any = importlib.import_module("strix.interface.main") + + +def _write_run(base: Path, name: str, *, state: bool = True) -> None: + run_dir = base / name + (run_dir / ".state").mkdir(parents=True) + (run_dir / "run.json").write_text( + json.dumps( + { + "run_name": name, + "status": "completed", + "start_time": "2026-10-04T10:00:00+00:00", + "targets_info": [ + { + "type": "web_application", + "details": {"target_url": "https://example.com"}, + "original": "https://example.com", + } + ], + } + ), + encoding="utf-8", + ) + if state: + (run_dir / ".state" / "agents.json").write_text("{}", encoding="utf-8") + + +def test_bare_resume_defers_to_the_picker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + _write_run(tmp_path / "strix_runs", "example-com_1111") + monkeypatch.setattr(sys, "argv", ["strix", "--resume"]) + + args = cli_main.parse_arguments() + + assert args.resume is None + assert args.resume_picker is True + assert args.needs_setup is False + assert args.resume is None + assert args.targets_info == [] + + +def test_bare_resume_with_no_runs_is_an_error( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(sys, "argv", ["strix", "--resume"]) + + with pytest.raises(SystemExit): + cli_main.parse_arguments() + + assert "no runs in ./strix_runs" in capsys.readouterr().err + + +def test_bare_resume_headless_lists_the_runs_instead_of_prompting( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.chdir(tmp_path) + _write_run(tmp_path / "strix_runs", "example-com_1111") + _write_run(tmp_path / "strix_runs", "example-com_2222", state=False) + monkeypatch.setattr(sys, "argv", ["strix", "-n", "--resume"]) + + with pytest.raises(SystemExit): + cli_main.parse_arguments() + + err = capsys.readouterr().err + assert "--resume needs a run name in headless mode" in err + assert "example-com_1111 completed 2026-10-04T10:00:00 https://example.com" in err + assert "example-com_2222 completed 2026-10-04T10:00:00 https://example.com" in err + assert "example-com_2222" in err + + +def test_bare_resume_still_rejects_targets( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.chdir(tmp_path) + _write_run(tmp_path / "strix_runs", "example-com_1111") + monkeypatch.setattr(sys, "argv", ["strix", "--resume", "-t", "https://example.com"]) + + with pytest.raises(SystemExit): + cli_main.parse_arguments() + + assert "Cannot combine --resume with --target" in capsys.readouterr().err + + +def test_named_resume_without_agent_state_is_an_error( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.chdir(tmp_path) + _write_run(tmp_path / "strix_runs", "example-com_1111", state=False) + monkeypatch.setattr(sys, "argv", ["strix", "--resume", "example-com_1111"]) + + with pytest.raises(SystemExit): + cli_main.parse_arguments() + + assert "never reached its first agent snapshot" in capsys.readouterr().err diff --git a/tests/test_main_launch.py b/tests/test_main_launch.py index 79ee656b..0651d126 100644 --- a/tests/test_main_launch.py +++ b/tests/test_main_launch.py @@ -13,11 +13,14 @@ cli_main: Any = importlib.import_module("strix.interface.main") report_state_module: Any = importlib.import_module("strix.report.state") -def _launch(monkeypatch: pytest.MonkeyPatch, *, needs_setup: bool) -> list[str]: +def _launch( + monkeypatch: pytest.MonkeyPatch, *, needs_setup: bool, resume_picker: bool = False +) -> list[str]: calls: list[str] = [] args = argparse.Namespace( non_interactive=False, needs_setup=needs_setup, + resume_picker=resume_picker, run_name=None, fail_on=None, ) @@ -36,6 +39,7 @@ def _launch(monkeypatch: pytest.MonkeyPatch, *, needs_setup: bool) -> list[str]: monkeypatch.setattr(cli_main, "validate_environment", lambda: None) monkeypatch.setattr(cli_main, "wait_for_import_warmup", lambda: None) monkeypatch.setattr(cli_main, "_bootstrap_scan", lambda _args: calls.append("bootstrap")) + monkeypatch.setattr(cli_main, "_pick_run_to_resume", lambda _args: calls.append("pick")) monkeypatch.setattr(cli_main, "run_tui", run_tui) monkeypatch.setattr(cli_main, "notify_update", lambda _console: None) @@ -53,6 +57,14 @@ def test_start_screen_launch_defers_the_model_check_to_the_tui( assert _launch(monkeypatch, needs_setup=True) == ["tui"] +def test_bare_resume_picks_a_run_before_the_preflight(monkeypatch: pytest.MonkeyPatch) -> None: + assert _launch(monkeypatch, needs_setup=False, resume_picker=True) == [ + "pick", + "bootstrap", + "tui", + ] + + def test_direct_launch_with_a_bad_key_prints_the_panel_and_exits_before_the_tui( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -84,7 +96,7 @@ def test_tui_startup_failure_marks_the_prepared_run_failed(monkeypatch: pytest.M calls: list[str] = [] report_state = SimpleNamespace(cleanup=lambda status: calls.append(f"cleanup:{status}")) args = argparse.Namespace( - non_interactive=False, needs_setup=False, run_name="run", fail_on=None + non_interactive=False, needs_setup=False, resume_picker=False, run_name="run", fail_on=None ) async def run_tui(_args: argparse.Namespace) -> None: diff --git a/tests/test_resume_picker.py b/tests/test_resume_picker.py new file mode 100644 index 00000000..bbf640ce --- /dev/null +++ b/tests/test_resume_picker.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +import io +from datetime import UTC, datetime, timedelta + +import pytest +from rich.console import Console + +from strix.interface.resume_picker import ( + KEY_BACKSPACE, + KEY_DOWN, + KEY_END, + KEY_ENTER, + KEY_ESCAPE, + KEY_INTERRUPT, + KEY_UP, + PickerUnavailableError, + ResumePicker, + filter_runs, + pick_run, + relative_time, + translate_key, +) +from strix.report.runs import RunSummary + + +NOW = datetime(2026, 10, 4, 16, 0, tzinfo=UTC) + + +def _run( + name: str, + target: str, + *, + minutes_ago: int, + status: str = "completed", + findings: int = 0, + resumable: bool = True, +) -> RunSummary: + started = (NOW - timedelta(minutes=minutes_ago)).isoformat() + return RunSummary( + run_name=name, + target=target, + started_at=started, + ended_at=started, + status=status, + findings=findings, + resumable=resumable, + ) + + +RUNS = [ + _run("example-com_e5e6", "https://example.com", minutes_ago=12), + _run( + "example-com_223e", "https://example.com", minutes_ago=40, status="interrupted", findings=2 + ), + _run( + "juice-shop_8a1f", + "https://juice-shop.herokuapp.com", + minutes_ago=60 * 26, + status="stopped", + findings=7, + ), + _run( + "strix_41c0", + "/home/me/code/strix (workspace)", + minutes_ago=60 * 50, + status="failed", + resumable=False, + ), +] + + +def _console(width: int = 110, height: int = 30) -> tuple[Console, io.StringIO]: + buffer = io.StringIO() + console = Console( + file=buffer, width=width, height=height, force_terminal=True, color_system="truecolor" + ) + return console, buffer + + +def _picker(runs: list[RunSummary] = RUNS) -> ResumePicker: + console, _ = _console() + return ResumePicker(runs, console=console, runs_dir="strix_runs", now=NOW) + + +def _plain(picker: ResumePicker) -> str: + return "\n".join(line.plain for line in picker.render()) + + +def test_translate_key_maps_control_characters() -> None: + assert translate_key("\r") == KEY_ENTER + assert translate_key("\n") == KEY_ENTER + assert translate_key("\x1b") == KEY_ESCAPE + assert translate_key("\x7f") == KEY_BACKSPACE + assert translate_key("\x03") == KEY_INTERRUPT + assert translate_key("a") == "a" + + +def test_relative_time_buckets() -> None: + assert relative_time((NOW - timedelta(seconds=5)).isoformat(), NOW) == "just now" + assert relative_time((NOW - timedelta(minutes=12)).isoformat(), NOW) == "12 min ago" + assert relative_time((NOW - timedelta(hours=3, minutes=1)).isoformat(), NOW) == "3 h ago" + assert relative_time((NOW - timedelta(days=2)).isoformat(), NOW) == "2 d ago" + assert relative_time("2026-01-02T15:04:00+00:00", NOW).startswith("Jan 2") + assert relative_time("2025-01-02T15:04:00+00:00", NOW).endswith("2025") + assert relative_time("not a date", NOW) == "not a date" + assert relative_time("", NOW) == "unknown" + + +def test_filter_matches_name_target_and_status() -> None: + assert [run.run_name for run in filter_runs(RUNS, "juice")] == ["juice-shop_8a1f"] + assert [run.run_name for run in filter_runs(RUNS, "INTERRUPTED")] == ["example-com_223e"] + assert [run.run_name for run in filter_runs(RUNS, "example.com")] == [ + "example-com_e5e6", + "example-com_223e", + ] + assert filter_runs(RUNS, "") == RUNS + + +def test_render_lists_every_run_with_its_metadata() -> None: + text = _plain(_picker()) + assert "Resume a run" in text + assert "4 runs in ./strix_runs" in text + for run in RUNS: + assert run.run_name in text + assert "12 min ago" in text + assert "https://juice-shop.herokuapp.com" in text + assert "interrupted" in text + assert "failed · no state" in text + assert text.splitlines()[3].startswith(" \u276f ") + + +def test_enter_returns_the_highlighted_run() -> None: + keys = iter([KEY_DOWN, KEY_ENTER]) + assert _picker().run(lambda: next(keys)) == RUNS[1] + + +def test_cursor_is_clamped_and_end_jumps_to_the_last_run() -> None: + picker = _picker() + picker.handle(KEY_UP) + assert picker.cursor == 0 + picker.handle(KEY_END) + assert picker.cursor == len(RUNS) - 1 + for _ in range(10): + picker.handle(KEY_DOWN) + assert picker.cursor == len(RUNS) - 1 + + +def test_escape_cancels_and_clears_the_search_first() -> None: + keys = iter(["j", "u", KEY_ESCAPE, KEY_ESCAPE]) + picker = _picker() + assert picker.run(lambda: next(keys)) is None + assert picker.filter == "" + + +def test_typing_filters_and_backspace_restores() -> None: + picker = _picker() + for char in "juice": + picker.handle(char) + assert [run.run_name for run in picker.rows] == ["juice-shop_8a1f"] + assert "search: juice" in _plain(picker) + for _ in range(5): + picker.handle(KEY_BACKSPACE) + assert len(picker.rows) == len(RUNS) + + +def test_runs_without_agent_state_cannot_be_picked() -> None: + picker = _picker() + picker.handle(KEY_END) + assert picker.handle(KEY_ENTER) == (False, None) + assert "no saved agent state" in _plain(picker) + + +def test_no_match_renders_a_hint_and_enter_does_nothing() -> None: + picker = _picker() + for char in "zzz": + picker.handle(char) + assert "no runs match 'zzz'" in _plain(picker) + assert picker.handle(KEY_ENTER) == (False, None) + + +def test_ctrl_c_raises_keyboard_interrupt() -> None: + with pytest.raises(KeyboardInterrupt): + _picker().handle(KEY_INTERRUPT) + + +def test_long_lists_scroll_to_keep_the_cursor_visible() -> None: + runs = [ + _run(f"run_{index:02d}", f"https://host{index}.example", minutes_ago=index) + for index in range(40) + ] + console, _ = _console(height=12) + picker = ResumePicker(runs, console=console, runs_dir="strix_runs", now=NOW) + text = _plain(picker) + assert "run_00" in text + assert "more above" not in text + assert "more below" in text + picker.handle(KEY_END) + text = _plain(picker) + assert "run_39" in text + assert "run_00" not in text + assert "more above" in text + assert "more below" not in text + + +def test_run_redraws_in_place_and_restores_the_cursor() -> None: + console, buffer = _console() + picker = ResumePicker(RUNS, console=console, runs_dir="strix_runs", now=NOW) + keys = iter([KEY_DOWN, KEY_ENTER]) + picker.run(lambda: next(keys)) + output = buffer.getvalue() + assert output.startswith("\x1b[?25l") + assert output.endswith("\x1b[?25h") + assert "\x1b[J" in output + assert "example-com_223e" in output + + +def test_pick_run_needs_a_terminal() -> None: + console, _ = _console() + with pytest.raises(PickerUnavailableError): + pick_run(RUNS, runs_dir="strix_runs", console=console, stdin=io.StringIO()) diff --git a/tests/test_run_summaries.py b/tests/test_run_summaries.py new file mode 100644 index 00000000..c7c1ee1f --- /dev/null +++ b/tests/test_run_summaries.py @@ -0,0 +1,124 @@ +"""Tests for the run listing behind the resume picker.""" + +from __future__ import annotations + +import json +import os +from typing import TYPE_CHECKING + +from strix.report.runs import list_run_summaries + + +if TYPE_CHECKING: + from pathlib import Path + + +def _write_run( + base: Path, + name: str, + record: object, + *, + modified: float, + state: bool = True, + findings: int | None = None, +) -> None: + run_dir = base / name + run_dir.mkdir(parents=True) + path = run_dir / "run.json" + path.write_text(record if isinstance(record, str) else json.dumps(record), encoding="utf-8") + if state: + (run_dir / ".state").mkdir() + (run_dir / ".state" / "agents.json").write_text("{}", encoding="utf-8") + if findings is not None: + (run_dir / "vulnerabilities.json").write_text( + json.dumps([{"id": index} for index in range(findings)]), encoding="utf-8" + ) + os.utime(path, (modified, modified)) + + +def test_lists_runs_newest_first_with_target_status_and_findings(tmp_path: Path) -> None: + base = tmp_path / "strix_runs" + _write_run( + base, + "example-com_1111", + { + "status": "completed", + "start_time": "2026-10-04T10:00:00+00:00", + "end_time": "2026-10-04T10:05:00+00:00", + "targets_info": [{"type": "web_application", "original": "https://example.com"}], + }, + modified=1_000, + findings=2, + ) + _write_run( + base, + "example-com_2222", + { + "status": "interrupted", + "start_time": "2026-10-04T11:00:00+00:00", + "targets_info": [ + {"type": "web_application", "original": "https://example.com"}, + {"type": "web_application", "original": "https://api.example.com"}, + ], + }, + modified=3_000, + ) + (base / "not-a-run").mkdir() + + runs = list_run_summaries(cwd=tmp_path) + + assert [run.run_name for run in runs] == ["example-com_2222", "example-com_1111"] + newest, oldest = runs + assert newest.target == "https://example.com +1 more" + assert newest.status == "interrupted" + assert newest.started_at == "2026-10-04T11:00:00+00:00" + assert newest.ended_at == "" + assert newest.findings == 0 + assert newest.resumable is True + assert oldest.target == "https://example.com" + assert oldest.findings == 2 + assert oldest.ended_at == "2026-10-04T10:05:00+00:00" + + +def test_target_less_runs_describe_their_workspace_or_instruction(tmp_path: Path) -> None: + base = tmp_path / "strix_runs" + _write_run( + base, + "pentest_aaaa", + {"status": "stopped", "targets_info": [], "workspace_mount": "/home/me/app"}, + modified=2_000, + ) + _write_run( + base, + "pentest_bbbb", + {"status": "failed", "targets_info": [], "user_instruction": "audit the login\nflow"}, + modified=1_000, + ) + + runs = {run.run_name: run for run in list_run_summaries(cwd=tmp_path)} + + assert runs["pentest_aaaa"].target == "/home/me/app (workspace)" + assert runs["pentest_bbbb"].target == "audit the login" + + +def test_runs_without_agent_state_or_with_a_broken_record_are_listed_not_resumable( + tmp_path: Path, +) -> None: + base = tmp_path / "strix_runs" + _write_run(base, "no-state_cccc", {"status": "failed"}, modified=2_000, state=False) + _write_run(base, "broken_dddd", "{not json", modified=1_000) + _write_run(base, "list_eeee", "[]", modified=500) + + runs = list_run_summaries(cwd=tmp_path) + + assert [(run.run_name, run.resumable) for run in runs] == [ + ("no-state_cccc", False), + ("broken_dddd", True), + ("list_eeee", True), + ] + assert runs[1].status == "unknown" + assert runs[1].target == "" + + +def test_missing_runs_dir_lists_nothing(tmp_path: Path) -> None: + assert list_run_summaries(cwd=tmp_path) == []