api/app/scoring.py

from __future__ import annotations
from typing import Any, Dict, Optional
from .models import (
    CheckResult, ScoreDimension, ScorecardResult,
    WorkerResults, ReproResult, ClaimReport,
    CAPInspectionResult, SecurityScanResult
)


def _check(name: str, passed: bool, note: str = "") -> CheckResult:
    return CheckResult(name=name, passed=passed, note=note)


def score_technical_execution(
    repo: Optional[ReproResult],
    cap: Optional[CAPInspectionResult],
    security: Optional[SecurityScanResult],
    input_data: Dict[str, Any],
) -> ScoreDimension:
    checks: list[CheckResult] = []

    # CAP provider checks
    github_url = input_data.get("github_url", "")
    agent_url = input_data.get("agent_listing_url", "")

    checks.append(_check("github_url_provided", bool(github_url or agent_url), "Input URL provided"))

    if repo:
        checks.append(_check("readme_exists", repo.files_present.get("README.md", False), "README.md present"))
        checks.append(_check("env_example_exists", repo.files_present.get(".env.example", False), ".env.example present"))
        checks.append(_check("docker_compose_exists", repo.files_present.get("docker-compose.yml", False), "docker-compose.yml present"))
        checks.append(_check("dockerfile_exists", repo.files_present.get("Dockerfile", False), "Dockerfile present"))
        checks.append(_check("tests_exist", repo.files_present.get("tests/", False), "tests/ directory present"))
        checks.append(_check("ci_exists", repo.files_present.get(".github/workflows/", False), "CI workflow present"))
        checks.append(_check("reproducible", repo.reproducible, f"Repo setup reproducible (score={repo.score:.0f})"))

    if cap:
        checks.append(_check("cap_schema_present", cap.checks.get("schema_present", False), "CAP capability schema present"))
        checks.append(_check("cap_pricing_defined", cap.checks.get("pricing_defined", False), "CAP pricing defined"))
        checks.append(_check("cap_proof_delivery", cap.checks.get("proof_delivery", False), "Proof-of-delivery fields present"))
        checks.append(_check("cap_output_machine_readable", cap.checks.get("output_machine_readable", False), "Output is machine-readable JSON"))

    if security:
        checks.append(_check("no_critical_secrets", security.risk_level != "critical", f"Security risk level: {security.risk_level}"))

    passed = sum(1 for c in checks if c.passed)
    total = max(len(checks), 1)
    score = (passed / total) * 100

    issues = []
    if repo and not repo.reproducible:
        issues.append("Repository setup is not reproducible")
    if cap and not cap.checks.get("proof_delivery", False):
        issues.append("Missing proof-of-delivery fields in CAP integration")
    if security and security.risk_level in ("high", "critical"):
        issues.append(f"Security risk level is {security.risk_level}")

    return ScoreDimension(
        score=round(score, 1),
        checks=checks,
        notes=f"{passed}/{total} checks passed. " + "; ".join(issues) if issues else f"{passed}/{total} checks passed.",
    )


def score_a2a_composability(
    cap: Optional[CAPInspectionResult],
    input_data: Dict[str, Any],
    a2a_calls_count: int = 0,
) -> ScoreDimension:
    checks: list[CheckResult] = []

    if cap:
        checks.append(_check("callable_by_agents", cap.checks.get("callable_by_agents", False), "Service can be called by another agent"))
        checks.append(_check("typed_inputs", cap.checks.get("typed_inputs", False), "Inputs are typed/minimal"))
        checks.append(_check("structured_output", cap.checks.get("output_machine_readable", False), "Outputs are structured JSON"))
        checks.append(_check("pricing_explicit", cap.checks.get("pricing_defined", False), "Pricing and SLA explicit"))
        checks.append(_check("a2a_dependency_documented", cap.checks.get("a2a_dependency", False), "A2A dependency documented"))
    else:
        # If no cap inspector ran, give partial credit for having A2A calls
        checks.append(_check("has_github_url", bool(input_data.get("github_url")), "GitHub URL provided (proxy for composability)"))

    checks.append(_check("makes_a2a_calls", a2a_calls_count > 0, f"Makes A2A calls to other agents ({a2a_calls_count} calls)"))
    checks.append(_check("failure_behavior_documented", cap.checks.get("failure_documented", False) if cap else False, "Failure/timeout behavior documented"))

    passed = sum(1 for c in checks if c.passed)
    total = max(len(checks), 1)
    return ScoreDimension(
        score=round((passed / total) * 100, 1),
        checks=checks,
        notes=f"{passed}/{total} A2A composability checks passed.",
    )


def score_innovation(
    cap: Optional[CAPInspectionResult],
    claims: Optional[ClaimReport],
    input_data: Dict[str, Any],
) -> ScoreDimension:
    checks: list[CheckResult] = []

    checks.append(_check("cap_integrated", cap is not None and cap.score > 50, "CAP integration is real (not superficial)"))
    checks.append(_check("unique_use_case", True, "Use case is distinct from generic API marketplace"))  # default true for CAPScore context
    checks.append(_check("evidence_provided", bool(input_data.get("agent_listing_url") or input_data.get("github_url")), "Evidence URLs provided"))

    if claims:
        supported = sum(1 for v in claims.verifications if v.status == "supported")
        checks.append(_check("claims_backed_by_evidence", supported > 0, f"{supported} claims backed by evidence"))

    checks.append(_check("machine_readable_output", cap.checks.get("output_machine_readable", False) if cap else False, "Output reusable by other agents"))

    passed = sum(1 for c in checks if c.passed)
    total = max(len(checks), 1)
    return ScoreDimension(
        score=round((passed / total) * 100, 1),
        checks=checks,
        notes=f"{passed}/{total} innovation checks passed.",
    )


def score_adoption_readiness(
    repo: Optional[ReproResult],
    cap: Optional[CAPInspectionResult],
    input_data: Dict[str, Any],
) -> ScoreDimension:
    checks: list[CheckResult] = []

    checks.append(_check("listing_url_present", bool(input_data.get("agent_listing_url")), "Agent Store listing URL present"))
    checks.append(_check("github_url_present", bool(input_data.get("github_url")), "GitHub URL present"))

    if repo:
        checks.append(_check("quickstart_exists", repo.files_present.get("README.md", False), "README with quickstart present"))
        checks.append(_check("env_documented", repo.files_present.get(".env.example", False), ".env.example with documented variables"))

    if cap:
        checks.append(_check("clear_pricing", cap.checks.get("pricing_defined", False), "Clear pricing for buyers"))
        checks.append(_check("sla_defined", cap.checks.get("sla_defined", False), "SLA defined"))

    passed = sum(1 for c in checks if c.passed)
    total = max(len(checks), 1)
    return ScoreDimension(
        score=round((passed / total) * 100, 1),
        checks=checks,
        notes=f"{passed}/{total} adoption readiness checks passed.",
    )


def score_presentation_readiness(
    repo: Optional[ReproResult],
    claims: Optional[ClaimReport],
    input_data: Dict[str, Any],
) -> ScoreDimension:
    checks: list[CheckResult] = []

    checks.append(_check("readme_present", repo.files_present.get("README.md", False) if repo else bool(input_data.get("github_url")), "README present"))
    checks.append(_check("demo_url_provided", bool(input_data.get("demo_url")), "Demo video URL provided"))
    checks.append(_check("claims_present", len(input_data.get("claims", [])) > 0 or bool(input_data.get("agent_listing_url")), "Claims or listing provided"))

    if claims:
        unsupported = sum(1 for v in claims.verifications if v.status == "unsupported")
        misleading = sum(1 for v in claims.verifications if v.status == "misleading")
        checks.append(_check("no_misleading_claims", misleading == 0, f"{misleading} misleading claims found"))
        checks.append(_check("few_unsupported_claims", unsupported <= 1, f"{unsupported} unsupported claims"))

    passed = sum(1 for c in checks if c.passed)
    total = max(len(checks), 1)
    return ScoreDimension(
        score=round((passed / total) * 100, 1),
        checks=checks,
        notes=f"{passed}/{total} presentation readiness checks passed.",
    )


def compute_scorecard(
    worker_results: WorkerResults,
    input_data: Dict[str, Any],
    a2a_calls_count: int = 0,
) -> ScorecardResult:
    tech = score_technical_execution(
        worker_results.repo, worker_results.cap, worker_results.security, input_data
    )
    a2a = score_a2a_composability(worker_results.cap, input_data, a2a_calls_count)
    innov = score_innovation(worker_results.cap, worker_results.claims, input_data)
    adopt = score_adoption_readiness(worker_results.repo, worker_results.cap, input_data)
    pres = score_presentation_readiness(worker_results.repo, worker_results.claims, input_data)

    overall = round(
        0.30 * tech.score +
        0.25 * a2a.score +
        0.20 * innov.score +
        0.15 * adopt.score +
        0.10 * pres.score,
        1
    )

    # Collect critical issues and top fixes
    critical_issues: list[str] = []
    top_fixes: list[str] = []

    if worker_results.repo:
        critical_issues.extend(worker_results.repo.issues[:3])
        top_fixes.extend(worker_results.repo.suggestions[:3])
    if worker_results.cap:
        critical_issues.extend(worker_results.cap.issues[:2])
    if worker_results.security and worker_results.security.risk_level in ("high", "critical"):
        critical_issues.append(f"Security risk level: {worker_results.security.risk_level}")

    # Deduplicate
    critical_issues = list(dict.fromkeys(critical_issues))[:5]
    top_fixes = list(dict.fromkeys(top_fixes))[:5]

    return ScorecardResult(
        overall_score=overall,
        technical_execution=tech,
        a2a_composability=a2a,
        innovation=innov,
        adoption_readiness=adopt,
        presentation_readiness=pres,
        critical_issues=critical_issues,
        top_fixes=top_fixes,
    )