from __future__ import annotations from collections.abc import Sequence from dataclasses import asdict, replace from typing import Any from repo_registry.acceptance import ( blocking_quality_gate_outcomes, evaluate_candidate_capability_quality, ) from repo_registry.core.models import ( AbilitySummary, AnalysisRunDiff, AnalysisRunDiffItem, AnalysisRunDiffSection, AnalysisRun, CapabilitySummary, CandidateAbility, CandidateCapability, CandidateEvidence, CandidateFeature, CandidateGraph, CharacteristicRebuildResult, ContentChunk, DependencyEdge, DependencyGraph, DependencyGraphViewProfile, DependencyImpactAnalysis, DependencyImpactItem, ExpectationGap, ObservedFact, Repository, RepositoryAbilityMap, ReviewDecision, ScanSummary, SearchResult, ) from repo_registry.candidate_graph.generator import CandidateGraphGenerator from repo_registry.candidate_graph.normalization import normalize_candidate_drafts from repo_registry.content_indexing.extractor import ContentExtractor from repo_registry.core.logging import log_operation from repo_registry.llm_extraction.extractor import LLMCandidateExtractor from repo_registry.llm_extraction.mapper import LLMExtractionMapper from repo_registry.repo_ingestion.git import GitIngestionService from repo_registry.repo_ingestion.metadata import RepositoryMetadataExtractor from repo_registry.repo_scanning.scanner import DeterministicScanner from repo_registry.semantic import EmbeddingProvider, cosine_similarity from repo_registry.storage.sqlite import RegistryStore class RegistryService: """Application service for the manual registry MVP.""" def __init__( self, store: RegistryStore, ingestion: GitIngestionService | None = None, llm_extractor: LLMCandidateExtractor | None = None, embedding_provider: EmbeddingProvider | None = None, ) -> None: self.store = store self.scanner = DeterministicScanner() self.ingestion = ingestion or GitIngestionService() self.metadata_extractor = RepositoryMetadataExtractor() self.candidate_generator = CandidateGraphGenerator() self.content_extractor = ContentExtractor() self.llm_extractor = llm_extractor self.llm_mapper = LLMExtractionMapper() self.embedding_provider = embedding_provider def register_repository( self, *, url: str, name: str | None = None, description: str | None = None, branch: str = "main", access_username: str | None = None, access_password: str | None = None, ) -> Repository: if name is None or description is None: checkout = self.ingestion.resolve( url, branch=branch, access_username=access_username, access_password=access_password, ) metadata = self.metadata_extractor.extract(checkout.source_path, url) else: metadata = None repository = self.store.create_repository( name=name or (metadata.name if metadata is not None else "repository"), url=url, description=description or (metadata.description if metadata is not None else None), branch=branch, ) log_operation( "repository_registered", repository_id=repository.id, repository_name=repository.name, branch=repository.branch, metadata_imported=metadata is not None, ) return repository def list_repositories(self) -> list[Repository]: return self.store.list_repositories() def get_repository(self, repository_id: int) -> Repository: return self.store.get_repository(repository_id) def update_repository( self, repository_id: int, *, name: str | None = None, description: str | None = None, branch: str | None = None, ) -> Repository: return self.store.update_repository( repository_id, name=name, description=description, branch=branch, ) def delete_repository(self, repository_id: int) -> None: self.store.delete_repository(repository_id) def analyze_repository( self, repository_id: int, *, source_path: str | None = None, use_cached_checkout: bool = False, use_llm_assistance: bool = True, trusted_auto_approve: bool = False, access_username: str | None = None, access_password: str | None = None, ) -> ScanSummary: repository = self.store.get_repository(repository_id) run = self.store.create_analysis_run(repository_id) self.store.update_repository_status(repository_id, "analyzing") log_operation( "analysis_started", repository_id=repository_id, analysis_run_id=run.id, source_override=source_path is not None, ) try: if source_path is None: if use_cached_checkout: checkout = self.ingestion.cached_checkout(repository.url) if checkout is None: raise RuntimeError( "cached checkout was requested, but no checkout exists " "for this repository" ) else: checkout = self.ingestion.resolve( repository.url, branch=repository.branch, access_username=access_username, access_password=access_password, ) scan_source = checkout.source_path else: scan_source = source_path scan_result = self.scanner.scan(scan_source) except Exception as exc: failed_run = self.store.fail_analysis_run(repository_id, run.id, str(exc)) log_operation( "analysis_failed", repository_id=repository_id, analysis_run_id=run.id, error=str(exc), ) return ScanSummary(analysis_run=failed_run, snapshot=None, facts=[]) completed_run = self.store.complete_analysis_run( repository_id, run.id, scan_result, ) snapshot = ( self.store.get_snapshot(completed_run.snapshot_id) if completed_run.snapshot_id is not None else None ) facts = self.store.list_observed_facts(repository_id, completed_run.id) chunks = self.content_extractor.extract(scan_result.source_path, facts) self.store.replace_content_chunks( repository_id, completed_run.id, completed_run.snapshot_id, chunks, ) stored_chunks = self.store.list_content_chunks(repository_id, completed_run.id) try: candidates, candidate_source = self._generate_candidates( repository, facts, stored_chunks, use_llm_assistance=use_llm_assistance, ) except Exception as exc: log_operation( "llm_extraction_failed", repository_id=repository_id, analysis_run_id=completed_run.id, error=str(exc), ) self.store.create_review_decision( repository_id, completed_run.id, action="llm_extraction_failed", notes=str(exc), ) candidates = self.candidate_generator.generate( repository, facts, stored_chunks, ) candidate_source = "deterministic" candidates = normalize_candidate_drafts(candidates) self.store.replace_candidate_graph(repository_id, completed_run.id, candidates) if "llm" in candidate_source: log_operation( "llm_extraction_used", repository_id=repository_id, analysis_run_id=completed_run.id, candidate_count=len(candidates), ) self.store.create_review_decision( repository_id, completed_run.id, action="llm_extraction_used", notes=( f"Generated {len(candidates)} candidate ability draft(s) " f"from {candidate_source} candidate generation." ), ) if trusted_auto_approve: self.trusted_auto_approve_candidate_graph( repository_id, completed_run.id, notes=( "Trusted auto-populate mode reviewed candidate graph " f"after {candidate_source} candidate generation." ), ) log_operation( "analysis_completed", repository_id=repository_id, analysis_run_id=completed_run.id, fact_count=len(facts), content_chunk_count=len(stored_chunks), candidate_count=len(candidates), candidate_source=candidate_source, ) return ScanSummary( analysis_run=completed_run, snapshot=snapshot, facts=facts, ) def _generate_candidates( self, repository: Repository, facts: list[ObservedFact], chunks: list[ContentChunk], *, use_llm_assistance: bool = True, ): deterministic = self.candidate_generator.generate(repository, facts, chunks) if use_llm_assistance and self.llm_extractor is not None: extracted = self.llm_extractor.extract(repository, chunks) if extracted: llm_candidates = self.llm_mapper.map(extracted, facts, chunks) return ( self._merge_llm_candidates(llm_candidates, deterministic), "llm+deterministic", ) return deterministic, "deterministic" def _merge_llm_candidates( self, llm_candidates: list, deterministic: list, ) -> list: if not deterministic: return [ ability for ability in llm_candidates if self._candidate_ability_has_trusted_sources(ability) ] merged_deterministic = list(deterministic) trusted_llm = [] folded_capabilities = [] for ability in llm_candidates: if self._candidate_ability_has_trusted_sources(ability): trusted_llm.append(ability) else: folded_capabilities.extend(ability.capabilities) if folded_capabilities: target = merged_deterministic[0] merged_deterministic[0] = replace( target, capabilities=[*target.capabilities, *folded_capabilities], ) return [*trusted_llm, *merged_deterministic] def _candidate_ability_has_trusted_sources(self, ability) -> bool: if not ability.source_refs: return False return any( ref.kind in {"intent", "documentation", "interface", "test", "example"} and not ref.path.lower().endswith("scope.md") for ref in ability.source_refs ) def list_analysis_runs(self, repository_id: int) -> list[AnalysisRun]: return self.store.list_analysis_runs(repository_id) def get_analysis_run(self, repository_id: int, analysis_run_id: int) -> AnalysisRun: return self.store.get_analysis_run(repository_id, analysis_run_id) def list_abilities(self) -> list[AbilitySummary]: return self.store.list_abilities() def list_capabilities(self) -> list[CapabilitySummary]: return self.store.list_capabilities() def list_review_decisions( self, repository_id: int, analysis_run_id: int | None = None, ) -> list[ReviewDecision]: return self.store.list_review_decisions(repository_id, analysis_run_id) def record_expectation_gap( self, repository_id: int, *, analysis_run_id: int | None = None, expected_type: str, expected_name: str, source: str, notes: str = "", ) -> ExpectationGap: gap = self.store.create_expectation_gap( repository_id, analysis_run_id, expected_type=expected_type, expected_name=expected_name, source=source, notes=notes, ) self.store.create_review_decision( repository_id, analysis_run_id, action="record_expectation_gap", notes=f"{source} expected {expected_type}: {expected_name}", ) return gap def list_expectation_gaps( self, repository_id: int, analysis_run_id: int | None = None, ) -> list[ExpectationGap]: return self.store.list_expectation_gaps(repository_id, analysis_run_id) def list_observed_facts( self, repository_id: int, analysis_run_id: int | None = None, ) -> list[ObservedFact]: return self.store.list_observed_facts(repository_id, analysis_run_id) def list_content_chunks( self, repository_id: int, analysis_run_id: int | None = None, ) -> list[ContentChunk]: return self.store.list_content_chunks(repository_id, analysis_run_id) def candidate_graph(self, repository_id: int, analysis_run_id: int) -> CandidateGraph: return self.store.get_candidate_graph(repository_id, analysis_run_id) def rebuild_characteristics_from_scratch( self, repository_id: int, *, dry_run: bool = True, confirm: bool = False, source_path: str | None = None, use_cached_checkout: bool = False, use_llm_assistance: bool = True, access_username: str | None = None, access_password: str | None = None, ) -> CharacteristicRebuildResult: if not dry_run and not confirm: raise ValueError("confirmed rebuild requires confirm=True") repository = self.store.get_repository(repository_id) previous_counts = self._approved_counts(repository_id) previous_ids = self._approved_ids(repository_id) summary = self.analyze_repository( repository_id, source_path=source_path, use_cached_checkout=use_cached_checkout, use_llm_assistance=use_llm_assistance, trusted_auto_approve=False, access_username=access_username, access_password=access_password, ) if summary.analysis_run.status != "completed": return CharacteristicRebuildResult( repository=repository, analysis_run=summary.analysis_run, dry_run=dry_run, confirmed=confirm, cleared_approved=False, previous_counts=previous_counts, previous_ids=previous_ids, candidate_counts={}, ) graph = self.store.get_candidate_graph(repository_id, summary.analysis_run.id) candidate_counts = self._candidate_counts(graph) cleared = False if not dry_run: self.store.clear_approved_characteristics(repository_id) self.store.update_repository_status(repository_id, "analyzed") cleared = True action = ( "rebuild_characteristics_from_scratch" if cleared else "dry_run_rebuild_characteristics_from_scratch" ) self.store.create_review_decision( repository_id, summary.analysis_run.id, action=action, notes=( f"Previous approved counts: {previous_counts}. " f"Previous approved IDs: {previous_ids}. " f"New candidate counts: {candidate_counts}." ), ) return CharacteristicRebuildResult( repository=repository, analysis_run=summary.analysis_run, dry_run=dry_run, confirmed=confirm, cleared_approved=cleared, previous_counts=previous_counts, previous_ids=previous_ids, candidate_counts=candidate_counts, ) def approve_candidate_graph( self, repository_id: int, analysis_run_id: int, *, notes: str = "", action: str = "approve_candidate_graph", ) -> RepositoryAbilityMap: graph = self.store.get_candidate_graph(repository_id, analysis_run_id) pending_abilities = [ ability for ability in graph.abilities if ability.status == "candidate" ] for ability in pending_abilities: approved_ability_id = self._ensure_approved_ability(repository_id, ability) for capability in ability.capabilities: if capability.status != "candidate": continue approved_capability_id = self._ensure_approved_capability( repository_id, approved_ability_id, ability.name, capability, ) for feature in capability.features: if feature.status != "candidate": continue self.store.create_feature( repository_id, approved_capability_id, name=feature.name, type=feature.type, location=feature.location, confidence=feature.confidence, source_refs=feature.source_refs, primary_class=feature.primary_class, attributes=feature.attributes, ) for evidence in capability.evidence: if evidence.status != "candidate": continue 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, ) if pending_abilities: self.store.mark_candidate_graph_status( repository_id, analysis_run_id, "approved", ) self.store.create_review_decision( repository_id, analysis_run_id, action=action, notes=notes, ) self.store.update_repository_status(repository_id, "indexed") return self.store.get_ability_map(repository_id) def trusted_auto_approve_candidate_graph( self, repository_id: int, analysis_run_id: int, *, notes: str = "", ) -> RepositoryAbilityMap: graph = self.store.get_candidate_graph(repository_id, analysis_run_id) approved_count = 0 skipped_count = 0 approved_reasons: list[str] = [] skipped_reasons: list[str] = [] for ability in graph.abilities: if ability.status != "candidate": continue candidate_capabilities = [ capability for capability in ability.capabilities if capability.status == "candidate" ] safe_capabilities = [] for capability in candidate_capabilities: safe, reason = self._trusted_auto_approve_capability_decision(capability) if safe: safe_capabilities.append(capability) approved_reasons.append(f"{capability.name}: {reason}") else: skipped_reasons.append(f"{capability.name}: {reason}") skipped_count += len(candidate_capabilities) - len(safe_capabilities) if not safe_capabilities: continue approved_ability_id = self._ensure_approved_ability(repository_id, ability) for capability in safe_capabilities: self._create_approved_capability_subtree( repository_id, approved_ability_id, capability, ) self.store.mark_candidate_capability_status( repository_id, analysis_run_id, capability.id, "approved", ) approved_count += 1 if len(safe_capabilities) == len(candidate_capabilities): self.store.mark_candidate_ability_status( repository_id, analysis_run_id, ability.id, "approved", ) if approved_count: self.store.update_repository_status(repository_id, "indexed") self.store.create_review_decision( repository_id, analysis_run_id, action="trusted_auto_approve_candidate_graph", notes=( f"{notes} Auto-approved {approved_count} safe candidate " f"capability(s); left {skipped_count} for review." f"{self._trusted_auto_approve_notes(approved_reasons, skipped_reasons)}" ).strip(), ) return self.store.get_ability_map(repository_id) def _trusted_auto_approve_capability_safe( self, capability: CandidateCapability, ) -> bool: safe, _reason = self._trusted_auto_approve_capability_decision(capability) return safe def _trusted_auto_approve_capability_decision( self, capability: CandidateCapability, ) -> tuple[bool, str]: gate_outcomes = evaluate_candidate_capability_quality(capability) blocking_outcomes = blocking_quality_gate_outcomes(gate_outcomes) if blocking_outcomes: criteria = ", ".join( sorted({outcome.criterion_id for outcome in blocking_outcomes}) ) return False, f"quality gates require review ({criteria})" has_source_refs = bool(capability.source_refs) or any( feature.source_refs for feature in capability.features ) if not has_source_refs: return False, "missing source references" if capability.primary_class == "repository-structure": return False, "structural/dependency context requires curator review" utility_relationships = self._candidate_utility_relationships(capability) eligible_relationships = {"owned", "facade", "adapter"} if not utility_relationships: return False, "missing utility relationship" if not (utility_relationships & eligible_relationships): relationships = ", ".join(sorted(utility_relationships)) return False, f"utility relationship is not eligible ({relationships})" if capability.primary_class == "llm-integration": return True, "eligible LLM utility relationship with source support" if capability.primary_class in {"interface", "API", "CLI", "callable", "api", "cli"}: if capability.confidence >= 0.55: return True, "owned interface with sufficient confidence" return False, "owned interface confidence below trusted threshold" if capability.features: if capability.confidence >= 0.55: return True, "eligible utility relationship with feature support" return False, "feature-backed capability confidence below trusted threshold" if capability.confidence >= 0.75: return True, "eligible utility relationship with high confidence" return False, "capability confidence below trusted threshold" def _candidate_utility_relationships( self, capability: CandidateCapability, ) -> set[str]: return { attribute.removeprefix("utility-") for attribute in capability.attributes if attribute.startswith("utility-") } def _trusted_auto_approve_notes( self, approved_reasons: list[str], skipped_reasons: list[str], ) -> str: details: list[str] = [] if approved_reasons: details.append("Approved: " + "; ".join(approved_reasons) + ".") if skipped_reasons: details.append("Skipped: " + "; ".join(skipped_reasons) + ".") if not details: return "" return " " + " ".join(details) def _approved_counts(self, repository_id: int) -> dict[str, int]: ability_map = self.store.get_ability_map(repository_id) capabilities = [ capability for ability in ability_map.abilities for capability in ability.capabilities ] features = [ feature for capability in capabilities for feature in capability.features ] evidence = [ item for capability in capabilities for item in capability.evidence ] return { "abilities": len(ability_map.abilities), "capabilities": len(capabilities), "features": len(features), "evidence": len(evidence), } def _approved_ids(self, repository_id: int) -> dict[str, list[int]]: ability_map = self.store.get_ability_map(repository_id) capabilities = [ capability for ability in ability_map.abilities for capability in ability.capabilities ] features = [ feature for capability in capabilities for feature in capability.features ] evidence = [ item for capability in capabilities for item in capability.evidence ] return { "abilities": [ability.id for ability in ability_map.abilities], "capabilities": [capability.id for capability in capabilities], "features": [feature.id for feature in features], "evidence": [item.id for item in evidence], } def _candidate_counts(self, graph: CandidateGraph) -> dict[str, int]: capabilities = [ capability for ability in graph.abilities for capability in ability.capabilities ] features = [ feature for capability in capabilities for feature in capability.features ] evidence = [ item for capability in capabilities for item in capability.evidence ] return { "abilities": len(graph.abilities), "capabilities": len(capabilities), "features": len(features), "evidence": len(evidence), } def accept_candidate_ability( self, repository_id: int, analysis_run_id: int, candidate_ability_id: int, *, notes: str = "", ) -> RepositoryAbilityMap: graph = self.store.get_candidate_graph(repository_id, analysis_run_id) ability = next( ( item for item in graph.abilities if item.id == candidate_ability_id and item.status == "candidate" ), None, ) if ability is None: raise ValueError(f"candidate ability {candidate_ability_id} is not pending") approved_ability_id = self._ensure_approved_ability(repository_id, ability) for capability in ability.capabilities: if capability.status == "candidate": self._create_approved_capability_subtree( repository_id, approved_ability_id, capability, ) self.store.mark_candidate_ability_status( repository_id, analysis_run_id, candidate_ability_id, "approved", ) self._record_candidate_acceptance( repository_id, analysis_run_id, "accept_candidate_ability", notes or f"Accepted candidate ability: {ability.name}", ) return self.store.get_ability_map(repository_id) def accept_candidate_capability( self, repository_id: int, analysis_run_id: int, candidate_capability_id: int, *, notes: str = "", ) -> RepositoryAbilityMap: graph = self.store.get_candidate_graph(repository_id, analysis_run_id) parent_ability, capability = self._candidate_capability_with_parent( graph, candidate_capability_id, ) if capability.status != "candidate": raise ValueError( f"candidate capability {candidate_capability_id} is not pending" ) approved_ability_id = self._ensure_approved_ability(repository_id, parent_ability) self._create_approved_capability_subtree( repository_id, approved_ability_id, capability, ) self.store.mark_candidate_capability_status( repository_id, analysis_run_id, candidate_capability_id, "approved", ) self._record_candidate_acceptance( repository_id, analysis_run_id, "accept_candidate_capability", notes or f"Accepted candidate capability: {capability.name}", ) return self.store.get_ability_map(repository_id) def accept_candidate_feature( self, repository_id: int, analysis_run_id: int, candidate_feature_id: int, *, notes: str = "", ) -> RepositoryAbilityMap: graph = self.store.get_candidate_graph(repository_id, analysis_run_id) parent_ability, parent_capability, feature = self._candidate_feature_with_parent( graph, candidate_feature_id, ) if feature.status != "candidate": raise ValueError(f"candidate feature {candidate_feature_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_feature( repository_id, approved_capability_id, name=feature.name, type=feature.type, location=feature.location, confidence=feature.confidence, source_refs=feature.source_refs, primary_class=feature.primary_class, attributes=feature.attributes, ) self.store.mark_candidate_feature_status( repository_id, analysis_run_id, candidate_feature_id, "approved", ) self._record_candidate_acceptance( repository_id, analysis_run_id, "accept_candidate_feature", notes or f"Accepted candidate feature: {feature.name}", ) 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, base_analysis_run_id: int, target_analysis_run_id: int, ) -> AnalysisRunDiff: repository = self.store.get_repository(repository_id) base_run = self.store.get_analysis_run(repository_id, base_analysis_run_id) target_run = self.store.get_analysis_run(repository_id, target_analysis_run_id) base_graph = self.store.get_candidate_graph(repository_id, base_analysis_run_id) target_graph = self.store.get_candidate_graph(repository_id, target_analysis_run_id) approved_map = self.store.get_ability_map(repository_id) return AnalysisRunDiff( repository=repository, base_run=base_run, target_run=target_run, facts=self._diff_items( self._fact_index( self.store.list_observed_facts(repository_id, base_analysis_run_id) ), self._fact_index( self.store.list_observed_facts(repository_id, target_analysis_run_id) ), ), chunks=self._diff_items( self._chunk_index( self.store.list_content_chunks(repository_id, base_analysis_run_id) ), self._chunk_index( self.store.list_content_chunks(repository_id, target_analysis_run_id) ), ), candidates=self._diff_items( self._candidate_index(base_graph.abilities), self._candidate_index(target_graph.abilities), ), approved_entries=self._diff_items( self._approved_index(approved_map.abilities), self._candidate_index(target_graph.abilities), ), ) def build_dependency_graph(self, repository_id: int) -> DependencyGraph: repository = self.store.get_repository(repository_id) ability_map = self.store.get_ability_map(repository_id) edges: list[DependencyEdge] = [] scope_key = self._dependency_key("scope", ability_map.scope.id) for ability in ability_map.abilities: ability_key = self._dependency_key("ability", ability.id) edges.append( self._dependency_edge( source_kind="ability", source_id=ability.id, source_key=ability_key, target_kind="scope", target_id=ability_map.scope.id, target_key=scope_key, dependency_type="summarizes", strength="strong", source="approved_characteristic", ) ) for capability in ability.capabilities: capability_key = self._dependency_key("capability", capability.id) edges.append( self._dependency_edge( source_kind="capability", source_id=capability.id, source_key=capability_key, target_kind="ability", target_id=ability.id, target_key=ability_key, dependency_type="realizes", strength="strong", source="approved_characteristic", ) ) edges.extend( self._capability_dependency_edges( capability, capability_key=capability_key, ) ) return DependencyGraph( repository=repository, scope=ability_map.scope, edges=edges, ) def analyze_dependency_impact( self, repository_id: int, base_analysis_run_id: int, target_analysis_run_id: int, ) -> DependencyImpactAnalysis: diff = self.diff_analysis_runs( repository_id, base_analysis_run_id, target_analysis_run_id, ) graph = self.build_dependency_graph(repository_id) changed_facts = [ item for section in ( diff.facts.added, diff.facts.removed, diff.facts.changed, diff.facts.weakened, ) for item in section ] changed_fact_keys = [item.key for item in changed_facts] fact_reasons = { item.key: f"{item.change_type} fact {item.key}" for item in changed_facts } adjacency: dict[str, list[DependencyEdge]] = {} for edge in graph.edges: adjacency.setdefault(edge.source_key, []).append(edge) queue: list[tuple[str, int, str]] = [ (key, 0, fact_reasons[key]) for key in changed_fact_keys ] impacts_by_key: dict[str, DependencyImpactItem] = {} visited_edges: set[tuple[str, str]] = set() while queue: source_key, depth, inherited_reason = queue.pop(0) for edge in adjacency.get(source_key, []): edge_marker = (edge.source_key, edge.target_key) if edge_marker in visited_edges: continue visited_edges.add(edge_marker) impact_depth = depth + 1 reason = ( f"{inherited_reason} -> {edge.target_kind} depends on " f"{edge.source_kind} via {edge.dependency_type}" ) current = impacts_by_key.get(edge.target_key) if current is None: impacts_by_key[edge.target_key] = DependencyImpactItem( item_kind=edge.target_kind, item_id=edge.target_id, item_key=edge.target_key, name=self._dependency_display_name( repository_id, edge.target_kind, edge.target_id, ), freshness_state="stale", ownership=edge.target_ownership, recommended_action=self._recommended_action( edge.target_ownership ), impact_depth=impact_depth, reasons=[reason], ) else: impacts_by_key[edge.target_key] = replace( current, impact_depth=min(current.impact_depth, impact_depth), reasons=[*current.reasons, reason], ) queue.append((edge.target_key, impact_depth, reason)) impacts = sorted( impacts_by_key.values(), key=lambda item: (item.impact_depth, item.item_kind, item.item_id), ) max_depth = max((item.impact_depth for item in impacts), default=0) return DependencyImpactAnalysis( repository=diff.repository, base_run=diff.base_run, target_run=diff.target_run, changed_fact_keys=changed_fact_keys, impacts=impacts, max_depth=max_depth, scope_impacted=any(item.item_kind == "scope" for item in impacts), propagation_breadth=len(impacts), graph=graph, ) def dependency_graph_elements( self, repository_id: int, *, base_analysis_run_id: int | None = None, target_analysis_run_id: int | None = None, profile_id: int | None = None, rules: list[dict[str, Any]] | None = None, manual_overrides: dict[str, str] | None = None, use_latest_profile: bool = True, ) -> dict[str, object]: impact = None if base_analysis_run_id is not None or target_analysis_run_id is not None: if base_analysis_run_id is None or target_analysis_run_id is None: raise ValueError( "base_analysis_run_id and target_analysis_run_id must be provided together" ) impact = self.analyze_dependency_impact( repository_id, base_analysis_run_id, target_analysis_run_id, ) graph = impact.graph else: graph = self.build_dependency_graph(repository_id) impact_by_key = ( {item.item_key: item for item in impact.impacts} if impact is not None else {} ) changed_fact_keys = set(impact.changed_fact_keys) if impact is not None else set() ability_map = self.store.get_ability_map(repository_id) facts_by_id = {fact.id: fact for fact in self.store.list_observed_facts(repository_id)} characteristic_index = self._dependency_characteristic_index(ability_map) nodes: dict[str, dict[str, object]] = {} edge_sources: dict[str, DependencyEdge] = {} profile = ( self.store.get_dependency_graph_profile(repository_id, profile_id) if profile_id is not None else self.store.latest_dependency_graph_profile(repository_id) if use_latest_profile and not rules and not manual_overrides else None ) merged_rules = [*(profile.filter_rules if profile is not None else []), *(rules or [])] merged_overrides = { **(profile.manual_overrides if profile is not None else {}), **(manual_overrides or {}), } graph_edges = [ display_edge for edge in graph.edges if (display_edge := self._dependency_display_edge(edge, facts_by_id)) is not None ] def ensure_node(kind: str, key: str, item_id: int | None) -> None: if key in nodes: return impact_item = impact_by_key.get(key) is_changed_fact = key in changed_fact_keys detail = characteristic_index.get(key, {}) fact = facts_by_id.get(item_id) if kind == "fact" and item_id else None if fact is not None: detail = { "name": fact.name, "label": ( f"{fact.path} ({fact.kind})" if key.startswith("fact:document:") else f"{fact.name} ({fact.kind}, {fact.path})" ), "description": fact.value, "primaryClass": fact.metadata.get("source_role", fact.kind), "attributes": self._dependency_fact_attributes(fact), "confidence": fact.metadata.get("confidence"), "path": fact.path, "value": fact.value, "metadata": fact.metadata, "sourceReferences": [ { "fact_id": fact.id, "path": fact.path, "kind": fact.kind, "name": fact.name, "line": fact.metadata.get("line"), } ], } nodes[key] = { "data": { "id": key, "key": key, "stableKey": key, "kind": kind, "layer": self._dependency_layer(kind), "label": detail.get("label") or self._dependency_node_label(repository_id, kind, key, item_id), "reviewState": "accepted", "name": detail.get("name") or self._dependency_node_label(repository_id, kind, key, item_id), "description": detail.get("description", ""), "primaryClass": detail.get("primaryClass", kind), "attributes": detail.get("attributes", []), "confidence": detail.get("confidence"), "visualSize": self._dependency_node_size(detail.get("confidence")), "ownership": self._ownership_for_kind(kind), "freshnessState": ( impact_item.freshness_state if impact_item is not None else "changed" if is_changed_fact else "current" ), "recommendedAction": ( impact_item.recommended_action if impact_item is not None else "" ), "impactDepth": ( impact_item.impact_depth if impact_item is not None else None ), "reasons": impact_item.reasons if impact_item is not None else [], "path": detail.get("path", ""), "value": detail.get("value", ""), "metadata": detail.get("metadata", {}), "sourceReferences": detail.get("sourceReferences", []), }, "classes": " ".join( class_name for class_name in ( kind, "stale" if impact_item is not None else "current", "changed" if is_changed_fact else "", ) if class_name ), } for edge in graph_edges: ensure_node(edge.source_kind, edge.source_key, edge.source_id) ensure_node(edge.target_kind, edge.target_key, edge.target_id) edges = [] for index, edge in enumerate(graph_edges): edge_id = f"{edge.source_key}->{edge.target_key}:{index}" source_data = nodes[edge.source_key]["data"] target_data = nodes[edge.target_key]["data"] edge_sources[edge_id] = edge edges.append( { "data": { "id": edge_id, "key": edge_id, "stableKey": edge_id, "kind": "edge", "layer": "dependency", "reviewState": "accepted", "source": edge.source_key, "target": edge.target_key, "sourceKind": edge.source_kind, "targetKind": edge.target_kind, "sourceLayer": self._dependency_layer(edge.source_kind), "targetLayer": self._dependency_layer(edge.target_kind), "dependencyType": edge.dependency_type, "strength": edge.strength, "edgeWidth": self._dependency_edge_width(edge.strength), "edgeSource": edge.source, "sameLayer": edge.same_layer, "freshnessState": ( "stale" if edge.target_key in impact_by_key else "changed" if edge.source_key in changed_fact_keys else "current" ), "sourceMetadata": { "key": edge.source_key, "kind": edge.source_kind, "layer": self._dependency_layer(edge.source_kind), "name": source_data.get("name", edge.source_key), }, "targetMetadata": { "key": edge.target_key, "kind": edge.target_kind, "layer": self._dependency_layer(edge.target_kind), "name": target_data.get("name", edge.target_key), }, "label": edge.dependency_type, }, "classes": " ".join( class_name for class_name in ( edge.dependency_type, edge.strength, "same-layer" if edge.same_layer else "", ) if class_name ), } ) elements = [*nodes.values(), *edges] visibility = self._evaluate_dependency_visibility( elements, merged_rules, merged_overrides, ) hidden_node_ids = { element["data"]["id"] for element in nodes.values() if visibility[element["data"]["id"]]["displayState"] == "hide" } blurred_node_ids = { element["data"]["id"] for element in nodes.values() if visibility[element["data"]["id"]]["displayState"] == "blur" } visible_elements: list[dict[str, object]] = [] hidden_elements: list[dict[str, object]] = [] orphaned_overrides = sorted( key for key in merged_overrides if key not in visibility ) for element in elements: element_id = element["data"]["id"] state = visibility[element_id] if "source" in element["data"] and ( element["data"]["source"] in hidden_node_ids or element["data"]["target"] in hidden_node_ids ): state = { **state, "displayState": "hide", "visibilityReason": "connected-node-hidden", } connected_to_blurred = ( "source" in element["data"] and ( element["data"]["source"] in blurred_node_ids or element["data"]["target"] in blurred_node_ids ) ) element["data"].update(state) element["data"]["connectedToBlurred"] = connected_to_blurred element["classes"] = " ".join( part for part in ( element.get("classes", ""), f"display-{state['displayState']}", "connects-blurred" if connected_to_blurred else "", "manual-override" if state["visibilitySource"] == "manual" else "", "rule-derived" if state["visibilitySource"] == "rule" else "", ) if part ) if state["displayState"] == "hide": hidden_elements.append(element) else: visible_elements.append(element) return { "repository": asdict(graph.repository), "scope": asdict(graph.scope), "mode": ( profile.default_mode if profile is not None and impact is None else "impact" if impact is not None else "full" ), "profile": asdict(profile) if profile is not None else None, "metrics": { "node_count": len( [ element for element in visible_elements if "source" not in element["data"] ] ), "edge_count": len( [ element for element in visible_elements if "source" in element["data"] ] ), "hidden_count": len(hidden_elements), "blurred_count": len( [ element for element in visible_elements if element["data"]["displayState"] == "blur" ] ), "propagation_breadth": impact.propagation_breadth if impact else 0, "max_depth": impact.max_depth if impact else 0, "scope_impacted": impact.scope_impacted if impact else False, }, "filter": { "rules": merged_rules, "manual_overrides": merged_overrides, "orphaned_overrides": orphaned_overrides, "precedence": "later rules override earlier rules; manual overrides win last", "connected_edge_behavior": "edges connected to hidden nodes are hidden", }, "changed_fact_keys": impact.changed_fact_keys if impact else [], "elements": visible_elements, "hidden_elements": hidden_elements, "impacts": [asdict(item) for item in impact.impacts] if impact else [], } def list_dependency_graph_profiles( self, repository_id: int, ) -> list[DependencyGraphViewProfile]: return self.store.list_dependency_graph_profiles(repository_id) def get_dependency_graph_profile( self, repository_id: int, profile_id: int, ) -> DependencyGraphViewProfile: return self.store.get_dependency_graph_profile(repository_id, profile_id) def create_dependency_graph_profile( self, repository_id: int, *, name: str, description: str = "", default_mode: str = "full", filter_rules: list[dict[str, Any]] | None = None, manual_overrides: dict[str, str] | None = None, ) -> DependencyGraphViewProfile: self._validate_dependency_graph_profile_payload( default_mode, filter_rules or [], manual_overrides or {}, ) return self.store.create_dependency_graph_profile( repository_id, name=name, description=description, default_mode=default_mode, filter_rules=filter_rules or [], manual_overrides=manual_overrides or {}, ) def update_dependency_graph_profile( self, repository_id: int, profile_id: int, *, name: str | None = None, description: str | None = None, default_mode: str | None = None, filter_rules: list[dict[str, Any]] | None = None, manual_overrides: dict[str, str] | None = None, ) -> DependencyGraphViewProfile: if default_mode is not None or filter_rules is not None or manual_overrides is not None: current = self.store.get_dependency_graph_profile(repository_id, profile_id) self._validate_dependency_graph_profile_payload( default_mode or current.default_mode, filter_rules if filter_rules is not None else current.filter_rules, manual_overrides if manual_overrides is not None else current.manual_overrides, ) return self.store.update_dependency_graph_profile( repository_id, profile_id, name=name, description=description, default_mode=default_mode, filter_rules=filter_rules, manual_overrides=manual_overrides, ) def duplicate_dependency_graph_profile( self, repository_id: int, profile_id: int, *, name: str | None = None, ) -> DependencyGraphViewProfile: profile = self.store.get_dependency_graph_profile(repository_id, profile_id) return self.store.create_dependency_graph_profile( repository_id, name=name or f"{profile.name} Copy", description=profile.description, default_mode=profile.default_mode, filter_rules=profile.filter_rules, manual_overrides=profile.manual_overrides, ) def delete_dependency_graph_profile( self, repository_id: int, profile_id: int, ) -> None: self.store.delete_dependency_graph_profile(repository_id, profile_id) def approve_analysis_run_changes( self, repository_id: int, analysis_run_id: int, *, notes: str = "", ) -> RepositoryAbilityMap: graph = self.store.get_candidate_graph(repository_id, analysis_run_id) self.store.replace_approved_from_candidate_graph(repository_id, graph) self.store.mark_candidate_graph_status(repository_id, analysis_run_id, "approved") self.store.create_review_decision( repository_id, analysis_run_id, action="approve_analysis_run_changes", notes=notes, ) self.store.update_repository_status(repository_id, "indexed") return self.store.get_ability_map(repository_id) def _create_approved_capability_subtree( self, repository_id: int, approved_ability_id: int, capability: CandidateCapability, ) -> int: approved_capability_id = self.store.create_capability( repository_id, approved_ability_id, name=capability.name, description=capability.description, inputs=capability.inputs, outputs=capability.outputs, confidence=capability.confidence, primary_class=capability.primary_class, attributes=capability.attributes, ) for feature in capability.features: if feature.status != "candidate": continue self.store.create_feature( repository_id, approved_capability_id, name=feature.name, type=feature.type, location=feature.location, confidence=feature.confidence, source_refs=feature.source_refs, primary_class=feature.primary_class, attributes=feature.attributes, ) for evidence in capability.evidence: if evidence.status != "candidate": continue 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, ) return approved_capability_id def _ensure_approved_ability( self, repository_id: int, candidate_ability: CandidateAbility, ) -> int: ability_map = self.store.get_ability_map(repository_id) for ability in ability_map.abilities: if ability.name == candidate_ability.name: return ability.id return self.store.create_ability( repository_id, name=candidate_ability.name, description=candidate_ability.description, confidence=candidate_ability.confidence, primary_class=candidate_ability.primary_class, attributes=candidate_ability.attributes, ) def _ensure_approved_capability( self, repository_id: int, approved_ability_id: int, approved_ability_name: str, candidate_capability: CandidateCapability, ) -> int: ability_map = self.store.get_ability_map(repository_id) for ability in ability_map.abilities: if ability.name != approved_ability_name: continue for capability in ability.capabilities: if capability.name == candidate_capability.name: return capability.id return self.store.create_capability( repository_id, approved_ability_id, name=candidate_capability.name, description=candidate_capability.description, inputs=candidate_capability.inputs, outputs=candidate_capability.outputs, confidence=candidate_capability.confidence, primary_class=candidate_capability.primary_class, attributes=candidate_capability.attributes, ) def _candidate_capability_with_parent( self, graph: CandidateGraph, candidate_capability_id: int, ) -> tuple[CandidateAbility, CandidateCapability]: for ability in graph.abilities: for capability in ability.capabilities: if capability.id == candidate_capability_id: 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, candidate_feature_id: int, ) -> tuple[CandidateAbility, CandidateCapability, CandidateFeature]: for ability in graph.abilities: for capability in ability.capabilities: for feature in capability.features: if feature.id == candidate_feature_id: 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, analysis_run_id: int, action: str, notes: str, ) -> None: self.store.create_review_decision( repository_id, analysis_run_id, action=action, notes=notes, ) self.store.update_repository_status(repository_id, "indexed") def reject_candidate_ability( self, repository_id: int, analysis_run_id: int, candidate_ability_id: int, *, notes: str = "", ) -> CandidateGraph: self.store.reject_candidate_ability( repository_id, analysis_run_id, candidate_ability_id, ) self.store.create_review_decision( repository_id, analysis_run_id, action="reject_candidate_ability", notes=notes, ) self.store.update_repository_status(repository_id, "reviewing") return self.store.get_candidate_graph(repository_id, analysis_run_id) def reject_candidate_capability( self, repository_id: int, analysis_run_id: int, candidate_capability_id: int, *, notes: str = "", ) -> CandidateGraph: self.store.reject_candidate_capability( repository_id, analysis_run_id, candidate_capability_id, ) self.store.create_review_decision( repository_id, analysis_run_id, action="reject_candidate_capability", notes=notes, ) self.store.update_repository_status(repository_id, "reviewing") return self.store.get_candidate_graph(repository_id, analysis_run_id) def reject_candidate_feature( self, repository_id: int, analysis_run_id: int, candidate_feature_id: int, *, notes: str = "", ) -> CandidateGraph: self.store.reject_candidate_feature( repository_id, analysis_run_id, candidate_feature_id, ) self.store.create_review_decision( repository_id, analysis_run_id, action="reject_candidate_feature", notes=notes, ) self.store.update_repository_status(repository_id, "reviewing") return self.store.get_candidate_graph(repository_id, analysis_run_id) def reject_candidate_evidence( self, repository_id: int, analysis_run_id: int, candidate_evidence_id: int, *, notes: str = "", ) -> CandidateGraph: self.store.reject_candidate_evidence( repository_id, analysis_run_id, candidate_evidence_id, ) self.store.create_review_decision( repository_id, analysis_run_id, action="reject_candidate_evidence", notes=notes, ) self.store.update_repository_status(repository_id, "reviewing") return self.store.get_candidate_graph(repository_id, analysis_run_id) def edit_candidate_ability( self, repository_id: int, analysis_run_id: int, candidate_ability_id: int, *, name: str, description: str, confidence: float, primary_class: str = "ability", attributes: Sequence[str] = (), notes: str = "", ) -> CandidateGraph: self.store.update_candidate_ability( repository_id, analysis_run_id, candidate_ability_id, name=name, description=description, confidence=confidence, primary_class=primary_class, attributes=list(attributes), ) self.store.create_review_decision( repository_id, analysis_run_id, action="edit_candidate_ability", notes=notes, ) self.store.update_repository_status(repository_id, "reviewing") return self.store.get_candidate_graph(repository_id, analysis_run_id) def edit_candidate_capability( self, repository_id: int, analysis_run_id: int, candidate_capability_id: int, *, name: str, description: str, confidence: float, primary_class: str = "capability", attributes: Sequence[str] = (), notes: str = "", ) -> CandidateGraph: self.store.update_candidate_capability( repository_id, analysis_run_id, candidate_capability_id, name=name, description=description, confidence=confidence, primary_class=primary_class, attributes=list(attributes), ) self.store.create_review_decision( repository_id, analysis_run_id, action="edit_candidate_capability", notes=notes, ) self.store.update_repository_status(repository_id, "reviewing") return self.store.get_candidate_graph(repository_id, analysis_run_id) def edit_candidate_feature( self, repository_id: int, analysis_run_id: int, candidate_feature_id: int, *, name: str, type: str, location: str, confidence: float, primary_class: str | None = None, attributes: Sequence[str] = (), notes: str = "", ) -> CandidateGraph: self.store.update_candidate_feature( repository_id, analysis_run_id, candidate_feature_id, name=name, type=type, location=location, confidence=confidence, primary_class=primary_class, attributes=list(attributes), ) self.store.create_review_decision( repository_id, analysis_run_id, action="edit_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_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 merge_candidate_ability( self, repository_id: int, analysis_run_id: int, source_ability_id: int, *, target_ability_id: int, notes: str = "", ) -> CandidateGraph: self.store.merge_candidate_ability( repository_id, analysis_run_id, source_ability_id, target_ability_id, ) self.store.create_review_decision( repository_id, analysis_run_id, action="merge_candidate_ability", notes=notes, ) self.store.update_repository_status(repository_id, "reviewing") return self.store.get_candidate_graph(repository_id, analysis_run_id) def merge_candidate_capability( self, repository_id: int, analysis_run_id: int, source_capability_id: int, *, target_capability_id: int, notes: str = "", ) -> CandidateGraph: self.store.merge_candidate_capability( repository_id, analysis_run_id, source_capability_id, target_capability_id, ) self.store.create_review_decision( repository_id, analysis_run_id, action="merge_candidate_capability", notes=notes, ) self.store.update_repository_status(repository_id, "reviewing") return self.store.get_candidate_graph(repository_id, analysis_run_id) def merge_candidate_feature( self, repository_id: int, analysis_run_id: int, source_feature_id: int, *, target_feature_id: int, notes: str = "", ) -> CandidateGraph: self.store.merge_candidate_feature( repository_id, analysis_run_id, source_feature_id, target_feature_id, ) self.store.create_review_decision( repository_id, analysis_run_id, action="merge_candidate_feature", notes=notes, ) self.store.update_repository_status(repository_id, "reviewing") return self.store.get_candidate_graph(repository_id, analysis_run_id) def merge_candidate_evidence( self, repository_id: int, analysis_run_id: int, source_evidence_id: int, *, target_evidence_id: int, notes: str = "", ) -> CandidateGraph: self.store.merge_candidate_evidence( repository_id, analysis_run_id, source_evidence_id, target_evidence_id, ) self.store.create_review_decision( repository_id, analysis_run_id, action="merge_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, *, name: str, description: str = "", confidence: float = 1.0, primary_class: str = "ability", attributes: Sequence[str] = (), ) -> int: self.store.get_repository(repository_id) return self.store.create_ability( repository_id, name=name, description=description, confidence=confidence, primary_class=primary_class, attributes=list(attributes), ) def update_ability( self, repository_id: int, ability_id: int, *, name: str | None = None, description: str | None = None, confidence: float | None = None, primary_class: str | None = None, attributes: Sequence[str] | None = None, ) -> RepositoryAbilityMap: self.store.update_ability( repository_id, ability_id, name=name, description=description, confidence=confidence, primary_class=primary_class, attributes=list(attributes) if attributes is not None else None, ) return self.store.get_ability_map(repository_id) def delete_ability( self, repository_id: int, ability_id: int, ) -> RepositoryAbilityMap: self.store.delete_ability(repository_id, ability_id) return self.store.get_ability_map(repository_id) def add_capability( self, repository_id: int, ability_id: int, *, name: str, description: str = "", inputs: Sequence[str] = (), outputs: Sequence[str] = (), confidence: float = 1.0, primary_class: str = "capability", attributes: Sequence[str] = (), ) -> int: self.store.ensure_ability(repository_id, ability_id) return self.store.create_capability( repository_id, ability_id, name=name, description=description, inputs=list(inputs), outputs=list(outputs), confidence=confidence, primary_class=primary_class, attributes=list(attributes), ) def update_capability( self, repository_id: int, capability_id: int, *, name: str | None = None, description: str | None = None, inputs: Sequence[str] | None = None, outputs: Sequence[str] | None = None, confidence: float | None = None, primary_class: str | None = None, attributes: Sequence[str] | None = None, ) -> RepositoryAbilityMap: self.store.update_capability( repository_id, capability_id, name=name, description=description, inputs=list(inputs) if inputs is not None else None, outputs=list(outputs) if outputs is not None else None, confidence=confidence, primary_class=primary_class, attributes=list(attributes) if attributes is not None else None, ) return self.store.get_ability_map(repository_id) def delete_capability( self, repository_id: int, capability_id: int, ) -> RepositoryAbilityMap: self.store.delete_capability(repository_id, capability_id) return self.store.get_ability_map(repository_id) def add_feature( self, repository_id: int, capability_id: int, *, name: str, type: str, location: str = "", confidence: float = 1.0, primary_class: str | None = None, attributes: Sequence[str] = (), ) -> int: self.store.ensure_capability(repository_id, capability_id) return self.store.create_feature( repository_id, capability_id, name=name, type=type, location=location, confidence=confidence, primary_class=primary_class, attributes=list(attributes), ) def update_feature( self, repository_id: int, feature_id: int, *, name: str | None = None, type: str | None = None, location: str | None = None, confidence: float | None = None, primary_class: str | None = None, attributes: Sequence[str] | None = None, ) -> RepositoryAbilityMap: self.store.update_feature( repository_id, feature_id, name=name, type=type, location=location, confidence=confidence, primary_class=primary_class, attributes=list(attributes) if attributes is not None else None, ) return self.store.get_ability_map(repository_id) def delete_feature( self, repository_id: int, feature_id: int, ) -> RepositoryAbilityMap: self.store.delete_feature(repository_id, feature_id) return self.store.get_ability_map(repository_id) def add_evidence( self, repository_id: int, capability_id: int, *, type: str, reference: str, strength: str = "medium", target_kind: str = "capability", target_id: int | None = None, reference_kind: str = "source", reference_id: int | None = None, ) -> int: self.store.ensure_capability(repository_id, capability_id) return self.store.create_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, ) def update_evidence( self, repository_id: int, evidence_id: int, *, type: str | None = None, reference: str | None = None, strength: str | None = None, target_kind: str | None = None, target_id: int | None = None, reference_kind: str | None = None, reference_id: int | None = None, ) -> RepositoryAbilityMap: self.store.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 self.store.get_ability_map(repository_id) def delete_evidence( self, repository_id: int, evidence_id: int, ) -> RepositoryAbilityMap: self.store.delete_evidence(repository_id, evidence_id) return self.store.get_ability_map(repository_id) def update_scope( self, repository_id: int, *, name: str | None = None, description: str | None = None, confidence: float | None = None, ) -> RepositoryAbilityMap: self.store.update_scope( repository_id, name=name, description=description, confidence=confidence, ) return self.store.get_ability_map(repository_id) def ability_map(self, repository_id: int) -> RepositoryAbilityMap: return self.store.get_ability_map(repository_id) def compare_repositories(self, repository_ids: Sequence[int]) -> dict[str, object]: maps = [self.store.get_ability_map(repository_id) for repository_id in repository_ids] ability_groups: dict[str, list[dict[str, object]]] = {} capability_groups: dict[str, list[dict[str, object]]] = {} for ability_map in maps: repository = ability_map.repository for ability in ability_map.abilities: ability_groups.setdefault(ability.name.lower(), []).append( { "repository_id": repository.id, "repository_name": repository.name, "confidence": ability.confidence, "confidence_label": ability.confidence_label, "capabilities": [ { "name": capability.name, "confidence": capability.confidence, "confidence_label": capability.confidence_label, "evidence_count": len(capability.evidence), } for capability in ability.capabilities ], "_name": ability.name, } ) for capability in ability.capabilities: capability_groups.setdefault(capability.name.lower(), []).append( { "repository_id": repository.id, "repository_name": repository.name, "ability_name": ability.name, "capability_name": capability.name, } ) abilities = [ { "name": repositories[0]["_name"], "repositories": [ { key: value for key, value in repository.items() if key != "_name" } for repository in repositories ], } for repositories in ability_groups.values() ] unique_capabilities = [ entries[0] for entries in capability_groups.values() if len({entry["repository_id"] for entry in entries}) == 1 ] return { "repositories": [asdict(ability_map.repository) for ability_map in maps], "abilities": sorted(abilities, key=lambda item: item["name"]), "unique_capabilities": sorted( unique_capabilities, key=lambda item: (item["repository_name"], item["capability_name"]), ), } def detect_capability_gaps( self, *, desired_ability: str, desired_capabilities: Sequence[str], repository_ids: Sequence[int] | None = None, ) -> dict[str, object]: repositories = ( [self.store.get_repository(repository_id) for repository_id in repository_ids] if repository_ids is not None else self.store.list_repositories() ) maps = [self.store.get_ability_map(repository.id) for repository in repositories] desired = [capability.strip() for capability in desired_capabilities if capability.strip()] capability_matches: dict[str, list[dict[str, object]]] = {name.lower(): [] for name in desired} duplicate_index: dict[str, set[str]] = {} weak: list[dict[str, object]] = [] for ability_map in maps: repository = ability_map.repository for ability in ability_map.abilities: for capability in ability.capabilities: key = capability.name.lower() duplicate_index.setdefault(key, set()).add(repository.name) if key in capability_matches: capability_matches[key].append( { "repository_id": repository.id, "repository_name": repository.name, "capability": capability, } ) strengths = {evidence.strength for evidence in capability.evidence} if "strong" not in strengths: weak.append( { "capability": capability.name, "repository_id": repository.id, "repository_name": repository.name, "evidence_count": len(capability.evidence), "strongest_evidence": self._strongest_evidence(strengths), "confidence": capability.confidence, "confidence_label": capability.confidence_label, } ) matched = [ { "capability": name, "repositories": [ match["repository_name"] for match in capability_matches[name.lower()] ], } for name in desired if capability_matches[name.lower()] ] missing = [name for name in desired if not capability_matches[name.lower()]] duplicates = [ { "capability": capability, "repositories": sorted(repositories), } for capability, repositories in duplicate_index.items() if len(repositories) > 1 and capability in capability_matches ] return { "desired_ability": desired_ability, "matched_capabilities": matched, "missing_capabilities": missing, "weakly_evidenced_capabilities": weak, "duplicate_capabilities": duplicates, } def export_registry_entry(self, repository_id: int) -> str: ability_map = self.store.get_ability_map(repository_id) lines = [ "repository:", f" name: {self._yaml_scalar(ability_map.repository.name)}", f" url: {self._yaml_scalar(ability_map.repository.url)}", f" branch: {self._yaml_scalar(ability_map.repository.branch)}", f" status: {self._yaml_scalar(ability_map.repository.status)}", "abilities:", ] for ability in ability_map.abilities: lines.extend( [ f" - name: {self._yaml_scalar(ability.name)}", f" description: {self._yaml_scalar(ability.description)}", f" confidence: {ability.confidence}", f" confidence_label: {self._yaml_scalar(ability.confidence_label)}", " capabilities:", ] ) for capability in ability.capabilities: lines.extend( [ f" - name: {self._yaml_scalar(capability.name)}", f" description: {self._yaml_scalar(capability.description)}", f" confidence: {capability.confidence}", f" confidence_label: {self._yaml_scalar(capability.confidence_label)}", f" inputs: {self._yaml_list(capability.inputs)}", f" outputs: {self._yaml_list(capability.outputs)}", " features:", ] ) for feature in capability.features: lines.extend( [ f" - name: {self._yaml_scalar(feature.name)}", f" type: {self._yaml_scalar(feature.type)}", f" location: {self._yaml_scalar(feature.location)}", f" confidence: {feature.confidence}", f" confidence_label: {self._yaml_scalar(feature.confidence_label)}", ] ) lines.append(" evidence:") for evidence in capability.evidence: lines.extend( [ f" - type: {self._yaml_scalar(evidence.type)}", f" reference: {self._yaml_scalar(evidence.reference)}", f" strength: {self._yaml_scalar(evidence.strength)}", ] ) return "\n".join(lines) + "\n" def _strongest_evidence(self, strengths: set[str]) -> str | None: for strength in ("strong", "medium", "weak"): if strength in strengths: return strength return None def _diff_items( self, base: dict[str, dict[str, object]], target: dict[str, dict[str, object]], ) -> AnalysisRunDiffSection: added: list[AnalysisRunDiffItem] = [] removed: list[AnalysisRunDiffItem] = [] changed: list[AnalysisRunDiffItem] = [] weakened: list[AnalysisRunDiffItem] = [] for key in sorted(target.keys() - base.keys()): added.append( AnalysisRunDiffItem( change_type="added", item_type=str(target[key]["item_type"]), key=key, target=target[key], ) ) for key in sorted(base.keys() - target.keys()): removed.append( AnalysisRunDiffItem( change_type="removed", item_type=str(base[key]["item_type"]), key=key, base=base[key], ) ) for key in sorted(base.keys() & target.keys()): if base[key] == target[key]: continue item = AnalysisRunDiffItem( change_type="weakened" if self._is_weakened(base[key], target[key]) else "changed", item_type=str(target[key]["item_type"]), key=key, base=base[key], target=target[key], ) if item.change_type == "weakened": weakened.append(item) else: changed.append(item) return AnalysisRunDiffSection( added=added, removed=removed, changed=changed, weakened=weakened, ) def _is_weakened( self, base: dict[str, object], target: dict[str, object], ) -> bool: base_confidence = base.get("confidence") target_confidence = target.get("confidence") if ( isinstance(base_confidence, int | float) and isinstance(target_confidence, int | float) and target_confidence < base_confidence ): return True base_strength = base.get("strength") target_strength = target.get("strength") strength_order = {"weak": 1, "medium": 2, "strong": 3} return ( isinstance(base_strength, str) and isinstance(target_strength, str) and strength_order.get(target_strength, 0) < strength_order.get(base_strength, 0) ) def _fact_index(self, facts: Sequence[ObservedFact]) -> dict[str, dict[str, object]]: return { f"fact:{fact.kind}:{fact.path}:{fact.name}": { "item_type": "fact", "id": fact.id, "kind": fact.kind, "path": fact.path, "name": fact.name, "value": fact.value, "metadata": fact.metadata, } for fact in facts } def _dependency_characteristic_index( self, ability_map: RepositoryAbilityMap, ) -> dict[str, dict[str, object]]: index: dict[str, dict[str, object]] = { self._dependency_key("scope", ability_map.scope.id): { "name": ability_map.scope.name, "description": ability_map.scope.description, "primaryClass": "scope", "attributes": ["scope"], "confidence": ability_map.scope.confidence, "sourceReferences": [], } } for ability in ability_map.abilities: index[self._dependency_key("ability", ability.id)] = { "name": ability.name, "description": ability.description, "primaryClass": ability.primary_class, "attributes": ability.attributes, "confidence": ability.confidence, "sourceReferences": [], } for capability in ability.capabilities: index[self._dependency_key("capability", capability.id)] = { "name": capability.name, "description": capability.description, "primaryClass": capability.primary_class, "attributes": capability.attributes, "confidence": capability.confidence, "sourceReferences": [], } for feature in capability.features: index[self._dependency_key("feature", feature.id)] = { "name": feature.name, "description": feature.location, "primaryClass": feature.primary_class or feature.type, "attributes": feature.attributes, "confidence": feature.confidence, "path": feature.location, "sourceReferences": [ asdict(source_ref) for source_ref in feature.source_refs ], } for evidence in capability.evidence: index[self._dependency_key("evidence", evidence.id)] = { "name": evidence.reference, "description": evidence.type, "primaryClass": evidence.type, "attributes": [evidence.type, evidence.strength], "confidence": self._evidence_confidence(evidence.strength), "sourceReferences": [ asdict(source_ref) for source_ref in evidence.source_refs ], } return index def _dependency_fact_attributes(self, fact: ObservedFact) -> list[str]: attributes = [fact.kind] for key in ("source_role", "classification", "language", "framework"): value = fact.metadata.get(key) if isinstance(value, str) and value: attributes.append(value) return sorted(set(attributes)) def _dependency_display_edge( self, edge: DependencyEdge, facts_by_id: dict[int, ObservedFact], ) -> DependencyEdge | None: if edge.source_kind != "fact" or edge.source_id is None: return edge fact = facts_by_id.get(edge.source_id) if fact is None: return edge if self._suppress_dependency_fact(fact): return None display_key = self._dependency_fact_display_key(fact) if display_key == edge.source_key: return edge return replace(edge, source_key=display_key) def _suppress_dependency_fact(self, fact: ObservedFact) -> bool: return ( fact.path.lower().endswith("scope.md") and fact.metadata.get("source_role") == "derived_scope" ) def _dependency_fact_display_key(self, fact: ObservedFact) -> str: document_paths = {"readme.md", "scope.md"} if fact.path.lower() in document_paths and fact.kind in { "documentation", "intent", "scope", }: return f"fact:document:{fact.path}" return f"fact:{fact.kind}:{fact.path}:{fact.name}" def _dependency_node_size(self, confidence: object) -> int: if not isinstance(confidence, int | float): return 36 bounded = max(0.0, min(float(confidence), 1.0)) return int(28 + (bounded * 28)) def _dependency_edge_width(self, strength: str) -> int: return {"weak": 1, "medium": 3, "strong": 5}.get(strength, 2) def _dependency_layer(self, kind: str) -> str: if kind in {"fact", "evidence", "feature", "capability", "ability", "scope"}: return kind return "dependency" def _evaluate_dependency_visibility( self, elements: list[dict[str, object]], rules: list[dict[str, Any]], manual_overrides: dict[str, str], ) -> dict[str, dict[str, str]]: visibility: dict[str, dict[str, str]] = {} for element in elements: data = element["data"] element_id = str(data["id"]) state = "show" source = "default" reason = "default" for index, rule in enumerate(rules): action = str(rule.get("action", "show")) if action not in {"show", "blur", "hide"}: continue if self._dependency_rule_matches(data, rule): state = action source = "rule" reason = str(rule.get("name") or f"rule:{index}") override = manual_overrides.get(element_id) if override in {"show", "blur", "hide"}: state = override source = "manual" reason = "manual-override" visibility[element_id] = { "displayState": state, "visibilitySource": source, "visibilityReason": reason, } return visibility def _dependency_rule_matches( self, data: dict[str, Any], rule: dict[str, Any], ) -> bool: match = rule.get("match", rule) if not isinstance(match, dict): return False for key, expected in match.items(): if key in {"action", "name", "description"}: continue if key == "text": text = " ".join( str(data.get(part, "")) for part in ("label", "name", "description", "path", "value") ).lower() if str(expected).lower() not in text: return False continue if key == "path": if str(expected).lower() not in str(data.get("path", "")).lower(): return False continue actual = data.get(key) if key == "dependencyType": actual = data.get("dependencyType") elif key == "reviewState": actual = data.get("reviewState") elif key == "sameLayer": actual = bool(data.get("sameLayer")) elif key == "attributes": actual_values = set(data.get("attributes") or []) expected_values = expected if isinstance(expected, list) else [expected] if not set(expected_values).intersection(actual_values): return False continue elif key == "confidence": if actual is None: return False threshold = float(expected) if float(actual) < threshold: return False continue if isinstance(expected, list): if actual not in expected: return False elif actual != expected: return False return True def _validate_dependency_graph_profile_payload( self, default_mode: str, filter_rules: list[dict[str, Any]], manual_overrides: dict[str, str], ) -> None: if default_mode not in {"full", "impact", "path"}: raise ValueError("default_mode must be one of full, impact, or path") for rule in filter_rules: action = rule.get("action") if action not in {"show", "blur", "hide"}: raise ValueError("filter rule action must be show, blur, or hide") invalid = { key: value for key, value in manual_overrides.items() if value not in {"show", "blur", "hide"} } if invalid: raise ValueError("manual override values must be show, blur, or hide") def _evidence_confidence(self, strength: str) -> float: return {"strong": 0.9, "medium": 0.6, "weak": 0.3}.get(strength, 0.5) def _capability_dependency_edges( self, capability, *, capability_key: str, ) -> list[DependencyEdge]: edges: list[DependencyEdge] = [] for feature in capability.features: feature_key = self._dependency_key("feature", feature.id) edges.append( self._dependency_edge( source_kind="feature", source_id=feature.id, source_key=feature_key, target_kind="capability", target_id=capability.id, target_key=capability_key, dependency_type="supports", strength="medium", source="approved_characteristic", ) ) for source_ref in feature.source_refs: edges.append( self._dependency_edge( source_kind="fact", source_id=source_ref.fact_id, source_key=self._source_ref_fact_key(source_ref), target_kind="feature", target_id=feature.id, target_key=feature_key, dependency_type="observes", strength="strong", source="source_ref", ) ) for evidence in capability.evidence: evidence_key = self._dependency_key("evidence", evidence.id) evidence_target_kind = evidence.target_kind or "capability" evidence_target_id = evidence.target_id or capability.id edges.append( self._dependency_edge( source_kind="evidence", source_id=evidence.id, source_key=evidence_key, target_kind=evidence_target_kind, target_id=evidence_target_id, target_key=self._dependency_key( evidence_target_kind, evidence_target_id, ), dependency_type="supports", strength=evidence.strength or "medium", source="approved_characteristic", ) ) for source_ref in evidence.source_refs: edges.append( self._dependency_edge( source_kind="fact", source_id=source_ref.fact_id, source_key=self._source_ref_fact_key(source_ref), target_kind="evidence", target_id=evidence.id, target_key=evidence_key, dependency_type="observes", strength=evidence.strength or "medium", source="source_ref", ) ) if evidence.reference_kind in {"feature", "capability", "ability", "scope"}: reference_id = evidence.reference_id if reference_id is not None: edges.append( self._dependency_edge( source_kind=evidence.reference_kind, source_id=reference_id, source_key=self._dependency_key( evidence.reference_kind, reference_id, ), target_kind=evidence.target_kind, target_id=evidence.target_id or capability.id, target_key=self._dependency_key( evidence.target_kind, evidence.target_id or capability.id, ), dependency_type="relates", strength=evidence.strength or "medium", source="approved_evidence", ) ) return edges def _dependency_edge( self, *, source_kind: str, source_id: int | None, source_key: str, target_kind: str, target_id: int, target_key: str, dependency_type: str, strength: str, source: str, ) -> DependencyEdge: return DependencyEdge( source_kind=source_kind, source_id=source_id, source_key=source_key, target_kind=target_kind, target_id=target_id, target_key=target_key, dependency_type=dependency_type, strength=strength, source=source, target_ownership=self._ownership_for_kind(target_kind), same_layer=source_kind == target_kind, ) def _dependency_key(self, kind: str, item_id: int) -> str: return f"{kind}:{item_id}" def _source_ref_fact_key(self, source_ref) -> str: return f"fact:{source_ref.kind}:{source_ref.path}:{source_ref.name}" def _ownership_for_kind(self, kind: str) -> str: if kind == "fact": return "deterministic" if kind in {"evidence", "feature", "capability"}: return "mixed" return "curator_owned" def _recommended_action(self, ownership: str) -> str: if ownership == "deterministic": return "recalculate" return "review" def _dependency_display_name( self, repository_id: int, kind: str, item_id: int, ) -> str: ability_map = self.store.get_ability_map(repository_id) if kind == "scope" and ability_map.scope.id == item_id: return ability_map.scope.name for ability in ability_map.abilities: if kind == "ability" and ability.id == item_id: return ability.name for capability in ability.capabilities: if kind == "capability" and capability.id == item_id: return capability.name for feature in capability.features: if kind == "feature" and feature.id == item_id: return feature.name for evidence in capability.evidence: if kind == "evidence" and evidence.id == item_id: return evidence.reference return f"{kind}:{item_id}" def _dependency_node_label( self, repository_id: int, kind: str, key: str, item_id: int | None, ) -> str: if item_id is not None and kind != "fact": return self._dependency_display_name(repository_id, kind, item_id) if kind == "fact": parts = key.split(":", 3) if len(parts) == 4: _, fact_kind, path, name = parts return f"{name} ({fact_kind}, {path})" if item_id is not None: return f"{kind}:{item_id}" return key def _chunk_index( self, chunks: Sequence[ContentChunk], ) -> dict[str, dict[str, object]]: return { f"chunk:{chunk.kind}:{chunk.path}:{chunk.start_line}:{chunk.end_line}": { "item_type": "chunk", "kind": chunk.kind, "path": chunk.path, "start_line": chunk.start_line, "end_line": chunk.end_line, "text": chunk.text, } for chunk in chunks } def _candidate_index( self, abilities: Sequence[CandidateAbility], ) -> dict[str, dict[str, object]]: index: dict[str, dict[str, object]] = {} for ability in abilities: ability_key = self._entry_key("ability", ability.name) index[ability_key] = { "item_type": "ability", "name": ability.name, "description": ability.description, "confidence": ability.confidence, "status": ability.status, } for capability in ability.capabilities: capability_key = self._entry_key( "capability", ability.name, capability.name, ) index[capability_key] = { "item_type": "capability", "ability_name": ability.name, "name": capability.name, "description": capability.description, "inputs": capability.inputs, "outputs": capability.outputs, "confidence": capability.confidence, "status": capability.status, } self._index_candidate_leaves(index, ability, capability) return index def _index_candidate_leaves( self, index: dict[str, dict[str, object]], ability: CandidateAbility, capability: CandidateCapability, ) -> None: for feature in capability.features: key = self._entry_key( "feature", ability.name, capability.name, feature.name, feature.type, feature.location, ) index[key] = self._feature_payload( feature, ability_name=ability.name, capability_name=capability.name, ) for evidence in capability.evidence: key = self._entry_key( "evidence", ability.name, capability.name, evidence.type, evidence.reference, ) index[key] = self._evidence_payload( evidence, ability_name=ability.name, capability_name=capability.name, ) def _approved_index(self, abilities) -> dict[str, dict[str, object]]: index: dict[str, dict[str, object]] = {} for ability in abilities: ability_key = self._entry_key("ability", ability.name) index[ability_key] = { "item_type": "ability", "name": ability.name, "description": ability.description, "confidence": ability.confidence, } for capability in ability.capabilities: capability_key = self._entry_key( "capability", ability.name, capability.name, ) index[capability_key] = { "item_type": "capability", "ability_name": ability.name, "name": capability.name, "description": capability.description, "inputs": capability.inputs, "outputs": capability.outputs, "confidence": capability.confidence, } for feature in capability.features: key = self._entry_key( "feature", ability.name, capability.name, feature.name, feature.type, feature.location, ) index[key] = self._feature_payload( feature, ability_name=ability.name, capability_name=capability.name, ) for evidence in capability.evidence: key = self._entry_key( "evidence", ability.name, capability.name, evidence.type, evidence.reference, ) index[key] = self._evidence_payload( evidence, ability_name=ability.name, capability_name=capability.name, ) return index def _feature_payload( self, feature: CandidateFeature, *, ability_name: str, capability_name: str, ) -> dict[str, object]: return { "item_type": "feature", "ability_name": ability_name, "capability_name": capability_name, "name": feature.name, "type": feature.type, "location": feature.location, "confidence": feature.confidence, } def _evidence_payload( self, evidence: CandidateEvidence, *, ability_name: str, capability_name: str, ) -> dict[str, object]: return { "item_type": "evidence", "ability_name": ability_name, "capability_name": capability_name, "type": evidence.type, "reference": evidence.reference, "strength": evidence.strength, } def _entry_key(self, *parts: str) -> str: return ":".join(part.strip().lower() for part in parts) def _yaml_list(self, values: Sequence[str]) -> str: return "[" + ", ".join(self._yaml_scalar(value) for value in values) + "]" def _yaml_scalar(self, value: object) -> str: text = "" if value is None else str(value) escaped = text.replace("\\", "\\\\").replace('"', '\\"') return f'"{escaped}"' def search( self, query: str, *, status: str | None = None, language: str | None = None, framework: str | None = None, ability: str | None = None, capability: str | None = None, ) -> list[SearchResult]: text_results = self.store.search( query, status=status, language=language, framework=framework, ability=ability, capability=capability, ) if self.embedding_provider is None: return text_results return self._hybrid_search( query, text_results, status=status, language=language, framework=framework, ability=ability, capability=capability, ) def _hybrid_search( self, query: str, text_results: list[SearchResult], *, status: str | None, language: str | None, framework: str | None, ability: str | None, capability: str | None, ) -> list[SearchResult]: query_vector = self.embedding_provider.embed(query) candidates = self._semantic_candidates( status=status, language=language, framework=framework, ability=ability, capability=capability, ) by_key = { self._search_result_key(result): replace( result, text_score=max(result.text_score, 1.0), hybrid_score=max(result.hybrid_score, result.confidence), ) for result in text_results } for text, result in candidates: vector_score = max( 0.0, cosine_similarity(query_vector, self.embedding_provider.embed(text)), ) if vector_score < 0.18: continue text_match = by_key.get(self._search_result_key(result)) text_score = text_match.text_score if text_match is not None else 0.0 hybrid_score = ( 0.55 * text_score + 0.35 * vector_score + 0.10 * result.confidence ) ranked = replace( text_match or result, vector_score=max(vector_score, (text_match or result).vector_score), text_score=text_score, hybrid_score=max(hybrid_score, (text_match or result).hybrid_score), matched_field=(text_match or result).matched_field or "semantic", ) by_key[self._search_result_key(ranked)] = ranked return sorted( by_key.values(), key=lambda result: ( -result.hybrid_score, -result.vector_score, -result.confidence, result.repository_name.lower(), result.match_type, result.match_name.lower(), ), ) def _semantic_candidates( self, *, status: str | None, language: str | None, framework: str | None, ability: str | None, capability: str | None, ) -> list[tuple[str, SearchResult]]: candidates: list[tuple[str, SearchResult]] = [] for repository in self.store.list_repositories(): if status and repository.status != status: continue facts = self.store.list_observed_facts(repository.id) if not self._repository_matches_observed_filter(facts, "language", language): continue if not self._repository_matches_observed_filter(facts, "framework", framework): continue ability_map = self.store.get_ability_map(repository.id) if not self._ability_map_matches_filter( ability_map, ability=ability, capability=capability, ): continue candidates.extend(self._approved_entry_candidates(ability_map)) candidates.extend(self._content_chunk_candidates(repository, ability_map)) return candidates def _approved_entry_candidates( self, ability_map: RepositoryAbilityMap, ) -> list[tuple[str, SearchResult]]: candidates: list[tuple[str, SearchResult]] = [] repository = ability_map.repository for ability in ability_map.abilities: ability_text = f"{ability.name} {ability.description}" candidates.append( ( ability_text, SearchResult( repository_id=repository.id, repository_name=repository.name, match_type="ability", match_name=ability.name, confidence=ability.confidence, confidence_label=ability.confidence_label, match_description=ability.description, matched_field="semantic", ability_id=ability.id, ability_name=ability.name, ), ) ) for capability in ability.capabilities: capability_text = " ".join( [ ability.name, capability.name, capability.description, " ".join(capability.inputs), " ".join(capability.outputs), ] ) candidates.append( ( capability_text, SearchResult( repository_id=repository.id, repository_name=repository.name, match_type="capability", match_name=capability.name, confidence=capability.confidence, confidence_label=capability.confidence_label, match_description=capability.description, matched_field="semantic", ability_id=ability.id, ability_name=ability.name, capability_id=capability.id, capability_name=capability.name, ), ) ) return candidates def _content_chunk_candidates( self, repository: Repository, ability_map: RepositoryAbilityMap, ) -> list[tuple[str, SearchResult]]: if not ability_map.abilities: return [] chunks = self.store.list_content_chunks(repository.id) candidates: list[tuple[str, SearchResult]] = [] for chunk in chunks: candidates.append( ( chunk.text, SearchResult( repository_id=repository.id, repository_name=repository.name, match_type="content_chunk", match_name=f"{chunk.path}:{chunk.start_line}-{chunk.end_line}", confidence=0.5, confidence_label="medium", match_description=chunk.text[:240], matched_field="semantic", source_reference=f"{chunk.path}:{chunk.start_line}", ), ) ) return candidates def _repository_matches_observed_filter( self, facts: Sequence[ObservedFact], kind: str, expected: str | None, ) -> bool: if not expected: return True expected_lower = expected.lower() return any( fact.kind == kind and expected_lower in fact.name.lower() for fact in facts ) def _ability_map_matches_filter( self, ability_map: RepositoryAbilityMap, *, ability: str | None, capability: str | None, ) -> bool: if not ability and not capability: return True ability_lower = ability.lower() if ability else None capability_lower = capability.lower() if capability else None for approved_ability in ability_map.abilities: ability_matches = ( ability_lower is None or ability_lower in approved_ability.name.lower() or ability_lower in approved_ability.description.lower() ) if not ability_matches: continue if capability_lower is None: return True for approved_capability in approved_ability.capabilities: if ( capability_lower in approved_capability.name.lower() or capability_lower in approved_capability.description.lower() ): return True return False def _search_result_key(self, result: SearchResult) -> tuple[object, ...]: return ( result.repository_id, result.match_type, result.ability_id, result.capability_id, result.match_name, result.source_reference, )