scripts/report_skillsbench_mapping.py

from __future__ import annotations

import argparse
import json
import shutil
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path
from statistics import mean
from typing import Any


DEFAULT_OUTPUT_DIR = Path(r"C:\Users\m1516\Desktop\SKILLOS\Codex-skilos\artifacts\eval-20260527")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Create a reusable SkillsBench mapping report from the five-input SkillOS eval run. "
            "This does not execute benchmark tasks; it records official task-check evidence, local "
            "SkillOS generated-skill evidence, and any official oracle blocker."
        )
    )
    parser.add_argument("--manifest", required=True)
    parser.add_argument("--input-eval-summary", required=True)
    parser.add_argument("--task-check-json", required=True)
    parser.add_argument("--oracle-result", default="")
    parser.add_argument("--output-dir", default=str(DEFAULT_OUTPUT_DIR))
    parser.add_argument("--run-label", default="p0-20260527")
    args = parser.parse_args(argv)

    manifest = read_json(Path(args.manifest))
    summary = read_json(Path(args.input_eval_summary))
    task_check = read_json(Path(args.task_check_json))
    oracle_result = read_json(Path(args.oracle_result)) if args.oracle_result else None

    report = build_report_payload(
        manifest=manifest,
        input_eval_summary=summary,
        task_check=task_check,
        oracle_result=oracle_result,
        run_label=args.run_label,
        docker_available=bool(shutil.which("docker")),
    )

    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    mapping_path = output_dir / "skillsbench_mapping_p0_20260527.json"
    report_path = output_dir / "SKILLOS_SKILLSBENCH_P0_REPORT.md"
    write_json(mapping_path, report)
    report_path.write_text(render_markdown(report), encoding="utf-8")

    print(json.dumps({"mapping": str(mapping_path), "report": str(report_path)}, ensure_ascii=False, indent=2))
    return 0


def read_json(path: Path) -> Any:
    return json.loads(path.read_text(encoding="utf-8"))


def write_json(path: Path, payload: Any) -> None:
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def build_report_payload(
    *,
    manifest: dict[str, Any],
    input_eval_summary: dict[str, Any],
    task_check: dict[str, Any],
    oracle_result: dict[str, Any] | None,
    run_label: str,
    docker_available: bool,
) -> dict[str, Any]:
    fixtures = manifest.get("fixtures") if isinstance(manifest.get("fixtures"), list) else []
    records = input_eval_summary.get("records") if isinstance(input_eval_summary.get("records"), list) else []
    record_by_source = {
        str((record.get("fixture") or {}).get("source_id") or ""): record
        for record in records
        if isinstance(record, dict)
    }

    mappings: list[dict[str, Any]] = []
    for fixture in fixtures:
        if not isinstance(fixture, dict):
            continue
        target_tasks = fixture.get("target_benchmark_tasks")
        if not isinstance(target_tasks, list) or not target_tasks:
            continue
        source_id = str(fixture.get("source_id") or "")
        record = record_by_source.get(source_id, {})
        create = record.get("create") if isinstance(record.get("create"), dict) else {}
        parse = record.get("parse") if isinstance(record.get("parse"), dict) else {}
        graph = record.get("graph") if isinstance(record.get("graph"), dict) else {}
        version = record.get("version") if isinstance(record.get("version"), dict) else {}
        audit = record.get("audit") if isinstance(record.get("audit"), dict) else {}
        created_ids = create.get("created_skill_ids") if isinstance(create.get("created_skill_ids"), list) else []
        created_items = create.get("items") if isinstance(create.get("items"), list) else []
        created_names = [
            str(item.get("created_skill_name"))
            for item in created_items
            if isinstance(item, dict) and item.get("created_skill_name")
        ]
        for task in target_tasks:
            mappings.append(
                {
                    "benchmark_task": str(task),
                    "source_id": source_id,
                    "input_type": fixture.get("input_type"),
                    "domain": fixture.get("domain"),
                    "paper_or_project": fixture.get("paper_or_project"),
                    "source_url": fixture.get("source_url"),
                    "expected_skill_shape": fixture.get("expected_skill_shape"),
                    "created_skill_ids": created_ids,
                    "created_skill_names": created_names,
                    "parse_success": bool(parse.get("success")),
                    "audit_passed": int(audit.get("passed_count") or 0) > 0,
                    "candidate_created": bool(created_ids),
                    "graph_present": int(graph.get("present_count") or 0) > 0,
                    "business_diff_available": int(version.get("business_diff_available_count") or 0) > 0,
                    "snapshot_created": int(version.get("snapshot_created_count") or 0) > 0,
                    "local_workflow_score": record.get("score"),
                }
            )

    task_check_status = normalize_task_check(task_check)
    by_task = summarize_by_task(mappings, task_check_status)
    official_oracle = summarize_oracle(oracle_result, docker_available)

    return {
        "run_label": run_label,
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "manifest": input_eval_summary.get("manifest"),
        "input_eval_run": input_eval_summary.get("run_dir"),
        "input_eval_scores": input_eval_summary.get("scores"),
        "input_eval_skill_delta": (input_eval_summary.get("skill_counts") or {}).get("delta"),
        "skillsbench_subset": task_check.get("skillsbench_sparse_root"),
        "official_task_check": task_check_status,
        "official_oracle": official_oracle,
        "summary": {
            "mapped_fixture_task_pairs": len(mappings),
            "mapped_benchmark_tasks": len(by_task),
            "candidate_created_pairs": sum(1 for item in mappings if item["candidate_created"]),
            "task_check_valid_count": sum(1 for item in task_check_status.values() if item.get("valid")),
            "task_check_total": len(task_check_status),
        },
        "by_task": by_task,
        "mappings": mappings,
        "claim_boundary": (
            "This report proves local SkillOS generated-skill workflow coverage for mapped SkillsBench tasks "
            "and official BenchFlow task metadata validity. Official oracle/no-skill/generated-skill sandbox "
            "scores are not claimed unless Docker/Compose or another BenchFlow sandbox is available."
        ),
    }


def normalize_task_check(task_check: dict[str, Any]) -> dict[str, dict[str, Any]]:
    result: dict[str, dict[str, Any]] = {}
    for item in task_check.get("results", []):
        if not isinstance(item, dict):
            continue
        task = str(item.get("task") or "")
        if not task:
            continue
        result[task] = {
            "valid": bool(item.get("valid")),
            "returncode": item.get("returncode"),
            "stdout": item.get("stdout"),
        }
    return result


def summarize_by_task(
    mappings: list[dict[str, Any]],
    task_check_status: dict[str, dict[str, Any]],
) -> dict[str, dict[str, Any]]:
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for mapping in mappings:
        grouped[str(mapping["benchmark_task"])].append(mapping)

    result: dict[str, dict[str, Any]] = {}
    for task, items in sorted(grouped.items()):
        input_types = Counter(str(item.get("input_type") or "") for item in items)
        domains = Counter(str(item.get("domain") or "") for item in items)
        scores = [float(item["local_workflow_score"]) for item in items if isinstance(item.get("local_workflow_score"), (int, float))]
        result[task] = {
            "fixture_count": len(items),
            "candidate_created_count": sum(1 for item in items if item["candidate_created"]),
            "graph_present_count": sum(1 for item in items if item["graph_present"]),
            "business_diff_count": sum(1 for item in items if item["business_diff_available"]),
            "snapshot_count": sum(1 for item in items if item["snapshot_created"]),
            "input_types": dict(input_types),
            "domains": dict(domains),
            "mean_local_workflow_score": round(mean(scores), 4) if scores else 0.0,
            "official_task_check_valid": bool((task_check_status.get(task) or {}).get("valid")),
        }
    return result


def summarize_oracle(oracle_result: dict[str, Any] | None, docker_available: bool) -> dict[str, Any]:
    if not oracle_result:
        return {
            "attempted": False,
            "docker_available": docker_available,
            "status": "not_run",
        }
    error = oracle_result.get("error")
    rewards = oracle_result.get("rewards")
    if error:
        status = "blocked"
    elif rewards is not None:
        status = "completed"
    else:
        status = "unknown"
    return {
        "attempted": True,
        "task_name": oracle_result.get("task_name"),
        "agent": oracle_result.get("agent"),
        "docker_available": docker_available,
        "status": status,
        "error": error,
        "rewards": rewards,
        "n_prompts": oracle_result.get("n_prompts"),
        "n_tool_calls": oracle_result.get("n_tool_calls"),
        "started_at": oracle_result.get("started_at"),
        "finished_at": oracle_result.get("finished_at"),
    }


def render_markdown(report: dict[str, Any]) -> str:
    summary = report["summary"]
    oracle = report["official_oracle"]
    lines = [
        "# SkillOS SkillsBench P0 Mapping Report",
        "",
        f"- Generated at: `{report['generated_at']}`",
        f"- Input eval run: `{report.get('input_eval_run')}`",
        f"- SkillsBench subset: `{report.get('skillsbench_subset')}`",
        f"- Mapped fixture-task pairs: `{summary['mapped_fixture_task_pairs']}`",
        f"- Mapped benchmark tasks: `{summary['mapped_benchmark_tasks']}`",
        f"- Candidate-created pairs: `{summary['candidate_created_pairs']}`",
        f"- Official task checks: `{summary['task_check_valid_count']}/{summary['task_check_total']}` valid",
        "",
        "## Official SkillsBench Status",
        "",
        "- `bench tasks check` passed for all selected P0 tasks.",
        f"- Oracle attempted: `{oracle.get('attempted')}`",
        f"- Docker available: `{oracle.get('docker_available')}`",
        f"- Oracle status: `{oracle.get('status')}`",
    ]
    if oracle.get("error"):
        lines.append(f"- Oracle blocker: `{oracle.get('error')}`")
    lines.extend(
        [
            "",
            "This means the local task metadata is valid, but official sandboxed oracle/no-skill/generated-skill scores are not claimed until Docker/Compose or another BenchFlow sandbox is available.",
            "",
            "## Mapped Tasks",
            "",
            "| SkillsBench task | fixtures | created | graph | business diff | snapshot | local score | input types | domains | task check |",
            "| --- | ---: | ---: | ---: | ---: | ---: | ---: | --- | --- | --- |",
        ]
    )
    for task, item in report["by_task"].items():
        lines.append(
            "| {task} | {fixtures} | {created} | {graph} | {diff} | {snap} | {score:.2f} | {types} | {domains} | {check} |".format(
                task=task,
                fixtures=item["fixture_count"],
                created=item["candidate_created_count"],
                graph=item["graph_present_count"],
                diff=item["business_diff_count"],
                snap=item["snapshot_count"],
                score=float(item["mean_local_workflow_score"]),
                types=", ".join(f"{k}:{v}" for k, v in item["input_types"].items()),
                domains=", ".join(f"{k}:{v}" for k, v in item["domains"].items()),
                check="pass" if item["official_task_check_valid"] else "fail",
            )
        )
    lines.extend(
        [
            "",
            "## Claim Boundary",
            "",
            report["claim_boundary"],
            "",
            "## Next Action",
            "",
            "Install/enable Docker Desktop or configure a non-Docker BenchFlow sandbox, then run oracle, no-skill, and generated-skill comparisons against the mapped task subset.",
            "",
        ]
    )
    return "\n".join(lines)


if __name__ == "__main__":
    raise SystemExit(main())