from __future__ import annotations
from dataclasses import asdict
from html import escape
from urllib.parse import quote_plus
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"),
access_username: str = Form(""),
access_password: str = Form(""),
explore_after_registration: str | None = Form(None),
use_llm_assistance: str | None = Form(None),
trusted_auto_approve: str | None = Form(None),
service: RegistryService = Depends(get_service),
):
try:
repository = service.register_repository(
url=url,
branch=branch or "main",
access_username=access_username or None,
access_password=access_password or None,
)
except (RuntimeError, ValueError) as exc:
return render_repository_index(
service,
error_message=str(exc),
status_code=400,
)
if explore_after_registration:
summary = service.analyze_repository(
repository.id,
use_llm_assistance=bool(use_llm_assistance),
trusted_auto_approve=bool(trusted_auto_approve),
access_username=access_username or None,
access_password=access_password or None,
)
return RedirectResponse(
f"/ui/repos/{repository.id}/analysis-runs/{summary.analysis_run.id}",
status_code=303,
)
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)
latest_candidate = latest_completed_candidate_graph(
service,
repository_id,
runs,
)
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 Characteristics
{render_graph_counts(
asdict(ability_map),
facts_count=None,
base_href=(
f"/ui/repos/{repository_id}/elements?scope=all"
f"&entry_filter=approved"
),
)}
{render_approved_registry_actions(repository_id, asdict(ability_map))}
Latest Candidate Graph
{render_latest_candidate_counts(repository_id, latest_candidate, service)}
Approved Characteristic Tree
{render_ability_map(asdict(ability_map), repository_id)}
Manual Characteristic Tuning
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}/scope/edit")
def edit_scope_from_form(
repository_id: int,
name: str = Form(...),
description: str = Form(""),
confidence: float = Form(1.0),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.update_scope(
repository_id,
name=name,
description=description,
confidence=confidence,
)
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"),
target_kind: str = Form("capability"),
target_id: int | None = Form(default=None),
reference_kind: str = Form("source"),
reference_id: int | None = Form(default=None),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.add_evidence(
repository_id,
capability_id,
type=type,
reference=reference,
strength=strength,
target_kind=target_kind,
target_id=target_id,
reference_kind=reference_kind,
reference_id=reference_id,
)
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"),
target_kind: str = Form("capability"),
target_id: int | None = Form(default=None),
reference_kind: str = Form("source"),
reference_id: int | None = Form(default=None),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.update_evidence(
repository_id,
evidence_id,
type=type,
reference=reference,
strength=strength,
target_kind=target_kind,
target_id=target_id,
reference_kind=reference_kind,
reference_id=reference_id,
)
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(""),
use_cached_checkout: str | None = Form(None),
use_llm_assistance: str | None = Form(None),
trusted_auto_approve: str | None = Form(None),
access_username: str = Form(""),
access_password: str = Form(""),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
summary = service.analyze_repository(
repository_id,
source_path=source_path or None,
use_cached_checkout=bool(use_cached_checkout),
use_llm_assistance=bool(use_llm_assistance),
trusted_auto_approve=bool(trusted_auto_approve),
access_username=access_username or None,
access_password=access_password 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)
expectation_gaps = service.list_expectation_gaps(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_graph_counts(
asdict(candidate_graph),
facts_count=len(facts),
base_href=(
f"/ui/repos/{repository_id}/elements?scope=all"
f"&entry_filter=candidate"
f"&analysis_run_id={analysis_run_id}"
),
facts_href=(
f"/ui/repos/{repository_id}/elements?scope=facts"
f"&analysis_run_id={analysis_run_id}&type=facts"
),
)}
{render_candidate_graph(asdict(candidate_graph), repository_id, analysis_run_id)}
Observed Facts
{render_count_pills(
hrefs={
"facts": (
f"/ui/repos/{repository_id}/elements?scope=facts"
f"&analysis_run_id={analysis_run_id}&type=facts"
)
},
facts=len(facts),
)}
Kind Name Path Value {fact_rows or 'No observed facts.
Review Decisions
{render_review_decisions(decisions)}
Content Chunks
{render_content_chunks(chunks)}
Expectation Gaps
{render_expectation_gaps(expectation_gaps)}
"""
return page(f"{repository.name} Run {analysis_run_id}", body)
@router.post("/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}/expectation-gaps")
def create_expectation_gap_from_form(
repository_id: int,
analysis_run_id: int,
expected_type: str = Form(...),
expected_name: str = Form(...),
source: str = Form("human"),
notes: str = Form(""),
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.record_expectation_gap(
repository_id,
analysis_run_id=analysis_run_id,
expected_type=expected_type,
expected_name=expected_name,
source=source,
notes=notes,
)
return RedirectResponse(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}",
status_code=303,
)
@router.get("/ui/repos/{repository_id}/elements")
def repository_element_listing(
repository_id: int,
scope: str = Query("all"),
type: str = Query("abilities"),
q: str = Query(""),
class_filter: str = Query(""),
entry_filter: str = Query(""),
candidate_status_filter: str = Query("active"),
analysis_run_id: int | None = Query(default=None),
service: RegistryService = Depends(get_service),
) -> HTMLResponse:
repository = service.get_repository(repository_id)
if scope in {"approved", "candidate"} and not entry_filter:
entry_filter = scope
title = element_listing_title(repository.name, scope, type)
if scope in {"all", "approved", "candidate"}:
elements = graph_element_rows(
asdict(service.ability_map(repository_id)),
type,
entry_state="approved",
)
candidate_elements: list[dict] = []
if analysis_run_id is None:
runs = service.list_analysis_runs(repository_id)
latest = latest_completed_candidate_graph(service, repository_id, runs)
if latest is not None:
analysis_run_id, candidate_graph = latest
candidate_elements = graph_element_rows(
asdict(candidate_graph),
type,
entry_state="candidate",
)
else:
candidate_graph = service.candidate_graph(repository_id, analysis_run_id)
candidate_elements = graph_element_rows(
asdict(candidate_graph),
type,
entry_state="candidate",
)
elements.extend(candidate_elements)
elif scope == "facts":
facts = service.list_observed_facts(repository_id, analysis_run_id)
elements = fact_element_rows(facts)
else:
elements = []
entry_scoped_elements = filter_element_rows(
elements,
"",
"",
entry_filter,
candidate_status_filter,
)
filtered = filter_element_rows(
entry_scoped_elements,
q,
class_filter,
candidate_status_filter="all",
)
rows = render_element_rows(
filtered,
repository_id=repository_id,
analysis_run_id=analysis_run_id,
)
filter_action = f"/ui/repos/{repository_id}/elements"
listing_scope = "facts" if scope == "facts" else "all"
body = f"""
Entry Class Name Parent Source Actions {rows}
"""
return page(title, 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-abilities/{candidate_ability_id}/accept"
)
def accept_candidate_ability_from_form(
repository_id: int,
analysis_run_id: int,
candidate_ability_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.accept_candidate_ability(
repository_id,
analysis_run_id,
candidate_ability_id,
notes="Accepted from web UI",
)
return RedirectResponse(f"/ui/repos/{repository_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-capabilities/{candidate_capability_id}/accept"
)
def accept_candidate_capability_from_form(
repository_id: int,
analysis_run_id: int,
candidate_capability_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.accept_candidate_capability(
repository_id,
analysis_run_id,
candidate_capability_id,
notes="Accepted from web UI",
)
return RedirectResponse(f"/ui/repos/{repository_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-features/{candidate_feature_id}/accept"
)
def accept_candidate_feature_from_form(
repository_id: int,
analysis_run_id: int,
candidate_feature_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.accept_candidate_feature(
repository_id,
analysis_run_id,
candidate_feature_id,
notes="Accepted from web UI",
)
return RedirectResponse(f"/ui/repos/{repository_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 latest_completed_candidate_graph(
service: RegistryService,
repository_id: int,
runs: list,
):
for run in sorted(runs, key=lambda item: item.id, reverse=True):
if run.status != "completed":
continue
try:
return run.id, service.candidate_graph(repository_id, run.id)
except NotFoundError:
continue
return None
def render_latest_candidate_counts(
repository_id: int,
latest_candidate,
service: RegistryService,
) -> str:
if latest_candidate is None:
return 'No completed candidate graph yet.
'
analysis_run_id, candidate_graph = latest_candidate
facts_count = len(service.list_observed_facts(repository_id, analysis_run_id))
return render_graph_counts(
active_candidate_graph(asdict(candidate_graph)),
facts_count=facts_count,
label_prefix="candidate",
base_href=(
f"/ui/repos/{repository_id}/elements?scope=all"
f"&entry_filter=candidate"
f"&analysis_run_id={analysis_run_id}"
),
facts_href=(
f"/ui/repos/{repository_id}/elements?scope=facts"
f"&analysis_run_id={analysis_run_id}&type=facts"
),
)
def active_candidate_graph(graph: dict) -> dict:
active_abilities = []
for ability in graph.get("abilities", []):
active_capabilities = []
for capability in ability.get("capabilities", []):
active_features = [
feature
for feature in capability.get("features", [])
if feature.get("status", "candidate") != "rejected"
]
active_evidence = [
evidence
for evidence in capability.get("evidence", [])
if evidence.get("status", "candidate") != "rejected"
]
capability_active = (
capability.get("status", "candidate") != "rejected"
and (active_features or active_evidence)
)
if capability_active:
active_capability = {**capability}
active_capability["features"] = active_features
active_capability["evidence"] = active_evidence
active_capabilities.append(active_capability)
if ability.get("status", "candidate") != "rejected" and active_capabilities:
active_ability = {**ability}
active_ability["capabilities"] = active_capabilities
active_abilities.append(active_ability)
active_graph = {**graph}
active_graph["abilities"] = active_abilities
return active_graph
def render_graph_counts(
graph: dict,
facts_count: int | None = None,
label_prefix: str = "",
base_href: str | None = None,
facts_href: str | None = None,
) -> str:
abilities = graph.get("abilities", [])
capabilities = [
capability
for ability in abilities
for capability in ability.get("capabilities", [])
]
features = [
feature
for capability in capabilities
for feature in capability.get("features", [])
]
supports = [
evidence
for capability in capabilities
for evidence in capability.get("evidence", [])
]
counts: dict[str, int] = {
"scopes": 1 if graph.get("scope") else 0,
"abilities": len(abilities),
"capabilities": len(capabilities),
"features": len(features),
"supports": len(supports),
}
if facts_count is not None:
counts["facts"] = facts_count
hrefs = (
{name: f"{base_href}&type={name}" for name in counts if name != "facts"}
if base_href
else {}
)
if facts_href:
hrefs["facts"] = facts_href
return render_count_pills(label_prefix=label_prefix, hrefs=hrefs, **counts)
def render_count_pills(
label_prefix: str = "",
hrefs: dict[str, str] | None = None,
**counts: int,
) -> str:
labels = {
"scopes": "scope",
"abilities": "abilities",
"capabilities": "capabilities",
"features": "features",
"supports": "supports",
"facts": "facts",
}
prefix = f"{label_prefix} " if label_prefix else ""
hrefs = hrefs or {}
items = []
for name, count in counts.items():
label = f"{count} {prefix}{labels[name]}"
if name in hrefs:
items.append(f'{label} ')
else:
items.append(f'{label} ')
return "".join(items)
def render_approved_registry_actions(repository_id: int, ability_map: dict) -> str:
abilities = ability_map.get("abilities", [])
if not abilities:
return ""
search_query = abilities[0]["name"]
return f"""
"""
def graph_element_rows(
graph: dict,
item_type: str,
*,
entry_state: str = "",
) -> list[dict]:
rows: list[dict] = []
scope = graph.get("scope")
if item_type in {"scopes", "scope"} and scope and entry_state == "approved":
rows.append(
element_row(
"scope",
scope["name"],
"",
[],
item_id=scope.get("id"),
item_kind="scope",
description=scope.get("description", ""),
confidence=scope.get("confidence", 1.0),
entry_state=entry_state,
)
)
for ability in graph.get("abilities", []):
if item_type == "abilities":
rows.append(
element_row(
ability.get("primary_class", "ability"),
ability["name"],
"",
ability.get("source_refs", []),
item_id=ability.get("id"),
item_kind="abilities",
description=ability.get("description", ""),
confidence=ability.get("confidence", 1.0),
status=ability.get("status", ""),
entry_state=entry_state,
)
)
for capability in ability.get("capabilities", []):
if item_type == "capabilities":
rows.append(
element_row(
capability.get("primary_class", "capability"),
capability["name"],
ability["name"],
capability.get("source_refs", []),
item_id=capability.get("id"),
item_kind="capabilities",
description=capability.get("description", ""),
confidence=capability.get("confidence", 1.0),
inputs=capability.get("inputs", []),
outputs=capability.get("outputs", []),
status=capability.get("status", ""),
entry_state=entry_state,
)
)
for feature in capability.get("features", []):
if item_type == "features":
rows.append(
element_row(
feature.get("type", "feature"),
feature["name"],
capability["name"],
feature.get("source_refs", []),
item_id=feature.get("id"),
item_kind="features",
confidence=feature.get("confidence", 1.0),
location=feature.get("location", ""),
status=feature.get("status", ""),
entry_state=entry_state,
)
)
for evidence in capability.get("evidence", []):
if item_type in {"supports", "evidence"}:
rows.append(
element_row(
evidence.get("strength", "medium"),
f"{evidence.get('type', 'support')}: {evidence.get('reference', '')}",
capability["name"],
evidence.get("source_refs", []),
item_id=evidence.get("id"),
item_kind="evidence",
support_type=evidence.get("type", ""),
reference=evidence.get("reference", ""),
strength=evidence.get("strength", "medium"),
status=evidence.get("status", ""),
target_kind=evidence.get("target_kind", "capability"),
target_id=evidence.get("target_id"),
reference_kind=evidence.get("reference_kind", "source"),
reference_id=evidence.get("reference_id"),
entry_state=entry_state,
)
)
return rows
def element_listing_title(repository_name: str, scope: str, item_type: str) -> str:
type_labels = {
"abilities": "Abilities",
"capabilities": "Capabilities",
"features": "Features",
"scopes": "Scopes",
"supports": "Supports",
"facts": "Facts",
}
if scope == "facts":
return f"{repository_name} · Observed Facts"
scope_labels = {
"all": "Registry",
"approved": "Approved",
"candidate": "Candidate",
}
scope_label = scope_labels.get(scope, scope.title())
type_label = type_labels.get(item_type, item_type.title())
return f"{repository_name} · {scope_label} {type_label}"
def fact_element_rows(facts: list) -> list[dict]:
return [
element_row(
fact.kind,
fact.name,
fact.path,
[
{
"path": fact.path,
"kind": fact.kind,
"name": fact.name,
"line": fact.metadata.get("line"),
}
],
)
for fact in facts
]
def element_row(
primary_class: str,
name: str,
parent: str,
source_refs: list[dict],
**metadata,
) -> dict:
return {
"primary_class": primary_class,
"name": name,
"parent": parent,
"source_refs": source_refs,
**metadata,
}
def filter_element_rows(
rows: list[dict],
query: str,
class_filter: str,
entry_filter: str = "",
candidate_status_filter: str = "",
) -> list[dict]:
query = query.strip().lower()
class_filter = class_filter.strip().lower()
entry_filter = entry_filter.strip().lower()
candidate_status_filter = candidate_status_filter.strip().lower()
filtered = []
for row in rows:
if entry_filter and row.get("entry_state", "").lower() != entry_filter:
continue
if not candidate_status_matches(row, candidate_status_filter):
continue
row_class = str(row["primary_class"]).lower()
if class_filter and class_filter not in row_class:
continue
haystack = " ".join(
[
str(row["primary_class"]),
str(row["name"]),
str(row["parent"]),
str(row.get("entry_state", "")),
str(row.get("status", "")),
str(row.get("support_type", "")),
str(row.get("reference", "")),
str(row.get("strength", "")),
str(row.get("target_kind", "")),
str(row.get("target_id", "")),
str(row.get("reference_kind", "")),
str(row.get("reference_id", "")),
source_refs_text(row["source_refs"]),
]
).lower()
if query and query not in haystack:
continue
filtered.append(row)
return filtered
def candidate_status_matches(row: dict, candidate_status_filter: str) -> bool:
if row.get("entry_state") != "candidate":
return True
status = str(row.get("status") or "candidate").lower()
if candidate_status_filter == "all":
return True
if candidate_status_filter in {"", "active"}:
return status != "rejected"
return status == candidate_status_filter
def render_element_rows(
rows: list[dict],
repository_id: int,
analysis_run_id: int | None,
) -> str:
if not rows:
return 'No matching elements.
{render_entry_badge(row)}
{escape(str(row["primary_class"]))} {escape(str(row["name"]))}
{escape(str(row["parent"]))}
{render_element_source_detail(row)}
{render_element_actions(row, repository_id, analysis_run_id)}
"""
def render_element_source_detail(row: dict) -> str:
if row.get("item_kind") == "evidence":
target = escape(str(row.get("target_kind") or "capability"))
target_id = row.get("target_id")
reference_kind = escape(str(row.get("reference_kind") or "source"))
reference_id = row.get("reference_id")
return (
f'supports {target}{f" #{target_id}" if target_id else ""} '
f' references {reference_kind}{f" #{reference_id}" if reference_id else ""}
'
f'{render_sources(row["source_refs"])}'
)
return render_sources(row["source_refs"])
def render_element_actions(
row: dict,
repository_id: int,
analysis_run_id: int | None,
) -> str:
item_id = row.get("item_id")
item_kind = row.get("item_kind")
if not item_id or not item_kind:
return ""
if row.get("entry_state") == "approved":
return render_approved_element_actions(row, repository_id)
if (
row.get("entry_state") == "candidate"
and row.get("item_kind") == "evidence"
and analysis_run_id is not None
and row.get("status", "candidate") == "candidate"
):
return render_candidate_support_element_actions(
row,
repository_id,
analysis_run_id,
)
if (
row.get("entry_state") == "candidate"
and analysis_run_id is not None
and row.get("status", "candidate") in {"candidate", "rejected"}
):
return render_candidate_element_actions(row, repository_id, analysis_run_id)
if row.get("entry_state") == "candidate" and row.get("status") == "approved":
return 'Accepted '
return ""
def render_entry_badge(row: dict) -> str:
entry_state = row.get("entry_state")
if not entry_state:
return 'fact '
status = row.get("status")
label = str(entry_state)
if entry_state == "candidate" and status and status != "candidate":
label = f"candidate: {status}"
return f'{escape(label)} '
def render_entry_filter(entry_filter: str) -> str:
options = [
("", "Approved and candidate"),
("approved", "Approved only"),
("candidate", "Candidate only"),
]
rendered_options = "".join(
f'{escape(label)} '
for value, label in options
)
return f"""
Entry
{rendered_options}
"""
def render_approved_element_actions(row: dict, repository_id: int) -> str:
item_id = row["item_id"]
item_kind = row["item_kind"]
if item_kind == "scope":
return f"""
"""
edit_action = f"/ui/repos/{repository_id}/{item_kind}/{item_id}/edit"
delete_action = f"/ui/repos/{repository_id}/{item_kind}/{item_id}/delete"
hidden_fields = render_element_hidden_fields(row)
edit_fields = render_element_edit_fields(row)
return f"""
"""
def render_candidate_support_element_actions(
row: dict,
repository_id: int,
analysis_run_id: int,
) -> str:
item_id = row["item_id"]
reject_action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/candidate-evidence/{item_id}/reject"
)
relink_action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/candidate-evidence/{item_id}/relink"
)
merge_action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/candidate-evidence/{item_id}/merge"
)
return f"""
"""
def render_candidate_element_actions(
row: dict,
repository_id: int,
analysis_run_id: int,
) -> str:
item_id = row["item_id"]
item_kind = row["item_kind"]
collection = f"candidate-{item_kind}"
reject_action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/{collection}/{item_id}/reject"
)
accept_action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/{collection}/{item_id}/accept"
)
status = row.get("status", "candidate")
edit = ""
if item_kind in {"abilities", "capabilities"}:
edit_action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/{collection}/{item_id}/edit"
)
edit = f"""
"""
remove = (
f"""
"""
if status == "candidate"
else ""
)
return f"""
{edit}
{remove}
"""
def render_candidate_status_filter(candidate_status_filter: str) -> str:
options = [
("active", "Hide rejected"),
("candidate", "Pending only"),
("approved", "Accepted only"),
("rejected", "Rejected only"),
("all", "Include rejected"),
]
rendered_options = "".join(
f'{escape(label)} '
for value, label in options
)
return f"""
Candidate status
{rendered_options}
"""
def render_element_edit_fields(row: dict) -> str:
item_kind = row["item_kind"]
name = escape(str(row["name"]))
if item_kind == "scope":
return f"""
Name
Description
Confidence
"""
if item_kind == "features":
feature_type = escape(str(row["primary_class"]))
location = escape(str(row.get("location", "")))
return f"""
Name
Type
Location
"""
if item_kind == "evidence":
return f"""
Supported characteristic kind
Supported characteristic ID
Support type
Reference
Reference kind
Reference ID
Strength
"""
return f'Name '
def render_element_hidden_fields(row: dict) -> str:
item_kind = row["item_kind"]
confidence = float(row.get("confidence", 1.0))
fields = [f'Name '
def render_candidate_hidden_fields(row: dict) -> str:
return (
f'{options} '
def render_optional_hidden(name: str, value: int | None) -> str:
if value is None:
return ""
return f'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_expectation_gaps(gaps: list) -> str:
if not gaps:
return 'No expectation gaps recorded for this run.
'
rows = "\n".join(
f"""
{escape(gap.expected_type)} {escape(gap.expected_name)}
{escape(gap.source)}
{escape(gap.notes)}
{escape(gap.status)}
"""
for gap in gaps
)
return f"""
Type Name Source Notes Status {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", [])
scope = ability_map["scope"]
scope_description = scope.get("description") or (
f"Scope root for the approved characteristics of {scope['name']}."
)
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)}
Features
{features or 'No approved features. '}
Evidence supporting this capability
{evidence or 'No support evidence yet. '}
"""
)
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_scope_form(scope: dict, repository_id: int) -> str:
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:
target_kind = escape(str(evidence.get("target_kind") or "capability"))
target_id = evidence.get("target_id")
reference_kind = escape(str(evidence.get("reference_kind") or "source"))
reference_id = evidence.get("reference_id")
return f"""
{escape(evidence["type"])}
{escape(evidence["strength"])}
supports {target_kind}{f' #{target_id}' if target_id else ''}
references {reference_kind}{f' #{reference_id}' if reference_id else ''}
{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)}
'