Agentswarm/orchestrator/quality.py

"""Quality instrumentation (benchmark Group B): turn a finished run's generated code into a real
TestPassRate, then into Q_quality via the masked/renormalized quality_score.

Pipeline:
  1. gather the files the specialist agents generated (from task results) — split implementation
     files from the swarm's own test files.
  2. grade against the fixture's HELD-OUT tests (authoritative) in the sandbox → TestPassRate.
  3. separately run the swarm's OWN tests as a non-grading signal (collaboration/robustness only).
  4. Q_quality = quality_score(test_pass_rate=<fixture>, code_review_score=None, user_acceptance=None).
     CodeReview / UserAcceptance are not collected here, so the masking rule renormalizes Q_quality
     onto the one present input (rule #9: absent ≠ fabricated 0).

SECURITY: step 2/3 execute model-generated code. They run ONLY when ENABLE_QUALITY_EVAL is set,
and only inside the isolated pod (see orchestrator/sandbox.py security model). Default OFF.
"""
from __future__ import annotations

import asyncio
import json
import logging
import os
from typing import Any, Dict, List, Optional

from benchmark.fixtures import Fixture, load_fixture
from benchmark.metrics import quality_score
from .sandbox import SandboxFile, SandboxIsolationError, assert_isolated, run_tests

logger = logging.getLogger(__name__)


def quality_eval_enabled() -> bool:
    return os.getenv("ENABLE_QUALITY_EVAL", "false").lower() in {"1", "true", "yes"}


def assert_quality_eval_safe() -> None:
    """Fail-closed startup check: if quality eval is ON, isolation MUST be confirmed.

    Called at orchestrator startup so a misconfiguration (eval enabled, isolation not confirmed)
    is a LOUD hard failure at boot — not a silent per-run skip and not an ops-only convention.
    """
    if quality_eval_enabled():
        assert_isolated()  # raises SandboxIsolationError if HEICODE_SANDBOX_ISOLATED is unset


def _is_test_file(path: str) -> bool:
    base = os.path.basename(path or "")
    return base.startswith("test_") and base.endswith(".py")


def collect_generated_files(tasks) -> Dict[str, List[SandboxFile]]:
    """Reconstruct files the agents wrote, from task results. Last writer wins per path.

    Returns {"impl": [...non-test files...], "agent_tests": [...the swarm's own test_*.py...]}.
    """
    impl: Dict[str, str] = {}
    agent_tests: Dict[str, str] = {}
    for task in tasks:
        result = getattr(task, "result", None)
        if not result:
            continue
        try:
            data = json.loads(result) if isinstance(result, str) else result
        except Exception:
            continue
        subtasks = data.get("subtasks") if isinstance(data, dict) else None
        file_groups = [data] + list(subtasks or []) if isinstance(data, dict) else []
        for group in file_groups:
            for f in (group.get("files") or []):
                if not isinstance(f, dict):
                    continue
                if (f.get("action") or "write") == "delete":
                    continue
                path = f.get("path")
                content = f.get("content")
                if not path or content is None:
                    continue
                (agent_tests if _is_test_file(path) else impl)[path] = content
    return {
        "impl": [SandboxFile(p, c) for p, c in impl.items()],
        "agent_tests": [SandboxFile(os.path.basename(p), c) for p, c in agent_tests.items()],
    }


async def evaluate_run_quality(run, tasks) -> Optional[Dict[str, Any]]:
    """Grade a completed run's code against its fixture. Returns a quality dict or None.

    None means: quality eval disabled, no fixture bound, or nothing to grade — caller leaves
    Q_quality/reward NaN (coverage=False). Never fabricates a score.
    """
    if not quality_eval_enabled():
        return None
    # Fail-closed: never execute generated code without confirmed isolation, even if a fixture
    # is bound and eval is enabled. Surface loudly rather than silently skipping.
    assert_isolated()  # raises SandboxIsolationError if HEICODE_SANDBOX_ISOLATED is unset
    fixture_id = (run.metadata or {}).get("benchmark_fixture_id")
    if not fixture_id:
        return None
    try:
        fixture: Fixture = load_fixture(fixture_id)
    except Exception as exc:
        logger.warning("quality eval: cannot load fixture %s: %s", fixture_id, exc)
        return None

    files = collect_generated_files(tasks)
    impl = files["impl"]
    agent_tests = files["agent_tests"]

    quality: Dict[str, Any] = {
        "fixture_id": fixture.id,
        "expects_code": fixture.expects_code,
        "test_pass_rate": None,
        "code_review_score": None,   # not collected (no reviewer wired)
        "user_acceptance": None,     # not collected (external Manager/human signal)
        "agent_test_pass_rate": None,  # SIGNAL ONLY — never part of the grade
        "target_time_seconds": fixture.target_time_seconds,
    }

    # 1) Authoritative grade: held-out fixture tests against the implementation files.
    if fixture.expects_code and fixture.test_files and impl:
        fix_files = [SandboxFile(t.path, t.content) for t in fixture.test_files]
        result = await asyncio.to_thread(run_tests, impl, fix_files)
        quality["test_pass_rate"] = result.pass_rate
        quality["fixture_total"] = result.total
        quality["fixture_passed"] = result.passed
        if result.error:
            quality["fixture_error"] = result.error

    # 2) Signal only: the swarm's own tests (collaboration/robustness, NOT the grade).
    if agent_tests and impl:
        sig = await asyncio.to_thread(run_tests, impl, agent_tests)
        quality["agent_test_pass_rate"] = sig.pass_rate

    # 3) Q_quality via the masked/renormalized blend (absent inputs drop out).
    quality["q_quality"] = quality_score(
        test_pass_rate=quality["test_pass_rate"],
        code_review_score=quality["code_review_score"],
        user_acceptance=quality["user_acceptance"],
    )
    return quality