coulomb/repo-scoping
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Candidate support/evidence acceptance

Browse Source
This commit is contained in:
Bernd Worsch
2026-04-29 17:52:42 +02:00
parent 0bb0c61f75
commit 466fd86d6d
6 changed files with 200 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
src/repo_registry/core/service.py
View File

@@ -379,6 +379,13 @@ class RegistryService:
type=evidence.type,
reference=evidence.reference,
strength=evidence.strength,
target_kind=evidence.target_kind,
target_id=self._approved_evidence_target_id(
evidence,
approved_capability_id,
),
reference_kind=evidence.reference_kind,
reference_id=evidence.reference_id,
source_refs=evidence.source_refs,
)
@@ -520,6 +527,58 @@ class RegistryService:
)
return self.store.get_ability_map(repository_id)
def accept_candidate_evidence(
self,
repository_id: int,
analysis_run_id: int,
candidate_evidence_id: int,
*,
notes: str = "",
) -> RepositoryAbilityMap:
graph = self.store.get_candidate_graph(repository_id, analysis_run_id)
parent_ability, parent_capability, evidence = (
self._candidate_evidence_with_parent(graph, candidate_evidence_id)
)
if evidence.status != "candidate":
raise ValueError(
f"candidate evidence {candidate_evidence_id} is not pending"
)
approved_ability_id = self._ensure_approved_ability(repository_id, parent_ability)
approved_capability_id = self._ensure_approved_capability(
repository_id,
approved_ability_id,
parent_ability.name,
parent_capability,
)
self.store.create_evidence(
repository_id,
approved_capability_id,
type=evidence.type,
reference=evidence.reference,
strength=evidence.strength,
target_kind=evidence.target_kind,
target_id=self._approved_evidence_target_id(
evidence,
approved_capability_id,
),
reference_kind=evidence.reference_kind,
reference_id=evidence.reference_id,
source_refs=evidence.source_refs,
)
self.store.mark_candidate_evidence_status(
repository_id,
analysis_run_id,
candidate_evidence_id,
"approved",
)
self._record_candidate_acceptance(
repository_id,
analysis_run_id,
"accept_candidate_evidence",
notes or f"Accepted candidate support: {evidence.reference}",
)
return self.store.get_ability_map(repository_id)
def diff_analysis_runs(
self,
repository_id: int,
@@ -618,6 +677,13 @@ class RegistryService:
type=evidence.type,
reference=evidence.reference,
strength=evidence.strength,
target_kind=evidence.target_kind,
target_id=self._approved_evidence_target_id(
evidence,
approved_capability_id,
),
reference_kind=evidence.reference_kind,
reference_id=evidence.reference_id,
source_refs=evidence.source_refs,
)
return approved_capability_id
@@ -673,6 +739,15 @@ class RegistryService:
return ability, capability
raise ValueError(f"candidate capability {candidate_capability_id} was not found")
def _approved_evidence_target_id(
self,
evidence: CandidateEvidence,
approved_capability_id: int,
) -> int | None:
if evidence.target_kind == "capability":
return approved_capability_id
return evidence.target_id
def _candidate_feature_with_parent(
self,
graph: CandidateGraph,
@@ -685,6 +760,18 @@ class RegistryService:
return ability, capability, feature
raise ValueError(f"candidate feature {candidate_feature_id} was not found")
def _candidate_evidence_with_parent(
self,
graph: CandidateGraph,
candidate_evidence_id: int,
) -> tuple[CandidateAbility, CandidateCapability, CandidateEvidence]:
for ability in graph.abilities:
for capability in ability.capabilities:
for evidence in capability.evidence:
if evidence.id == candidate_evidence_id:
return ability, capability, evidence
raise ValueError(f"candidate evidence {candidate_evidence_id} was not found")
def _record_candidate_acceptance(
self,
repository_id: int,

23
src/repo_registry/storage/sqlite.py
View File

@@ -677,6 +677,29 @@ class RegistryStore:
f"{repository_id} analysis run {analysis_run_id}"
)
def mark_candidate_evidence_status(
self,
repository_id: int,
analysis_run_id: int,
candidate_evidence_id: int,
status: str,
) -> None:
with self.connect() as connection:
cursor = connection.execute(
"""
UPDATE candidate_evidence
SET status = ?
WHERE id = ? AND repository_id = ? AND analysis_run_id = ?
""",
(status, candidate_evidence_id, repository_id, analysis_run_id),
)
if cursor.rowcount == 0:
raise NotFoundError(
"candidate evidence "
f"{candidate_evidence_id} was not found for repository "
f"{repository_id} analysis run {analysis_run_id}"
)
def _mark_candidate_children_status(
self,
connection: sqlite3.Connection,

26
src/repo_registry/web_ui/views.py
View File

@@ -1367,6 +1367,25 @@ def reject_candidate_evidence_from_form(
)
@router.post(
"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-evidence/{candidate_evidence_id}/accept"
)
def accept_candidate_evidence_from_form(
repository_id: int,
analysis_run_id: int,
candidate_evidence_id: int,
service: RegistryService = Depends(get_service),
) -> RedirectResponse:
service.accept_candidate_evidence(
repository_id,
analysis_run_id,
candidate_evidence_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-abilities/{candidate_ability_id}/edit"
@@ -2376,6 +2395,10 @@ def render_candidate_support_element_actions(
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/candidate-evidence/{item_id}/reject"
)
accept_action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/candidate-evidence/{item_id}/accept"
)
relink_action = (
f"/ui/repos/{repository_id}/analysis-runs/{analysis_run_id}"
f"/candidate-evidence/{item_id}/relink"
@@ -2385,6 +2408,9 @@ def render_candidate_support_element_actions(
f"/candidate-evidence/{item_id}/merge"
)
return f"""
<form method="post" action="{accept_action}">
<button type="submit">Accept</button>
</form>
<form method="post" action="{reject_action}">
<button class="secondary" type="submit">Remove</button>
</form>