cross-eval/cross_eval/discovery.py

"""Repository/service discovery helpers for autonomous execution prompts."""
from __future__ import annotations

import json
from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class RepoDiscovery:
    languages: set[str] = field(default_factory=set)
    package_managers: set[str] = field(default_factory=set)
    databases: set[str] = field(default_factory=set)
    services: set[str] = field(default_factory=set)
    hints: list[str] = field(default_factory=list)


def _read_text(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError):
        return ""


def _add_if_contains(target: set[str], content: str, mapping: dict[str, str]) -> None:
    lowered = content.lower()
    for needle, name in mapping.items():
        if needle in lowered:
            target.add(name)


def discover_repo(project_root: Path, env_names: set[str] | None = None) -> RepoDiscovery:
    """Infer runtime-relevant stack hints from common manifest/config files."""
    discovery = RepoDiscovery()
    env_names = {name.upper() for name in (env_names or set())}

    file_map = {
        "pyproject": project_root / "pyproject.toml",
        "requirements": project_root / "requirements.txt",
        "package": project_root / "package.json",
        "docker_compose": project_root / "docker-compose.yml",
        "docker_compose_alt": project_root / "docker-compose.yaml",
        "compose": project_root / "compose.yaml",
        "prisma": project_root / "prisma" / "schema.prisma",
    }

    if file_map["pyproject"].exists() or file_map["requirements"].exists():
        discovery.languages.add("python")
    if file_map["package"].exists():
        discovery.languages.add("node")

    if file_map["pyproject"].exists():
        discovery.package_managers.add("pip")
    if file_map["package"].exists():
        try:
            package_json = json.loads(_read_text(file_map["package"]) or "{}")
        except json.JSONDecodeError:
            package_json = {}
        pm = package_json.get("packageManager")
        if isinstance(pm, str) and pm:
            discovery.package_managers.add(pm.split("@", 1)[0])
        else:
            discovery.package_managers.add("npm")

    manifests = {
        name: _read_text(path)
        for name, path in file_map.items()
        if path.exists()
    }
    combined = "\n".join(manifests.values())

    _add_if_contains(
        discovery.databases,
        combined,
        {
            "psycopg": "postgresql",
            "asyncpg": "postgresql",
            "postgres": "postgresql",
            "mysql": "mysql",
            "pymongo": "mongodb",
            "mongodb": "mongodb",
            "mongoengine": "mongodb",
            "clickhouse": "clickhouse",
            "clickhouse-driver": "clickhouse",
            "clickhouse_connect": "clickhouse",
            "redis": "redis",
        },
    )

    if file_map["package"].exists():
        try:
            package_json = json.loads(_read_text(file_map["package"]) or "{}")
        except json.JSONDecodeError:
            package_json = {}
        deps = {
            **(package_json.get("dependencies") or {}),
            **(package_json.get("devDependencies") or {}),
        }
        dep_blob = "\n".join(deps.keys()).lower()
        _add_if_contains(
            discovery.databases,
            dep_blob,
            {
                "pg": "postgresql",
                "mysql": "mysql",
                "mongoose": "mongodb",
                "mongodb": "mongodb",
                "@clickhouse/client": "clickhouse",
                "redis": "redis",
                "prisma": "postgresql",
            },
        )

    for env_name in env_names:
        if "CLICKHOUSE" in env_name or env_name.startswith("CH_"):
            discovery.databases.add("clickhouse")
        if "POSTGRES" in env_name or env_name.startswith("PG") or env_name == "DATABASE_URL":
            discovery.databases.add("postgresql")
        if "MYSQL" in env_name:
            discovery.databases.add("mysql")
        if "MONGO" in env_name:
            discovery.databases.add("mongodb")
        if "REDIS" in env_name:
            discovery.databases.add("redis")

    compose_blob = "\n".join(
        manifests.get(key, "")
        for key in ("docker_compose", "docker_compose_alt", "compose")
    ).lower()
    _add_if_contains(
        discovery.services,
        compose_blob,
        {
            "clickhouse": "clickhouse",
            "postgres": "postgresql",
            "mysql": "mysql",
            "mongo": "mongodb",
            "redis": "redis",
        },
    )

    if file_map["prisma"].exists():
        discovery.hints.append("Prisma schema detected.")
    if (project_root / "alembic.ini").exists():
        discovery.hints.append("Alembic migration config detected.")
    if (project_root / "docker").exists() or discovery.services:
        discovery.hints.append("Containerized services may be available for local verification.")

    return discovery


def format_repo_discovery(discovery: RepoDiscovery) -> str:
    """Render discovery results into a compact prompt summary."""
    lines: list[str] = []
    if discovery.languages:
        lines.append("Detected languages: " + ", ".join(sorted(discovery.languages)))
    if discovery.package_managers:
        lines.append("Likely package managers: " + ", ".join(sorted(discovery.package_managers)))
    if discovery.databases:
        lines.append("Detected databases/services in code or env: " + ", ".join(sorted(discovery.databases)))
    if discovery.services:
        lines.append("Detected local service containers: " + ", ".join(sorted(discovery.services)))
    if discovery.hints:
        lines.extend(discovery.hints)
    if not lines:
        return "No strong runtime/service signals were detected from repository manifests."
    return "\n".join(lines)