from __future__ import annotations
from dataclasses import asdict
from html import escape
from fastapi import APIRouter, Depends, Form, HTTPException, Query
from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse
from repo_registry.core.service import RegistryService
from repo_registry.storage.sqlite import NotFoundError
from repo_registry.web_api.app import get_service
router = APIRouter(include_in_schema=False)
def page(title: str, body: str) -> HTMLResponse:
return HTMLResponse(
f"""
{escape(title)} · Repository Ability Registry
{body}
"""
)
def render_repository_index(
service: RegistryService,
*,
error_message: str | None = None,
status_code: int = 200,
) -> HTMLResponse:
repositories = service.list_repositories()
rows = "\n".join(
f"""
{escape(repo.name)} {escape(repo.status)} {escape(repo.branch)}
{escape(repo.url)}
"""
for repo in repositories
)
error = (
f"""
Registration failed.
{escape(error_message)}
"""
if error_message
else ""
)
body = f"""
Repositories
{error}
Name Status Branch Source {rows or 'No repositories yet.
"""
response = page("Repositories", body)
response.status_code = status_code
return response
@router.get("/ui")
def repository_index(service: RegistryService = Depends(get_service)) -> HTMLResponse:
return render_repository_index(service)
@router.get("/ui/discovery")
def discovery_page(service: RegistryService = Depends(get_service)) -> HTMLResponse:
repositories = service.list_repositories()
return page(
"Discovery",
f"""
""",
)
@router.get("/ui/discovery/compare")
def discovery_compare_page(
repository_ids: list[int] = Query(default=[]),
service: RegistryService = Depends(get_service),
) -> HTMLResponse:
if len(repository_ids) < 2:
return page(
"Repository Comparison",
"""
Select at least two repositories with approved profiles.
""",
)
try:
comparison = service.compare_repositories(repository_ids)
except NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
body = f"""
Compared Repositories
{render_compared_repositories(comparison["repositories"])}
Shared Abilities
{render_compared_abilities(comparison["abilities"])}
Unique Capabilities
{render_unique_capabilities(comparison["unique_capabilities"])}
"""
return page("Repository Comparison", body)
@router.post("/ui/discovery/gaps")
def discovery_gap_report_page(
desired_ability: str = Form(...),
desired_capabilities: str = Form(...),
repository_ids: list[int] = Form(default=[]),
service: RegistryService = Depends(get_service),
) -> HTMLResponse:
capabilities = split_capability_lines(desired_capabilities)
try:
report = service.detect_capability_gaps(
desired_ability=desired_ability,
desired_capabilities=capabilities,
repository_ids=repository_ids or None,
)
except NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
body = f"""
{escape(report["desired_ability"])}
{render_gap_report(report)}
"""
return page("Capability Gap Report", body)
@router.get("/ui/search")
def search_page(
q: str = "",
status: str = "",
language: str = "",
framework: str = "",
ability: str = "",
capability: str = "",
service: RegistryService = Depends(get_service),
) -> HTMLResponse:
results = (
service.search(
q,
status=status or None,
language=language or None,
framework=framework or None,
ability=ability or None,
capability=capability or None,
)
if q.strip()
else []
)
rows = "\n".join(
f"""
{escape(result.repository_name)} {escape(result.match_type)}
{escape(result.match_name)}
{render_search_context(asdict(result))}
{escape(result.matched_field)}
{result.confidence:.2f} {escape(result.confidence_label)}
"""
for result in results
)
empty = (
'No matches. Enter a need, capability, or repository name.
Repository Match Name Field Confidence {rows or empty}
"""
return page("Search", body)
@router.post("/ui/repos")
def create_repository_from_form(
url: str = Form(...),
branch: str = Form("main"),
service: RegistryService = Depends(get_service),
):
try:
repository = service.register_repository(
url=url,
branch=branch or "main",
)
except (RuntimeError, ValueError) as exc:
return render_repository_index(
service,
error_message=str(exc),
status_code=400,
)
return RedirectResponse(f"/ui/repos/{repository.id}", status_code=303)
@router.get("/ui/repos/{repository_id}")
def repository_detail(
repository_id: int,
service: RegistryService = Depends(get_service),
) -> HTMLResponse:
try:
repository = service.get_repository(repository_id)
except NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
runs = service.list_analysis_runs(repository_id)
ability_map = service.ability_map(repository_id)
decisions = service.list_review_decisions(repository_id)
facts = service.list_observed_facts(repository_id)
languages = sorted({fact.name for fact in facts if fact.kind == "language"})
frameworks = sorted({fact.name for fact in facts if fact.kind == "framework"})
run_rows = "\n".join(
f"""
#{run.id} {escape(run.status)} {escape(run.started_at)}
{render_run_compare_link(repository_id, run.id, runs)}
{escape(run.error_message or '')}
"""
for run in runs
)
body = f"""
{escape(repository.description or '')}
{escape(repository.status)} {escape(repository.url)}
{render_repository_facts(languages, frameworks)}
Repository Metadata
Run Analysis
Analysis Runs
Run Status Started Compare Error {run_rows or 'No runs yet.
Approved Ability Map
{render_ability_map(asdict(ability_map), repository_id)}
Review Decisions
{render_review_decisions(decisions)}
"""
return page(repository.name, body)
@router.get("/ui/repos/{repository_id}/export")
def export_repository_from_ui(
repository_id: int,
service: RegistryService = Depends(get_service),
) -> PlainTextResponse:
try:
content = service.export_registry_entry(repository_id)
except NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
return PlainTextResponse(content, media_type="application/x-yaml")
@router.post("/ui/repos/{repository_id}/edit")
def edit_repository_from_form(
repository_id: int,
name: str = Form(...),
description: str = Form(""),
branch: str = Form("main"),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.update_repository(
repository_id,
name=name,
description=description,
branch=branch or "main",
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/delete")
def delete_repository_from_form(
repository_id: int,
confirm_name: str = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
repository = service.get_repository(repository_id)
if confirm_name != repository.name:
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
service.delete_repository(repository_id)
return RedirectResponse("/ui", status_code=303)
@router.post("/ui/repos/{repository_id}/abilities")
def create_ability_from_form(
repository_id: int,
name: str = Form(...),
description: str = Form(""),
confidence: float = Form(1.0),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.add_ability(
repository_id,
name=name,
description=description,
confidence=confidence,
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/capabilities")
def create_capability_from_form(
repository_id: int,
ability_id: int = Form(...),
name: str = Form(...),
description: str = Form(""),
inputs: str = Form(""),
outputs: str = Form(""),
confidence: float = Form(1.0),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.add_capability(
repository_id,
ability_id,
name=name,
description=description,
inputs=split_csv(inputs),
outputs=split_csv(outputs),
confidence=confidence,
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/features")
def create_feature_from_form(
repository_id: int,
capability_id: int = Form(...),
name: str = Form(...),
type: str = Form(...),
location: str = Form(""),
confidence: float = Form(1.0),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.add_feature(
repository_id,
capability_id,
name=name,
type=type,
location=location,
confidence=confidence,
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/evidence")
def create_evidence_from_form(
repository_id: int,
capability_id: int = Form(...),
type: str = Form(...),
reference: str = Form(...),
strength: str = Form("medium"),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.add_evidence(
repository_id,
capability_id,
type=type,
reference=reference,
strength=strength,
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/abilities/{ability_id}/edit")
def edit_ability_from_form(
repository_id: int,
ability_id: int,
name: str = Form(...),
description: str = Form(""),
confidence: float = Form(1.0),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.update_ability(
repository_id,
ability_id,
name=name,
description=description,
confidence=confidence,
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/abilities/{ability_id}/delete")
def delete_ability_from_form(
repository_id: int,
ability_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.delete_ability(repository_id, ability_id)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/capabilities/{capability_id}/edit")
def edit_capability_from_form(
repository_id: int,
capability_id: int,
name: str = Form(...),
description: str = Form(""),
inputs: str = Form(""),
outputs: str = Form(""),
confidence: float = Form(1.0),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.update_capability(
repository_id,
capability_id,
name=name,
description=description,
inputs=split_csv(inputs),
outputs=split_csv(outputs),
confidence=confidence,
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/capabilities/{capability_id}/delete")
def delete_capability_from_form(
repository_id: int,
capability_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.delete_capability(repository_id, capability_id)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/features/{feature_id}/edit")
def edit_feature_from_form(
repository_id: int,
feature_id: int,
name: str = Form(...),
type: str = Form(...),
location: str = Form(""),
confidence: float = Form(1.0),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.update_feature(
repository_id,
feature_id,
name=name,
type=type,
location=location,
confidence=confidence,
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/features/{feature_id}/delete")
def delete_feature_from_form(
repository_id: int,
feature_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.delete_feature(repository_id, feature_id)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/evidence/{evidence_id}/edit")
def edit_evidence_from_form(
repository_id: int,
evidence_id: int,
type: str = Form(...),
reference: str = Form(...),
strength: str = Form("medium"),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.update_evidence(
repository_id,
evidence_id,
type=type,
reference=reference,
strength=strength,
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/evidence/{evidence_id}/delete")
def delete_evidence_from_form(
repository_id: int,
evidence_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.delete_evidence(repository_id, evidence_id)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/analysis-runs")
def create_analysis_run_from_form(
repository_id: int,
source_path: str = Form(""),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
summary = service.analyze_repository(
repository_id,
source_path=source_path or None,
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{summary.analysis_run.id}",
status_code=303,
)
@router.get("/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}")
def analysis_run_detail(
repository_id: int,
analysis_run_id: int,
service: RegistryService = Depends(get_service),
) -> HTMLResponse:
repository = service.get_repository(repository_id)
candidate_graph = service.candidate_graph(repository_id, analysis_run_id)
facts = service.list_observed_facts(repository_id, analysis_run_id)
chunks = service.list_content_chunks(repository_id, analysis_run_id)
decisions = service.list_review_decisions(repository_id, analysis_run_id)
fact_rows = "\n".join(
f"""
{escape(fact.kind)}
{escape(fact.name)}
{escape(fact.path)}
{escape(fact.value)}
"""
for fact in facts
)
body = f"""
{escape(repository.name)} · Run #{analysis_run_id}
{render_run_detail_compare_link(repository_id, analysis_run_id, service.list_analysis_runs(repository_id))}
Repository
Candidate Graph
{render_candidate_graph(asdict(candidate_graph), repository_id, analysis_run_id)}
Observed Facts
Kind Name Path Value {fact_rows or 'No observed facts.
Review Decisions
{render_review_decisions(decisions)}
Content Chunks
{render_content_chunks(chunks)}
"""
return page(f"{repository.name} Run {analysis_run_id}", body)
@router.get("/ui/repos/{repository_id}/analysis-runs/{base_analysis_run_id}/diff/{target_analysis_run_id}")
def analysis_run_diff_detail(
repository_id: int,
base_analysis_run_id: int,
target_analysis_run_id: int,
service: RegistryService = Depends(get_service),
) -> HTMLResponse:
try:
diff = service.diff_analysis_runs(
repository_id,
base_analysis_run_id,
target_analysis_run_id,
)
except NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
body = f"""
Comparing run #{base_analysis_run_id} to run #{target_analysis_run_id}.
Approved Registry Impact
{render_diff_section(asdict(diff.approved_entries))}
Candidate Claims
{render_diff_section(asdict(diff.candidates))}
Observed Facts
{render_diff_section(asdict(diff.facts))}
Content Chunks
{render_diff_section(asdict(diff.chunks))}
"""
return page(f"{diff.repository.name} Change Review", body)
@router.post("/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-graph/approve")
def approve_candidate_graph_from_form(
repository_id: int,
analysis_run_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.approve_candidate_graph(
repository_id,
analysis_run_id,
notes="Approved from web UI",
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post("/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}/changes/approve")
def approve_analysis_run_changes_from_form(
repository_id: int,
analysis_run_id: int,
notes: str = Form("Approved target run changes from web UI"),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.approve_analysis_run_changes(
repository_id,
analysis_run_id,
notes=notes,
)
return RedirectResponse(f"/ui/repos/{repository_id}", status_code=303)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-abilities/{candidate_ability_id}/reject"
)
def reject_candidate_ability_from_form(
repository_id: int,
analysis_run_id: int,
candidate_ability_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.reject_candidate_ability(
repository_id,
analysis_run_id,
candidate_ability_id,
notes="Rejected from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-capabilities/{candidate_capability_id}/reject"
)
def reject_candidate_capability_from_form(
repository_id: int,
analysis_run_id: int,
candidate_capability_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.reject_candidate_capability(
repository_id,
analysis_run_id,
candidate_capability_id,
notes="Rejected from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-features/{candidate_feature_id}/reject"
)
def reject_candidate_feature_from_form(
repository_id: int,
analysis_run_id: int,
candidate_feature_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.reject_candidate_feature(
repository_id,
analysis_run_id,
candidate_feature_id,
notes="Rejected from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-evidence/{candidate_evidence_id}/reject"
)
def reject_candidate_evidence_from_form(
repository_id: int,
analysis_run_id: int,
candidate_evidence_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.reject_candidate_evidence(
repository_id,
analysis_run_id,
candidate_evidence_id,
notes="Rejected from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-abilities/{candidate_ability_id}/edit"
)
def edit_candidate_ability_from_form(
repository_id: int,
analysis_run_id: int,
candidate_ability_id: int,
name: str = Form(...),
description: str = Form(""),
confidence: float = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.edit_candidate_ability(
repository_id,
analysis_run_id,
candidate_ability_id,
name=name,
description=description,
confidence=confidence,
notes="Edited from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-capabilities/{candidate_capability_id}/edit"
)
def edit_candidate_capability_from_form(
repository_id: int,
analysis_run_id: int,
candidate_capability_id: int,
name: str = Form(...),
description: str = Form(""),
confidence: float = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.edit_candidate_capability(
repository_id,
analysis_run_id,
candidate_capability_id,
name=name,
description=description,
confidence=confidence,
notes="Edited from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-capabilities/{candidate_capability_id}/relink"
)
def relink_candidate_capability_from_form(
repository_id: int,
analysis_run_id: int,
candidate_capability_id: int,
target_ability_id: int = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.relink_candidate_capability(
repository_id,
analysis_run_id,
candidate_capability_id,
target_ability_id=target_ability_id,
notes="Relinked from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-features/{candidate_feature_id}/relink"
)
def relink_candidate_feature_from_form(
repository_id: int,
analysis_run_id: int,
candidate_feature_id: int,
target_capability_id: int = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.relink_candidate_feature(
repository_id,
analysis_run_id,
candidate_feature_id,
target_capability_id=target_capability_id,
notes="Relinked from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-evidence/{candidate_evidence_id}/relink"
)
def relink_candidate_evidence_from_form(
repository_id: int,
analysis_run_id: int,
candidate_evidence_id: int,
target_capability_id: int = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.relink_candidate_evidence(
repository_id,
analysis_run_id,
candidate_evidence_id,
target_capability_id=target_capability_id,
notes="Relinked from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-abilities/{source_ability_id}/merge"
)
def merge_candidate_ability_from_form(
repository_id: int,
analysis_run_id: int,
source_ability_id: int,
target_ability_id: int = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.merge_candidate_ability(
repository_id,
analysis_run_id,
source_ability_id,
target_ability_id=target_ability_id,
notes="Merged from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-capabilities/{source_capability_id}/merge"
)
def merge_candidate_capability_from_form(
repository_id: int,
analysis_run_id: int,
source_capability_id: int,
target_capability_id: int = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.merge_candidate_capability(
repository_id,
analysis_run_id,
source_capability_id,
target_capability_id=target_capability_id,
notes="Merged from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-features/{source_feature_id}/merge"
)
def merge_candidate_feature_from_form(
repository_id: int,
analysis_run_id: int,
source_feature_id: int,
target_feature_id: int = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.merge_candidate_feature(
repository_id,
analysis_run_id,
source_feature_id,
target_feature_id=target_feature_id,
notes="Merged from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-evidence/{source_evidence_id}/merge"
)
def merge_candidate_evidence_from_form(
repository_id: int,
analysis_run_id: int,
source_evidence_id: int,
target_evidence_id: int = Form(...),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.merge_candidate_evidence(
repository_id,
analysis_run_id,
source_evidence_id,
target_evidence_id=target_evidence_id,
notes="Merged from web UI",
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
def render_run_compare_link(repository_id: int, run_id: int, runs: list) -> str:
base_run = next((run for run in runs if run.id < run_id), None)
if base_run is None:
return 'Baseline '
href = f"/ui/repos/{repository_id}/analysis-runs/{base_run.id}/diff/{run_id}"
return f'Compare to #{base_run.id} '
def render_run_detail_compare_link(
repository_id: int,
analysis_run_id: int,
runs: list,
) -> str:
base_run = next((run for run in runs if run.id < analysis_run_id), None)
if base_run is None:
return ""
href = f"/ui/repos/{repository_id}/analysis-runs/{base_run.id}/diff/{analysis_run_id}"
return f'Compare to #{base_run.id} '
def render_diff_section(section: dict) -> str:
groups = [
("added", "Added"),
("removed", "Removed"),
("changed", "Changed"),
("weakened", "Weakened"),
]
rendered = "".join(
render_diff_group(section.get(key, []), title)
for key, title in groups
)
return rendered or 'No differences.
'
def render_diff_group(items: list[dict], title: str) -> str:
if not items:
return ""
rows = "\n".join(
f"""
{escape(item['item_type'])} {escape(item['key'])}
{render_diff_payload(item.get('base'))}
{render_diff_payload(item.get('target'))}
"""
for item in items
)
return f"""
{escape(title)}
"""
def render_diff_payload(payload: dict | None) -> str:
if payload is None:
return 'None '
preferred = [
"name",
"description",
"confidence",
"strength",
"path",
"location",
"reference",
"value",
"text",
]
parts = []
for key in preferred:
if key not in payload or payload[key] in (None, "", []):
continue
value = str(payload[key])
if key == "text" and len(value) > 180:
value = value[:180] + "..."
parts.append(
f'{escape(key)} : {escape(value)}
'
)
return "".join(parts) or 'No display fields. '
def render_repository_checkbox_list(
service: RegistryService,
repositories: list,
) -> str:
if not repositories:
return 'No repositories registered.
'
rows = []
for repository in repositories:
ability_map = service.ability_map(repository.id)
approved_count = len(ability_map.abilities)
disabled = " disabled" if approved_count == 0 else ""
status = (
f"{approved_count} approved "
f"{'ability' if approved_count == 1 else 'abilities'}"
if approved_count
else "No approved profile"
)
rows.append(
f"""
{escape(status)}
"""
)
return f'{"".join(rows)}
'
def render_compared_repositories(repositories: list[dict]) -> str:
if not repositories:
return 'No repositories selected.
'
rows = "\n".join(
f"""
{escape(repository['name'])} {escape(repository['status'])} {escape(repository['branch'])}
"""
for repository in repositories
)
return f"""
"""
def render_compared_abilities(abilities: list[dict]) -> str:
if not abilities:
return 'No shared approved abilities.
'
rows = []
for ability in abilities:
repositories = "".join(
f"""
{escape(repository['repository_name'])}
{repository['confidence']:.2f} {escape(repository['confidence_label'])}
{render_compared_capability_pills(repository['capabilities'])}
"""
for repository in ability["repositories"]
)
rows.append(
f"""
{escape(ability['name'])}
"""
)
return f''
def render_compared_capability_pills(capabilities: list[dict]) -> str:
if not capabilities:
return ""
return " ".join(
f'{escape(capability["name"])} · {capability["evidence_count"]} evidence '
for capability in capabilities
)
def render_unique_capabilities(capabilities: list[dict]) -> str:
if not capabilities:
return 'No unique approved capabilities.
'
rows = "\n".join(
f"""
{escape(capability['repository_name'])} {escape(capability['ability_name'])}
{escape(capability['capability_name'])}
"""
for capability in capabilities
)
return f"""
Repository Ability Capability {rows}
"""
def render_gap_report(report: dict) -> str:
return f"""
Matched
{render_gap_matches(report["matched_capabilities"])}
Missing
{render_gap_list(report["missing_capabilities"], "No missing capabilities.")}
Weak Evidence
{render_weak_evidence(report["weakly_evidenced_capabilities"])}
Duplicates
{render_duplicate_capabilities(report["duplicate_capabilities"])}
"""
def render_gap_matches(matches: list[dict]) -> str:
if not matches:
return 'No desired capabilities matched.
'
rows = "\n".join(
f"""
{escape(match['capability'])} {escape(', '.join(match['repositories']))}
"""
for match in matches
)
return f""
def render_gap_list(values: list[str], empty: str) -> str:
if not values:
return f'{escape(empty)}
'
return "" + "".join(f"{escape(value)} " for value in values) + " "
def render_weak_evidence(items: list[dict]) -> str:
if not items:
return 'No weak evidence flags.
'
rows = "\n".join(
f"""
{escape(item['capability'])}
{escape(item['repository_name'])}
{escape(str(item['strongest_evidence'] or 'none'))}
{item['confidence']:.2f} {escape(item['confidence_label'])}
"""
for item in items
)
return f"""
Capability Repository Evidence Confidence {rows}
"""
def render_duplicate_capabilities(items: list[dict]) -> str:
if not items:
return 'No duplicate capabilities.
'
rows = "\n".join(
f"""
{escape(item['capability'])}
{escape(', '.join(item['repositories']))}
"""
for item in items
)
return f""
def split_capability_lines(value: str) -> list[str]:
normalized = value.replace(",", "\n")
return [line.strip() for line in normalized.splitlines() if line.strip()]
def render_candidate_graph(graph: dict, repository_id: int, analysis_run_id: int) -> str:
abilities = graph.get("abilities", [])
if not abilities:
return 'No candidates generated.
'
items = []
for ability in abilities:
capabilities = "".join(
render_candidate_capability(capability, repository_id, analysis_run_id)
for capability in ability["capabilities"]
)
items.append(
f"""
{escape(ability['name'])}
ID {ability['id']}
{escape(ability['status'])}
{ability['confidence']:.2f} {escape(ability['confidence_label'])}
{render_candidate_ability_actions(ability, repository_id, analysis_run_id)}
{escape(ability['description'])}
{render_candidate_edit_form('candidate-abilities', ability, repository_id, analysis_run_id)}
{render_candidate_merge_form('candidate-abilities', ability, repository_id, analysis_run_id, 'target_ability_id', 'Merge into ability ID')}
{render_sources(ability['source_refs'])}
"""
)
return f''
def render_repository_facts(languages: list[str], frameworks: list[str]) -> str:
if not languages and not frameworks:
return ""
language_pills = "".join(
f'Language: {escape(language)} '
for language in languages
)
framework_pills = "".join(
f'Framework: {escape(framework)} '
for framework in frameworks
)
return f'{language_pills}{framework_pills}
'
def split_csv(value: str) -> list[str]:
return [item.strip() for item in value.split(",") if item.strip()]
def render_review_decisions(decisions: list) -> str:
if not decisions:
return 'No review decisions yet.
'
rows = "\n".join(
f"""
{escape(decision.action)}
{escape(decision.created_at)}
{escape(decision.notes)}
"""
for decision in decisions
)
return f"""
Action Created Notes {rows}
"""
def render_content_chunks(chunks: list) -> str:
if not chunks:
return 'No content chunks extracted.
'
rows = "\n".join(
f"""
{escape(chunk.kind)} {escape(chunk.path)}:{chunk.start_line}-{chunk.end_line}
{escape(chunk.text[:500])}
"""
for chunk in chunks
)
return f"""
"""
def render_candidate_ability_actions(
ability: dict,
repository_id: int,
analysis_run_id: int,
) -> str:
if ability["status"] != "candidate":
return ""
action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/candidate-abilities/{ability['id']}/reject"
)
return f"""
"""
def render_candidate_edit_form(
collection: str,
candidate: dict,
repository_id: int,
analysis_run_id: int,
) -> str:
if candidate["status"] != "candidate":
return ""
action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/{collection}/{candidate['id']}/edit"
)
confidence = f"{candidate['confidence']:.2f}"
return f"""
"""
def render_candidate_capability(
capability: dict,
repository_id: int,
analysis_run_id: int,
) -> str:
features = "".join(
render_candidate_feature(feature, repository_id, analysis_run_id)
for feature in capability["features"]
)
evidence = "".join(
render_candidate_evidence(item, repository_id, analysis_run_id)
for item in capability["evidence"]
)
return f"""
{escape(capability['name'])}
ID {capability['id']}
{escape(capability['status'])}
{capability['confidence']:.2f} {escape(capability['confidence_label'])}
{render_candidate_reject_form('candidate-capabilities', capability, repository_id, analysis_run_id)}
{escape(capability['description'])}
{render_candidate_edit_form('candidate-capabilities', capability, repository_id, analysis_run_id)}
{render_candidate_relink_form('candidate-capabilities', capability, repository_id, analysis_run_id, 'target_ability_id', 'Target ability ID')}
{render_candidate_merge_form('candidate-capabilities', capability, repository_id, analysis_run_id, 'target_capability_id', 'Merge into capability ID')}
{render_sources(capability['source_refs'])}
Features
{features or 'No feature candidates. '}
Evidence
{evidence or 'No evidence candidates. '}
"""
def render_candidate_feature(
feature: dict,
repository_id: int,
analysis_run_id: int,
) -> str:
return f"""
{escape(feature["name"])}
ID {feature["id"]}
{escape(feature["status"])}
{escape(feature["type"])}
{escape(feature["location"])}
{render_candidate_reject_form('candidate-features', feature, repository_id, analysis_run_id)}
{render_candidate_relink_form('candidate-features', feature, repository_id, analysis_run_id, 'target_capability_id', 'Target capability ID')}
{render_candidate_merge_form('candidate-features', feature, repository_id, analysis_run_id, 'target_feature_id', 'Merge into feature ID')}
"""
def render_candidate_evidence(
evidence: dict,
repository_id: int,
analysis_run_id: int,
) -> str:
return f"""
{escape(evidence["type"])}
ID {evidence["id"]}
{escape(evidence["status"])}
{escape(evidence["strength"])}
{escape(evidence["reference"])}
{render_candidate_reject_form('candidate-evidence', evidence, repository_id, analysis_run_id)}
{render_candidate_relink_form('candidate-evidence', evidence, repository_id, analysis_run_id, 'target_capability_id', 'Target capability ID')}
{render_candidate_merge_form('candidate-evidence', evidence, repository_id, analysis_run_id, 'target_evidence_id', 'Merge into evidence ID')}
"""
def render_candidate_reject_form(
collection: str,
candidate: dict,
repository_id: int,
analysis_run_id: int,
) -> str:
if candidate["status"] != "candidate":
return ""
action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/{collection}/{candidate['id']}/reject"
)
return f"""
"""
def render_candidate_relink_form(
collection: str,
candidate: dict,
repository_id: int,
analysis_run_id: int,
field_name: str,
label: str,
) -> str:
if candidate["status"] != "candidate":
return ""
action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/{collection}/{candidate['id']}/relink"
)
return f"""
"""
def render_candidate_merge_form(
collection: str,
candidate: dict,
repository_id: int,
analysis_run_id: int,
field_name: str,
label: str,
) -> str:
if candidate["status"] != "candidate":
return ""
action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/{collection}/{candidate['id']}/merge"
)
return f"""
"""
def render_ability_map(ability_map: dict, repository_id: int) -> str:
abilities = ability_map.get("abilities", [])
if not abilities:
return 'No approved entries yet.
'
items = []
for ability in abilities:
capabilities = []
for capability in ability["capabilities"]:
features = "".join(
render_approved_feature(feature, repository_id)
for feature in capability["features"]
)
evidence = "".join(
render_approved_evidence(item, repository_id)
for item in capability["evidence"]
)
capabilities.append(
f"""
{escape(capability['name'])}
ID {capability['id']}
{capability['confidence']:.2f} {escape(capability['confidence_label'])}
{escape(capability['description'])}
{render_approved_capability_forms(capability, repository_id)}
"""
)
items.append(
f"""
{escape(ability['name'])}
ID {ability['id']}
{ability['confidence']:.2f} {escape(ability['confidence_label'])}
{escape(ability['description'])}
{render_approved_ability_forms(ability, repository_id)}
"""
)
return f''
def render_approved_ability_forms(ability: dict, repository_id: int) -> str:
return f"""
"""
def render_approved_capability_forms(capability: dict, repository_id: int) -> str:
inputs = ", ".join(capability["inputs"])
outputs = ", ".join(capability["outputs"])
return f"""
"""
def render_approved_feature(feature: dict, repository_id: int) -> str:
return f"""
{escape(feature["name"])}
{escape(feature["type"])}
{feature["confidence"]:.2f} {escape(feature["confidence_label"])}
{escape(feature["location"])}
{render_sources(feature.get("source_refs", []))}
"""
def render_approved_evidence(evidence: dict, repository_id: int) -> str:
return f"""
{escape(evidence["type"])}
{escape(evidence["strength"])}
{escape(evidence["reference"])}
{render_sources(evidence.get("source_refs", []))}
"""
def search_result_href(result: dict) -> str:
href = f"/ui/repos/{result['repository_id']}"
if result.get("capability_id"):
return f"{href}#capability-{result['capability_id']}"
if result.get("ability_id"):
return f"{href}#ability-{result['ability_id']}"
return href
def render_sources(source_refs: list[dict]) -> str:
if not source_refs:
return ""
sources = ", ".join(
f'{escape(ref["kind"])}:{escape(source_ref_label(ref))} '
for ref in source_refs[:5]
)
if len(source_refs) > 5:
sources += f' +{len(source_refs) - 5} more '
return f"{sources}
"
def source_ref_label(ref: dict) -> str:
label = ref["path"] or ref["name"]
if ref.get("line"):
label = f"{label}:{ref['line']}"
return label
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'{" · ".join(details)}
'