scripts/run_input_skill_eval.py

from __future__ import annotations

import argparse
import hashlib
import json
import re
import sys
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any


REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_RUN_ROOT = REPO_ROOT / "artifacts" / "input-skill-eval-runs"
ALLOWED_INPUT_TYPES = {"trajectory", "document", "api_doc", "script", "past_skills"}


@dataclass(frozen=True)
class FixtureCase:
    source_id: str
    input_type: str
    domain: str
    source_url: str
    paper_or_project: str
    expected_skill_shape: str
    license_note: str
    local_path: Path
    content: str
    content_sha256: str
    manifest: dict[str, Any]
    truncated: bool = False


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Run reusable SkillOS five-input evaluation from a manifest. "
            "Raw fixture files are read-only; all outputs go under an isolated run directory."
        )
    )
    parser.add_argument("--manifest", required=True, help="JSON manifest containing a fixtures array.")
    parser.add_argument("--fixture-root", default="", help="Root for relative fixture content_file paths.")
    parser.add_argument("--api-base", default="http://127.0.0.1:8001/api/v1")
    parser.add_argument("--run-root", default=str(DEFAULT_RUN_ROOT))
    parser.add_argument("--run-id", default=datetime.now().strftime("%Y%m%d-%H%M%S"))
    parser.add_argument("--limit-per-type", type=int, default=0)
    parser.add_argument("--max-chars", type=int, default=14000)
    parser.add_argument("--request-timeout", type=int, default=180)
    parser.add_argument("--create-candidates", action="store_true")
    parser.add_argument("--max-candidates-per-fixture", type=int, default=1)
    parser.add_argument("--snapshot", action="store_true", help="Create a version snapshot for created candidates.")
    parser.add_argument("--fail-under", type=float, default=0.0)
    args = parser.parse_args(argv)

    manifest_path = Path(args.manifest).resolve()
    fixture_root = Path(args.fixture_root).resolve() if args.fixture_root else manifest_path.parent
    run_dir = Path(args.run_root).resolve() / f"input-skill-eval-{args.run_id}"
    raw_dir = run_dir / "raw"
    raw_dir.mkdir(parents=True, exist_ok=True)

    cases = load_manifest(manifest_path, fixture_root=fixture_root, max_chars=args.max_chars)
    cases = limit_cases(cases, args.limit_per_type)
    client = ApiClient(args.api_base.rstrip("/"), raw_dir)
    summary: dict[str, Any] = {
        "started_at": datetime.now().isoformat(timespec="seconds"),
        "api_base": client.base_url,
        "manifest": str(manifest_path),
        "fixture_root": str(fixture_root),
        "run_dir": str(run_dir),
        "run_id": args.run_id,
        "isolation": {
            "raw_fixtures_read_only": True,
            "created_candidates": bool(args.create_candidates),
            "outputs_are_run_scoped": True,
        },
        "options": {
            "limit_per_type": args.limit_per_type,
            "max_chars": args.max_chars,
            "create_candidates": args.create_candidates,
            "max_candidates_per_fixture": args.max_candidates_per_fixture,
            "snapshot": args.snapshot,
        },
        "connectivity": {},
        "fixture_count": len(cases),
        "records": [],
        "scores": {},
        "notes": [],
    }

    summary["connectivity"]["skills_before"] = client.get("/skills?limit=1000", label="skills_before_eval")
    before_count = count_skills(summary["connectivity"]["skills_before"])

    for case in cases:
        record = run_fixture_case(
            case,
            client,
            run_id=args.run_id,
            request_timeout=args.request_timeout,
            create_candidates=args.create_candidates,
            max_candidates_per_fixture=args.max_candidates_per_fixture,
            snapshot=args.snapshot,
        )
        record["score"] = score_fixture_record(record)
        summary["records"].append(record)

    summary["connectivity"]["skills_after"] = client.get("/skills?limit=1000", label="skills_after_eval")
    after_count = count_skills(summary["connectivity"]["skills_after"])
    summary["skill_counts"] = {
        "before": before_count,
        "after": after_count,
        "delta": after_count - before_count,
    }
    summary["scores"] = summarize_scores(summary["records"])
    summary["finished_at"] = datetime.now().isoformat(timespec="seconds")

    write_json(run_dir / "summary.json", summary)
    (run_dir / "REPORT.md").write_text(render_report(summary), encoding="utf-8")
    print(json.dumps({
        "run_dir": str(run_dir),
        "fixture_count": len(cases),
        "overall": summary["scores"]["overall"],
        "by_input_type": summary["scores"]["by_input_type"],
        "skill_delta": summary["skill_counts"]["delta"],
    }, indent=2, ensure_ascii=False))
    if args.fail_under and summary["scores"]["overall"] < args.fail_under:
        return 1
    return 0


def load_manifest(manifest_path: Path, *, fixture_root: Path, max_chars: int = 0) -> list[FixtureCase]:
    data = json.loads(manifest_path.read_text(encoding="utf-8"))
    entries = data.get("fixtures") if isinstance(data, dict) else data
    if not isinstance(entries, list):
        raise ValueError("Manifest must be a JSON array or an object with a fixtures array.")
    cases: list[FixtureCase] = []
    for idx, entry in enumerate(entries):
        if not isinstance(entry, dict):
            raise ValueError(f"Fixture entry {idx} must be an object.")
        input_type = str(entry.get("input_type") or entry.get("source_type") or "").strip()
        if input_type not in ALLOWED_INPUT_TYPES:
            raise ValueError(f"Fixture {idx} has unsupported input_type {input_type!r}.")
        source_id = str(entry.get("source_id") or f"{input_type}-{idx + 1}").strip()
        path_text = str(entry.get("content_file") or entry.get("local_path") or entry.get("path") or "").strip()
        if not path_text:
            raise ValueError(f"Fixture {source_id} must provide content_file/local_path/path.")
        local_path = Path(path_text)
        if not local_path.is_absolute():
            local_path = fixture_root / local_path
        content, truncated = read_fixture_text(local_path, max_chars=max_chars)
        sha = hashlib.sha256(content.encode("utf-8")).hexdigest()
        cases.append(FixtureCase(
            source_id=source_id,
            input_type=input_type,
            domain=str(entry.get("domain") or "other"),
            source_url=str(entry.get("source_url") or ""),
            paper_or_project=str(entry.get("paper_or_project") or ""),
            expected_skill_shape=str(entry.get("expected_skill_shape") or ""),
            license_note=str(entry.get("license_note") or ""),
            local_path=local_path,
            content=content,
            content_sha256=sha,
            manifest=dict(entry),
            truncated=truncated,
        ))
    return cases


def limit_cases(cases: list[FixtureCase], limit_per_type: int) -> list[FixtureCase]:
    if limit_per_type <= 0:
        return cases
    counts: dict[str, int] = {}
    selected: list[FixtureCase] = []
    for case in cases:
        count = counts.get(case.input_type, 0)
        if count >= limit_per_type:
            continue
        selected.append(case)
        counts[case.input_type] = count + 1
    return selected


def read_fixture_text(path: Path, *, max_chars: int) -> tuple[str, bool]:
    content = path.read_text(encoding="utf-8", errors="replace")
    if max_chars <= 0 or len(content) <= max_chars:
        return content, False
    marker = (
        "\n\n[Fixture truncated for repeatable local evaluation. "
        f"Original file: {path.name}; retained_chars={max_chars}.]\n"
    )
    keep = max(1000, max_chars - len(marker))
    return content[:keep] + marker, True


def run_fixture_case(
    case: FixtureCase,
    client: "ApiClient",
    *,
    run_id: str,
    request_timeout: int,
    create_candidates: bool,
    max_candidates_per_fixture: int,
    snapshot: bool,
) -> dict[str, Any]:
    label_base = safe_filename(f"{case.input_type}_{case.source_id}")
    record: dict[str, Any] = {
        "fixture": fixture_case_summary(case),
        "parse": {},
        "audit": {"status": "not_run"},
        "create": {"status": "not_run"},
        "graph": {"status": "not_run"},
        "version": {"status": "not_run"},
        "harness": {"status": "not_run"},
        "skillsbench": {
            "status": "not_run",
            "note": "Reserved for SkillsBench/BenchFlow adapter. This runner keeps the field explicit instead of faking benchmark execution.",
        },
    }
    parse_result = client.post(
        "/ingest/parse",
        {
            "source_type": case.input_type,
            "content": case.content,
            "metadata": {
                "eval_run_id": run_id,
                "source_id": case.source_id,
                "input_type": case.input_type,
                "domain": case.domain,
                "source_url": case.source_url,
                "paper_or_project": case.paper_or_project,
                "expected_skill_shape": case.expected_skill_shape,
                "license_note": case.license_note,
                "content_sha256": case.content_sha256,
                "origin_file": str(case.local_path),
                "ephemeral_preview_only": not create_candidates,
                "max_candidates": max(1, max_candidates_per_fixture),
            },
        },
        label=f"parse_{label_base}",
        timeout_s=request_timeout,
    )
    record["parse"] = analyze_parse(case, parse_result)

    units = parse_result.get("units") if isinstance(parse_result, dict) else []
    if not create_candidates or not units:
        return record

    created: list[dict[str, Any]] = []
    audits: list[dict[str, Any]] = []
    graph_checks: list[dict[str, Any]] = []
    version_checks: list[dict[str, Any]] = []
    for idx, unit in enumerate(units[:max(1, max_candidates_per_fixture)]):
        payload = candidate_review_payload(unit, case, run_id=run_id)
        audit_result = client.post(
            "/ingest/audit-candidate",
            payload,
            label=f"audit_{label_base}_{idx + 1}",
            timeout_s=request_timeout,
        )
        audits.append(analyze_audit(payload, audit_result))
        create_result = client.post(
            "/ingest/create-candidate",
            payload,
            label=f"create_{label_base}_{idx + 1}",
            timeout_s=request_timeout,
        )
        created_info = analyze_create(create_result)
        created.append(created_info)
        skill_id = created_info.get("created_skill_id")
        if not skill_id:
            continue
        graph_checks.append(check_graph_for_skill(client, str(skill_id), label=f"graph_{label_base}_{idx + 1}"))
        version_checks.append(check_version_for_skill(
            client,
            str(skill_id),
            label=f"version_{label_base}_{idx + 1}",
            snapshot=snapshot,
        ))

    record["audit"] = summarize_audits(audits)
    record["create"] = summarize_creates(created)
    record["graph"] = summarize_graph_checks(graph_checks)
    record["version"] = summarize_version_checks(version_checks)
    return record


def analyze_parse(case: FixtureCase, result: dict[str, Any]) -> dict[str, Any]:
    units = result.get("units") or []
    unit_infos = [analyze_unit(unit) for unit in units if isinstance(unit, dict)]
    return {
        "http_status": result.get("_http_status"),
        "success": bool(result.get("success")),
        "unit_count": int(result.get("unit_count") or len(unit_infos)),
        "error_count": len(result.get("errors") or []),
        "errors": result.get("errors") or [],
        "elapsed_ms": result.get("_elapsed_ms"),
        "schema_completeness": average([unit["schema_completeness"] for unit in unit_infos]),
        "ctx2skill_evidence_completeness": average([unit["ctx2skill_evidence_completeness"] for unit in unit_infos]),
        "layer_correctness": expected_layer_score(case, unit_infos),
        "graph_relation_preview_count": sum(unit["graph_relation_preview_count"] for unit in unit_infos),
        "units": unit_infos,
    }


def analyze_unit(unit: dict[str, Any]) -> dict[str, Any]:
    meta = unit.get("metadata") or {}
    interface = meta.get("candidate_interface") if isinstance(meta.get("candidate_interface"), dict) else {}
    evidence = meta.get("ctx2skill_evidence") if isinstance(meta.get("ctx2skill_evidence"), dict) else {}
    relations = meta.get("candidate_relations") if isinstance(meta.get("candidate_relations"), dict) else {}
    graph_preview = meta.get("graph_relation_preview") or []
    input_schema = interface.get("input_schema") if isinstance(interface.get("input_schema"), dict) else {}
    output_schema = interface.get("output_schema") if isinstance(interface.get("output_schema"), dict) else {}
    schema_bits = [
        bool(input_schema),
        bool(output_schema),
        bool(interface.get("preconditions")),
        bool(interface.get("postconditions")),
    ]
    evidence_bits = [
        bool(evidence),
        bool(evidence.get("challenges")),
        bool(evidence.get("judge_results")),
        bool(evidence.get("selected_candidate") or evidence.get("selected_reason")),
    ]
    return {
        "unit_id": unit.get("unit_id"),
        "proposed_skill_name": unit.get("proposed_skill_name"),
        "proposed_type": unit.get("proposed_type"),
        "confidence": unit.get("confidence"),
        "schema_completeness": round(sum(1 for bit in schema_bits if bit) / len(schema_bits), 2),
        "ctx2skill_evidence_completeness": round(sum(1 for bit in evidence_bits if bit) / len(evidence_bits), 2),
        "challenge_count": len(evidence.get("challenges") or []),
        "judge_result_count": len(evidence.get("judge_results") or []),
        "candidate_score_count": len(evidence.get("candidate_scores") or []),
        "layering_reason_present": bool(meta.get("layering_reason")),
        "graph_relation_preview_count": len(graph_preview) if isinstance(graph_preview, list) else 0,
        "dependency_count": len(relations.get("dependency_ids") or []),
        "component_count": len(relations.get("component_ids") or []),
        "parent_count": len(relations.get("parent_skill_ids") or []),
    }


def expected_layer_score(case: FixtureCase, units: list[dict[str, Any]]) -> float:
    if not units:
        return 0.0
    expected = case.expected_skill_shape.strip().lower()
    if expected not in {"atomic", "functional", "strategic"}:
        return 0.5
    matches = sum(1 for unit in units if unit.get("proposed_type") == expected)
    return round(matches / len(units), 2)


def candidate_review_payload(unit: dict[str, Any], case: FixtureCase, *, run_id: str) -> dict[str, Any]:
    meta = unit.get("metadata") or {}
    interface = meta.get("candidate_interface") if isinstance(meta.get("candidate_interface"), dict) else {}
    implementation = meta.get("candidate_implementation") if isinstance(meta.get("candidate_implementation"), dict) else {}
    relations = meta.get("candidate_relations") if isinstance(meta.get("candidate_relations"), dict) else {}
    evaluation = meta.get("candidate_evaluation") if isinstance(meta.get("candidate_evaluation"), dict) else None
    source_type = unit.get("source_type") or case.input_type
    base_name = normalize_skill_name(unit.get("proposed_skill_name") or f"{case.source_id}_{source_type}_candidate")
    source_suffix = normalize_skill_name(case.source_id)
    if source_suffix and source_suffix not in base_name:
        name = normalize_skill_name(f"{base_name}_{source_suffix}")[:120]
    else:
        name = base_name
    description = unit.get("proposed_description") or unit.get("summary") or name
    if evaluation:
        evaluation = dict(evaluation)
        evaluation.setdefault("test_case_refs", [f"{case.source_id}:{unit.get('unit_id', 'unit')}"])
        evaluation.setdefault("benchmark_task_ids", list(case.manifest.get("target_benchmark_tasks") or []))
        evaluation.setdefault("validation_summary", "Created by isolated input-skill evaluation runner.")
    else:
        evaluation = {
            "verifier_specs": [{"type": "json_exists", "path": "output.result"}],
            "test_case_refs": [f"{case.source_id}:{unit.get('unit_id', 'unit')}"],
            "benchmark_task_ids": list(case.manifest.get("target_benchmark_tasks") or []),
            "validation_summary": "Created by isolated input-skill evaluation runner.",
        }
    tags = meta.get("candidate_tags") if isinstance(meta.get("candidate_tags"), list) else []
    return {
        "source_type": source_type,
        "unit_id": unit.get("unit_id") or f"{case.source_id}:unit",
        "raw_content": unit.get("raw_content") or "",
        "name": name,
        "description": description,
        "skill_type": unit.get("proposed_type") if unit.get("proposed_type") in {"atomic", "functional", "strategic"} else "atomic",
        "tags": unique_strings([
            *tags,
            *(unit.get("index_keywords") or [])[:4],
            "input-skill-eval",
            f"eval-run:{run_id}",
            f"source-id:{case.source_id}",
            f"domain:{case.domain}",
        ])[:12],
        "input_schema": interface.get("input_schema") or {"type": "object", "properties": {}},
        "output_schema": interface.get("output_schema") or {"type": "object", "properties": {"result": {"type": "object"}}},
        "preconditions": interface.get("preconditions") or [],
        "postconditions": interface.get("postconditions") or ["Candidate returns a structured result."],
        "prompt_template": implementation.get("prompt_template") or unit.get("summary") or description,
        "provenance": {
            "source_type": source_type,
            "source_ids": unique_strings([case.source_id, unit.get("unit_id") or "unit"]),
            "parent_skill_ids": relations.get("parent_skill_ids") or [],
            "created_by_agent": "input_skill_eval_runner",
            "creation_context": {
                "eval_run_id": run_id,
                "fixture_import_mode": "isolated_eval_candidate",
                "source_id": case.source_id,
                "input_type": case.input_type,
                "domain": case.domain,
                "source_url": case.source_url,
                "paper_or_project": case.paper_or_project,
                "expected_skill_shape": case.expected_skill_shape,
                "license_note": case.license_note,
                "local_path": str(case.local_path),
                "content_sha256": case.content_sha256,
                "ctx2skill_evidence": meta.get("ctx2skill_evidence"),
            },
        },
        "evaluation": evaluation,
        "dependency_ids": relations.get("dependency_ids") or [],
        "component_ids": relations.get("component_ids") or [],
        "sub_skill_ids": relations.get("sub_skill_ids") or relations.get("component_ids") or [],
        "parent_skill_ids": relations.get("parent_skill_ids") or [],
        "tool_calls": implementation.get("tool_calls") or [],
        "author": "input_skill_eval_runner",
    }


def analyze_audit(payload: dict[str, Any], result: dict[str, Any]) -> dict[str, Any]:
    return {
        "candidate_name": payload.get("name"),
        "http_status": result.get("_http_status"),
        "passed": bool(result.get("passed")),
        "audit_score": result.get("audit_score"),
        "issues": result.get("issues") or [],
        "warnings": result.get("warnings") or [],
    }


def analyze_create(result: dict[str, Any]) -> dict[str, Any]:
    created = result.get("created_skill") if isinstance(result.get("created_skill"), dict) else {}
    return {
        "http_status": result.get("_http_status"),
        "success": bool(result.get("success")),
        "created_skill_id": created.get("skill_id"),
        "created_skill_name": created.get("name"),
        "state": created.get("state"),
        "audit_passed": (result.get("audit") or {}).get("passed") if isinstance(result.get("audit"), dict) else None,
    }


def check_graph_for_skill(client: "ApiClient", skill_id: str, *, label: str) -> dict[str, Any]:
    result = client.get(f"/graph/view?view=skill_only&limit=500", label=label)
    nodes = result.get("nodes") or []
    edges = result.get("edges") or []
    skill_present = any(str(node.get("skill_id") or node.get("id") or "").endswith(skill_id) for node in nodes)
    relation_count = sum(
        1 for edge in edges
        if skill_id in str(edge.get("source") or edge.get("source_id") or "")
        or skill_id in str(edge.get("target") or edge.get("target_id") or "")
    )
    return {
        "skill_id": skill_id,
        "http_status": result.get("_http_status"),
        "skill_present": skill_present,
        "relation_count": relation_count,
        "node_count": len(nodes),
        "edge_count": len(edges),
    }


def check_version_for_skill(client: "ApiClient", skill_id: str, *, label: str, snapshot: bool) -> dict[str, Any]:
    diff = client.get(f"/lifecycle/{skill_id}/diff", label=f"{label}_diff")
    result = {
        "skill_id": skill_id,
        "http_status": diff.get("_http_status"),
        "business_diff_available": "business_diff" in diff or "business_summary" in diff or "history" in diff,
        "raw_diff_available": "raw_diff" in diff or "changes" in diff or "snapshot_diff" in diff or "history" in diff,
        "snapshot_created": False,
    }
    if snapshot:
        snap = client.post(
            f"/lifecycle/{skill_id}/snapshot",
            {
                "description": "Input-skill eval snapshot.",
                "author": "input_skill_eval_runner",
            },
            label=f"{label}_snapshot",
            timeout_s=120,
        )
        result["snapshot_http_status"] = snap.get("_http_status")
        result["snapshot_created"] = snap.get("_http_status") in {200, 201} and not snap.get("detail")
    return result


def summarize_audits(items: list[dict[str, Any]]) -> dict[str, Any]:
    if not items:
        return {"status": "not_run"}
    return {
        "status": "completed",
        "count": len(items),
        "passed_count": sum(1 for item in items if item.get("passed")),
        "average_audit_score": average([
            float(item.get("audit_score") or 0.0)
            for item in items
            if isinstance(item.get("audit_score"), (int, float))
        ]),
        "items": items,
    }


def summarize_creates(items: list[dict[str, Any]]) -> dict[str, Any]:
    if not items:
        return {"status": "not_run"}
    created_ids = [item.get("created_skill_id") for item in items if item.get("created_skill_id")]
    return {
        "status": "completed",
        "count": len(items),
        "success_count": sum(1 for item in items if item.get("success")),
        "created_skill_ids": created_ids,
        "http_statuses": [item.get("http_status") for item in items],
        "items": items,
    }


def summarize_graph_checks(items: list[dict[str, Any]]) -> dict[str, Any]:
    if not items:
        return {"status": "not_run"}
    return {
        "status": "completed",
        "count": len(items),
        "present_count": sum(1 for item in items if item.get("skill_present")),
        "relation_count": sum(int(item.get("relation_count") or 0) for item in items),
        "items": items,
    }


def summarize_version_checks(items: list[dict[str, Any]]) -> dict[str, Any]:
    if not items:
        return {"status": "not_run"}
    return {
        "status": "completed",
        "count": len(items),
        "business_diff_available_count": sum(1 for item in items if item.get("business_diff_available")),
        "snapshot_created_count": sum(1 for item in items if item.get("snapshot_created")),
        "items": items,
    }


def score_fixture_record(record: dict[str, Any]) -> float:
    parse = record.get("parse") or {}
    score = 0.0
    if parse.get("http_status") == 200 and parse.get("success"):
        score += 0.18
    if parse.get("unit_count", 0) > 0:
        score += 0.10
    score += 0.08 * clamp01(float(parse.get("schema_completeness") or 0.0))
    score += 0.08 * clamp01(float(parse.get("ctx2skill_evidence_completeness") or 0.0))
    score += 0.06 * clamp01(float(parse.get("layer_correctness") or 0.0))

    audit = record.get("audit") or {}
    if audit.get("status") == "completed":
        score += 0.10 * ratio(audit.get("passed_count"), audit.get("count"))
    elif audit.get("http_status") == 200 and audit.get("passed"):
        score += 0.10

    create = record.get("create") or {}
    if create.get("status") == "completed":
        score += 0.12 * ratio(create.get("success_count"), create.get("count"))
    elif create.get("http_status") in {200, 201} and create.get("success"):
        score += 0.12

    graph = record.get("graph") or {}
    if graph.get("status") == "completed":
        score += 0.10 * ratio(graph.get("present_count"), graph.get("count"))
        if int(graph.get("relation_count") or 0) > 0:
            score += 0.05
    elif graph.get("http_status") == 200 and graph.get("skill_present"):
        score += 0.10
        if int(graph.get("relation_count") or 0) > 0:
            score += 0.05

    version = record.get("version") or {}
    if version.get("status") == "completed":
        score += 0.06 * ratio(version.get("business_diff_available_count"), version.get("count"))
        if int(version.get("snapshot_created_count") or 0) > 0:
            score += 0.03
    elif version.get("http_status") == 200:
        if version.get("business_diff_available"):
            score += 0.06
        if version.get("snapshot_created"):
            score += 0.03

    harness = record.get("harness") or {}
    if harness.get("status") == "verified":
        score += 0.12
    elif harness.get("positive_pass") or harness.get("negative_rejected"):
        score += 0.06

    skillsbench = record.get("skillsbench") or {}
    if skillsbench.get("status") in {"mapped", "completed"} and skillsbench.get("verifier_ran"):
        score += 0.10
        if skillsbench.get("generated_pass"):
            score += 0.05

    return round(min(score, 1.0), 3)


def summarize_scores(records: list[dict[str, Any]]) -> dict[str, Any]:
    by_type_values: dict[str, list[float]] = {}
    by_domain_values: dict[str, list[float]] = {}
    for record in records:
        score = float(record.get("score") or 0.0)
        fixture = record.get("fixture") or {}
        by_type_values.setdefault(str(fixture.get("input_type") or "unknown"), []).append(score)
        by_domain_values.setdefault(str(fixture.get("domain") or "unknown"), []).append(score)
    return {
        "overall": average([float(record.get("score") or 0.0) for record in records]),
        "by_input_type": {key: average(values) for key, values in by_type_values.items()},
        "by_domain": {key: average(values) for key, values in by_domain_values.items()},
        "sample_count": len(records),
    }


def render_report(summary: dict[str, Any]) -> str:
    lines = [
        "# SkillOS Input-to-Skill Evaluation Report",
        "",
        f"- Started: `{summary['started_at']}`",
        f"- Finished: `{summary.get('finished_at', '')}`",
        f"- API base: `{summary['api_base']}`",
        f"- Manifest: `{summary['manifest']}`",
        f"- Fixture root: `{summary['fixture_root']}`",
        f"- Run directory: `{summary['run_dir']}`",
        f"- Created candidates: `{summary['options']['create_candidates']}`",
        "",
        "## Isolation",
        "",
        "- Raw fixture files are read only.",
        "- API responses and reports are written only under this run directory.",
        "- Candidate creation is disabled unless `--create-candidates` is passed.",
        "- SkillsBench results are explicit fields and remain `not_run` until the benchmark adapter executes.",
        "",
        "## Scores",
        "",
        f"- Overall: `{summary['scores']['overall']}`",
        f"- By input type: `{json.dumps(summary['scores']['by_input_type'], ensure_ascii=False)}`",
        f"- By domain: `{json.dumps(summary['scores']['by_domain'], ensure_ascii=False)}`",
        f"- Skill delta: `{summary['skill_counts']['delta']}`",
        "",
        "## Fixtures",
        "",
    ]
    for record in summary["records"]:
        fixture = record["fixture"]
        parse = record["parse"]
        lines.extend([
            f"### {fixture['source_id']} - {record['score']}",
            f"- Input/domain: `{fixture['input_type']}` / `{fixture['domain']}`",
            f"- Source: `{fixture.get('source_url', '')}`",
            f"- Paper/project: `{fixture.get('paper_or_project', '')}`",
            f"- Local file: `{fixture['local_path']}`",
            f"- Parse: http=`{parse.get('http_status')}`, success=`{parse.get('success')}`, "
            f"units=`{parse.get('unit_count')}`, schema=`{parse.get('schema_completeness')}`, "
            f"ctx2skill=`{parse.get('ctx2skill_evidence_completeness')}`, layer=`{parse.get('layer_correctness')}`",
            f"- Audit: `{record.get('audit', {}).get('status')}`",
            f"- Create: `{record.get('create', {}).get('status')}`",
            f"- Graph: `{record.get('graph', {}).get('status')}`",
            f"- Version: `{record.get('version', {}).get('status')}`",
            f"- Harness: `{record.get('harness', {}).get('status')}`",
            f"- SkillsBench: `{record.get('skillsbench', {}).get('status')}`",
            "",
        ])
    lines.append("Raw API envelopes are saved under `raw/` next to this report.")
    lines.append("")
    return "\n".join(lines)


class ApiClient:
    def __init__(self, base_url: str, raw_dir: Path) -> None:
        self.base_url = base_url
        self.raw_dir = raw_dir

    def get(self, path: str, *, label: str, timeout_s: int = 30) -> dict[str, Any]:
        return self._request("GET", self.base_url + path, None, label, timeout_s)

    def post(self, path: str, payload: dict[str, Any], *, label: str, timeout_s: int = 120) -> dict[str, Any]:
        return self._request("POST", self.base_url + path, payload, label, timeout_s)

    def _request(
        self,
        method: str,
        url: str,
        payload: dict[str, Any] | None,
        label: str,
        timeout_s: int,
    ) -> dict[str, Any]:
        data = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8")
        req = urllib.request.Request(url, data=data, method=method)
        req.add_header("Accept", "application/json")
        if payload is not None:
            req.add_header("Content-Type", "application/json")
        start = time.perf_counter()
        try:
            with urllib.request.urlopen(req, timeout=timeout_s) as resp:
                body = resp.read().decode("utf-8", errors="replace")
                status = resp.status
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")
            status = exc.code
        except Exception as exc:
            body = json.dumps({"error": str(exc)}, ensure_ascii=False)
            status = 0
        elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
        parsed = parse_body(body)
        envelope = {
            "label": label,
            "method": method,
            "url": url,
            "status_code": status,
            "elapsed_ms": elapsed_ms,
            "request_preview": preview_payload(payload),
            "body": parsed,
        }
        write_json(self.raw_dir / f"{safe_filename(label)}.json", envelope)
        if isinstance(parsed, dict):
            result = dict(parsed)
            result["_http_status"] = status
            result["_elapsed_ms"] = elapsed_ms
            return result
        return {"_http_status": status, "_elapsed_ms": elapsed_ms, "body": parsed}


def fixture_case_summary(case: FixtureCase) -> dict[str, Any]:
    return {
        "source_id": case.source_id,
        "input_type": case.input_type,
        "domain": case.domain,
        "source_url": case.source_url,
        "paper_or_project": case.paper_or_project,
        "expected_skill_shape": case.expected_skill_shape,
        "license_note": case.license_note,
        "local_path": str(case.local_path),
        "content_sha256": case.content_sha256,
        "truncated": case.truncated,
    }


def count_skills(payload: Any) -> int:
    if isinstance(payload, list):
        return len(payload)
    if isinstance(payload, dict):
        body = payload.get("body")
        if isinstance(body, list):
            return len(body)
        for key in ("value", "skills", "items"):
            value = payload.get(key)
            if isinstance(value, list):
                return len(value)
        if isinstance(payload.get("total"), int):
            return int(payload["total"])
    return 0


def write_json(path: Path, data: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")


def parse_body(body: str) -> Any:
    try:
        return json.loads(body)
    except json.JSONDecodeError:
        return body


def preview_payload(payload: dict[str, Any] | None) -> Any:
    if payload is None:
        return None
    preview = dict(payload)
    content = preview.get("content")
    if isinstance(content, str):
        preview["content"] = f"{content[:500]}... [chars={len(content)}]"
    return preview


def normalize_skill_name(value: str) -> str:
    text = re.sub(r"[^a-zA-Z0-9_]+", "_", str(value or "").strip().lower())
    text = re.sub(r"_+", "_", text).strip("_")
    return text[:120] or "input_skill_eval_candidate"


def safe_filename(value: str) -> str:
    normalized = normalize_skill_name(value).replace("/", "_")
    if len(normalized) <= 96:
        return normalized
    digest = hashlib.sha1(str(value).encode("utf-8")).hexdigest()[:10]
    return f"{normalized[:84].rstrip('_')}_{digest}"


def unique_strings(values: list[Any]) -> list[str]:
    seen: set[str] = set()
    result: list[str] = []
    for value in values:
        text = str(value or "").strip()
        if not text or text in seen:
            continue
        seen.add(text)
        result.append(text)
    return result


def average(values: list[float]) -> float:
    return round(sum(values) / len(values), 3) if values else 0.0


def ratio(numerator: Any, denominator: Any) -> float:
    try:
        denom = float(denominator or 0)
        if denom <= 0:
            return 0.0
        return clamp01(float(numerator or 0) / denom)
    except (TypeError, ValueError):
        return 0.0


def clamp01(value: float) -> float:
    return min(max(value, 0.0), 1.0)


if __name__ == "__main__":
    sys.exit(main())