Skip to content
Open
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
28 changes: 17 additions & 11 deletions harness/local_harness/benchmark/analyze_misses.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,19 @@

from local_harness.config import BENCHMARK_DIR, MODEL, REPO_ROOT, RESULTS_DIR, STATE_FILE, atomic_write_json

# REPO_ROOT is the harness/ directory; the scanner's phase prompts live in the
# vulnhunt skill at the top of the repository.
PROJECT_ROOT = os.path.dirname(REPO_ROOT)
PROMPTS_DIR = os.path.join(PROJECT_ROOT, "vulnhunt", "phases")

# Prompt files per loss phase, relative to PROJECT_ROOT.
PHASE_TO_PROMPT = {
"phase1": ["skill/phases/phase1_recon.md"],
"phase2_inj": ["skill/phases/phase2_shared.md", "skill/phases/phase2_class_inj.md"],
"phase2_nav": ["skill/phases/phase2_shared.md", "skill/phases/phase2_class_nav.md"],
"phase2_log": ["skill/phases/phase2_shared.md", "skill/phases/phase2_class_log.md"],
"phase2b": ["skill/phases/phase2b_verify.md"],
"unknown": ["skill/phases/phase2_hunt.md"],
"phase1": ["vulnhunt/phases/phase1_recon.md"],
"phase2_inj": ["vulnhunt/phases/phase2_shared.md", "vulnhunt/phases/phase2_class_inj.md"],
"phase2_nav": ["vulnhunt/phases/phase2_shared.md", "vulnhunt/phases/phase2_class_nav.md"],
"phase2_log": ["vulnhunt/phases/phase2_shared.md", "vulnhunt/phases/phase2_class_log.md"],
"phase2b": ["vulnhunt/phases/phase2b_verify.md"],
"unknown": ["vulnhunt/phases/phase2_hunt.md"],
}

ANALYSIS_JSON = os.path.join(RESULTS_DIR, "miss_analysis.json")
Expand All @@ -46,7 +52,7 @@
1. Start with the ground truth vulnerability definition to understand exactly what should have been found
2. Read the scanner results in the VULNHUNT_RESULTS directory to see what WAS found and what wasn't
3. Dive into the scan log (benchmark_scan.log - JSONL format, one JSON object per line) to trace where the scanner's reasoning went wrong. Use grep to find relevant sections rather than reading the whole file.
4. Read the relevant prompt file(s) in skill/phases/ to identify what instruction gap led to the miss
4. Read the relevant prompt file(s) in vulnhunt/phases/ to identify what instruction gap led to the miss

DIAGNOSIS CONSTRAINTS:
- Identify the ROOT CAUSE in the scanner's prompts/instructions
Expand All @@ -59,7 +65,7 @@
Output ONLY valid JSON:
{
"root_cause": "Why the scanner missed this — be specific about which instruction/rule/gate caused the miss",
"prompt_file": "The primary prompt file to change (relative path like skill/phases/phase2_shared.md)",
"prompt_file": "The primary prompt file to change (relative path like vulnhunt/phases/phase2_shared.md)",
"section_to_change": "Quote the relevant section heading or existing text",
"suggested_change": "The specific text to add, modify, or delete — as short and general as possible",
"change_type": "add|edit|delete",
Expand Down Expand Up @@ -227,7 +233,7 @@ def build_diagnostic_prompt(finding, phase_key, evidence, results_dir, repo_dir)
gt_files = glob.glob(os.path.join(BENCHMARK_DIR, f"{finding['repo_name']}*.json"))
gt_path = gt_files[0] if gt_files else "unknown"
prompt_files = PHASE_TO_PROMPT.get(phase_key, PHASE_TO_PROMPT["unknown"])
prompt_paths = [os.path.join(REPO_ROOT, p) for p in prompt_files]
prompt_paths = [os.path.join(PROJECT_ROOT, p) for p in prompt_files]
scan_log_path = os.path.join(repo_dir, "benchmark_scan.log")

return f"""Finding {finding['finding_id']} was NOT detected. Figure out where the scanner went amiss.
Expand All @@ -242,7 +248,7 @@ def build_diagnostic_prompt(finding, phase_key, evidence, results_dir, repo_dir)
- Scanner results directory: {results_dir}/
- Scan log (JSONL, use grep): {scan_log_path}
- Relevant prompt file(s): {', '.join(prompt_paths)}
- All prompt files: {os.path.join(REPO_ROOT, 'skill', 'phases')}/
- All prompt files: {PROMPTS_DIR}/

## Pre-analysis Hint (may be wrong — verify)
Lost at **{phase_key}** phase. Evidence: {evidence[:500]}
Expand Down Expand Up @@ -275,7 +281,7 @@ def invoke_diagnostic(finding, phase_key, evidence, results_dir, repo_dir):
"--strict-mcp-config",
"--add-dir", repo_dir,
"--add-dir", results_dir,
"--add-dir", os.path.join(REPO_ROOT, "skill", "phases"),
"--add-dir", PROMPTS_DIR,
"--add-dir", BENCHMARK_DIR,
]

Expand Down
39 changes: 39 additions & 0 deletions harness/tests/test_analyze_misses.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,18 @@
"""Tests for local_harness.benchmark.analyze_misses."""

import json
import os
import subprocess
import types
from pathlib import Path

import pytest

import local_harness.benchmark.analyze_misses as am

# Top of the repository: harness/tests/ -> harness/ -> repository root.
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]


def _proc(returncode=0, stdout="", stderr=""):
return types.SimpleNamespace(returncode=returncode, stdout=stdout, stderr=stderr)
Expand Down Expand Up @@ -173,6 +178,40 @@ def fake_run(cmd, *a, **k):
assert "--allowedTools" not in cmd


def test_phase_prompt_files_exist():
"""Every prompt file named for a loss phase exists in the repository."""
missing = [
prompt_file
for prompt_files in am.PHASE_TO_PROMPT.values()
for prompt_file in prompt_files
if not (REPOSITORY_ROOT / prompt_file).is_file()
]
assert not missing, missing


def test_invoke_diagnostic_add_dirs_exist(monkeypatch, tmp_path):
"""The session can only read inside its --add-dir roots, and claude accepts
a missing --add-dir without an error, so every root must exist.
"""
captured = {}

def fake_run(cmd, *a, **k):
captured["cmd"] = cmd
return _proc(0, stdout='{"root_cause":"rc"}')

monkeypatch.setattr(am.subprocess, "run", fake_run)
results_dir = tmp_path / "rd"
repo_dir = tmp_path / "repo"
results_dir.mkdir()
repo_dir.mkdir()
finding = {"finding_id": "F1", "type": "SQLi", "description": "d", "repo_name": "r"}
am.invoke_diagnostic(finding, "phase1", "ev", str(results_dir), str(repo_dir))
cmd = captured["cmd"]
add_dirs = [cmd[i + 1] for i, tok in enumerate(cmd) if tok == "--add-dir"]
missing = [d for d in add_dirs if not os.path.isdir(d)]
assert not missing, missing


def test_invoke_diagnostic_timeout(monkeypatch):
def boom(*a, **k):
raise subprocess.TimeoutExpired(cmd="claude", timeout=1)
Expand Down