feat(learning-harness): curated repository tools and a parameterised hunter loop

Task 1.5. Four tools join the three the hunter already had. Resolving a
definition by symbol returns its file, line span and body, decorators included,
so a detector stops guessing what a callee or a route handler does. Listing
entry points, flows and obligations answers with identifiers and line spans
only, so a structural tool can never become a second way to dump the
repository. Every path a caller supplies is clamped to the scan root first.

The loop now takes the system prompt, the user prompt, the files to start from,
the tool subset, the character limit and the tags its findings carry, each
defaulting to today's behaviour, and it replays the provider's reasoning field
on the next turn because the provider otherwise drops its own chain of thought
across a tool call. A detector configuration is exactly this set of knobs.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
This commit is contained in:
Marius Ciepluch
2026-09-06 09:22:23 +02:00
co-authored by Claude Opus 5
parent 6b58c81ba0
commit 6965c39b7b
3 changed files with 448 additions and 21 deletions
+170 -1
View File
@@ -5,7 +5,7 @@ import re
from collections.abc import Iterable, Iterator, Mapping
from pathlib import Path
from .preprocess import enumerate_source_files
from .preprocess import FileTarget, enumerate_source_files
class PathEscapesRepo(ValueError):
@@ -128,3 +128,172 @@ def _as_line(value: object) -> int | None:
if isinstance(value, int):
return value
return None
# --- curated structural tools (learning-harness, Req 6.4, 6.5) --------------------------------------
# A detector reads code with `read_file` and `read_definition`; the three list tools below answer with
# identifiers and line spans only, so they can never become a second way to dump the repository. Every
# path a caller supplies is clamped to the scan root first. No caching: a detector calls these a handful
# of times per region, and a stale view of a tree that a round may have rewritten is worse than a reparse.
def read_definition(root: Path, symbol: str, *, max_chars: int, path: str | None = None) -> dict[str, object] | None:
"""The definition of one named function: where it is, how far it runs, and its text. None when absent."""
resolved_root = root.resolve()
scope = clamp_repo_path(resolved_root, path) if path is not None else None
for relative, target in _iter_clamped_source_files(resolved_root):
if scope is not None and target != scope:
continue
text = target.read_text(errors="ignore")
span = _named_span(relative, text, symbol)
if span is None:
continue
lines = text.splitlines()
start, end = span
while start > 1 and lines[start - 2].lstrip().startswith("@"):
start -= 1 # a handler's decorators are part of what the detector must see
body = "\n".join(lines[start - 1 : end])
return {"path": relative, "name": symbol, "start_line": start, "end_line": end, "text": body[: max(max_chars, 0)]}
return None
def entry_points(root: Path, *, path: str | None = None) -> list[dict[str, object]]:
"""Handlers the mapper names, with their access classification and evidence. No source text."""
from .mapping import analyze_entry_points
resolved_root = root.resolve()
relative = _relative_or_none(resolved_root, path)
targets = _repo_targets(resolved_root)
rows: list[dict[str, object]] = []
for record in analyze_entry_points(resolved_root, targets):
if relative is not None and record.path != relative:
continue
rows.append(
{
"path": record.path,
"function": record.function_name,
"name": record.name,
"kind": record.kind,
"line": record.line,
"end_line": record.end_line,
"access_level": record.access_level,
"evidence": list(record.access_evidence),
}
)
return rows
def flows(root: Path, *, path: str, function: str) -> list[dict[str, object]]:
"""Adjudicated source-to-sink records inside one function: which source, which sink, what the overlay said."""
from .findings import quick_scan_findings
from .rank import rank_targets
from .semantic import adjudicate
resolved_root = root.resolve()
relative = _relative_or_none(resolved_root, path)
targets = _repo_targets(resolved_root)
span = _span_in(resolved_root, relative, function)
if span is None:
return []
findings = [item for item in quick_scan_findings(resolved_root, targets, rank_targets(targets), None) if item.status != "shadow"]
rows: list[dict[str, object]] = []
for record in adjudicate(root=resolved_root, targets=targets, findings=findings):
if record.path != relative or record.line is None or not span[0] <= record.line <= span[1]:
continue
rows.append(
{
"path": record.path,
"function": function,
"line": record.line,
"sources": list(record.sources),
"sinks": list(record.sinks),
"disposition": record.disposition,
"cwe": record.cwe,
}
)
return rows
def obligations(root: Path, *, path: str, function: str) -> list[dict[str, object]]:
"""What one function owes and has not discharged: the operation, the missing discharger, why it is owed."""
from .config import ObligationsConfig
from .semantic.facts import FactLoadError, load_facts
from .semantic.ir import parse_file
from .semantic.obligations import check_obligations, load_obligation_facts
from .semantic.obligations.dominance import OrderDominance
resolved_root = root.resolve()
relative = _relative_or_none(resolved_root, path)
targets = _repo_targets(resolved_root)
try:
flow_facts = load_facts()
except FactLoadError:
return []
irs: dict[str, tuple[object, str]] = {}
texts: dict[str, str] = {}
for target in targets:
try:
text = (resolved_root / target.path).read_text(errors="ignore")
except OSError:
continue
texts[target.path] = text
irs[target.path] = (parse_file(target.path, text, target.language), text)
from .mapping import analyze_entry_points
result = check_obligations(
irs=irs, # type: ignore[arg-type]
entries=analyze_entry_points(resolved_root, targets),
facts=load_obligation_facts(),
flow_facts=flow_facts,
policy=None,
paths=(),
dominance=OrderDominance(texts=texts),
store_shapes=(),
min_siblings=ObligationsConfig().min_siblings,
)
return [
{
"path": finding.operation.path,
"function": finding.operation.function,
"line": finding.operation.line,
"operation": finding.operation.kind,
"resource": finding.operation.resource,
"missing": finding.missing,
"label": finding.label,
"evidence": list(finding.evidence),
}
for finding in result.findings
if (relative is None or finding.operation.path == relative) and finding.operation.function == function
]
def _repo_targets(root: Path) -> list[FileTarget]:
from .preprocess import preprocess_repository
_, targets = preprocess_repository(root)
return list(targets)
def _relative_or_none(root: Path, path: str | None) -> str | None:
if path is None:
return None
return clamp_repo_path(root, path).relative_to(root).as_posix()
def _named_span(relative: str, text: str, symbol: str) -> tuple[int, int] | None:
from .preprocess import detect_language
from .semantic.functions import named_function_ranges, spans_named
ranges = named_function_ranges(relative, text, detect_language(Path(relative)))
spans = spans_named(ranges or (), symbol)
return spans[0] if spans else None
def _span_in(root: Path, relative: str | None, function: str) -> tuple[int, int] | None:
if relative is None:
return None
try:
text = (root / relative).read_text(errors="ignore")
except OSError:
return None
return _named_span(relative, text, function)