from __future__ import annotations import re from dataclasses import dataclass, field from repo_registry.core.models import ContentChunk, ObservedFact, Repository, SourceReference @dataclass(frozen=True) class CandidateEvidenceDraft: type: str reference: str strength: str source_refs: list[SourceReference] @dataclass(frozen=True) class CandidateFeatureDraft: name: str type: str location: str confidence: float source_refs: list[SourceReference] @dataclass(frozen=True) class CandidateCapabilityDraft: name: str description: str inputs: list[str] outputs: list[str] confidence: float source_refs: list[SourceReference] features: list[CandidateFeatureDraft] = field(default_factory=list) evidence: list[CandidateEvidenceDraft] = field(default_factory=list) @dataclass(frozen=True) class CandidateAbilityDraft: name: str description: str confidence: float source_refs: list[SourceReference] capabilities: list[CandidateCapabilityDraft] = field(default_factory=list) class CandidateGraphGenerator: """Build conservative review candidates from observed facts.""" def generate( self, repository: Repository, facts: list[ObservedFact], chunks: list[ContentChunk] | None = None, ) -> list[CandidateAbilityDraft]: if not facts: return [] chunks = chunks or [] docs = self._facts(facts, "documentation") tests = self._facts(facts, "test") examples = self._facts(facts, "example") interfaces = self._facts(facts, "interface") manifests = self._facts(facts, "manifest") frameworks = self._facts(facts, "framework") languages = self._facts(facts, "language") llm_providers = self._facts(facts, "llm_provider") credential_configs = self._facts(facts, "credential_config") provider_registries = self._facts(facts, "provider_registry") fallback_policies = self._facts(facts, "fallback_policy") ability_sources = docs or manifests or languages ability = CandidateAbilityDraft( name=self._ability_name(repository, chunks), description=self._ability_description(chunks), confidence=self._ability_confidence( docs=docs, interfaces=interfaces, tests=tests, examples=examples, frameworks=frameworks, languages=languages, ), source_refs=self._source_refs(ability_sources), capabilities=[], ) capabilities: list[CandidateCapabilityDraft] = [] if interfaces: capabilities.append( self._interface_capability(interfaces, tests, examples, docs, chunks) ) if llm_providers or provider_registries or fallback_policies: capabilities.append( self._llm_provider_capability( llm_providers, credential_configs, provider_registries, fallback_policies, tests, examples, docs, ) ) if manifests or frameworks or languages: capabilities.append( CandidateCapabilityDraft( name="Describe Repository Structure", description=( "Summarize detected languages, package manifests, and framework " "hints as structural context for review." ), inputs=[], outputs=["repository structure summary"], confidence=self._structure_confidence( manifests=manifests, frameworks=frameworks, languages=languages, docs=docs, ), source_refs=self._source_refs(manifests + frameworks + languages), evidence=self._evidence(tests, examples, docs), ) ) return [ CandidateAbilityDraft( name=ability.name, description=ability.description, confidence=ability.confidence, source_refs=ability.source_refs, capabilities=capabilities, ) ] def _interface_capability( self, interfaces: list[ObservedFact], tests: list[ObservedFact], examples: list[ObservedFact], docs: list[ObservedFact], chunks: list[ContentChunk], ) -> CandidateCapabilityDraft: features = self._interface_features(interfaces, chunks) return CandidateCapabilityDraft( name="Expose Repository Interface", description=self._interface_description(chunks), inputs=self._interface_inputs(interfaces), outputs=self._interface_outputs(interfaces), confidence=self._interface_confidence( interfaces=interfaces, tests=tests, examples=examples, docs=docs, ), source_refs=self._source_refs(interfaces), features=features, evidence=self._evidence(tests, examples, docs), ) def _llm_provider_capability( self, providers: list[ObservedFact], credentials: list[ObservedFact], registries: list[ObservedFact], fallback_policies: list[ObservedFact], tests: list[ObservedFact], examples: list[ObservedFact], docs: list[ObservedFact], ) -> CandidateCapabilityDraft: provider_names = sorted({fact.name for fact in providers}) provider_summary = ", ".join(provider_names) if provider_names else "LLM providers" features = [ CandidateFeatureDraft( name=f"Use {provider} Models", type="integration", location=self._grouped_location( [fact for fact in providers if fact.name == provider] ), confidence=0.75, source_refs=self._source_refs( [fact for fact in providers if fact.name == provider] ), ) for provider in provider_names ] if credentials: features.append( CandidateFeatureDraft( name="Configure LLM Provider Credentials", type="configuration", location=self._grouped_location(credentials), confidence=0.7, source_refs=self._source_refs(credentials), ) ) if registries: features.append( CandidateFeatureDraft( name="Maintain LLM Provider Registry", type="backend", location=self._grouped_location(registries), confidence=0.65, source_refs=self._source_refs(registries), ) ) if fallback_policies: features.append( CandidateFeatureDraft( name="Apply LLM Provider Fallback Policy", type="backend", location=self._grouped_location(fallback_policies), confidence=0.6, source_refs=self._source_refs(fallback_policies), ) ) return CandidateCapabilityDraft( name="Route LLM Requests Across Providers", description=( "Expose or configure model-provider integrations detected from " f"source-linked provider hints: {provider_summary}." ), inputs=["LLM request", "provider configuration"], outputs=["provider-specific model response"], confidence=self._llm_provider_confidence( providers=providers, credentials=credentials, registries=registries, fallback_policies=fallback_policies, docs=docs, ), source_refs=self._source_refs( providers + credentials + registries + fallback_policies ), features=features, evidence=self._evidence(tests, examples, docs), ) def _interface_features( self, interfaces: list[ObservedFact], chunks: list[ContentChunk], ) -> list[CandidateFeatureDraft]: by_type: dict[str, list[ObservedFact]] = {} for fact in interfaces: by_type.setdefault(self._feature_type(fact), []).append(fact) features: list[CandidateFeatureDraft] = [] for feature_type, facts in sorted(by_type.items()): if len(facts) == 1: fact = facts[0] features.append( CandidateFeatureDraft( name=self._feature_name(fact, chunks), type=feature_type, location=fact.path, confidence=0.65 if fact.value else 0.45, source_refs=self._source_refs([fact]), ) ) continue features.append( CandidateFeatureDraft( name=self._grouped_interface_feature_name( feature_type, facts, chunks, ), type=feature_type, location=self._grouped_location(facts), confidence=self._grouped_interface_confidence(facts), source_refs=self._source_refs(facts), ) ) return features def _grouped_interface_feature_name( self, feature_type: str, facts: list[ObservedFact], chunks: list[ContentChunk], ) -> str: summary = self._grouped_interface_summary(facts, chunks) if feature_type == "API": return f"HTTP API surface: {summary}" if feature_type == "CLI": return f"CLI command surface: {summary}" return f"Callable interface surface: {summary}" def _grouped_interface_summary( self, facts: list[ObservedFact], chunks: list[ContentChunk], ) -> str: names = [self._feature_name(fact, chunks) for fact in facts] compact_names = [name for name in names if name] if not compact_names: return f"{len(facts)} entry points" visible = compact_names[:3] suffix = f", +{len(compact_names) - 3} more" if len(compact_names) > 3 else "" return f"{', '.join(visible)}{suffix}" def _grouped_location(self, facts: list[ObservedFact]) -> str: paths = sorted({fact.path for fact in facts if fact.path}) if not paths: return "" if len(paths) == 1: return paths[0] return "multiple files" def _grouped_interface_confidence(self, facts: list[ObservedFact]) -> float: valued = sum(1 for fact in facts if fact.value) return 0.7 if valued == len(facts) else 0.55 def _evidence( self, tests: list[ObservedFact], examples: list[ObservedFact], docs: list[ObservedFact], ) -> list[CandidateEvidenceDraft]: evidence: list[CandidateEvidenceDraft] = [] for fact in tests: evidence.append( CandidateEvidenceDraft( type="test", reference=fact.path, strength="strong", source_refs=self._source_refs([fact]), ) ) for fact in examples: evidence.append( CandidateEvidenceDraft( type="example", reference=fact.path, strength="strong", source_refs=self._source_refs([fact]), ) ) for fact in docs: evidence.append( CandidateEvidenceDraft( type="documentation", reference=fact.path, strength="medium", source_refs=self._source_refs([fact]), ) ) return evidence def _feature_type(self, fact: ObservedFact) -> str: lower = f"{fact.name} {fact.path} {fact.value}".lower() if "cli" in lower or "command" in lower: return "CLI" if "api" in lower or "route" in lower or "@app." in lower or "@router." in lower: return "API" return "interface" def _interface_inputs(self, interfaces: list[ObservedFact]) -> list[str]: feature_types = {self._feature_type(fact) for fact in interfaces} inputs: list[str] = [] if "API" in feature_types: inputs.append("HTTP request") if "CLI" in feature_types: inputs.append("CLI arguments") if not inputs: inputs.append("caller input") return inputs def _interface_outputs(self, interfaces: list[ObservedFact]) -> list[str]: feature_types = {self._feature_type(fact) for fact in interfaces} outputs: list[str] = [] if "API" in feature_types: outputs.append("HTTP response") if "CLI" in feature_types: outputs.append("command output") if not outputs: outputs.append("callable interface result") return outputs def _feature_name(self, fact: ObservedFact, chunks: list[ContentChunk]) -> str: route_name = self._route_feature_name(fact.value) if route_name: return route_name if self._feature_type(fact) == "CLI": function_name = self._function_name_near_fact(fact, chunks) if function_name: return f"CLI command {function_name}" return fact.value or fact.name def _route_feature_name(self, value: str) -> str: match = re.search(r"@(?:app|router)\.(get|post|put|patch|delete)\((['\"])(.*?)\2", value) if match is None: return "" method = match.group(1).upper() path = match.group(3) return f"{method} {path}" def _function_name_near_fact( self, fact: ObservedFact, chunks: list[ContentChunk], ) -> str: line = fact.metadata.get("line") for chunk in chunks: if chunk.path != fact.path or chunk.kind != "interface": continue if isinstance(line, int) and not (chunk.start_line <= line <= chunk.end_line): continue match = re.search(r"^\s*def\s+([a-zA-Z_][a-zA-Z0-9_]*)\s*\(", chunk.text, re.MULTILINE) if match is not None: return match.group(1) return "" def _ability_confidence( self, *, docs: list[ObservedFact], interfaces: list[ObservedFact], tests: list[ObservedFact], examples: list[ObservedFact], frameworks: list[ObservedFact], languages: list[ObservedFact], ) -> float: return self._confidence( 0.25, [ (0.20, bool(docs)), (0.15, bool(interfaces)), (0.15, bool(tests)), (0.10, bool(examples)), (0.10, bool(frameworks)), (0.05, bool(languages)), ], ) def _interface_confidence( self, *, interfaces: list[ObservedFact], tests: list[ObservedFact], examples: list[ObservedFact], docs: list[ObservedFact], ) -> float: return self._confidence( 0.30, [ (0.20, bool(interfaces)), (0.15, bool(tests)), (0.10, bool(examples)), (0.10, bool(docs)), (0.05, len(interfaces) > 1), ], ) def _structure_confidence( self, *, manifests: list[ObservedFact], frameworks: list[ObservedFact], languages: list[ObservedFact], docs: list[ObservedFact], ) -> float: return self._confidence( 0.25, [ (0.20, bool(manifests)), (0.15, bool(frameworks)), (0.10, bool(languages)), (0.05, bool(docs)), ], ) def _llm_provider_confidence( self, *, providers: list[ObservedFact], credentials: list[ObservedFact], registries: list[ObservedFact], fallback_policies: list[ObservedFact], docs: list[ObservedFact], ) -> float: return self._confidence( 0.35, [ (0.20, bool(providers)), (0.10, len({fact.name for fact in providers}) > 1), (0.10, bool(credentials)), (0.10, bool(registries)), (0.10, bool(fallback_policies)), (0.05, bool(docs)), ], ) def _confidence( self, base: float, factors: list[tuple[float, bool]], ) -> float: score = base + sum(weight for weight, applies in factors if applies) return min(1.0, round(score, 2)) def _ability_description(self, chunks: list[ContentChunk]) -> str: doc_summary = self._document_summary(chunks) if doc_summary: return ( "Candidate repository purpose inferred from repository content: " f"{doc_summary} Review is required before treating this as an " "approved domain ability." ) return ( "Candidate repository purpose inferred from observed repository " "documentation, manifests, languages, and interfaces. Review is " "required before treating this as an approved domain ability." ) def _ability_name( self, repository: Repository, chunks: list[ContentChunk], ) -> str: purpose_text = self._document_purpose_sentence(chunks) or repository.description if purpose_text: normalized = self._imperative_purpose(purpose_text) if normalized: return normalized return f"Support {self._humanize_identifier(repository.name)}" def _document_purpose_sentence(self, chunks: list[ContentChunk]) -> str: for chunk in chunks: if chunk.kind != "documentation": continue lines = [line.strip() for line in chunk.text.splitlines() if line.strip()] paragraph = next((line for line in lines if not line.startswith("#")), "") if paragraph: return paragraph return "" def _imperative_purpose(self, text: str) -> str: cleaned = re.sub(r"\s+", " ", text.strip()) cleaned = re.split(r"[.!?]\s+", cleaned, maxsplit=1)[0] cleaned = re.sub(r"^[A-Z][A-Za-z0-9_-]*\s+(?:is|provides|offers)\s+", "", cleaned) cleaned = cleaned.strip(" .:-") if not cleaned: return "" words = cleaned.split() if not words: return "" words[0] = self._imperative_verb(words[0]) return self._title_from_words(words[:8]) def _imperative_verb(self, word: str) -> str: lower = word.lower().strip(",;:") irregular = { "does": "do", "has": "have", "is": "be", } if lower in irregular: return irregular[lower] if lower.endswith("ies") and len(lower) > 4: return f"{lower[:-3]}y" if lower.endswith(("des", "ses", "tes", "ves", "zes")) and len(lower) > 4: return lower[:-1] if lower.endswith("es") and len(lower) > 3: return lower[:-2] if lower.endswith("s") and len(lower) > 3: return lower[:-1] return lower def _title_from_words(self, words: list[str]) -> str: cleaned_words = [ re.sub(r"[^A-Za-z0-9_/{}-]", "", word) for word in words ] return " ".join( word[:1].upper() + word[1:] for word in cleaned_words if word ) def _humanize_identifier(self, value: str) -> str: spaced = re.sub(r"[_-]+", " ", value) spaced = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", " ", spaced) return self._title_from_words(spaced.split()) def _interface_description(self, chunks: list[ContentChunk]) -> str: interface_summary = self._interface_summary(chunks) if interface_summary: return ( "Expose one or more likely user-facing API or CLI entry points. " f"Source context: {interface_summary} Review is required to name " "the concrete domain behavior." ) return ( "Expose one or more likely user-facing API or CLI entry points. " "Review is required to name the concrete domain behavior." ) def _document_summary(self, chunks: list[ContentChunk]) -> str: for chunk in chunks: if chunk.kind != "documentation": continue lines = [line.strip() for line in chunk.text.splitlines() if line.strip()] if not lines: continue heading = next((line.lstrip("#").strip() for line in lines if line.startswith("#")), "") paragraph = next((line for line in lines if not line.startswith("#")), "") if heading and paragraph: return f"{heading}. {paragraph}" return heading or paragraph return "" def _interface_summary(self, chunks: list[ContentChunk]) -> str: for chunk in chunks: if chunk.kind != "interface": continue lines = [line.strip() for line in chunk.text.splitlines() if line.strip()] if not lines: continue return " ".join(lines[:3]) return "" def _facts(self, facts: list[ObservedFact], kind: str) -> list[ObservedFact]: return [fact for fact in facts if fact.kind == kind] def _source_refs(self, facts: list[ObservedFact]) -> list[SourceReference]: return [ SourceReference( fact_id=fact.id, path=fact.path, kind=fact.kind, name=fact.name, line=fact.metadata.get("line"), ) for fact in facts ]