783 lines
31 KiB
Python
783 lines
31 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from io import StringIO
|
|
import importlib.util
|
|
import logging
|
|
import math
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Iterable, Sequence, TextIO
|
|
from band_aider.judgement.repo_exec import RepoExec
|
|
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.ai_client.token_budget_estimator import TokenBudgetEstimator
|
|
from band_aider.judgement.check import Check
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.lexical_anchor_check import LexicalAnchorCheck
|
|
from band_aider.judgement.source_discovery_strategy import (
|
|
SourceDiscoveryStrategy,
|
|
searchable_beat_texts,
|
|
)
|
|
|
|
MODEL2VEC_SIMILARITY_THRESHOLD = 0.72
|
|
MODEL2VEC_MODEL_NAME = "minishlab/potion-retrieval-32M"
|
|
MODEL2VEC_CACHE_DIR_ENV = "BAND_AIDER_MODEL2VEC_CACHE_DIR"
|
|
MODEL2VEC_REQUEST_CHUNK_TOKEN_BUDGET = 256
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class RequestChunker:
|
|
"""Split structured request text into bounded, heading-preserving chunks."""
|
|
|
|
token_budget: int = MODEL2VEC_REQUEST_CHUNK_TOKEN_BUDGET
|
|
estimator: TokenBudgetEstimator = field(default_factory=TokenBudgetEstimator)
|
|
|
|
def chunk(self, prompt: str) -> tuple[str, ...]:
|
|
"""Return ordered, non-empty, stable-deduplicated request chunks."""
|
|
if not isinstance(prompt, str) or not prompt.strip():
|
|
return ()
|
|
text = prompt.replace("\r\n", "\n").replace("\r", "\n")
|
|
sections = self._sections(text)
|
|
chunks = [chunk for heading, body in sections for chunk in self._section_chunks(heading, body)]
|
|
return self._deduplicate(chunks)
|
|
|
|
def _sections(self, text: str) -> list[tuple[str, str]]:
|
|
"""Return prompt sections split at markdown or banner headings."""
|
|
lines = text.split("\n")
|
|
sections: list[tuple[str, list[str]]] = []
|
|
heading = ""
|
|
body: list[str] = []
|
|
index = 0
|
|
while index < len(lines):
|
|
line = lines[index]
|
|
candidate = line.strip()
|
|
banner = (
|
|
index + 2 < len(lines)
|
|
and candidate
|
|
and re.fullmatch(r"[=-]{3,}", lines[index + 1].strip() or " ")
|
|
and lines[index + 2].strip() == candidate
|
|
)
|
|
if self._is_heading(candidate) or banner:
|
|
if heading or any(item.strip() for item in body):
|
|
sections.append((heading, body))
|
|
heading = candidate if not banner else candidate
|
|
body = []
|
|
index += 3 if banner else 1
|
|
continue
|
|
body.append(line)
|
|
index += 1
|
|
if heading or any(item.strip() for item in body):
|
|
sections.append((heading, body))
|
|
return [(heading, "\n".join(body).strip()) for heading, body in sections]
|
|
|
|
def _is_heading(self, line: str) -> bool:
|
|
"""Return whether a line is a markdown or conservative plain heading."""
|
|
if not line or line.startswith("```"):
|
|
return False
|
|
if re.match(r"^#{1,6}\s+\S", line):
|
|
return True
|
|
return len(line) <= 100 and not line.endswith((".", ":", ";", ",", "?", "!")) and " " in line
|
|
|
|
def _section_chunks(self, heading: str, body: str) -> list[str]:
|
|
"""Return greedy minimum-sized chunks for one heading and its body."""
|
|
paragraphs = [part.strip() for part in re.split(r"\n\s*\n", body) if part.strip()]
|
|
if not paragraphs:
|
|
return [heading] if heading else []
|
|
chunks: list[str] = []
|
|
current = ""
|
|
for paragraph in paragraphs:
|
|
if self.estimator.estimate(self._join(heading, "", paragraph)) >= self.token_budget:
|
|
if current:
|
|
chunks.append(self._join(heading, "", current))
|
|
current = ""
|
|
chunks.extend(self._split_text(heading, paragraph))
|
|
continue
|
|
current = paragraph if not current else f"{current}\n\n{paragraph}"
|
|
if self.estimator.estimate(self._join(heading, "", current)) >= self.token_budget:
|
|
chunks.append(self._join(heading, "", current))
|
|
current = ""
|
|
if current:
|
|
chunks.append(self._join(heading, "", current))
|
|
return chunks
|
|
|
|
def _split_text(self, heading: str, text: str) -> list[str]:
|
|
"""Greedily split oversized text on lines before using hard slices."""
|
|
if self.estimator.estimate(self._join(heading, "", text)) <= self.token_budget:
|
|
return [self._join(heading, "", text)]
|
|
lines = [line.strip() for line in text.splitlines() if line.strip()]
|
|
chunks: list[str] = []
|
|
current = ""
|
|
for line in lines:
|
|
candidate = self._join(heading, current, line)
|
|
if current and self.estimator.estimate(candidate) >= self.token_budget:
|
|
chunks.append(self._join(heading, "", current))
|
|
current = line
|
|
else:
|
|
current = line if not current else f"{current}\n{line}"
|
|
if current:
|
|
chunks.append(self._join(heading, "", current))
|
|
return [chunk for item in chunks for chunk in self._hard_slice(heading, item)]
|
|
|
|
def _hard_slice(self, heading: str, text: str) -> list[str]:
|
|
"""Split an individually oversized line into sequential contextual pieces."""
|
|
if self.estimator.estimate(text) <= self.token_budget:
|
|
return [text]
|
|
prefix = f"{heading}\n\n" if heading else ""
|
|
body = text[len(prefix):] if prefix and text.startswith(prefix) else text
|
|
ratio = max(1, int(self.estimator.estimate(body) / max(1, len(body))))
|
|
width = max(1, int(self.token_budget / ratio))
|
|
return [f"{prefix}{body[start:start + width]}".strip() for start in range(0, len(body), width)]
|
|
|
|
def _join(self, heading: str, current: str, value: str) -> str:
|
|
"""Join heading context and body text without semantic rewriting."""
|
|
body = f"{current}\n\n{value}" if current else value
|
|
return f"{heading}\n\n{body}".strip() if heading else body.strip()
|
|
|
|
def _deduplicate(self, chunks: Iterable[str]) -> tuple[str, ...]:
|
|
"""Remove empty and exact duplicate chunks while preserving order."""
|
|
seen: set[str] = set()
|
|
result: list[str] = []
|
|
for chunk in chunks:
|
|
if chunk and chunk not in seen:
|
|
seen.add(chunk)
|
|
result.append(chunk)
|
|
return tuple(result)
|
|
|
|
|
|
def default_model2vec_cache_dir() -> Path:
|
|
"""Return the Band-Aider-owned cache root for Model2Vec assets."""
|
|
|
|
configured = os.getenv(MODEL2VEC_CACHE_DIR_ENV, "").strip()
|
|
if configured:
|
|
return Path(configured).expanduser()
|
|
return Path.home() / ".cache" / "band-aider" / "model2vec"
|
|
|
|
|
|
def model2vec_package_available() -> bool:
|
|
"""Return whether the Model2Vec package is importable."""
|
|
|
|
return importlib.util.find_spec("model2vec") is not None
|
|
|
|
|
|
def _snapshot_download(
|
|
model_name: str,
|
|
cache_dir: Path,
|
|
*,
|
|
local_files_only: bool,
|
|
) -> str:
|
|
"""Resolve one model snapshot path, optionally without network access."""
|
|
|
|
from huggingface_hub import snapshot_download
|
|
|
|
return snapshot_download(
|
|
repo_id=model_name,
|
|
cache_dir=str(cache_dir),
|
|
local_files_only=local_files_only,
|
|
)
|
|
|
|
|
|
def resolve_model2vec_model_path(
|
|
model_name: str = MODEL2VEC_MODEL_NAME,
|
|
cache_dir: Path | None = None,
|
|
) -> Path | None:
|
|
"""Return the cached local model path when present, else None."""
|
|
|
|
if not model2vec_package_available():
|
|
return None
|
|
try:
|
|
snapshot_path = _snapshot_download(
|
|
model_name,
|
|
cache_dir or default_model2vec_cache_dir(),
|
|
local_files_only=True,
|
|
)
|
|
except Exception:
|
|
return None
|
|
return Path(snapshot_path)
|
|
|
|
|
|
def model2vec_model_cached(
|
|
model_name: str = MODEL2VEC_MODEL_NAME,
|
|
cache_dir: Path | None = None,
|
|
) -> bool:
|
|
"""Return whether the configured Model2Vec model is already cached locally."""
|
|
|
|
return resolve_model2vec_model_path(model_name=model_name, cache_dir=cache_dir) is not None
|
|
|
|
|
|
def ensure_model2vec_model_downloaded(
|
|
model_name: str = MODEL2VEC_MODEL_NAME,
|
|
cache_dir: Path | None = None,
|
|
) -> Path:
|
|
"""Download the configured Model2Vec model into the Band-Aider cache."""
|
|
|
|
if not model2vec_package_available():
|
|
raise RuntimeError("model2vec is not installed.")
|
|
snapshot_path = _snapshot_download(
|
|
model_name,
|
|
cache_dir or default_model2vec_cache_dir(),
|
|
local_files_only=False,
|
|
)
|
|
return Path(snapshot_path)
|
|
|
|
|
|
@dataclass(init=False)
|
|
class Model2VecSimilarityCheck(Check):
|
|
"""Model2VecSimilarityCheck compares request text with searchable beats by cosine similarity."""
|
|
|
|
request_text: str
|
|
searchable_beats: Iterable[str]
|
|
model_name: str
|
|
cache_dir: Path
|
|
threshold: float
|
|
score: float | None
|
|
matched: bool
|
|
reason: str
|
|
error: str
|
|
best_candidate: str
|
|
best_request_chunk: str
|
|
best_request_chunk_index: int | None
|
|
_model: Any | None = field(default=None, init=False, repr=False)
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
changes: list[tuple[Check, bool]],
|
|
request_text: str,
|
|
searchable_beats: Iterable[str],
|
|
model_name: str = MODEL2VEC_MODEL_NAME,
|
|
cache_dir: Path | None = None,
|
|
threshold: float = MODEL2VEC_SIMILARITY_THRESHOLD,
|
|
) -> None:
|
|
"""Initialize one semantic similarity check with injected request and beats."""
|
|
|
|
self.changes = changes
|
|
self.request_text = request_text
|
|
self.searchable_beats = searchable_beats
|
|
self.model_name = model_name
|
|
self.cache_dir = cache_dir or default_model2vec_cache_dir()
|
|
self.threshold = threshold
|
|
self.score = None
|
|
self.matched = False
|
|
self.reason = ""
|
|
self.error = ""
|
|
self.best_candidate = ""
|
|
self.best_request_chunk = ""
|
|
self.best_request_chunk_index = None
|
|
self._model = None
|
|
|
|
def __call__(
|
|
self,
|
|
request_text: str | None = None,
|
|
searchable_beats: Iterable[str] | None = None,
|
|
) -> bool:
|
|
"""Return whether the best valid candidate score reaches the configured threshold."""
|
|
|
|
try:
|
|
text = self.request_text if request_text is None else request_text
|
|
beats = tuple(self.searchable_beats if searchable_beats is None else searchable_beats)
|
|
result = self._matches(text, beats)
|
|
self.matched = result
|
|
self.changes.append((self, result))
|
|
logger.info(
|
|
"Model2VecSimilarityCheck matched=%s reason=%s score=%s threshold=%s "
|
|
"candidate=%s error=%s",
|
|
result,
|
|
self.reason,
|
|
self.score if self.score is not None else "none",
|
|
self.threshold,
|
|
self.best_candidate or "none",
|
|
self.error or "none",
|
|
)
|
|
return result
|
|
except Exception as error:
|
|
self._record_error(error)
|
|
self.changes.append((self, False))
|
|
return False
|
|
|
|
def score_candidates(
|
|
self, request_text: str, searchable_beats: Iterable[str]
|
|
) -> list[tuple[str, float]]:
|
|
"""Score every supplied candidate using this check's cached model."""
|
|
if not isinstance(request_text, str) or not request_text.strip():
|
|
return []
|
|
beats = tuple(beat.strip() for beat in searchable_beats if isinstance(beat, str) and beat.strip())
|
|
if not beats or not model2vec_package_available():
|
|
return []
|
|
model = self._load_model()
|
|
request_vector = self._validated_vector(model.encode(request_text), context="request")
|
|
values = self._candidate_vectors(model.encode(list(beats)), len(beats))
|
|
scored = []
|
|
for beat, value in zip(beats, values):
|
|
try:
|
|
scored.append((beat, self._cosine_similarity(request_vector, self._validated_vector(value, context="candidate"))))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return scored
|
|
|
|
def match_request_chunks(
|
|
self,
|
|
request_chunks: Iterable[str],
|
|
searchable_beats: Iterable[str] | None = None,
|
|
) -> bool:
|
|
"""Match every request chunk against every beat while retaining the winning pair."""
|
|
try:
|
|
chunks = tuple(chunk.strip() for chunk in request_chunks if isinstance(chunk, str) and chunk.strip())
|
|
beats = tuple(self.searchable_beats if searchable_beats is None else searchable_beats)
|
|
beats = tuple(beat.strip() for beat in beats if isinstance(beat, str) and beat.strip())
|
|
if not chunks:
|
|
self.reason = "empty_request_chunks"
|
|
return False
|
|
if not beats:
|
|
self.reason = "empty_searchable_beats"
|
|
return False
|
|
if not model2vec_package_available():
|
|
self.reason = "model2vec_package_unavailable"
|
|
return False
|
|
model = self._load_model()
|
|
request_values = self._candidate_vectors(model.encode(list(chunks)), len(chunks))
|
|
beat_values = self._candidate_vectors(model.encode(list(beats)), len(beats))
|
|
best: tuple[float, int, str] | None = None
|
|
for chunk_index, request_value in enumerate(request_values):
|
|
try:
|
|
request_vector = self._validated_vector(request_value, context="request chunk")
|
|
except (TypeError, ValueError) as error:
|
|
logger.warning("Model2VecSimilarityCheck skipped invalid request chunk vector: %s: %s", type(error).__name__, error)
|
|
continue
|
|
for beat, beat_value in zip(beats, beat_values):
|
|
try:
|
|
score = self._cosine_similarity(request_vector, self._validated_vector(beat_value, context="candidate"))
|
|
except (TypeError, ValueError) as error:
|
|
logger.warning("Model2VecSimilarityCheck skipped invalid candidate vector: %s: %s", type(error).__name__, error)
|
|
continue
|
|
if best is None or score > best[0]:
|
|
best = (score, chunk_index, beat)
|
|
if best is None:
|
|
self.reason = "no_valid_candidate_vectors"
|
|
return False
|
|
self.score, self.best_request_chunk_index, self.best_candidate = best
|
|
self.best_request_chunk = chunks[self.best_request_chunk_index]
|
|
self.reason = "matched" if self.score >= self.threshold else "below_threshold"
|
|
return self.score >= self.threshold
|
|
except Exception as error:
|
|
self._record_error(error)
|
|
return False
|
|
|
|
def _matches(self, request_text: str, beats: tuple[str, ...]) -> bool:
|
|
"""Return whether request and beat vectors produce an accepted cosine score."""
|
|
|
|
if not isinstance(request_text, str) or not request_text.strip():
|
|
self.reason = "empty_request_text"
|
|
return False
|
|
usable_beats = tuple(beat.strip() for beat in beats if isinstance(beat, str) and beat.strip())
|
|
if not usable_beats:
|
|
self.reason = "empty_searchable_beats"
|
|
return False
|
|
if not model2vec_package_available():
|
|
self.reason = "model2vec_package_unavailable"
|
|
return False
|
|
|
|
model = self._load_model()
|
|
request_vector = self._validated_vector(model.encode(request_text), context="request")
|
|
encoded = model.encode(list(usable_beats))
|
|
candidate_vectors = self._candidate_vectors(encoded, len(usable_beats))
|
|
|
|
best_score: float | None = None
|
|
best_candidate = ""
|
|
for beat, vector_value in zip(usable_beats, candidate_vectors):
|
|
try:
|
|
vector = self._validated_vector(vector_value, context="candidate")
|
|
score = self._cosine_similarity(request_vector, vector)
|
|
except (TypeError, ValueError) as error:
|
|
logger.warning(
|
|
"Model2VecSimilarityCheck skipped invalid candidate vector: %s: %s",
|
|
type(error).__name__,
|
|
error,
|
|
)
|
|
continue
|
|
if best_score is None or score > best_score:
|
|
best_score = score
|
|
best_candidate = beat
|
|
|
|
if best_score is None:
|
|
self.reason = "no_valid_candidate_vectors"
|
|
return False
|
|
|
|
self.score = best_score
|
|
self.best_candidate = best_candidate
|
|
self.reason = "matched" if best_score >= self.threshold else "below_threshold"
|
|
return best_score >= self.threshold
|
|
|
|
def _load_model(self) -> Any:
|
|
"""Return the cached local static embedding model."""
|
|
|
|
if self._model is not None:
|
|
return self._model
|
|
local_path = resolve_model2vec_model_path(
|
|
model_name=self.model_name,
|
|
cache_dir=self.cache_dir,
|
|
)
|
|
if local_path is None:
|
|
raise FileNotFoundError(
|
|
f"Model2Vec model {self.model_name} is not cached under {self.cache_dir}."
|
|
)
|
|
from model2vec import StaticModel
|
|
|
|
self._model = StaticModel.from_pretrained(str(local_path), force_download=False)
|
|
return self._model
|
|
|
|
def _candidate_vectors(self, encoded: object, expected_count: int) -> tuple[object, ...]:
|
|
"""Return candidate vector objects from one encoder result."""
|
|
|
|
if isinstance(encoded, (str, bytes)):
|
|
raise TypeError("candidate encoder output must be a vector sequence")
|
|
try:
|
|
values = tuple(encoded) # type: ignore[arg-type]
|
|
except TypeError as error:
|
|
raise TypeError("candidate encoder output must be iterable") from error
|
|
if len(values) != expected_count:
|
|
raise ValueError(
|
|
f"candidate encoder output count {len(values)} did not match beats {expected_count}"
|
|
)
|
|
return values
|
|
|
|
def _validated_vector(self, vector: object, *, context: str) -> tuple[float, ...]:
|
|
"""Return one finite nonzero numeric vector or raise a normalized validation error."""
|
|
|
|
if isinstance(vector, (str, bytes)):
|
|
raise TypeError(f"{context} vector must be a numeric sequence")
|
|
try:
|
|
values = tuple(float(value) for value in vector) # type: ignore[arg-type]
|
|
except TypeError as error:
|
|
raise TypeError(f"{context} vector must be iterable") from error
|
|
except ValueError as error:
|
|
raise ValueError(f"{context} vector contains non-numeric values") from error
|
|
if not values:
|
|
raise ValueError(f"{context} vector is empty")
|
|
if any(math.isnan(value) for value in values):
|
|
raise ValueError(f"{context} vector contains NaN")
|
|
if any(math.isinf(value) for value in values):
|
|
raise ValueError(f"{context} vector contains infinity")
|
|
if math.sqrt(sum(value * value for value in values)) <= 0:
|
|
raise ValueError(f"{context} vector has zero norm")
|
|
return values
|
|
|
|
def _cosine_similarity(self, left: Sequence[float], right: Sequence[float]) -> float:
|
|
"""Return cosine similarity between two validated dense vectors."""
|
|
|
|
if len(left) != len(right):
|
|
raise ValueError(f"vector dimension mismatch: {len(left)} != {len(right)}")
|
|
numerator = 0.0
|
|
left_norm = 0.0
|
|
right_norm = 0.0
|
|
for left_value, right_value in zip(left, right):
|
|
numerator += left_value * right_value
|
|
left_norm += left_value * left_value
|
|
right_norm += right_value * right_value
|
|
if left_norm <= 0 or right_norm <= 0:
|
|
raise ValueError("cosine similarity requires nonzero vector norms")
|
|
score = numerator / ((left_norm ** 0.5) * (right_norm ** 0.5))
|
|
if math.isnan(score) or math.isinf(score):
|
|
raise ValueError("cosine similarity produced a non-finite score")
|
|
return score
|
|
|
|
def _record_error(self, error: BaseException) -> None:
|
|
"""Store and log normalized diagnostics for one semantic failure."""
|
|
|
|
self.matched = False
|
|
self.reason = "error"
|
|
self.error = f"{type(error).__name__}: {error}"
|
|
logger.exception("Model2VecSimilarityCheck failed closed: %s", self.error)
|
|
|
|
|
|
LexicalCheckFactory = Callable[[Iterable[str], Iterable[str]], LexicalAnchorCheck]
|
|
SemanticCheckFactory = Callable[[str, Iterable[str]], Model2VecSimilarityCheck]
|
|
|
|
|
|
def create_lexical_anchor_check(
|
|
anchors: Iterable[str],
|
|
beats: Iterable[str],
|
|
) -> LexicalAnchorCheck:
|
|
"""Create one fresh lexical anchor check for parser prefiltering."""
|
|
|
|
return LexicalAnchorCheck(
|
|
changes=[],
|
|
anchors=anchors,
|
|
candidate_beats=beats,
|
|
)
|
|
|
|
|
|
def create_model2vec_similarity_check(
|
|
request_text: str,
|
|
beats: Iterable[str],
|
|
) -> Model2VecSimilarityCheck:
|
|
"""Create one fresh Model2Vec similarity check for parser prefiltering."""
|
|
|
|
return Model2VecSimilarityCheck(
|
|
changes=[],
|
|
request_text=request_text,
|
|
searchable_beats=beats,
|
|
)
|
|
|
|
|
|
@dataclass(init=False)
|
|
class NarrowParser(CodeParser):
|
|
"""NarrowParser gates parser delegation through lexical-or-semantic source prefiltering."""
|
|
|
|
_inner: CodeParser = field(init=False, repr=False)
|
|
source_discovery_strategy: SourceDiscoveryStrategy
|
|
anchors: tuple[str, ...]
|
|
lexical_check_factory: LexicalCheckFactory
|
|
semantic_check_factory: SemanticCheckFactory
|
|
semantic_threshold: float
|
|
request_chunker: RequestChunker
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
inner: CodeParser,
|
|
source_discovery_strategy: SourceDiscoveryStrategy,
|
|
anchors: Iterable[str],
|
|
lexical_check_factory: LexicalCheckFactory | None = None,
|
|
semantic_check_factory: SemanticCheckFactory | None = None,
|
|
semantic_threshold: float = MODEL2VEC_SIMILARITY_THRESHOLD,
|
|
request_chunker: RequestChunker | None = None,
|
|
) -> None:
|
|
"""Initialize one narrow parser wrapper with injected strategy and check factories."""
|
|
|
|
self._inner = inner
|
|
self.source_discovery_strategy = source_discovery_strategy
|
|
self.anchors = tuple(anchors)
|
|
self.lexical_check_factory = lexical_check_factory or create_lexical_anchor_check
|
|
self.semantic_check_factory = semantic_check_factory or create_model2vec_similarity_check
|
|
self.semantic_threshold = semantic_threshold
|
|
self.request_chunker = request_chunker or RequestChunker()
|
|
|
|
@property
|
|
def inner(self) -> CodeParser:
|
|
"""Return the wrapped parser."""
|
|
|
|
return self._inner
|
|
|
|
# Forward the execution adapter to callers that require it
|
|
@property
|
|
def repo_exec(self) -> RepoExec:
|
|
return self._inner.repo_exec
|
|
|
|
def __getstate__(self) -> dict[str, object]:
|
|
"""Return durable parser state while stripping runtime-owned factory state."""
|
|
|
|
return dict(self.__dict__)
|
|
|
|
def parse_file(
|
|
self,
|
|
file_name: str,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Parse one file path into ClassMeta metadata."""
|
|
|
|
source = self.repo_exec.read_text(Path(file_name), encoding="utf-8")
|
|
return self._parse_source(file_name, source, prompt, content_type)
|
|
|
|
def parse_stream(
|
|
self,
|
|
file_name: str,
|
|
stream: TextIO,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Parse one readable stream into ClassMeta metadata."""
|
|
|
|
source = stream.read()
|
|
return self._parse_source(file_name, source, prompt, content_type)
|
|
|
|
def parse_files(
|
|
self,
|
|
file_names: list[str],
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> list[ClassMeta]:
|
|
"""Parse multiple file paths into ClassMeta metadata."""
|
|
|
|
parsed: list[ClassMeta] = []
|
|
for file_name in file_names:
|
|
source = self.repo_exec.read_text(Path(file_name), encoding="utf-8")
|
|
parsed.append(self._parse_source(file_name, source, prompt, content_type))
|
|
return parsed
|
|
|
|
def _parse_source(
|
|
self,
|
|
file_name: str,
|
|
source: str,
|
|
prompt: str,
|
|
content_type: ContentType,
|
|
) -> ClassMeta:
|
|
"""Route one source string through narrow parse handling."""
|
|
|
|
beats = self._searchable_beats(file_name, source)
|
|
if not beats:
|
|
self._log_decision(
|
|
file_name=file_name,
|
|
beats=0,
|
|
lexical_status="skipped",
|
|
lexical_result=False,
|
|
semantic_status="skipped",
|
|
semantic_result=False,
|
|
decision="skip",
|
|
reason="empty_searchable_beats",
|
|
)
|
|
return self._skip_meta(file_name, content_type)
|
|
|
|
lexical_check = self.lexical_check_factory(self.anchors, beats)
|
|
lexical_result = lexical_check()
|
|
if lexical_result:
|
|
self._log_decision(
|
|
file_name=file_name,
|
|
beats=len(beats),
|
|
lexical_status="complete",
|
|
lexical_result=True,
|
|
semantic_status="skipped",
|
|
semantic_result=False,
|
|
decision="forward",
|
|
lexical_check=lexical_check,
|
|
)
|
|
return self.inner.parse_stream(file_name, StringIO(source), prompt, content_type)
|
|
|
|
if not source.strip():
|
|
self._log_decision(
|
|
file_name=file_name,
|
|
beats=len(beats),
|
|
lexical_status="complete",
|
|
lexical_result=False,
|
|
semantic_status="skipped",
|
|
semantic_result=False,
|
|
decision="skip",
|
|
reason="empty_source",
|
|
lexical_check=lexical_check,
|
|
)
|
|
return self._skip_meta(file_name, content_type)
|
|
|
|
semantic_check = self.semantic_check_factory(prompt, beats)
|
|
semantic_check.threshold = self.semantic_threshold
|
|
request_chunks = self.request_chunker.chunk(prompt)
|
|
semantic_result = semantic_check.match_request_chunks(request_chunks, beats)
|
|
decision = "forward" if semantic_result else "skip"
|
|
self._log_decision(
|
|
file_name=file_name,
|
|
beats=len(beats),
|
|
lexical_status="complete",
|
|
lexical_result=False,
|
|
semantic_status="complete" if not semantic_check.error else "error",
|
|
semantic_result=semantic_result,
|
|
decision=decision,
|
|
lexical_check=lexical_check,
|
|
semantic_check=semantic_check,
|
|
request_chunks=len(request_chunks),
|
|
)
|
|
if semantic_result:
|
|
return self.inner.parse_stream(file_name, StringIO(source), prompt, content_type)
|
|
return self._skip_meta(file_name, content_type)
|
|
|
|
def _searchable_beats(self, file_name: str, source: str) -> tuple[str, ...]:
|
|
"""Return language-strategy searchable beats while failing closed on extraction errors."""
|
|
|
|
try:
|
|
beats = searchable_beat_texts(
|
|
self.source_discovery_strategy.searchable_beats(file_name, source)
|
|
)
|
|
except Exception as error:
|
|
logger.exception(
|
|
"Prefilter scope=narrow file=%s beat_extraction=error error=%s: %s decision=skip",
|
|
file_name,
|
|
type(error).__name__,
|
|
error,
|
|
)
|
|
return ()
|
|
return beats
|
|
|
|
def _skip_meta(self, file_name: str, content_type: ContentType) -> ClassMeta:
|
|
"""Return conservative metadata when the prefilter decides to skip parsing."""
|
|
|
|
return ClassMeta(
|
|
location=str(file_name),
|
|
summary=f"Search prefilter skipped advisory parse for {file_name}.",
|
|
relevance=[],
|
|
relevant_symbols={},
|
|
content_type=content_type,
|
|
)
|
|
|
|
def _log_decision(
|
|
self,
|
|
*,
|
|
file_name: str,
|
|
beats: int,
|
|
lexical_status: str,
|
|
lexical_result: bool,
|
|
semantic_status: str,
|
|
semantic_result: bool,
|
|
decision: str,
|
|
reason: str = "",
|
|
lexical_check: LexicalAnchorCheck | None = None,
|
|
semantic_check: Model2VecSimilarityCheck | None = None,
|
|
request_chunks: int = 0,
|
|
) -> None:
|
|
"""Log one structured narrow prefilter decision."""
|
|
|
|
best_anchor = lexical_check.best_anchor if lexical_check is not None else ""
|
|
best_lexical_candidate = lexical_check.best_candidate if lexical_check is not None else ""
|
|
edit_distance = lexical_check.edit_distance if lexical_check is not None else None
|
|
allowed_edit_distance = lexical_check.allowed_edit_distance if lexical_check is not None else None
|
|
comparison_count = lexical_check.comparison_count if lexical_check is not None else 0
|
|
lexical_reason = lexical_check.reason if lexical_check is not None else ""
|
|
lexical_error = lexical_check.error if lexical_check is not None else ""
|
|
score = semantic_check.score if semantic_check is not None else None
|
|
request_chunk_index = semantic_check.best_request_chunk_index if semantic_check is not None else None
|
|
request_chunk = semantic_check.best_request_chunk if semantic_check is not None else ""
|
|
best_semantic_candidate = semantic_check.best_candidate if semantic_check is not None else ""
|
|
threshold = semantic_check.threshold if semantic_check is not None else MODEL2VEC_SIMILARITY_THRESHOLD
|
|
semantic_reason = semantic_check.reason if semantic_check is not None else ""
|
|
semantic_error = semantic_check.error if semantic_check is not None else ""
|
|
|
|
logger.info(
|
|
"Prefilter scope=narrow file=%s anchors=%s beats=%s lexical=%s "
|
|
"lexical_status=%s lexical_anchor=%s lexical_candidate=%s edit_distance=%s "
|
|
"allowed_distance=%s comparisons=%s request_chunks=%s semantic=%s semantic_status=%s "
|
|
"cosine=%s threshold=%s request_chunk_index=%s request_chunk=%s candidate=%s "
|
|
"reason=%s error=%s decision=%s",
|
|
file_name,
|
|
len(self.anchors),
|
|
beats,
|
|
lexical_result,
|
|
lexical_status,
|
|
best_anchor or "none",
|
|
best_lexical_candidate or "none",
|
|
edit_distance if edit_distance is not None else "none",
|
|
allowed_edit_distance if allowed_edit_distance is not None else "none",
|
|
comparison_count,
|
|
request_chunks,
|
|
semantic_result if semantic_status != "skipped" else "skipped",
|
|
semantic_status,
|
|
score if score is not None else "none",
|
|
request_chunk_index if request_chunk_index is not None else "none",
|
|
self._preview(request_chunk),
|
|
best_semantic_candidate or "none",
|
|
threshold,
|
|
reason or lexical_reason or semantic_reason or "none",
|
|
lexical_error or semantic_error or "none",
|
|
decision,
|
|
)
|
|
|
|
@staticmethod
|
|
def _preview(value: str, limit: int = 160) -> str:
|
|
"""Return a bounded single-line diagnostic preview."""
|
|
# TODO: Deliberate with a human whether to remove this arbitrary diagnostic-preview cap.
|
|
text = " ".join(value.split())
|
|
return text[:limit] + ("..." if len(text) > limit else "") if text else "none"
|