diff --git a/Makefile b/Makefile index d20df1d7..07923b80 100644 --- a/Makefile +++ b/Makefile @@ -69,7 +69,9 @@ help: @printf "\n" @printf " $(BOLD)make list-risk-files$(RESET) List top-scoring risky files from index\n" @printf " $(BOLD)make sweep$(RESET) Run deep sweep on top-scoring files\n" - @printf " $(BOLD)make sweep FILE=\"src/foo.*\"$(RESET) Run deep sweep on specific file(s)\n" + @printf " $(BOLD)make sweep FILES=\"src/a.*,src/b.*\"$(RESET) Run deep sweep on patterns (comma-separated)\n" + @printf " $(BOLD)make sweep FILE=\"src/foo.*\"$(RESET) Run deep sweep on single file pattern\n" + @printf " $(BOLD)make sweep FILES=... EXCLUDE=...$(RESET) Exclude patterns (comma-separated globs)\n" @printf "\n" @printf " $(BOLD)$(CYAN)Phase controls:$(RESET)\n" @printf "\n" @@ -222,11 +224,11 @@ list-risk-files: env-check @$(PYTHON) tools/list-risk-files.py sweep: env-check - @if [ -n "$(FILE)" ]; then \ - $(PYTHON) tools/run-sweep.py --file "$(FILE)"; \ - else \ - $(PYTHON) tools/run-sweep.py; \ - fi + @$(PYTHON) tools/run-sweep.py \ + $(if $(FILES),--files "$(FILES)") \ + $(if $(FILE),--file "$(FILE)") \ + $(if $(EXCLUDE),--exclude "$(EXCLUDE)") \ + $(if $(filter 1,$(RESET) $(RESTART)),--reset) # --------------------------------------------------------------------------- # Raw opencode debug target (non-workflow) diff --git a/docs/file-risk-sweeps.md b/docs/file-risk-sweeps.md index 69c8f45d..a45a6479 100644 --- a/docs/file-risk-sweeps.md +++ b/docs/file-risk-sweeps.md @@ -36,10 +36,21 @@ Show only paths for scripting: While the global Phase 2 agent (`make phase-2`) focuses on macro-level architectural flaws and cross-component issues, you can run an optional **Deep Sweep** (Phase 2 sweep mode) to perform exhaustive, line-by-line vulnerability hunting on specific high-risk files. Each sweep run creates Phase 2 candidate findings under `itemdb/findings/PENDING/` and writes a Phase 2 run summary. -Run a sweep on specific files (supports glob patterns): +Run a sweep on specific files (supports glob patterns, comma-separated): make sweep FILE="src/path/to/file.ext" make sweep FILE="src/**/*.cs" + make sweep FILES="src/a.py,src/**/*.cs" + make sweep FILES="src/controllers/upload.php,src/models/user.py" + +Exclude files with `EXCLUDE` (comma-separated globs, applied before the sweep): + + make sweep FILES="src/**/*.py" EXCLUDE="src/vendor/**,src/__pycache__/**" + make sweep EXCLUDE="src/vendor/**" + +`EXCLUDE` also works with index-based sweeps: + + make sweep EXCLUDE="src/vendor/**" Run a sweep sequentially across the top indexed files (score 4+): @@ -51,7 +62,17 @@ Preview selected files and generated prompts without invoking OpenCode: The sweep runner is sequential by default. It invokes the normal `auditor` agent with a specialized prompt (`prompts/phase-2-sweep.md`) that forces the model to read related dependencies and imports to establish complete source-to-sink context. -Generated temporary prompts are written under: +### Resume on interruption + +The sweep runner tracks successfully scanned files in `tmp/sweep-state.txt` (one path per line). If a sweep is interrupted or a file fails, re-running the same command will skip already-completed files and resume where it left off. + +To force a fresh sweep of all files, use `RESET=1` or `RESTART=1`: + + make sweep FILES="src/a.py,src/b.py" RESET=1 + +You can also inspect or manually edit `tmp/sweep-state.txt` to remove files you want to re-sweep. + +### Generated files tmp/file-sweep-prompts/ diff --git a/tests/test_run_sweep.py b/tests/test_run_sweep.py index bc9dd1ff..bffe9745 100644 --- a/tests/test_run_sweep.py +++ b/tests/test_run_sweep.py @@ -370,3 +370,189 @@ def fake_run(command, **kwargs): assert cmd[:3] == ["opencode", "run", "--agent"] finally: self._teardown_summary_env(module, orig_swp, orig_tmp_dir, orig_root) + + +class TestSweepStateFile: + def test_load_completed_empty_when_no_file(self, tmp_path): + module = _load_run_sweep() + orig_state = module.STATE_FILE + module.STATE_FILE = tmp_path / "nonexistent.txt" + try: + result = module.load_completed() + assert result == set() + finally: + module.STATE_FILE = orig_state + + def test_load_completed_reads_paths(self, tmp_path): + module = _load_run_sweep() + state_file = tmp_path / "sweep-state.txt" + state_file.write_text("src/a.py\nsrc/b.cs\n\n \n") + orig_state = module.STATE_FILE + module.STATE_FILE = state_file + try: + result = module.load_completed() + assert result == {"src/a.py", "src/b.cs"} + finally: + module.STATE_FILE = orig_state + + def test_mark_done_appends(self, tmp_path): + module = _load_run_sweep() + state_file = tmp_path / "sweep-state.txt" + orig_state = module.STATE_FILE + module.STATE_FILE = state_file + try: + module.mark_done("src/a.py") + module.mark_done("src/b.cs") + content = state_file.read_text(encoding="utf-8") + assert content == "src/a.py\nsrc/b.cs\n" + finally: + module.STATE_FILE = orig_state + + def test_clear_state_removes_file(self, tmp_path): + module = _load_run_sweep() + state_file = tmp_path / "sweep-state.txt" + state_file.write_text("src/a.py\n") + orig_state = module.STATE_FILE + module.STATE_FILE = state_file + try: + module.clear_state() + assert not state_file.exists() + finally: + module.STATE_FILE = orig_state + + def test_clear_state_noop_when_no_file(self, tmp_path): + module = _load_run_sweep() + orig_state = module.STATE_FILE + module.STATE_FILE = tmp_path / "nonexistent.txt" + try: + module.clear_state() + finally: + module.STATE_FILE = orig_state + + +class TestSweepFilesArg: + def test_files_arg_splits_comma(self): + module = _load_run_sweep() + parser = module.argparse.ArgumentParser() + parser.add_argument("--file", action="append", default=[]) + parser.add_argument("--files", default=None) + parsed = parser.parse_args(["--files", "src/a.py, src/b.cs , src/**/*.php"]) + if parsed.files: + for pat in parsed.files.split(","): + stripped = pat.strip() + if stripped: + parsed.file.append(stripped) + assert "src/a.py" in parsed.file + assert "src/b.cs" in parsed.file + assert "src/**/*.php" in parsed.file + assert len(parsed.file) == 3 + + def test_files_arg_handles_empty_tokens(self): + module = _load_run_sweep() + parser = module.argparse.ArgumentParser() + parser.add_argument("--file", action="append", default=[]) + parser.add_argument("--files", default=None) + parsed = parser.parse_args(["--files", "src/a.py,,src/b.cs,"]) + if parsed.files: + for pat in parsed.files.split(","): + stripped = pat.strip() + if stripped: + parsed.file.append(stripped) + assert parsed.file == ["src/a.py", "src/b.cs"] + + def test_files_and_file_can_be_combined(self): + module = _load_run_sweep() + parser = module.argparse.ArgumentParser() + parser.add_argument("--file", action="append", default=[]) + parser.add_argument("--files", default=None) + parsed = parser.parse_args([ + "--file", "src/x.py", + "--files", "src/a.py,src/b.cs", + "--file", "src/y.py", + ]) + if parsed.files: + for pat in parsed.files.split(","): + stripped = pat.strip() + if stripped: + parsed.file.append(stripped) + assert parsed.file == ["src/x.py", "src/y.py", "src/a.py", "src/b.cs"] + + +class TestSweepExclude: + def test_exclude_glob_removes_matching_files(self, tmp_path): + module = _load_run_sweep() + src = tmp_path / "src" + vendor = src / "vendor" + vendor.mkdir(parents=True) + (src / "a.py").write_text("") + (vendor / "x.py").write_text("") + (vendor / "y.py").write_text("") + (src / "b.cs").write_text("") + orig_root = module.ROOT + module.ROOT = tmp_path + try: + exclude_set = set() + for m in module.glob.glob("src/vendor/*", root_dir=str(tmp_path)): + exclude_set.add(m) + candidates = ["src/a.py", "src/vendor/x.py", "src/vendor/y.py", "src/b.cs"] + result = [f for f in candidates if f not in exclude_set] + assert result == ["src/a.py", "src/b.cs"] + finally: + module.ROOT = orig_root + + def test_exclude_glob_empty_when_no_match(self, tmp_path): + module = _load_run_sweep() + src = tmp_path / "src" + src.mkdir(parents=True) + (src / "a.py").write_text("") + orig_root = module.ROOT + module.ROOT = tmp_path + try: + exclude_set = set() + for m in module.glob.glob("src/nonexistent/*", root_dir=str(tmp_path)): + exclude_set.add(m) + assert exclude_set == set() + finally: + module.ROOT = orig_root + + def test_exclude_glob_supports_recursive(self, tmp_path): + module = _load_run_sweep() + vendor = tmp_path / "src" / "vendor" + vendor.mkdir(parents=True) + (vendor / "x.py").write_text("") + deep = vendor / "deep" + deep.mkdir() + (deep / "z.py").write_text("") + orig_root = module.ROOT + module.ROOT = tmp_path + try: + exclude_set = set() + for m in module.glob.glob("src/vendor/**", root_dir=str(tmp_path), recursive=True): + exclude_set.add(m) + assert "src/vendor/x.py" in exclude_set + assert "src/vendor/deep/z.py" in exclude_set + finally: + module.ROOT = orig_root + + +class TestSweepMarkDoneWithSummary: + def test_mark_done_writes_to_state(self, tmp_path): + module = _load_run_sweep() + state_file = tmp_path / "sweep-state.txt" + orig_state = module.STATE_FILE + module.STATE_FILE = state_file + try: + module.mark_done("src/bar.py") + assert "src/bar.py" in module.load_completed() + finally: + module.STATE_FILE = orig_state + + def test_load_completed_empty_when_no_mark_done(self, tmp_path): + module = _load_run_sweep() + state_file = tmp_path / "sweep-state.txt" + orig_state = module.STATE_FILE + module.STATE_FILE = state_file + try: + assert module.load_completed() == set() + finally: + module.STATE_FILE = orig_state diff --git a/tools/run-sweep.py b/tools/run-sweep.py index f1f93a8a..adeb4417 100755 --- a/tools/run-sweep.py +++ b/tools/run-sweep.py @@ -12,8 +12,10 @@ ./tools/run-sweep.py --file src/app/controllers/upload.php ./tools/run-sweep.py --file "src/**/*.cs" + ./tools/run-sweep.py --files "src/a.py,src/**/*.cs" ./tools/run-sweep.py --min-score 4 --limit 5 ./tools/run-sweep.py --min-score 5 --dry-run + ./tools/run-sweep.py --reset """ from __future__ import annotations @@ -42,6 +44,7 @@ PROMPT_TEMPLATE = ROOT / "prompts" / "phase-2-sweep.md" SWEEP_SUMMARY_PROMPT = ROOT / "prompts" / "phase-2-sweep-summary.md" TMP_DIR = ROOT / "tmp" / "file-sweep-prompts" +STATE_FILE = ROOT / "tmp" / "sweep-state.txt" def load_risk_entries(index_path: Path) -> list[dict[str, Any]]: @@ -117,6 +120,26 @@ def slugify(path: str) -> str: return value[:120] or "target" +def load_completed() -> set[str]: + if not STATE_FILE.exists(): + return set() + return { + line.strip() + for line in STATE_FILE.read_text(encoding="utf-8").splitlines() + if line.strip() + } + + +def mark_done(file_path: str) -> None: + STATE_FILE.parent.mkdir(parents=True, exist_ok=True) + with open(STATE_FILE, "a", encoding="utf-8") as fh: + fh.write(file_path + "\n") + + +def clear_state() -> None: + STATE_FILE.unlink(missing_ok=True) + + def build_prompt_for_file(file_path: str) -> Path: if not PROMPT_TEMPLATE.exists(): raise FileNotFoundError(f"missing prompt template: {PROMPT_TEMPLATE.relative_to(ROOT)}") @@ -229,6 +252,9 @@ def run_sweep_summary(files: list[str], per_file_summaries: list[str]) -> int: def main() -> int: parser = argparse.ArgumentParser(description="Run sequential CodeCome file-scoped sweeps") parser.add_argument("--file", action="append", default=[], help="Specific file or glob to sweep. May be repeated.") + parser.add_argument("--files", default=None, help="Comma-separated list of file patterns (convenience; added to --file args)") + parser.add_argument("--exclude", default=None, help="Comma-separated list of glob patterns to exclude from sweep") + parser.add_argument("--reset", action="store_true", help="Clear sweep progress state and start fresh") parser.add_argument("--index", default=str(DEFAULT_INDEX), help="Path to file-risk-index.yml") parser.add_argument("--min-score", type=int, default=4, help="Minimum risk score when selecting from the index") parser.add_argument("--limit", type=int, default=None, help="Maximum number of indexed files to sweep") @@ -236,23 +262,84 @@ def main() -> int: parser.add_argument("--skip-gates", action="store_true", help="Skip readiness and sandbox gates") args = parser.parse_args() + if args.files: + for pat in args.files.split(","): + stripped = pat.strip() + if stripped: + args.file.append(stripped) + + if args.reset: + clear_state() + print(C.info("Sweep state cleared — starting fresh.")) + index_path = Path(args.index) if not index_path.is_absolute(): index_path = ROOT / index_path try: if args.file: - files = expand_file_args(args.file) + candidates = expand_file_args(args.file) else: - files = select_from_index(index_path, args.min_score, args.limit) + candidates = select_from_index(index_path, args.min_score, args.limit) except Exception as exc: # noqa: BLE001 print(C.fail(str(exc)), file=sys.stderr) return 1 - if not files: + if not candidates: print(C.warn("No files selected for sweep.")) return 0 + if args.exclude: + exclude_set: set[str] = set() + for pat in args.exclude.split(","): + pat = pat.strip() + if not pat: + continue + for m in glob.glob(pat, root_dir=str(ROOT), recursive=True): + exclude_set.add(m) + before = len(candidates) + candidates = [f for f in candidates if f not in exclude_set] + removed = before - len(candidates) + if removed: + print(C.info(f"Excluded {removed} file(s) matching --exclude")) + + if not candidates: + print(C.warn("No files selected for sweep after exclusions.")) + return 0 + + completed = load_completed() + files = [f for f in candidates if f not in completed] + skipped = [f for f in candidates if f in completed] + + if skipped: + print(C.info(f"Skipping {len(skipped)} already-scanned file(s):")) + for f in skipped: + print(f" {C.SYM_BULLET} {f}") + + if not files: + print(C.warn("All selected files have already been scanned. Use --reset to start fresh.")) + if not args.dry_run and skipped: + prior_summaries: list[str] = [] + seen = set() + for f in skipped: + prior = sorted( + (ROOT / "runs").glob(f"phase-2-summary-sweep-{slugify(f)}-*.md"), + key=lambda p: p.stat().st_mtime, + reverse=True, + ) + if prior: + rel = str(prior[0].relative_to(ROOT)) + if rel not in seen: + prior_summaries.append(rel) + seen.add(rel) + if prior_summaries: + print(C.header("Sweep Summary (Aggregate Rollup)")) + code = run_sweep_summary(skipped, prior_summaries) + if code != 0: + print(C.fail(f"Sweep aggregate summary failed with exit code {code}"), file=sys.stderr) + return code + return 0 + print(C.header("Selected files")) for path in files: print(f"{C.SYM_BULLET} {path}") @@ -270,18 +357,42 @@ def main() -> int: if code != 0: print(C.fail(f"Sweep failed for {file_path} with exit code {code}"), file=sys.stderr) return code + if not args.dry_run: + summaries = list( + (ROOT / "runs").glob(f"phase-2-summary-sweep-{slugify(file_path)}-*.md") + ) + recent = [s for s in summaries if s.stat().st_mtime >= sweep_start_time] + if recent: + mark_done(file_path) + else: + print(C.warn(f"No sweep summary found for {file_path} — will re-sweep on next run")) if not args.dry_run: - fresh_summaries = [ - str(s.relative_to(ROOT)) - for f in files + all_files = files + skipped + fresh_summaries: list[str] = [] + seen = set() + for f in files: for s in sorted( (ROOT / "runs").glob(f"phase-2-summary-sweep-{slugify(f)}-*.md"), key=lambda p: p.stat().st_mtime, + ): + rel = str(s.relative_to(ROOT)) + if s.stat().st_mtime >= sweep_start_time and rel not in seen: + fresh_summaries.append(rel) + seen.add(rel) + for f in skipped: + prior = sorted( + (ROOT / "runs").glob(f"phase-2-summary-sweep-{slugify(f)}-*.md"), + key=lambda p: p.stat().st_mtime, + reverse=True, ) - if s.stat().st_mtime >= sweep_start_time - ] - code = run_sweep_summary(files, fresh_summaries) + if prior: + rel = str(prior[0].relative_to(ROOT)) + if rel not in seen: + fresh_summaries.append(rel) + seen.add(rel) + + code = run_sweep_summary(all_files, fresh_summaries if fresh_summaries else []) if code != 0: print(C.fail(f"Sweep aggregate summary failed with exit code {code}"), file=sys.stderr) return code