Milestone 0 plus the manual-registry spine from Milestone 1

README.md

@@ -1,3 +1,48 @@
-# repo-seed
+# Repository Ability Registry
 
-A git repository template to bootstrap coulomb projects from.
+The Repository Ability Registry maps repositories from usefulness to implementation:
+
+```text
+Ability -> Capability -> Feature -> Evidence -> Code location
+```
+
+The first implementation slice is a Python registry core plus FastAPI HTTP API for manual repository profiles. It deliberately separates the manual/canonical registry path from the later analyzer pipeline.
+
+## Local Development
+
+Create an environment and install dependencies:
+
+```bash
+python3 -m venv .venv
+. .venv/bin/activate
+python -m pip install -e ".[dev]"
+```
+
+Run tests:
+
+```bash
+pytest
+```
+
+Run the API:
+
+```bash
+uvicorn repo_registry.web_api.app:app --reload
+```
+
+The API creates a local SQLite database at `var/repo-registry.sqlite3` by default.
+
+## First API Loop
+
+```bash
+curl -X POST http://127.0.0.1:8000/repos \
+  -H 'content-type: application/json' \
+  -d '{"name":"MailRouter","url":"https://example.com/mail-router.git"}'
+```
+
+Then add abilities, capabilities, features, and evidence under that repository and inspect:
+
+```bash
+curl http://127.0.0.1:8000/repos/1/ability-map
+curl 'http://127.0.0.1:8000/search?q=classify'
+```

migrations/0001_initial.sql

@@ -0,0 +1,58 @@
+CREATE TABLE IF NOT EXISTS repositories (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  name TEXT NOT NULL,
+  url TEXT NOT NULL UNIQUE,
+  description TEXT,
+  branch TEXT NOT NULL DEFAULT 'main',
+  status TEXT NOT NULL DEFAULT 'registered',
+  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS approved_abilities (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  repository_id INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,
+  name TEXT NOT NULL,
+  description TEXT NOT NULL DEFAULT '',
+  confidence REAL NOT NULL DEFAULT 1.0,
+  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS approved_capabilities (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  repository_id INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,
+  ability_id INTEGER NOT NULL REFERENCES approved_abilities(id) ON DELETE CASCADE,
+  name TEXT NOT NULL,
+  description TEXT NOT NULL DEFAULT '',
+  inputs TEXT NOT NULL DEFAULT '[]',
+  outputs TEXT NOT NULL DEFAULT '[]',
+  confidence REAL NOT NULL DEFAULT 1.0,
+  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS approved_features (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  repository_id INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,
+  capability_id INTEGER NOT NULL REFERENCES approved_capabilities(id) ON DELETE CASCADE,
+  name TEXT NOT NULL,
+  type TEXT NOT NULL,
+  location TEXT NOT NULL DEFAULT '',
+  confidence REAL NOT NULL DEFAULT 1.0,
+  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS approved_evidence (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  repository_id INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,
+  capability_id INTEGER NOT NULL REFERENCES approved_capabilities(id) ON DELETE CASCADE,
+  type TEXT NOT NULL,
+  reference TEXT NOT NULL,
+  strength TEXT NOT NULL DEFAULT 'medium',
+  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_repositories_status ON repositories(status);
+CREATE INDEX IF NOT EXISTS idx_abilities_repository ON approved_abilities(repository_id);
+CREATE INDEX IF NOT EXISTS idx_capabilities_repository ON approved_capabilities(repository_id);
+CREATE INDEX IF NOT EXISTS idx_features_repository ON approved_features(repository_id);
+CREATE INDEX IF NOT EXISTS idx_evidence_repository ON approved_evidence(repository_id);

pyproject.toml

@@ -0,0 +1,28 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "repo-registry"
+version = "0.1.0"
+description = "Repository Ability Registry"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+  "fastapi>=0.115",
+  "uvicorn[standard]>=0.30",
+  "pydantic-settings>=2.4",
+]
+
+[project.optional-dependencies]
+dev = [
+  "httpx>=0.28",
+  "pytest>=7.4",
+]
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["src"]

src/repo_registry/__init__.py

@@ -0,0 +1,5 @@
+"""Repository Ability Registry."""
+
+__all__ = ["__version__"]
+
+__version__ = "0.1.0"

src/repo_registry/core/__init__.py

@@ -0,0 +1 @@
+"""Core registry domain objects and services."""

src/repo_registry/core/models.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True)
+class Repository:
+    id: int
+    name: str
+    url: str
+    description: str | None
+    branch: str
+    status: str
+
+
+@dataclass(frozen=True)
+class Evidence:
+    id: int
+    type: str
+    reference: str
+    strength: str
+
+
+@dataclass(frozen=True)
+class Feature:
+    id: int
+    name: str
+    type: str
+    location: str
+    confidence: float
+
+
+@dataclass(frozen=True)
+class Capability:
+    id: int
+    name: str
+    description: str
+    inputs: list[str]
+    outputs: list[str]
+    confidence: float
+    features: list[Feature] = field(default_factory=list)
+    evidence: list[Evidence] = field(default_factory=list)
+
+
+@dataclass(frozen=True)
+class Ability:
+    id: int
+    name: str
+    description: str
+    confidence: float
+    capabilities: list[Capability] = field(default_factory=list)
+
+
+@dataclass(frozen=True)
+class RepositoryAbilityMap:
+    repository: Repository
+    abilities: list[Ability]
+
+
+@dataclass(frozen=True)
+class SearchResult:
+    repository_id: int
+    repository_name: str
+    match_type: str
+    match_name: str
+    confidence: float

src/repo_registry/core/service.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from repo_registry.core.models import Repository, RepositoryAbilityMap, SearchResult
+from repo_registry.storage.sqlite import RegistryStore
+
+
+class RegistryService:
+    """Application service for the manual registry MVP."""
+
+    def __init__(self, store: RegistryStore) -> None:
+        self.store = store
+
+    def register_repository(
+        self,
+        *,
+        name: str,
+        url: str,
+        description: str | None = None,
+        branch: str = "main",
+    ) -> Repository:
+        return self.store.create_repository(
+            name=name,
+            url=url,
+            description=description,
+            branch=branch,
+        )
+
+    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 add_ability(
+        self,
+        repository_id: int,
+        *,
+        name: str,
+        description: str = "",
+        confidence: float = 1.0,
+    ) -> int:
+        self.store.get_repository(repository_id)
+        return self.store.create_ability(
+            repository_id,
+            name=name,
+            description=description,
+            confidence=confidence,
+        )
+
+    def add_capability(
+        self,
+        repository_id: int,
+        ability_id: int,
+        *,
+        name: str,
+        description: str = "",
+        inputs: Sequence[str] = (),
+        outputs: Sequence[str] = (),
+        confidence: float = 1.0,
+    ) -> 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,
+        )
+
+    def add_feature(
+        self,
+        repository_id: int,
+        capability_id: int,
+        *,
+        name: str,
+        type: str,
+        location: str = "",
+        confidence: float = 1.0,
+    ) -> 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,
+        )
+
+    def add_evidence(
+        self,
+        repository_id: int,
+        capability_id: int,
+        *,
+        type: str,
+        reference: str,
+        strength: str = "medium",
+    ) -> int:
+        self.store.ensure_capability(repository_id, capability_id)
+        return self.store.create_evidence(
+            repository_id,
+            capability_id,
+            type=type,
+            reference=reference,
+            strength=strength,
+        )
+
+    def ability_map(self, repository_id: int) -> RepositoryAbilityMap:
+        return self.store.get_ability_map(repository_id)
+
+    def search(self, query: str) -> list[SearchResult]:
+        return self.store.search(query)

src/repo_registry/storage/__init__.py

@@ -0,0 +1 @@
+"""Persistence adapters."""

src/repo_registry/storage/sqlite.py

@@ -0,0 +1,338 @@
+from __future__ import annotations
+
+import json
+import sqlite3
+from pathlib import Path
+from typing import Any
+
+from repo_registry.core.models import (
+    Ability,
+    Capability,
+    Evidence,
+    Feature,
+    Repository,
+    RepositoryAbilityMap,
+    SearchResult,
+)
+
+
+class NotFoundError(ValueError):
+    pass
+
+
+class RegistryStore:
+    def __init__(self, database_path: str | Path) -> None:
+        self.database_path = str(database_path)
+
+    def initialize(self) -> None:
+        migration_path = Path(__file__).parents[3] / "migrations" / "0001_initial.sql"
+        with self.connect() as connection:
+            connection.executescript(migration_path.read_text(encoding="utf-8"))
+
+    def connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.database_path)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA foreign_keys = ON")
+        return connection
+
+    def create_repository(
+        self,
+        *,
+        name: str,
+        url: str,
+        description: str | None,
+        branch: str,
+    ) -> Repository:
+        with self.connect() as connection:
+            cursor = connection.execute(
+                """
+                INSERT INTO repositories (name, url, description, branch)
+                VALUES (?, ?, ?, ?)
+                """,
+                (name, url, description, branch),
+            )
+            repository_id = int(cursor.lastrowid)
+        return self.get_repository(repository_id)
+
+    def list_repositories(self) -> list[Repository]:
+        with self.connect() as connection:
+            rows = connection.execute(
+                """
+                SELECT id, name, url, description, branch, status
+                FROM repositories
+                ORDER BY created_at DESC, id DESC
+                """
+            ).fetchall()
+        return [self._repository_from_row(row) for row in rows]
+
+    def get_repository(self, repository_id: int) -> Repository:
+        with self.connect() as connection:
+            row = connection.execute(
+                """
+                SELECT id, name, url, description, branch, status
+                FROM repositories
+                WHERE id = ?
+                """,
+                (repository_id,),
+            ).fetchone()
+        if row is None:
+            raise NotFoundError(f"repository {repository_id} was not found")
+        return self._repository_from_row(row)
+
+    def create_ability(
+        self,
+        repository_id: int,
+        *,
+        name: str,
+        description: str,
+        confidence: float,
+    ) -> int:
+        with self.connect() as connection:
+            cursor = connection.execute(
+                """
+                INSERT INTO approved_abilities
+                  (repository_id, name, description, confidence)
+                VALUES (?, ?, ?, ?)
+                """,
+                (repository_id, name, description, confidence),
+            )
+            return int(cursor.lastrowid)
+
+    def ensure_ability(self, repository_id: int, ability_id: int) -> None:
+        with self.connect() as connection:
+            row = connection.execute(
+                """
+                SELECT id FROM approved_abilities
+                WHERE id = ? AND repository_id = ?
+                """,
+                (ability_id, repository_id),
+            ).fetchone()
+        if row is None:
+            raise NotFoundError(
+                f"ability {ability_id} was not found for repository {repository_id}"
+            )
+
+    def create_capability(
+        self,
+        repository_id: int,
+        ability_id: int,
+        *,
+        name: str,
+        description: str,
+        inputs: list[str],
+        outputs: list[str],
+        confidence: float,
+    ) -> int:
+        with self.connect() as connection:
+            cursor = connection.execute(
+                """
+                INSERT INTO approved_capabilities
+                  (repository_id, ability_id, name, description, inputs, outputs, confidence)
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    repository_id,
+                    ability_id,
+                    name,
+                    description,
+                    json.dumps(inputs),
+                    json.dumps(outputs),
+                    confidence,
+                ),
+            )
+            return int(cursor.lastrowid)
+
+    def ensure_capability(self, repository_id: int, capability_id: int) -> None:
+        with self.connect() as connection:
+            row = connection.execute(
+                """
+                SELECT id FROM approved_capabilities
+                WHERE id = ? AND repository_id = ?
+                """,
+                (capability_id, repository_id),
+            ).fetchone()
+        if row is None:
+            raise NotFoundError(
+                f"capability {capability_id} was not found for repository {repository_id}"
+            )
+
+    def create_feature(
+        self,
+        repository_id: int,
+        capability_id: int,
+        *,
+        name: str,
+        type: str,
+        location: str,
+        confidence: float,
+    ) -> int:
+        with self.connect() as connection:
+            cursor = connection.execute(
+                """
+                INSERT INTO approved_features
+                  (repository_id, capability_id, name, type, location, confidence)
+                VALUES (?, ?, ?, ?, ?, ?)
+                """,
+                (repository_id, capability_id, name, type, location, confidence),
+            )
+            return int(cursor.lastrowid)
+
+    def create_evidence(
+        self,
+        repository_id: int,
+        capability_id: int,
+        *,
+        type: str,
+        reference: str,
+        strength: str,
+    ) -> int:
+        with self.connect() as connection:
+            cursor = connection.execute(
+                """
+                INSERT INTO approved_evidence
+                  (repository_id, capability_id, type, reference, strength)
+                VALUES (?, ?, ?, ?, ?)
+                """,
+                (repository_id, capability_id, type, reference, strength),
+            )
+            return int(cursor.lastrowid)
+
+    def get_ability_map(self, repository_id: int) -> RepositoryAbilityMap:
+        repository = self.get_repository(repository_id)
+        with self.connect() as connection:
+            ability_rows = connection.execute(
+                """
+                SELECT id, name, description, confidence
+                FROM approved_abilities
+                WHERE repository_id = ?
+                ORDER BY id
+                """,
+                (repository_id,),
+            ).fetchall()
+            capability_rows = connection.execute(
+                """
+                SELECT id, ability_id, name, description, inputs, outputs, confidence
+                FROM approved_capabilities
+                WHERE repository_id = ?
+                ORDER BY id
+                """,
+                (repository_id,),
+            ).fetchall()
+            feature_rows = connection.execute(
+                """
+                SELECT id, capability_id, name, type, location, confidence
+                FROM approved_features
+                WHERE repository_id = ?
+                ORDER BY id
+                """,
+                (repository_id,),
+            ).fetchall()
+            evidence_rows = connection.execute(
+                """
+                SELECT id, capability_id, type, reference, strength
+                FROM approved_evidence
+                WHERE repository_id = ?
+                ORDER BY id
+                """,
+                (repository_id,),
+            ).fetchall()
+
+        features_by_capability: dict[int, list[Feature]] = {}
+        for row in feature_rows:
+            features_by_capability.setdefault(row["capability_id"], []).append(
+                Feature(
+                    id=row["id"],
+                    name=row["name"],
+                    type=row["type"],
+                    location=row["location"],
+                    confidence=row["confidence"],
+                )
+            )
+
+        evidence_by_capability: dict[int, list[Evidence]] = {}
+        for row in evidence_rows:
+            evidence_by_capability.setdefault(row["capability_id"], []).append(
+                Evidence(
+                    id=row["id"],
+                    type=row["type"],
+                    reference=row["reference"],
+                    strength=row["strength"],
+                )
+            )
+
+        capabilities_by_ability: dict[int, list[Capability]] = {}
+        for row in capability_rows:
+            capabilities_by_ability.setdefault(row["ability_id"], []).append(
+                Capability(
+                    id=row["id"],
+                    name=row["name"],
+                    description=row["description"],
+                    inputs=json.loads(row["inputs"]),
+                    outputs=json.loads(row["outputs"]),
+                    confidence=row["confidence"],
+                    features=features_by_capability.get(row["id"], []),
+                    evidence=evidence_by_capability.get(row["id"], []),
+                )
+            )
+
+        abilities = [
+            Ability(
+                id=row["id"],
+                name=row["name"],
+                description=row["description"],
+                confidence=row["confidence"],
+                capabilities=capabilities_by_ability.get(row["id"], []),
+            )
+            for row in ability_rows
+        ]
+        return RepositoryAbilityMap(repository=repository, abilities=abilities)
+
+    def search(self, query: str) -> list[SearchResult]:
+        needle = f"%{query.strip()}%"
+        if needle == "%%":
+            return []
+
+        with self.connect() as connection:
+            rows = connection.execute(
+                """
+                SELECT r.id AS repository_id, r.name AS repository_name,
+                       'repository' AS match_type, r.name AS match_name,
+                       1.0 AS confidence
+                FROM repositories r
+                WHERE r.name LIKE ? OR COALESCE(r.description, '') LIKE ?
+                UNION ALL
+                SELECT r.id, r.name, 'ability', a.name, a.confidence
+                FROM approved_abilities a
+                JOIN repositories r ON r.id = a.repository_id
+                WHERE a.name LIKE ? OR a.description LIKE ?
+                UNION ALL
+                SELECT r.id, r.name, 'capability', c.name, c.confidence
+                FROM approved_capabilities c
+                JOIN repositories r ON r.id = c.repository_id
+                WHERE c.name LIKE ? OR c.description LIKE ?
+                ORDER BY confidence DESC, repository_name ASC, match_name ASC
+                """,
+                (needle, needle, needle, needle, needle, needle),
+            ).fetchall()
+
+        return [
+            SearchResult(
+                repository_id=row["repository_id"],
+                repository_name=row["repository_name"],
+                match_type=row["match_type"],
+                match_name=row["match_name"],
+                confidence=row["confidence"],
+            )
+            for row in rows
+        ]
+
+    @staticmethod
+    def _repository_from_row(row: sqlite3.Row) -> Repository:
+        return Repository(
+            id=row["id"],
+            name=row["name"],
+            url=row["url"],
+            description=row["description"],
+            branch=row["branch"],
+            status=row["status"],
+        )

src/repo_registry/web_api/__init__.py

@@ -0,0 +1 @@
+"""HTTP API package."""

src/repo_registry/web_api/app.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+from dataclasses import asdict
+from pathlib import Path
+
+from fastapi import Depends, FastAPI, HTTPException
+from pydantic import BaseModel, Field
+
+from repo_registry.core.service import RegistryService
+from repo_registry.storage.sqlite import NotFoundError, RegistryStore
+
+
+class Settings(BaseModel):
+    database_path: str = Field(default="var/repo-registry.sqlite3")
+
+
+def get_settings() -> Settings:
+    return Settings()
+
+
+def get_service(settings: Settings = Depends(get_settings)) -> RegistryService:
+    database_path = Path(settings.database_path)
+    database_path.parent.mkdir(parents=True, exist_ok=True)
+    store = RegistryStore(database_path)
+    store.initialize()
+    return RegistryService(store)
+
+
+class RepositoryCreate(BaseModel):
+    name: str
+    url: str
+    description: str | None = None
+    branch: str = "main"
+
+
+class AbilityCreate(BaseModel):
+    name: str
+    description: str = ""
+    confidence: float = Field(default=1.0, ge=0.0, le=1.0)
+
+
+class CapabilityCreate(BaseModel):
+    ability_id: int
+    name: str
+    description: str = ""
+    inputs: list[str] = Field(default_factory=list)
+    outputs: list[str] = Field(default_factory=list)
+    confidence: float = Field(default=1.0, ge=0.0, le=1.0)
+
+
+class FeatureCreate(BaseModel):
+    capability_id: int
+    name: str
+    type: str
+    location: str = ""
+    confidence: float = Field(default=1.0, ge=0.0, le=1.0)
+
+
+class EvidenceCreate(BaseModel):
+    capability_id: int
+    type: str
+    reference: str
+    strength: str = "medium"
+
+
+app = FastAPI(title="Repository Ability Registry", version="0.1.0")
+
+
+@app.get("/health")
+def health() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.post("/repos", status_code=201)
+def create_repository(
+    payload: RepositoryCreate,
+    service: RegistryService = Depends(get_service),
+) -> dict[str, object]:
+    try:
+        repository = service.register_repository(**payload.model_dump())
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    return asdict(repository)
+
+
+@app.get("/repos")
+def list_repositories(
+    service: RegistryService = Depends(get_service),
+) -> list[dict[str, object]]:
+    return [asdict(repository) for repository in service.list_repositories()]
+
+
+@app.get("/repos/{repository_id}")
+def get_repository(
+    repository_id: int,
+    service: RegistryService = Depends(get_service),
+) -> dict[str, object]:
+    try:
+        return asdict(service.get_repository(repository_id))
+    except NotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+
+@app.post("/repos/{repository_id}/abilities", status_code=201)
+def create_ability(
+    repository_id: int,
+    payload: AbilityCreate,
+    service: RegistryService = Depends(get_service),
+) -> dict[str, int]:
+    try:
+        ability_id = service.add_ability(repository_id, **payload.model_dump())
+    except NotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    return {"id": ability_id}
+
+
+@app.post("/repos/{repository_id}/capabilities", status_code=201)
+def create_capability(
+    repository_id: int,
+    payload: CapabilityCreate,
+    service: RegistryService = Depends(get_service),
+) -> dict[str, int]:
+    try:
+        capability_id = service.add_capability(repository_id, **payload.model_dump())
+    except NotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    return {"id": capability_id}
+
+
+@app.post("/repos/{repository_id}/features", status_code=201)
+def create_feature(
+    repository_id: int,
+    payload: FeatureCreate,
+    service: RegistryService = Depends(get_service),
+) -> dict[str, int]:
+    try:
+        feature_id = service.add_feature(repository_id, **payload.model_dump())
+    except NotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    return {"id": feature_id}
+
+
+@app.post("/repos/{repository_id}/evidence", status_code=201)
+def create_evidence(
+    repository_id: int,
+    payload: EvidenceCreate,
+    service: RegistryService = Depends(get_service),
+) -> dict[str, int]:
+    try:
+        evidence_id = service.add_evidence(repository_id, **payload.model_dump())
+    except NotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    return {"id": evidence_id}
+
+
+@app.get("/repos/{repository_id}/ability-map")
+def get_ability_map(
+    repository_id: int,
+    service: RegistryService = Depends(get_service),
+) -> dict[str, object]:
+    try:
+        return asdict(service.ability_map(repository_id))
+    except NotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+
+@app.get("/search")
+def search(
+    q: str,
+    service: RegistryService = Depends(get_service),
+) -> list[dict[str, object]]:
+    return [asdict(result) for result in service.search(q)]

tests/test_registry_service.py

@@ -0,0 +1,98 @@
+from repo_registry.core.service import RegistryService
+from repo_registry.storage.sqlite import NotFoundError, RegistryStore
+
+
+def make_service(tmp_path):
+    store = RegistryStore(tmp_path / "registry.sqlite3")
+    store.initialize()
+    return RegistryService(store)
+
+
+def test_manual_registry_builds_ability_map(tmp_path):
+    service = make_service(tmp_path)
+
+    repository = service.register_repository(
+        name="MailRouter",
+        url="https://example.com/mail-router.git",
+        description="Routes incoming customer email",
+    )
+    ability_id = service.add_ability(
+        repository.id,
+        name="Business Email Routing",
+        description="Route inbound messages to the right department.",
+        confidence=0.92,
+    )
+    capability_id = service.add_capability(
+        repository.id,
+        ability_id,
+        name="Classify Incoming Email",
+        description="Classify messages into intent categories.",
+        inputs=["subject", "body"],
+        outputs=["intent", "confidence"],
+        confidence=0.88,
+    )
+    service.add_feature(
+        repository.id,
+        capability_id,
+        name="POST /api/classify-email",
+        type="REST endpoint",
+        location="src/routes/classify_email.py",
+        confidence=0.84,
+    )
+    service.add_evidence(
+        repository.id,
+        capability_id,
+        type="unit_test",
+        reference="tests/test_email_classification.py",
+        strength="strong",
+    )
+
+    ability_map = service.ability_map(repository.id)
+
+    assert ability_map.repository.name == "MailRouter"
+    assert ability_map.abilities[0].name == "Business Email Routing"
+    capability = ability_map.abilities[0].capabilities[0]
+    assert capability.name == "Classify Incoming Email"
+    assert capability.inputs == ["subject", "body"]
+    assert capability.features[0].location == "src/routes/classify_email.py"
+    assert capability.evidence[0].strength == "strong"
+
+
+def test_search_matches_approved_abilities_and_capabilities(tmp_path):
+    service = make_service(tmp_path)
+    repository = service.register_repository(
+        name="MailRouter",
+        url="https://example.com/mail-router.git",
+    )
+    ability_id = service.add_ability(
+        repository.id,
+        name="Business Email Routing",
+        description="Route inbound messages.",
+    )
+    service.add_capability(
+        repository.id,
+        ability_id,
+        name="Classify Incoming Email",
+        description="Classify messages into intent categories.",
+    )
+
+    results = service.search("classify")
+
+    assert len(results) == 1
+    assert results[0].repository_name == "MailRouter"
+    assert results[0].match_type == "capability"
+    assert results[0].match_name == "Classify Incoming Email"
+
+
+def test_capability_must_belong_to_repository(tmp_path):
+    service = make_service(tmp_path)
+    first = service.register_repository(name="First", url="https://example.com/first.git")
+    second = service.register_repository(name="Second", url="https://example.com/second.git")
+    ability_id = service.add_ability(first.id, name="Document Classification")
+
+    try:
+        service.add_capability(second.id, ability_id, name="Classify Document")
+    except NotFoundError as exc:
+        assert "ability" in str(exc)
+    else:
+        raise AssertionError("expected a NotFoundError")

tests/test_web_api.py

@@ -0,0 +1,69 @@
+from fastapi.testclient import TestClient
+
+from repo_registry.web_api.app import Settings, app, get_settings
+
+
+def test_api_manual_registry_loop(tmp_path):
+    def override_settings():
+        return Settings(database_path=str(tmp_path / "api.sqlite3"))
+
+    app.dependency_overrides[get_settings] = override_settings
+    client = TestClient(app)
+    try:
+        repository_response = client.post(
+            "/repos",
+            json={
+                "name": "MailRouter",
+                "url": "https://example.com/mail-router.git",
+                "description": "Routes incoming customer email",
+            },
+        )
+        assert repository_response.status_code == 201
+        repository_id = repository_response.json()["id"]
+
+        ability_response = client.post(
+            f"/repos/{repository_id}/abilities",
+            json={
+                "name": "Business Email Routing",
+                "description": "Route inbound messages.",
+            },
+        )
+        assert ability_response.status_code == 201
+        ability_id = ability_response.json()["id"]
+
+        capability_response = client.post(
+            f"/repos/{repository_id}/capabilities",
+            json={
+                "ability_id": ability_id,
+                "name": "Classify Incoming Email",
+                "inputs": ["subject", "body"],
+                "outputs": ["intent"],
+            },
+        )
+        assert capability_response.status_code == 201
+        capability_id = capability_response.json()["id"]
+
+        feature_response = client.post(
+            f"/repos/{repository_id}/features",
+            json={
+                "capability_id": capability_id,
+                "name": "POST /api/classify-email",
+                "type": "REST endpoint",
+                "location": "src/routes/classify_email.py",
+            },
+        )
+        assert feature_response.status_code == 201
+
+        map_response = client.get(f"/repos/{repository_id}/ability-map")
+        assert map_response.status_code == 200
+        ability_map = map_response.json()
+        assert ability_map["repository"]["name"] == "MailRouter"
+        assert ability_map["abilities"][0]["capabilities"][0]["name"] == (
+            "Classify Incoming Email"
+        )
+
+        search_response = client.get("/search", params={"q": "email"})
+        assert search_response.status_code == 200
+        assert search_response.json()
+    finally:
+        app.dependency_overrides.clear()