Adds scan resumption capability to Strix, allowing users to resume interrupted scans from automatic checkpoints. Fixes #29

.gitignore

@@ -97,3 +97,5 @@ Thumbs.db
 schema.graphql
 
 .opencode/
+
+.claude/

README.md

@@ -12,7 +12,6 @@
 
 [![Python](https://img.shields.io/pypi/pyversions/strix-agent?color=3776AB)](https://pypi.org/project/strix-agent/)
 [![PyPI](https://img.shields.io/pypi/v/strix-agent?color=10b981)](https://pypi.org/project/strix-agent/)
-![PyPI Downloads](https://static.pepy.tech/personalized-badge/strix-agent?period=total&units=INTERNATIONAL_SYSTEM&left_color=GREY&right_color=RED&left_text=Downloads)
 [![License](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](LICENSE)
 
 [![GitHub Stars](https://img.shields.io/github/stars/usestrix/strix)](https://github.com/usestrix/strix)
@@ -167,6 +166,9 @@ strix --target api.your-app.com --instruction "Focus on business logic flaws and
 
 # Provide detailed instructions through file (e.g., rules of engagement, scope, exclusions)
 strix --target api.your-app.com --instruction-file ./instruction.md
+
+# Resume an interrupted scan
+strix --target https://your-app.com --run-name my-scan --resume
 ```
 
 ### 🤖 Headless Mode

strix/agents/base_agent.py

@@ -205,6 +205,24 @@ async def agent_loop(self, task: str) -> dict[str, Any]:  # noqa: PLR0912, PLR09
 
             try:
                 should_finish = await self._process_iteration(tracer)
+
+                # Save checkpoint after successful iteration
+                try:
+                    from strix.telemetry.checkpoint import save_checkpoint
+                    from strix.telemetry.tracer import get_global_tracer
+
+                    tracer_instance = get_global_tracer()
+                    if tracer_instance and hasattr(self, "state"):
+                        run_dir = tracer_instance.get_run_dir()
+                        scan_config = tracer_instance.scan_config or {}
+                        save_checkpoint(run_dir, self.state, scan_config)
+                except Exception as exc:  # noqa: BLE001
+                    logger.debug(
+                        "Checkpoint save failed (non-fatal): %s",
+                        exc,
+                        exc_info=True,
+                    )
+
                 if should_finish:
                     if self.non_interactive:
                         self.state.set_completed({"success": True})

strix/interface/cli.py

@@ -17,7 +17,7 @@
 from .utils import build_final_stats_text, build_live_stats_text, get_severity_color
 
 
-async def run_cli(args: Any) -> None:  # noqa: PLR0915
+async def run_cli(args: Any) -> None:  # noqa: PLR0915, PLR0912
     console = Console()
 
     start_text = Text()
@@ -85,6 +85,64 @@ async def run_cli(args: Any) -> None:  # noqa: PLR0915
     if getattr(args, "local_sources", None):
         agent_config["local_sources"] = args.local_sources
 
+    # Check for resume from checkpoint
+    from pathlib import Path
+
+    from pydantic import ValidationError
+
+    from strix.agents.state import AgentState
+    from strix.telemetry.checkpoint import can_resume, load_checkpoint
+
+    resume_from_checkpoint = False
+    restored_state = None
+
+    if getattr(args, "resume", False):
+        run_dir = Path.cwd() / "strix_runs" / args.run_name
+
+        if run_dir.exists() and can_resume(run_dir, scan_config):
+            checkpoint = load_checkpoint(run_dir)
+
+            if checkpoint:
+                try:
+                    agent_state_data = checkpoint["agent_state"]
+                    restored_state = AgentState(**agent_state_data)
+                    resume_from_checkpoint = True
+
+                    console.print()
+                    resume_text = Text()
+                    resume_text.append("✓ ", style="bold green")
+                    resume_text.append("Resuming from checkpoint at iteration ", style="green")
+                    resume_text.append(
+                        f"{restored_state.iteration}/{restored_state.max_iterations}",
+                        style="bold green",
+                    )
+                    console.print(resume_text)
+                    console.print()
+
+                except ValidationError as e:
+                    warn_text = Text()
+                    warn_text.append("⚠ ", style="bold yellow")
+                    warn_text.append(
+                        f"Checkpoint validation failed: {e}. Starting fresh scan.", style="yellow"
+                    )
+                    console.print()
+                    console.print(warn_text)
+                    console.print()
+        elif getattr(args, "resume", False):
+            warn_text = Text()
+            warn_text.append("⚠ ", style="bold yellow")
+            warn_text.append(
+                "--resume flag provided but no valid checkpoint found. Starting fresh scan.",
+                style="yellow",
+            )
+            console.print()
+            console.print(warn_text)
+            console.print()
+
+    # Add restored state to agent config if resuming
+    if resume_from_checkpoint and restored_state:
+        agent_config["state"] = restored_state
+
     tracer = Tracer(args.run_name)
     tracer.set_scan_config(scan_config)
 

strix/interface/main.py

@@ -343,6 +343,16 @@ def parse_arguments() -> argparse.Namespace:
         ),
     )
 
+    parser.add_argument(
+        "--resume",
+        action="store_true",
+        help=(
+            "Resume an interrupted scan from checkpoint. "
+            "Requires --run-name to match the interrupted scan. "
+            "If no valid checkpoint is found, starts a fresh scan."
+        ),
+    )
+
     args = parser.parse_args()
 
     if args.instruction and args.instruction_file:

strix/interface/tui.py

@@ -81,12 +81,15 @@ class SplashScreen(Static):  # type: ignore[misc]
         " ╚══════╝   ╚═╝   ╚═╝  ╚═╝╚═╝╚═╝  ╚═╝"
     )
 
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
+    def __init__(
+        self, *args: Any, resume_info: dict[str, Any] | None = None, **kwargs: Any
+    ) -> None:
         super().__init__(*args, **kwargs)
         self._animation_step = 0
         self._animation_timer: Timer | None = None
         self._panel_static: Static | None = None
         self._version = "dev"
+        self._resume_info = resume_info
 
     def compose(self) -> ComposeResult:
         self._version = get_package_version()
@@ -116,15 +119,22 @@ def _animate_start_line(self) -> None:
         self._panel_static.update(panel)
 
     def _build_panel(self, start_line: Text) -> Panel:
-        content = Group(
+        content_parts = [
             Align.center(Text(self.BANNER.strip("\n"), style=self.PRIMARY_GREEN, justify="center")),
             Align.center(Text(" ")),
             Align.center(self._build_welcome_text()),
             Align.center(self._build_version_text()),
             Align.center(self._build_tagline_text()),
-            Align.center(Text(" ")),
-            Align.center(start_line.copy()),
-        )
+        ]
+
+        if self._resume_info:
+            content_parts.append(Align.center(Text(" ")))
+            content_parts.append(Align.center(self._build_resume_text()))
+
+        content_parts.append(Align.center(Text(" ")))
+        content_parts.append(Align.center(start_line.copy()))
+
+        content = Group(*content_parts)
 
         return Panel.fit(content, border_style=self.PRIMARY_GREEN, padding=(1, 6))
 
@@ -140,6 +150,17 @@ def _build_version_text(self) -> Text:
     def _build_tagline_text(self) -> Text:
         return Text("Open-source AI hackers for your apps", style=Style(color="white", dim=True))
 
+    def _build_resume_text(self) -> Text:
+        if not self._resume_info:
+            return Text("")
+
+        text = Text("✓ Resuming from iteration ", style=Style(color="#fbbf24", bold=True))
+        text.append(
+            f"{self._resume_info['iteration']}/{self._resume_info['max_iterations']}",
+            style=Style(color=self.PRIMARY_GREEN, bold=True),
+        )
+        return text
+
     def _build_start_line_text(self, phase: int) -> Text:
         emphasize = phase % 2 == 1
         base_style = Style(color="white", dim=not emphasize, bold=emphasize)
@@ -280,6 +301,51 @@ def __init__(self, args: argparse.Namespace):
         self.scan_config = self._build_scan_config(args)
         self.agent_config = self._build_agent_config(args)
 
+        # Check for resume from checkpoint
+        from pathlib import Path
+
+        from pydantic import ValidationError
+
+        from strix.agents.state import AgentState
+        from strix.telemetry.checkpoint import can_resume, load_checkpoint
+
+        self.resume_info: dict[str, Any] | None = None
+
+        if getattr(args, "resume", False):
+            run_dir = Path.cwd() / "strix_runs" / args.run_name
+
+            if run_dir.exists() and can_resume(run_dir, self.scan_config):
+                checkpoint = load_checkpoint(run_dir)
+
+                if checkpoint:
+                    try:
+                        agent_state_data = checkpoint["agent_state"]
+                        restored_state = AgentState(**agent_state_data)
+                        self.agent_config["state"] = restored_state
+
+                        self.resume_info = {
+                            "iteration": restored_state.iteration,
+                            "max_iterations": restored_state.max_iterations,
+                        }
+
+                        import logging
+
+                        logging.info(
+                            f"Resuming from checkpoint at iteration "
+                            f"{restored_state.iteration}/{restored_state.max_iterations}"
+                        )
+
+                    except ValidationError as e:
+                        import logging
+
+                        logging.warning(f"Checkpoint validation failed: {e}. Starting fresh scan.")
+            elif getattr(args, "resume", False):
+                import logging
+
+                logging.warning(
+                    "--resume flag provided but no valid checkpoint found. Starting fresh scan."
+                )
+
         self.tracer = Tracer(self.scan_config["run_name"])
         self.tracer.set_scan_config(self.scan_config)
         set_global_tracer(self.tracer)
@@ -348,7 +414,7 @@ def signal_handler(_signum: int, _frame: Any) -> None:
 
     def compose(self) -> ComposeResult:
         if self.show_splash:
-            yield SplashScreen(id="splash_screen")
+            yield SplashScreen(id="splash_screen", resume_info=self.resume_info)
 
     def watch_show_splash(self, show_splash: bool) -> None:
         if not show_splash and self.is_mounted: