search and inspection polish

This commit is contained in:
2026-04-25 23:59:38 +02:00
parent cc0eef21be
commit b8627c0e1d
5 changed files with 252 additions and 22 deletions

View File

@@ -168,6 +168,14 @@ class SearchResult:
match_type: str
match_name: str
confidence: float
match_description: str = ""
matched_field: str = ""
ability_id: int | None = None
ability_name: str | None = None
capability_id: int | None = None
capability_name: str | None = None
evidence_level: str | None = None
source_reference: str | None = None
@dataclass(frozen=True)

View File

@@ -1276,43 +1276,201 @@ class RegistryStore:
return RepositoryAbilityMap(repository=repository, abilities=abilities)
def search(self, query: str) -> list[SearchResult]:
needle = f"%{query.strip()}%"
if needle == "%%":
term = query.strip()
needle = f"%{term}%"
if not term:
return []
with self.connect() as connection:
rows = connection.execute(
repository_rows = connection.execute(
"""
SELECT r.id AS repository_id, r.name AS repository_name,
'repository' AS match_type, r.name AS match_name,
1.0 AS confidence
r.description
FROM repositories r
WHERE r.name LIKE ? OR COALESCE(r.description, '') LIKE ?
UNION ALL
SELECT r.id, r.name, 'ability', a.name, a.confidence
""",
(needle, needle),
).fetchall()
ability_rows = connection.execute(
"""
SELECT r.id AS repository_id, r.name AS repository_name,
a.id AS ability_id, a.name AS ability_name,
a.description AS ability_description, a.confidence
FROM approved_abilities a
JOIN repositories r ON r.id = a.repository_id
WHERE a.name LIKE ? OR a.description LIKE ?
UNION ALL
SELECT r.id, r.name, 'capability', c.name, c.confidence
""",
(needle, needle),
).fetchall()
capability_rows = connection.execute(
"""
SELECT r.id AS repository_id, r.name AS repository_name,
a.id AS ability_id, a.name AS ability_name,
c.id AS capability_id, c.name AS capability_name,
c.description AS capability_description, c.confidence
FROM approved_capabilities c
JOIN approved_abilities a ON a.id = c.ability_id
JOIN repositories r ON r.id = c.repository_id
WHERE c.name LIKE ? OR c.description LIKE ?
ORDER BY confidence DESC, repository_name ASC, match_name ASC
""",
(needle, needle, needle, needle, needle, needle),
(needle, needle),
).fetchall()
feature_rows = connection.execute(
"""
SELECT r.id AS repository_id, r.name AS repository_name,
a.id AS ability_id, a.name AS ability_name,
c.id AS capability_id, c.name AS capability_name,
f.name AS feature_name, f.type AS feature_type,
f.location, f.confidence
FROM approved_features f
JOIN approved_capabilities c ON c.id = f.capability_id
JOIN approved_abilities a ON a.id = c.ability_id
JOIN repositories r ON r.id = f.repository_id
WHERE f.name LIKE ? OR f.type LIKE ? OR f.location LIKE ?
""",
(needle, needle, needle),
).fetchall()
evidence_rows = connection.execute(
"""
SELECT r.id AS repository_id, r.name AS repository_name,
a.id AS ability_id, a.name AS ability_name,
c.id AS capability_id, c.name AS capability_name,
e.type AS evidence_type, e.reference, e.strength
FROM approved_evidence e
JOIN approved_capabilities c ON c.id = e.capability_id
JOIN approved_abilities a ON a.id = c.ability_id
JOIN repositories r ON r.id = e.repository_id
WHERE e.type LIKE ? OR e.reference LIKE ? OR e.strength LIKE ?
""",
(needle, needle, needle),
).fetchall()
return [
SearchResult(
repository_id=row["repository_id"],
repository_name=row["repository_name"],
match_type=row["match_type"],
match_name=row["match_name"],
confidence=row["confidence"],
results: list[SearchResult] = []
for row in repository_rows:
matched_field = (
"name" if self._matches(row["repository_name"], term) else "description"
)
for row in rows
]
results.append(
SearchResult(
repository_id=row["repository_id"],
repository_name=row["repository_name"],
match_type="repository",
match_name=row["repository_name"],
match_description=row["description"] or "",
matched_field=matched_field,
confidence=1.0,
)
)
for row in ability_rows:
matched_field = (
"name" if self._matches(row["ability_name"], term) else "description"
)
results.append(
SearchResult(
repository_id=row["repository_id"],
repository_name=row["repository_name"],
match_type="ability",
match_name=row["ability_name"],
match_description=row["ability_description"],
matched_field=matched_field,
confidence=row["confidence"],
ability_id=row["ability_id"],
ability_name=row["ability_name"],
)
)
for row in capability_rows:
matched_field = (
"name"
if self._matches(row["capability_name"], term)
else "description"
)
results.append(
SearchResult(
repository_id=row["repository_id"],
repository_name=row["repository_name"],
match_type="capability",
match_name=row["capability_name"],
match_description=row["capability_description"],
matched_field=matched_field,
confidence=row["confidence"],
ability_id=row["ability_id"],
ability_name=row["ability_name"],
capability_id=row["capability_id"],
capability_name=row["capability_name"],
)
)
for row in feature_rows:
matched_field = self._first_matched_field(
term,
{
"name": row["feature_name"],
"type": row["feature_type"],
"location": row["location"],
},
)
results.append(
SearchResult(
repository_id=row["repository_id"],
repository_name=row["repository_name"],
match_type="feature",
match_name=row["feature_name"],
match_description=row["feature_type"],
matched_field=matched_field,
confidence=row["confidence"],
ability_id=row["ability_id"],
ability_name=row["ability_name"],
capability_id=row["capability_id"],
capability_name=row["capability_name"],
source_reference=row["location"],
)
)
for row in evidence_rows:
matched_field = self._first_matched_field(
term,
{
"type": row["evidence_type"],
"reference": row["reference"],
"strength": row["strength"],
},
)
results.append(
SearchResult(
repository_id=row["repository_id"],
repository_name=row["repository_name"],
match_type="evidence",
match_name=row["reference"],
match_description=row["evidence_type"],
matched_field=matched_field,
confidence=self._evidence_confidence(row["strength"]),
ability_id=row["ability_id"],
ability_name=row["ability_name"],
capability_id=row["capability_id"],
capability_name=row["capability_name"],
evidence_level=row["strength"],
source_reference=row["reference"],
)
)
return sorted(
results,
key=lambda result: (
-result.confidence,
result.repository_name.lower(),
result.match_type,
result.match_name.lower(),
),
)
def _matches(self, value: str | None, term: str) -> bool:
return term.lower() in (value or "").lower()
def _first_matched_field(self, term: str, values: dict[str, str | None]) -> str:
for field, value in values.items():
if self._matches(value, term):
return field
return ""
def _evidence_confidence(self, strength: str) -> float:
return {"strong": 0.9, "medium": 0.6, "weak": 0.3}.get(strength, 0.5)
def _insert_facts(
self,

View File

@@ -184,7 +184,11 @@ def search_page(
<tr>
<td><a href="/ui/repos/{result.repository_id}">{escape(result.repository_name)}</a></td>
<td><span class="pill">{escape(result.match_type)}</span></td>
<td>{escape(result.match_name)}</td>
<td>
<strong>{escape(result.match_name)}</strong>
{render_search_context(asdict(result))}
</td>
<td>{escape(result.matched_field)}</td>
<td>{result.confidence:.2f}</td>
</tr>
"""
@@ -208,7 +212,7 @@ def search_page(
</section>
<section class="panel" style="margin-top:18px">
<table>
<thead><tr><th>Repository</th><th>Match</th><th>Name</th><th>Confidence</th></tr></thead>
<thead><tr><th>Repository</th><th>Match</th><th>Name</th><th>Field</th><th>Confidence</th></tr></thead>
<tbody>{rows or empty}</tbody>
</table>
</section>
@@ -917,3 +921,20 @@ def render_sources(source_refs: list[dict]) -> str:
if len(source_refs) > 5:
sources += f' <span class="muted">+{len(source_refs) - 5} more</span>'
return f"<p>{sources}</p>"
def render_search_context(result: dict) -> str:
details = []
if result.get("ability_name"):
details.append(f"Ability: {escape(result['ability_name'])}")
if result.get("capability_name"):
details.append(f"Capability: {escape(result['capability_name'])}")
if result.get("evidence_level"):
details.append(f"Evidence: {escape(result['evidence_level'])}")
if result.get("source_reference"):
details.append(f"Source: {escape(result['source_reference'])}")
if result.get("match_description"):
details.append(escape(result["match_description"]))
if not details:
return ""
return f'<p class="muted">{" · ".join(details)}</p>'

View File

@@ -96,6 +96,47 @@ def test_search_matches_approved_abilities_and_capabilities(tmp_path):
assert capabilities[0].name == "Classify Incoming Email"
def test_search_matches_features_and_evidence_with_context(tmp_path):
service = make_service(tmp_path)
repository = service.register_repository(
name="MailRouter",
url="https://example.com/mail-router-feature.git",
description="Manual test repository.",
)
ability_id = service.add_ability(repository.id, name="Business Email Routing")
capability_id = service.add_capability(
repository.id,
ability_id,
name="Classify Incoming Email",
)
service.add_feature(
repository.id,
capability_id,
name="POST /api/classify-email",
type="REST endpoint",
location="src/routes/classify_email.py",
)
service.add_evidence(
repository.id,
capability_id,
type="unit_test",
reference="tests/test_email_classification.py",
strength="strong",
)
feature_results = service.search("classify_email")
evidence_results = service.search("unit_test")
assert feature_results[0].match_type == "feature"
assert feature_results[0].matched_field == "location"
assert feature_results[0].ability_name == "Business Email Routing"
assert feature_results[0].capability_name == "Classify Incoming Email"
assert feature_results[0].source_reference == "src/routes/classify_email.py"
assert evidence_results[0].match_type == "evidence"
assert evidence_results[0].evidence_level == "strong"
assert evidence_results[0].confidence == 0.9
def test_register_repository_imports_metadata_when_name_is_omitted(tmp_path):
source = tmp_path / "metadata-source"
source.mkdir()

View File

@@ -211,6 +211,7 @@ def test_api_analysis_run_loop(tmp_path):
search_response = client.get("/search", params={"q": "frontend"})
assert search_response.status_code == 200
assert search_response.json()
assert "matched_field" in search_response.json()[0]
abilities_response = client.get("/abilities")
assert abilities_response.status_code == 200
@@ -303,6 +304,7 @@ def test_ui_register_analyze_and_approve_loop(tmp_path):
search_response = client.get("/ui/search", params={"q": "repository"})
assert search_response.status_code == 200
assert "UI Repo" in search_response.text
assert "Field" in search_response.text
finally:
app.dependency_overrides.clear()