lambdagent/agent-lint-action/lint_runner.py

#!/usr/bin/env python3
"""
GitHub Action runner for lambdagent agent config lint.

Scans directories for agent YAML/JSON configs, runs lint + type check + cost
estimation, and outputs results as GitHub Action annotations + summary.
"""

import glob
import json
import os
import sys
import yaml


def find_configs(paths: str) -> list:
    """Find all agent YAML configs in given paths."""
    configs = []
    for path in paths.strip().split("\n"):
        path = path.strip()
        if not path:
            continue
        if os.path.isfile(path):
            configs.append(path)
        elif os.path.isdir(path):
            for pattern in ["**/*.yml", "**/*.yaml", "**/*.json"]:
                for f in glob.glob(os.path.join(path, pattern), recursive=True):
                    if "node_modules" in f or ".git" in f:
                        continue
                    # Check if it's an agent config (has 'type' field)
                    try:
                        with open(f) as fh:
                            cfg = yaml.safe_load(fh)
                        if cfg and isinstance(cfg, dict) and "type" in cfg:
                            configs.append(f)
                    except Exception:
                        pass
    return configs


def run_lint(config_path: str) -> dict:
    """Run lint on a single config file."""
    try:
        from lambdagent.fromconfig.lint import lint_config, detect_framework
        from lambdagent.fromconfig.schema import validate_schema

        with open(config_path) as f:
            cfg = yaml.safe_load(f)

        framework = detect_framework(cfg)
        lint_results = lint_config(cfg)
        schema_errors = validate_schema(cfg)

        return {
            "path": config_path,
            "framework": framework,
            "lint": [
                {"rule": r.rule, "level": r.level, "message": r.message}
                for r in lint_results
            ],
            "schema": [
                {"rule": e[1], "level": e[0], "message": e[2]}
                for e in schema_errors
            ],
        }
    except ImportError:
        # lambdagent not installed — run basic YAML validation only
        return {
            "path": config_path,
            "framework": "unknown",
            "lint": [],
            "schema": [],
            "warning": "lambdagent not installed; only basic YAML validation available",
        }
    except Exception as e:
        return {
            "path": config_path,
            "framework": "unknown",
            "lint": [],
            "schema": [],
            "error": str(e),
        }


def run_cost_estimate(config_path: str) -> dict:
    """Estimate cost for a config."""
    try:
        from lambdagent.fromconfig import from_config
        from lambdagent.cost_grade import estimate_cost
        term = from_config(config_path)
        grade = estimate_cost(term)
        return {
            "tokens": grade.tokens,
            "cost_usd": round(grade.money, 4),
            "probability": round(grade.probability, 4),
        }
    except Exception as e:
        return {"error": str(e)}


def main():
    paths = os.environ.get("INPUT_PATHS", ".")
    fail_on = os.environ.get("INPUT_FAIL_ON", "error").upper()
    cost_threshold = os.environ.get("INPUT_COST_THRESHOLD", "")

    configs = find_configs(paths)
    if not configs:
        print("No agent config files found.")
        _set_output("total-errors", "0")
        _set_output("total-warnings", "0")
        _set_output("total-files", "0")
        return

    total_errors = 0
    total_warnings = 0
    all_results = []

    for config_path in configs:
        result = run_lint(config_path)
        cost = run_cost_estimate(config_path)
        result["cost"] = cost
        all_results.append(result)

        for finding in result["lint"] + result["schema"]:
            level = finding["level"]
            if level == "ERROR":
                total_errors += 1
                _annotation("error", config_path, finding["message"], finding.get("rule", ""))
            elif level == "WARN":
                total_warnings += 1
                _annotation("warning", config_path, finding["message"], finding.get("rule", ""))

        # Cost threshold check
        if cost_threshold and cost.get("cost_usd", 0) > float(cost_threshold):
            total_errors += 1
            _annotation(
                "error", config_path,
                f"Estimated cost ${cost['cost_usd']:.2f} exceeds threshold ${float(cost_threshold):.2f}",
                "COST"
            )

    # Summary
    print(f"\n{'='*60}")
    print(f"Agent Config Lint Results")
    print(f"{'='*60}")
    print(f"Files scanned: {len(configs)}")
    print(f"Errors:        {total_errors}")
    print(f"Warnings:      {total_warnings}")

    for r in all_results:
        errors = [f for f in r["lint"] + r["schema"] if f["level"] == "ERROR"]
        warns = [f for f in r["lint"] + r["schema"] if f["level"] == "WARN"]
        status = "PASS" if not errors else "FAIL"
        cost_str = ""
        if r.get("cost", {}).get("cost_usd"):
            cost_str = f" | Cost: ${r['cost']['cost_usd']:.2f}"
        print(f"  {'FAIL' if errors else 'PASS'} {r['path']} ({r['framework']}) "
              f"— {len(errors)} errors, {len(warns)} warnings{cost_str}")

    _set_output("total-errors", str(total_errors))
    _set_output("total-warnings", str(total_warnings))
    _set_output("total-files", str(len(configs)))

    # Fail if needed
    level_order = {"ERROR": 0, "WARN": 1, "INFO": 2}
    fail_order = level_order.get(fail_on, 0)

    should_fail = False
    if fail_order <= 0 and total_errors > 0:
        should_fail = True
    elif fail_order <= 1 and (total_errors + total_warnings) > 0:
        should_fail = True

    if should_fail:
        print(f"\nFailing because {total_errors} errors and {total_warnings} warnings found "
              f"(fail-on: {fail_on})")
        sys.exit(1)


def _annotation(level: str, file: str, message: str, rule: str):
    """Emit GitHub Actions annotation."""
    print(f"::{level} file={file}::[{rule}] {message}")


def _set_output(name: str, value: str):
    """Set GitHub Actions output."""
    output_file = os.environ.get("GITHUB_OUTPUT")
    if output_file:
        with open(output_file, "a") as f:
            f.write(f"{name}={value}\n")


if __name__ == "__main__":
    main()