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

relinking workflow

Browse Source
This commit is contained in:
Bernd Worsch
2026-04-25 23:44:25 +02:00
parent 71beb0d458
commit 1d6d103bc2
6 changed files with 558 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -322,6 +322,78 @@ class RegistryService:
self.store.update_repository_status(repository_id, "reviewing")
return self.store.get_candidate_graph(repository_id, analysis_run_id)
def relink_candidate_capability(
self,
repository_id: int,
analysis_run_id: int,
candidate_capability_id: int,
*,
target_ability_id: int,
notes: str = "",
) -> CandidateGraph:
self.store.relink_candidate_capability(
repository_id,
analysis_run_id,
candidate_capability_id,
target_ability_id,
)
self.store.create_review_decision(
repository_id,
analysis_run_id,
action="relink_candidate_capability",
notes=notes,
)
self.store.update_repository_status(repository_id, "reviewing")
return self.store.get_candidate_graph(repository_id, analysis_run_id)
def relink_candidate_feature(
self,
repository_id: int,
analysis_run_id: int,
candidate_feature_id: int,
*,
target_capability_id: int,
notes: str = "",
) -> CandidateGraph:
self.store.relink_candidate_feature(
repository_id,
analysis_run_id,
candidate_feature_id,
target_capability_id,
)
self.store.create_review_decision(
repository_id,
analysis_run_id,
action="relink_candidate_feature",
notes=notes,
)
self.store.update_repository_status(repository_id, "reviewing")
return self.store.get_candidate_graph(repository_id, analysis_run_id)
def relink_candidate_evidence(
self,
repository_id: int,
analysis_run_id: int,
candidate_evidence_id: int,
*,
target_capability_id: int,
notes: str = "",
) -> CandidateGraph:
self.store.relink_candidate_evidence(
repository_id,
analysis_run_id,
candidate_evidence_id,
target_capability_id,
)
self.store.create_review_decision(
repository_id,
analysis_run_id,
action="relink_candidate_evidence",
notes=notes,
)
self.store.update_repository_status(repository_id, "reviewing")
return self.store.get_candidate_graph(repository_id, analysis_run_id)
def add_ability(
self,
repository_id: int,

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

@@ -576,6 +576,133 @@ class RegistryStore:
f"{repository_id} analysis run {analysis_run_id}"
)
def relink_candidate_capability(
self,
repository_id: int,
analysis_run_id: int,
candidate_capability_id: int,
target_ability_id: int,
) -> None:
self._ensure_candidate_row(
table="candidate_abilities",
label="target candidate ability",
repository_id=repository_id,
analysis_run_id=analysis_run_id,
candidate_id=target_ability_id,
)
with self.connect() as connection:
cursor = connection.execute(
"""
UPDATE candidate_capabilities
SET ability_id = ?
WHERE id = ? AND repository_id = ? AND analysis_run_id = ?
""",
(
target_ability_id,
candidate_capability_id,
repository_id,
analysis_run_id,
),
)
if cursor.rowcount == 0:
raise NotFoundError(
"candidate capability "
f"{candidate_capability_id} was not found for repository "
f"{repository_id} analysis run {analysis_run_id}"
)
def relink_candidate_feature(
self,
repository_id: int,
analysis_run_id: int,
candidate_feature_id: int,
target_capability_id: int,
) -> None:
self._relink_candidate_leaf(
table="candidate_features",
label="candidate feature",
repository_id=repository_id,
analysis_run_id=analysis_run_id,
candidate_id=candidate_feature_id,
target_capability_id=target_capability_id,
)
def relink_candidate_evidence(
self,
repository_id: int,
analysis_run_id: int,
candidate_evidence_id: int,
target_capability_id: int,
) -> None:
self._relink_candidate_leaf(
table="candidate_evidence",
label="candidate evidence",
repository_id=repository_id,
analysis_run_id=analysis_run_id,
candidate_id=candidate_evidence_id,
target_capability_id=target_capability_id,
)
def _ensure_candidate_row(
self,
*,
table: str,
label: str,
repository_id: int,
analysis_run_id: int,
candidate_id: int,
) -> None:
with self.connect() as connection:
row = connection.execute(
f"""
SELECT id FROM {table}
WHERE id = ? AND repository_id = ? AND analysis_run_id = ?
""",
(candidate_id, repository_id, analysis_run_id),
).fetchone()
if row is None:
raise NotFoundError(
f"{label} {candidate_id} was not found for repository "
f"{repository_id} analysis run {analysis_run_id}"
)
def _relink_candidate_leaf(
self,
*,
table: str,
label: str,
repository_id: int,
analysis_run_id: int,
candidate_id: int,
target_capability_id: int,
) -> None:
self._ensure_candidate_row(
table="candidate_capabilities",
label="target candidate capability",
repository_id=repository_id,
analysis_run_id=analysis_run_id,
candidate_id=target_capability_id,
)
with self.connect() as connection:
cursor = connection.execute(
f"""
UPDATE {table}
SET capability_id = ?
WHERE id = ? AND repository_id = ? AND analysis_run_id = ?
""",
(
target_capability_id,
candidate_id,
repository_id,
analysis_run_id,
),
)
if cursor.rowcount == 0:
raise NotFoundError(
f"{label} {candidate_id} was not found for repository "
f"{repository_id} analysis run {analysis_run_id}"
)
def _reject_candidate_leaf(
self,
*,

82
src/repo_registry/web_api/app.py
View File

@@ -84,6 +84,16 @@ class CandidateEdit(BaseModel):
notes: str = ""
class CandidateCapabilityRelink(BaseModel):
target_ability_id: int
notes: str = ""
class CandidateLeafRelink(BaseModel):
target_capability_id: int
notes: str = ""
app = FastAPI(title="Repository Ability Registry", version="0.1.0")
@@ -344,6 +354,78 @@ def edit_candidate_capability(
raise HTTPException(status_code=404, detail=str(exc)) from exc
@app.post(
"/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-capabilities/{candidate_capability_id}/relink"
)
def relink_candidate_capability(
repository_id: int,
analysis_run_id: int,
candidate_capability_id: int,
payload: CandidateCapabilityRelink,
service: RegistryService = Depends(get_service),
) -> dict[str, object]:
try:
return asdict(
service.relink_candidate_capability(
repository_id,
analysis_run_id,
candidate_capability_id,
**payload.model_dump(),
)
)
except NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@app.post(
"/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-features/{candidate_feature_id}/relink"
)
def relink_candidate_feature(
repository_id: int,
analysis_run_id: int,
candidate_feature_id: int,
payload: CandidateLeafRelink,
service: RegistryService = Depends(get_service),
) -> dict[str, object]:
try:
return asdict(
service.relink_candidate_feature(
repository_id,
analysis_run_id,
candidate_feature_id,
**payload.model_dump(),
)
)
except NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@app.post(
"/repos/{repository_id}/analysis-runs/{analysis_run_id}"
"/candidate-evidence/{candidate_evidence_id}/relink"
)
def relink_candidate_evidence(
repository_id: int,
analysis_run_id: int,
candidate_evidence_id: int,
payload: CandidateLeafRelink,
service: RegistryService = Depends(get_service),
) -> dict[str, object]:
try:
return asdict(
service.relink_candidate_evidence(
repository_id,
analysis_run_id,
candidate_evidence_id,
**payload.model_dump(),
)
)
except NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@app.post("/repos/{repository_id}/abilities", status_code=201)
def create_ability(
repository_id: int,

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

@@ -452,6 +452,78 @@ def edit_candidate_capability_from_form(
)
@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,
)
def render_candidate_graph(graph: dict, repository_id: int, analysis_run_id: int) -> str:
abilities = graph.get("abilities", [])
if not abilities:
@@ -541,6 +613,7 @@ def render_candidate_capability(
{render_candidate_reject_form('candidate-capabilities', capability, repository_id, analysis_run_id)}
<p class="muted">{escape(capability['description'])}</p>
{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_sources(capability['source_refs'])}
<h3>Features</h3>
<ul>{features or '<li class="muted">No feature candidates.</li>'}</ul>
@@ -562,6 +635,7 @@ def render_candidate_feature(
<span class="pill">{escape(feature["type"])}</span>
<span class="source">{escape(feature["location"])}</span>
{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')}
</li>
"""
@@ -578,6 +652,7 @@ def render_candidate_evidence(
<span class="pill">{escape(evidence["strength"])}</span>
<span class="source">{escape(evidence["reference"])}</span>
{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')}
</li>
"""
@@ -601,6 +676,28 @@ def render_candidate_reject_form(
"""
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"""
<form style="display:inline-grid; grid-template-columns: 120px auto; gap: 6px; align-items: end;" method="post" action="{action}">
<label>{label}<input name="{field_name}" type="number" min="1" required></label>
<button class="secondary" type="submit">Relink</button>
</form>
"""
def render_ability_map(ability_map: dict) -> str:
abilities = ability_map.get("abilities", [])
if not abilities:

108
tests/test_registry_service.py
View File

@@ -372,6 +372,114 @@ def test_reject_candidate_feature_and_evidence_excludes_only_those_leaves(tmp_pa
assert len(approved_capability.evidence) == len(capability.evidence) - 1
def test_relink_candidate_capability_to_another_ability_before_approval(tmp_path):
source = tmp_path / "repo"
source.mkdir()
(source / "README.md").write_text("# Relink Capability\n", encoding="utf-8")
(source / "app.py").write_text(
"from fastapi import FastAPI\n"
"app = FastAPI()\n"
'@app.get("/health")\n'
"def health():\n"
" return {}\n",
encoding="utf-8",
)
service = make_service(tmp_path)
repository = service.register_repository(name="Relink Capability", url=str(source))
summary = service.analyze_repository(repository.id)
graph = service.candidate_graph(repository.id, summary.analysis_run.id)
capability = graph.abilities[0].capabilities[0]
with service.store.connect() as connection:
cursor = connection.execute(
"""
INSERT INTO candidate_abilities
(repository_id, analysis_run_id, name, description, confidence)
VALUES (?, ?, ?, ?, ?)
""",
(
repository.id,
summary.analysis_run.id,
"Operations Visibility",
"Curator-created target ability.",
0.72,
),
)
target_ability_id = int(cursor.lastrowid)
relinked_graph = service.relink_candidate_capability(
repository.id,
summary.analysis_run.id,
capability.id,
target_ability_id=target_ability_id,
notes="Move interface under the operational ability.",
)
ability_map = service.approve_candidate_graph(repository.id, summary.analysis_run.id)
target_candidate = [
ability for ability in relinked_graph.abilities if ability.id == target_ability_id
][0]
assert target_candidate.capabilities[0].id == capability.id
approved_target = [
ability for ability in ability_map.abilities if ability.name == "Operations Visibility"
][0]
assert approved_target.capabilities[0].name == capability.name
def test_relink_candidate_feature_and_evidence_to_another_capability(tmp_path):
source = tmp_path / "repo"
source.mkdir()
(source / "README.md").write_text("# Relink Leaves\n", encoding="utf-8")
(source / "requirements.txt").write_text("fastapi\n", encoding="utf-8")
(source / "tests").mkdir()
(source / "tests" / "test_health.py").write_text(
"def test_health(): pass\n",
encoding="utf-8",
)
(source / "app.py").write_text(
"from fastapi import FastAPI\n"
"app = FastAPI()\n"
'@app.get("/health")\n'
"def health():\n"
" return {}\n",
encoding="utf-8",
)
service = make_service(tmp_path)
repository = service.register_repository(name="Relink Leaves", url=str(source))
summary = service.analyze_repository(repository.id)
graph = service.candidate_graph(repository.id, summary.analysis_run.id)
source_capability = graph.abilities[0].capabilities[0]
target_capability = graph.abilities[0].capabilities[1]
feature = source_capability.features[0]
evidence = source_capability.evidence[0]
service.relink_candidate_feature(
repository.id,
summary.analysis_run.id,
feature.id,
target_capability_id=target_capability.id,
)
service.relink_candidate_evidence(
repository.id,
summary.analysis_run.id,
evidence.id,
target_capability_id=target_capability.id,
)
ability_map = service.approve_candidate_graph(repository.id, summary.analysis_run.id)
approved_capabilities = {
capability.name: capability for capability in ability_map.abilities[0].capabilities
}
assert approved_capabilities[source_capability.name].features == []
assert feature.name in {
item.name for item in approved_capabilities[target_capability.name].features
}
assert evidence.reference in {
item.reference for item in approved_capabilities[target_capability.name].evidence
}
def test_analyze_repository_failure_is_recorded(tmp_path):
service = make_service(tmp_path)
repository = service.register_repository(

72
tests/test_web_api.py
View File

@@ -371,3 +371,75 @@ def test_api_rejects_candidate_capability_feature_and_evidence(tmp_path):
)
finally:
app.dependency_overrides.clear()
def test_api_relinks_candidate_feature_and_evidence(tmp_path):
source = tmp_path / "repo"
source.mkdir()
(source / "README.md").write_text("# API Relink Leaves\n", encoding="utf-8")
(source / "requirements.txt").write_text("fastapi\n", encoding="utf-8")
(source / "tests").mkdir()
(source / "tests" / "test_status.py").write_text(
"def test_status(): pass\n",
encoding="utf-8",
)
(source / "app.py").write_text(
"from fastapi import FastAPI\n"
"app = FastAPI()\n"
'@app.get("/status")\n'
"def status():\n"
" return {}\n",
encoding="utf-8",
)
def override_settings():
return Settings(
database_path=str(tmp_path / "api-relink.sqlite3"),
checkout_root=str(tmp_path / "api-relink-checkouts"),
)
app.dependency_overrides[get_settings] = override_settings
client = TestClient(app)
try:
repository_response = client.post(
"/repos",
json={"name": "API Relink Leaves", "url": str(source)},
)
repository_id = repository_response.json()["id"]
run_response = client.post(f"/repos/{repository_id}/analysis-runs", json={})
run_id = run_response.json()["analysis_run"]["id"]
graph_response = client.get(
f"/repos/{repository_id}/analysis-runs/{run_id}/candidate-graph"
)
capabilities = graph_response.json()["abilities"][0]["capabilities"]
source_capability = capabilities[0]
target_capability = capabilities[1]
feature_id = source_capability["features"][0]["id"]
evidence_id = source_capability["evidence"][0]["id"]
feature_response = client.post(
f"/repos/{repository_id}/analysis-runs/{run_id}"
f"/candidate-features/{feature_id}/relink",
json={
"target_capability_id": target_capability["id"],
"notes": "Move feature",
},
)
assert feature_response.status_code == 200
relinked_capabilities = feature_response.json()["abilities"][0]["capabilities"]
assert relinked_capabilities[0]["features"] == []
assert relinked_capabilities[1]["features"][0]["id"] == feature_id
evidence_response = client.post(
f"/repos/{repository_id}/analysis-runs/{run_id}"
f"/candidate-evidence/{evidence_id}/relink",
json={
"target_capability_id": target_capability["id"],
"notes": "Move evidence",
},
)
assert evidence_response.status_code == 200
relinked_capabilities = evidence_response.json()["abilities"][0]["capabilities"]
assert relinked_capabilities[1]["evidence"][0]["id"] == evidence_id
finally:
app.dependency_overrides.clear()