fix(cli): restore staged API specifications on resume

strix/interface/cli_args.py

@@ -6,6 +6,7 @@
 import os
 import sys
 from pathlib import Path
+from typing import Any
 
 from strix.config import apply_config_override
 from strix.config.settings import DEFAULT_MAX_TURNS
@@ -16,6 +17,7 @@
     check_mountable_dir,
     collect_local_sources,
     resolve_workspace_files,
+    stage_api_specs,
     validate_config_file,
 )
 from strix.report.runs import list_run_summaries
@@ -464,7 +466,8 @@ def load_resume_state(args: argparse.Namespace) -> None:
     except (RuntimeError, TypeError) as exc:
         raise ResumeError(f"--resume {args.resume}: run.json unreadable: {exc}") from exc
 
-    args.targets_info = state.get("targets_info") or []
+    targets_info: Any = state.get("targets_info") or []
+    args.targets_info = targets_info
     # A target-less run has no targets_info at all. It is driven by its
     # instruction, over a mounted working directory or over nothing when the
     # mount was declined, so either of those is enough to resume it.
@@ -499,6 +502,10 @@ def load_resume_state(args: argparse.Namespace) -> None:
     if not getattr(args, "user_instruction", None):
         args.user_instruction = state.get("user_instruction") or None
     args.local_sources = collect_local_sources(args.targets_info)
+    try:
+        args.local_sources.extend(stage_api_specs(targets_info, args.resume, resume=True))
+    except ValueError as exc:
+        raise ResumeError(f"--resume {args.resume}: {exc}") from exc
     # Remount the workspace the run was started with. The user already confirmed
     # this directory, so the target mount guard does not apply to it; it only has
     # to still be there.

strix/interface/utils.py

@@ -1514,20 +1514,24 @@ def write_fetched_collection(collection: dict[str, Any], collection_uid: str) ->
     return str(path)
 
 
-def stage_api_specs(targets_info: list[dict[str, Any]], run_name: str) -> list[dict[str, Any]]:
+def stage_api_specs(
+    targets_info: list[dict[str, Any]], run_name: str, *, resume: bool = False
+) -> list[dict[str, Any]]:
     """Copy every ``api_spec`` target into one directory for the sandbox.
 
     A spec is a single file the agent reads, not a tree it works in, so it is
     copied to a per-run staging directory that is exposed at
     ``/workspace/api-specs`` instead of mounting its host location. Each target's
-    ``workspace_path`` records where the agent will find it.
+    ``workspace_path`` records where the agent will find it. On resume, reuse
+    the staged files without replacing them with potentially changed inputs.
     """
     specs = [t for t in targets_info if t.get("type") == "api_spec"]
     if not specs:
         return []
 
     staging = Path(tempfile.gettempdir()) / "strix_api_specs" / run_name
-    staging.mkdir(parents=True, exist_ok=True)
+    if not resume:
+        staging.mkdir(parents=True, exist_ok=True)
 
     used: set[str] = set()
     for target in specs:
@@ -1540,7 +1544,16 @@ def stage_api_specs(targets_info: list[dict[str, Any]], run_name: str) -> list[d
             count += 1
             name = f"{stem}-{count}{suffix}"
         used.add(name)
-        shutil.copy2(source, staging / name)
+        staged = staging / name
+        if resume:
+            if not staged.is_file():
+                raise ValueError(
+                    f"staged API specification at {staged} is missing. "
+                    "Restore the staged file before resuming, or start a fresh scan "
+                    "with the original specification target."
+                )
+        else:
+            shutil.copy2(source, staged)
         details["workspace_path"] = f"/workspace/{API_SPEC_WORKSPACE_SUBDIR}/{name}"
 
     return [

tests/test_api_spec_targets.py

@@ -4,14 +4,18 @@
 
 import argparse
 import json
+import sys
 from pathlib import Path
 from typing import Any
 
 import pytest
 
 from strix.core.inputs import build_root_task, build_scope_context
-from strix.interface.scan_setup import build_targets_info
+from strix.core.paths import run_dir_for, runtime_state_dir
+from strix.interface.cli_args import parse_arguments
+from strix.interface.scan_setup import build_targets_info, prepare_run
 from strix.interface.utils import infer_target_type, stage_api_specs
+from strix.runtime.session_manager import build_bind_mounts, build_manifest_entries
 
 
 OPENAPI = {
@@ -133,6 +137,56 @@ def test_stage_api_specs_without_specs_returns_nothing() -> None:
     assert stage_api_specs([{"type": "web_application", "details": {}}], "run") == []
 
 
+def test_resume_restores_staged_api_specs(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr("strix.interface.utils.tempfile.gettempdir", lambda: str(tmp_path))
+    first = _write_spec(tmp_path / "a")
+    second = _write_spec(tmp_path / "b")
+    monkeypatch.setattr(sys, "argv", ["strix", "-n", "-t", str(first), "-t", str(second)])
+    original = parse_arguments()
+    prepare_run(original)
+    state_dir = runtime_state_dir(run_dir_for(original.run_name))
+    state_dir.mkdir(parents=True)
+    (state_dir / "agents.json").write_text("{}", encoding="utf-8")
+    # Resume must use the staged snapshot, even if its inputs change or disappear.
+    first.write_text("changed after the scan started", encoding="utf-8")
+    second.unlink()
+    monkeypatch.setattr(sys, "argv", ["strix", "-n", "--resume", original.run_name])
+
+    resumed = parse_arguments()
+    prepare_run(resumed)
+
+    assert resumed.local_sources == original.local_sources
+    assert build_bind_mounts(resumed.local_sources) == build_bind_mounts(original.local_sources)
+    assert "api-specs" in build_manifest_entries(resumed.local_sources)
+    staging = Path(resumed.local_sources[0]["source_path"])
+    for name in ("openapi.json", "openapi-2.json"):
+        assert json.loads((staging / name).read_text(encoding="utf-8")) == OPENAPI
+    assert resumed.targets_info == original.targets_info
+
+
+def test_resume_rejects_a_missing_staged_api_spec(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr("strix.interface.utils.tempfile.gettempdir", lambda: str(tmp_path))
+    spec = _write_spec(tmp_path / "src")
+    monkeypatch.setattr(sys, "argv", ["strix", "-n", "-t", str(spec)])
+    original = parse_arguments()
+    prepare_run(original)
+    state_dir = runtime_state_dir(run_dir_for(original.run_name))
+    state_dir.mkdir(parents=True)
+    (state_dir / "agents.json").write_text("{}", encoding="utf-8")
+    (Path(original.local_sources[0]["source_path"]) / spec.name).unlink()
+    monkeypatch.setattr(sys, "argv", ["strix", "-n", "--resume", original.run_name])
+
+    with pytest.raises(SystemExit) as exc:
+        parse_arguments()
+
+    assert exc.value.code == 2
+    assert "staged API specification" in capsys.readouterr().err
+
+
 def test_build_root_task_points_at_the_spec_file(tmp_path: Path) -> None:
     task = build_root_task({"targets": [_staged_target(tmp_path)]})
     assert "API Specifications" in task