Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down
25 changes: 23 additions & 2 deletions docs/file-risk-sweeps.md
Original file line number Diff line number Diff line change
Expand Up @@ -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+):

Expand All @@ -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/

Expand Down
186 changes: 186 additions & 0 deletions tests/test_run_sweep.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
pruiz marked this conversation as resolved.
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
Loading
Loading