Initial public upload from Projekte/AgentArena

3b0d6b3 verified about 1 month ago

15.4 kB

import asyncio
import json
import logging
import uuid
import time
import os
from datetime import datetime

from sqlalchemy.orm import Session

from database import SessionLocal
from models import Task, Agent, Run, RunStatus
from core.sandbox import Sandbox
from adapters import get_adapter
from adapters.base import RunContext
from core.metrics import (
    compute_cost,
    parse_pytest_output,
    compute_trajectory_efficiency,
    analyze_trajectory,
    compute_diff_minimality,
    compute_composite_score,
)
from core.judge import Judge, JudgeResult
from core.alignment import ALIGNMENT_RUBRIC_VERSION

logger = logging.getLogger(__name__)


class Runner:
    def __init__(self):
        self.judge = Judge()
        self._active_runs: dict[int, dict] = {}  # run_id -> progress info

    async def run_single(self, db: Session, task: Task, agent: Agent) -> Run:
        """Execute a single benchmark run."""
        agent_config = json.loads(agent.config_json or "{}")
        replicate_index = self._next_replicate_index(db, task.id, agent.id)
        # 1. Create Run record
        run = Run(
            task_id=task.id,
            agent_id=agent.id,
            status=RunStatus.running,
            task_family=self._task_family(task),
            seed=str(agent_config.get("seed", "")) or None,
            replicate_index=replicate_index,
            experiment_id=agent_config.get("experiment_id"),
            arm_id=agent_config.get("arm_id") or agent.name,
            alignment_rubric_version=ALIGNMENT_RUBRIC_VERSION,
            created_at=datetime.utcnow(),
        )
        db.add(run)
        db.commit()
        db.refresh(run)

        self._active_runs[run.id] = {"status": "starting", "started_at": time.time()}
        sandbox = None
        judge_sandbox = None
        start_time = time.time()

        try:
            # 2. Create agent sandbox
            self._active_runs[run.id]["status"] = "creating_sandbox"
            sandbox = await asyncio.to_thread(Sandbox.create, task)

            # 3. Get adapter and build RunContext
            self._active_runs[run.id]["status"] = "running_agent"
            adapter = get_adapter(agent.adapter_type, agent_config)

            ctx = RunContext(
                work_dir="/workspace",
                task_prompt=task.description or task.title,
                success_cmd=task.success_cmd,
                time_limit_s=task.time_limit_s,
                token_limit=task.token_limit,
                repo_url=task.repo_url or "",
                repo_ref=task.repo_ref or "main",
            )

            # 4. Run adapter — returns RunResult dataclass
            adapter_result = await adapter.run(ctx, sandbox)

            # 5. Get diff — prefer adapter-provided diff (e.g. ClaudeCode runs on host),
            #    fall back to sandbox.get_diff() for container-based adapters.
            self._active_runs[run.id]["status"] = "collecting_diff"
            if adapter_result.diff_text:
                diff_text = adapter_result.diff_text
            else:
                diff_text = await asyncio.to_thread(sandbox.get_diff)

            # 5b. Destroy agent container — judge must not share it
            await asyncio.to_thread(sandbox.destroy)
            sandbox = None

            # 6. Create isolated judge container (fresh clone + patch applied)
            self._active_runs[run.id]["status"] = "creating_judge_sandbox"
            judge_sandbox = await asyncio.to_thread(
                Sandbox.create_judge_sandbox, task, diff_text
            )

            # Abort if patch failed to apply — running judge on unpatched repo gives false negatives
            if diff_text and diff_text.strip() and not judge_sandbox.patch_applied:
                patch_output = judge_sandbox.patch_output or "unknown"
                run.invalid_patch_flag = 1
                raise RuntimeError(f"Patch failed to apply in judge container: {patch_output}")

            # 7. Judge in the isolated container
            self._active_runs[run.id]["status"] = "judging"
            judge_result = await asyncio.to_thread(
                self.judge.judge_run, judge_sandbox, task.success_cmd
            )

            # 7. Compute metrics
            # Use adapter-measured duration (only CLI time), not wall-clock
            # which includes sandbox creation + judge overhead.
            agent_duration_s = adapter_result.duration_s
            end_to_end_duration_s = time.time() - start_time
            tokens_input = adapter_result.tokens_input
            tokens_output = adapter_result.tokens_output
            cost_usd = adapter_result.cost_usd

            trajectory = adapter_result.trajectory

            # Trajectory and diff analysis
            traj_metrics = analyze_trajectory(trajectory)
            diff_metrics = compute_diff_minimality(diff_text)

            functional_success = 1 if judge_result.functional_score >= 1.0 else 0
            quality_score = (
                judge_result.regression_score * 0.5
                + judge_result.lint_score * 0.25
                + judge_result.complexity_score * 0.25
            )
            final_status = RunStatus.success if functional_success else RunStatus.failure
            tests_total = judge_result.tests_total if judge_result.tests_total > 0 else adapter_result.tests_total
            tests_passed = judge_result.tests_passed if judge_result.tests_total > 0 else adapter_result.tests_passed
            test_pass_rate = (tests_passed / tests_total) if tests_total > 0 else None

            # Build full metrics JSON
            judge_metrics = {
                "functional_score": judge_result.functional_score,
                "functional_success": functional_success,
                "regression_score": judge_result.regression_score,
                "regression_tests_total": judge_result.regression_tests_total,
                "regression_tests_passed": judge_result.regression_tests_passed,
                "lint_score": judge_result.lint_score,
                "lint_delta": judge_result.lint_delta,
                "complexity_score": judge_result.complexity_score,
                "complexity_delta": judge_result.complexity_delta,
                "quality_score": quality_score,
            }

            # Normalize cost to 0-1 efficiency score (lower cost = higher score)
            efficiency = 1.0 / (1.0 + cost_usd / 0.50) if cost_usd > 0 else 1.0

            composite_score = compute_composite_score({
                "functionality": judge_result.functional_score,
                "directness_score": traj_metrics.get("directness_score", 0),
                "efficiency": efficiency,
                "self_correction_rate": traj_metrics.get("self_correction_rate", 0),
                "churn": diff_metrics.get("churn", 0),
                "loop_count": traj_metrics.get("loop_count", 0),
                "total_steps": traj_metrics.get("total_steps", 1),
            })

            full_metrics = {
                "trajectory": traj_metrics,
                "diff": diff_metrics,
                "judge": judge_metrics,
                "composite_score": composite_score,
                "eval": {
                    "functional_success": functional_success,
                    "test_pass_rate": test_pass_rate,
                    "tests_total": tests_total,
                    "tests_passed": tests_passed,
                    "agent_duration_s": agent_duration_s,
                    "end_to_end_duration_s": round(end_to_end_duration_s, 2),
                    "timeout_flag": 1 if adapter_result.status == RunStatus.timeout else 0,
                    "tool_failure_flag": 1 if adapter_result.errors_count > 0 else 0,
                    "invalid_patch_flag": 0,
                    "task_family": run.task_family,
                    "seed": run.seed,
                    "replicate_index": replicate_index,
                    "experiment_id": run.experiment_id,
                    "arm_id": run.arm_id,
                    "alignment_rubric_version": ALIGNMENT_RUBRIC_VERSION,
                },
            }

            # 8. Update Run record
            run.status = final_status
            run.duration_s = agent_duration_s
            run.agent_duration_s = agent_duration_s
            run.end_to_end_duration_s = round(end_to_end_duration_s, 2)
            run.tokens_input = tokens_input
            run.tokens_output = tokens_output
            run.cost_usd = cost_usd
            run.tests_total = tests_total
            run.tests_passed = tests_passed
            run.functional_success = functional_success
            run.test_pass_rate = test_pass_rate
            run.quality_score = round(quality_score, 4)
            run.errors_count = adapter_result.errors_count
            run.tool_calls = adapter_result.tool_calls
            run.diff_text = diff_text
            run.trajectory = json.dumps(trajectory)
            run.judge_score = judge_result.overall_score
            run.judge_notes = judge_result.overall_notes
            run.timeout_flag = 1 if adapter_result.status == RunStatus.timeout else 0
            run.tool_failure_flag = 1 if adapter_result.errors_count > 0 else 0
            run.invalid_patch_flag = 0
            run.metrics_json = json.dumps(full_metrics)
            db.commit()
            db.refresh(run)

            # 8b. Fire-and-forget LLM judge (non-blocking)
            if os.environ.get("MINIMAX_API_KEY"):
                asyncio.create_task(
                    _run_llm_judges(db, run.id, task, agent, full_metrics, diff_text, adapter_result.raw_output)
                )

        except asyncio.TimeoutError:
            run.status = RunStatus.timeout
            run.duration_s = time.time() - start_time
            run.agent_duration_s = run.duration_s
            run.end_to_end_duration_s = run.duration_s
            run.judge_notes = f"Run timed out after {task.time_limit_s}s"
            run.timeout_flag = 1
            db.commit()
        except Exception as e:
            run.status = RunStatus.error
            run.duration_s = time.time() - start_time
            run.agent_duration_s = run.duration_s
            run.end_to_end_duration_s = run.duration_s
            run.judge_notes = f"Error: {str(e)}"
            run.tool_failure_flag = 1
            db.commit()
        finally:
            # 9. Destroy sandboxes (agent may already be gone, judge always cleaned up)
            if sandbox is not None:
                await asyncio.to_thread(sandbox.destroy)
            if judge_sandbox is not None:
                await asyncio.to_thread(judge_sandbox.destroy)
            self._active_runs.pop(run.id, None)

        db.refresh(run)
        return run

    async def _run_single_for_k(
        self, db: Session, task: Task, agent: Agent, group_id: str
    ) -> Run:
        """Run a single instance for pass@k, tagging with group_id."""
        run = await self.run_single(db, task, agent)
        run.pass_at_k_group = group_id
        db.commit()
        db.refresh(run)
        return run

    async def run_pass_at_k(
        self, db: Session, task: Task, agent: Agent, k: int = 3
    ) -> list[Run]:
        """Run k parallel benchmark runs for pass@k calculation."""
        group_id = str(uuid.uuid4())[:8]
        task_id, agent_id = task.id, agent.id

        async def _single(i):
            run_db = SessionLocal()
            try:
                t = run_db.query(Task).filter(Task.id == task_id).first()
                a = run_db.query(Agent).filter(Agent.id == agent_id).first()
                run = await self.run_single(run_db, t, a)
                run.pass_at_k_group = group_id
                run_db.commit()
                run_db.refresh(run)
                return run
            finally:
                run_db.close()

        results = await asyncio.gather(*[_single(i) for i in range(k)], return_exceptions=True)
        # Log exceptions but don't crash the entire batch
        clean = []
        for i, r in enumerate(results):
            if isinstance(r, Exception):
                logger.error("pass@k run %d/%d failed: %s", i + 1, k, r)
            else:
                clean.append(r)
        return clean

    def get_progress(self, run_id: int) -> dict | None:
        return self._active_runs.get(run_id)

    def _task_family(self, task: Task) -> str:
        slug = task.slug or ""
        if "_" in slug:
            return slug.split("_", 1)[0]
        if "-" in slug:
            return slug.split("-", 1)[0]
        return slug or "unknown"

    def _next_replicate_index(self, db: Session, task_id: int, agent_id: int) -> int:
        return (
            db.query(Run)
            .filter(Run.task_id == task_id, Run.agent_id == agent_id)
            .count()
            + 1
        )


async def _run_llm_judges(
    db: Session, run_id: int, task: Task, agent: Agent,
    full_metrics: dict, diff_text: str, raw_output: str,
) -> None:
    """Fire-and-forget LLM judges — updates metrics_json async."""
    try:
        from core.llm_judge import llm_judge
        from core.alignment import llm_alignment_judge

        judge_metrics = full_metrics.get("judge", {})
        run_data = {
            "task_title": task.title,
            "task_description": task.description or "",
            "agent_name": agent.name,
            "adapter_type": agent.adapter_type,
            "status": "success" if judge_metrics.get("functional_score", 0) >= 1.0 else "failure",
            "functional_score": judge_metrics.get("functional_score"),
            "functional_success": judge_metrics.get("functional_success"),
            "regression_score": judge_metrics.get("regression_score"),
            "lint_score": judge_metrics.get("lint_score"),
            "complexity_score": judge_metrics.get("complexity_score"),
            "composite_score": full_metrics.get("composite_score"),
            "trajectory": full_metrics.get("trajectory", {}),
            "diff_text": diff_text or "",
            "judge_notes": "",
            "tests_total": full_metrics.get("eval", {}).get("tests_total", 0),
            "tests_passed": full_metrics.get("eval", {}).get("tests_passed", 0),
            "raw_output": raw_output or "",
        }

        quality_result = await llm_judge(run_data)
        alignment_result = await llm_alignment_judge(run_data)
        if quality_result:
            full_metrics["llm_judge"] = quality_result
        if alignment_result:
            full_metrics["alignment"] = alignment_result
        if quality_result or alignment_result:
            from database import SessionLocal
            judge_db = SessionLocal()
            try:
                from models import Run
                r = judge_db.query(Run).filter(Run.id == run_id).first()
                if r:
                    import json
                    if alignment_result:
                        r.alignment_score_llm = alignment_result.get("alignment_score")
                        r.alignment_rubric_version = alignment_result.get("rubric_version")
                    r.metrics_json = json.dumps(full_metrics)
                    judge_db.commit()
            finally:
                judge_db.close()
    except Exception as e:
        import logging
        logging.getLogger(__name__).warning("LLM judge background task failed: %s", e)

	import asyncio
	import json
	import logging
	import uuid
	import time
	import os
	from datetime import datetime

	from sqlalchemy.orm import Session

	from database import SessionLocal
	from models import Task, Agent, Run, RunStatus
	from core.sandbox import Sandbox
	from adapters import get_adapter
	from adapters.base import RunContext
	from core.metrics import (
	compute_cost,
	parse_pytest_output,
	compute_trajectory_efficiency,
	analyze_trajectory,
	compute_diff_minimality,
	compute_composite_score,
	)
	from core.judge import Judge, JudgeResult
	from core.alignment import ALIGNMENT_RUBRIC_VERSION

	logger = logging.getLogger(__name__)


	class Runner:
	def __init__(self):
	self.judge = Judge()
	self._active_runs: dict[int, dict] = {} # run_id -> progress info

	async def run_single(self, db: Session, task: Task, agent: Agent) -> Run:
	"""Execute a single benchmark run."""
	agent_config = json.loads(agent.config_json or "{}")
	replicate_index = self._next_replicate_index(db, task.id, agent.id)
	# 1. Create Run record
	run = Run(
	task_id=task.id,
	agent_id=agent.id,
	status=RunStatus.running,
	task_family=self._task_family(task),
	seed=str(agent_config.get("seed", "")) or None,
	replicate_index=replicate_index,
	experiment_id=agent_config.get("experiment_id"),
	arm_id=agent_config.get("arm_id") or agent.name,
	alignment_rubric_version=ALIGNMENT_RUBRIC_VERSION,
	created_at=datetime.utcnow(),
	)
	db.add(run)
	db.commit()
	db.refresh(run)

	self._active_runs[run.id] = {"status": "starting", "started_at": time.time()}
	sandbox = None
	judge_sandbox = None
	start_time = time.time()

	try:
	# 2. Create agent sandbox
	self._active_runs[run.id]["status"] = "creating_sandbox"
	sandbox = await asyncio.to_thread(Sandbox.create, task)

	# 3. Get adapter and build RunContext
	self._active_runs[run.id]["status"] = "running_agent"
	adapter = get_adapter(agent.adapter_type, agent_config)

	ctx = RunContext(
	work_dir="/workspace",
	task_prompt=task.description or task.title,
	success_cmd=task.success_cmd,
	time_limit_s=task.time_limit_s,
	token_limit=task.token_limit,
	repo_url=task.repo_url or "",
	repo_ref=task.repo_ref or "main",
	)

	# 4. Run adapter — returns RunResult dataclass
	adapter_result = await adapter.run(ctx, sandbox)

	# 5. Get diff — prefer adapter-provided diff (e.g. ClaudeCode runs on host),
	# fall back to sandbox.get_diff() for container-based adapters.
	self._active_runs[run.id]["status"] = "collecting_diff"
	if adapter_result.diff_text:
	diff_text = adapter_result.diff_text
	else:
	diff_text = await asyncio.to_thread(sandbox.get_diff)

	# 5b. Destroy agent container — judge must not share it
	await asyncio.to_thread(sandbox.destroy)
	sandbox = None

	# 6. Create isolated judge container (fresh clone + patch applied)
	self._active_runs[run.id]["status"] = "creating_judge_sandbox"
	judge_sandbox = await asyncio.to_thread(
	Sandbox.create_judge_sandbox, task, diff_text
	)

	# Abort if patch failed to apply — running judge on unpatched repo gives false negatives
	if diff_text and diff_text.strip() and not judge_sandbox.patch_applied:
	patch_output = judge_sandbox.patch_output or "unknown"
	run.invalid_patch_flag = 1
	raise RuntimeError(f"Patch failed to apply in judge container: {patch_output}")

	# 7. Judge in the isolated container
	self._active_runs[run.id]["status"] = "judging"
	judge_result = await asyncio.to_thread(
	self.judge.judge_run, judge_sandbox, task.success_cmd
	)

	# 7. Compute metrics
	# Use adapter-measured duration (only CLI time), not wall-clock
	# which includes sandbox creation + judge overhead.
	agent_duration_s = adapter_result.duration_s
	end_to_end_duration_s = time.time() - start_time
	tokens_input = adapter_result.tokens_input
	tokens_output = adapter_result.tokens_output
	cost_usd = adapter_result.cost_usd

	trajectory = adapter_result.trajectory

	# Trajectory and diff analysis
	traj_metrics = analyze_trajectory(trajectory)
	diff_metrics = compute_diff_minimality(diff_text)

	functional_success = 1 if judge_result.functional_score >= 1.0 else 0
	quality_score = (
	judge_result.regression_score * 0.5
	+ judge_result.lint_score * 0.25
	+ judge_result.complexity_score * 0.25
	)
	final_status = RunStatus.success if functional_success else RunStatus.failure
	tests_total = judge_result.tests_total if judge_result.tests_total > 0 else adapter_result.tests_total
	tests_passed = judge_result.tests_passed if judge_result.tests_total > 0 else adapter_result.tests_passed
	test_pass_rate = (tests_passed / tests_total) if tests_total > 0 else None

	# Build full metrics JSON
	judge_metrics = {
	"functional_score": judge_result.functional_score,
	"functional_success": functional_success,
	"regression_score": judge_result.regression_score,
	"regression_tests_total": judge_result.regression_tests_total,
	"regression_tests_passed": judge_result.regression_tests_passed,
	"lint_score": judge_result.lint_score,
	"lint_delta": judge_result.lint_delta,
	"complexity_score": judge_result.complexity_score,
	"complexity_delta": judge_result.complexity_delta,
	"quality_score": quality_score,
	}

	# Normalize cost to 0-1 efficiency score (lower cost = higher score)
	efficiency = 1.0 / (1.0 + cost_usd / 0.50) if cost_usd > 0 else 1.0

	composite_score = compute_composite_score({
	"functionality": judge_result.functional_score,
	"directness_score": traj_metrics.get("directness_score", 0),
	"efficiency": efficiency,
	"self_correction_rate": traj_metrics.get("self_correction_rate", 0),
	"churn": diff_metrics.get("churn", 0),
	"loop_count": traj_metrics.get("loop_count", 0),
	"total_steps": traj_metrics.get("total_steps", 1),
	})

	full_metrics = {
	"trajectory": traj_metrics,
	"diff": diff_metrics,
	"judge": judge_metrics,
	"composite_score": composite_score,
	"eval": {
	"functional_success": functional_success,
	"test_pass_rate": test_pass_rate,
	"tests_total": tests_total,
	"tests_passed": tests_passed,
	"agent_duration_s": agent_duration_s,
	"end_to_end_duration_s": round(end_to_end_duration_s, 2),
	"timeout_flag": 1 if adapter_result.status == RunStatus.timeout else 0,
	"tool_failure_flag": 1 if adapter_result.errors_count > 0 else 0,
	"invalid_patch_flag": 0,
	"task_family": run.task_family,
	"seed": run.seed,
	"replicate_index": replicate_index,
	"experiment_id": run.experiment_id,
	"arm_id": run.arm_id,
	"alignment_rubric_version": ALIGNMENT_RUBRIC_VERSION,
	},
	}

	# 8. Update Run record
	run.status = final_status
	run.duration_s = agent_duration_s
	run.agent_duration_s = agent_duration_s
	run.end_to_end_duration_s = round(end_to_end_duration_s, 2)
	run.tokens_input = tokens_input
	run.tokens_output = tokens_output
	run.cost_usd = cost_usd
	run.tests_total = tests_total
	run.tests_passed = tests_passed
	run.functional_success = functional_success
	run.test_pass_rate = test_pass_rate
	run.quality_score = round(quality_score, 4)
	run.errors_count = adapter_result.errors_count
	run.tool_calls = adapter_result.tool_calls
	run.diff_text = diff_text
	run.trajectory = json.dumps(trajectory)
	run.judge_score = judge_result.overall_score
	run.judge_notes = judge_result.overall_notes
	run.timeout_flag = 1 if adapter_result.status == RunStatus.timeout else 0
	run.tool_failure_flag = 1 if adapter_result.errors_count > 0 else 0
	run.invalid_patch_flag = 0
	run.metrics_json = json.dumps(full_metrics)
	db.commit()
	db.refresh(run)

	# 8b. Fire-and-forget LLM judge (non-blocking)
	if os.environ.get("MINIMAX_API_KEY"):
	asyncio.create_task(
	_run_llm_judges(db, run.id, task, agent, full_metrics, diff_text, adapter_result.raw_output)
	)

	except asyncio.TimeoutError:
	run.status = RunStatus.timeout
	run.duration_s = time.time() - start_time
	run.agent_duration_s = run.duration_s
	run.end_to_end_duration_s = run.duration_s
	run.judge_notes = f"Run timed out after {task.time_limit_s}s"
	run.timeout_flag = 1
	db.commit()
	except Exception as e:
	run.status = RunStatus.error
	run.duration_s = time.time() - start_time
	run.agent_duration_s = run.duration_s
	run.end_to_end_duration_s = run.duration_s
	run.judge_notes = f"Error: {str(e)}"
	run.tool_failure_flag = 1
	db.commit()
	finally:
	# 9. Destroy sandboxes (agent may already be gone, judge always cleaned up)
	if sandbox is not None:
	await asyncio.to_thread(sandbox.destroy)
	if judge_sandbox is not None:
	await asyncio.to_thread(judge_sandbox.destroy)
	self._active_runs.pop(run.id, None)

	db.refresh(run)
	return run

	async def _run_single_for_k(
	self, db: Session, task: Task, agent: Agent, group_id: str
	) -> Run:
	"""Run a single instance for pass@k, tagging with group_id."""
	run = await self.run_single(db, task, agent)
	run.pass_at_k_group = group_id
	db.commit()
	db.refresh(run)
	return run

	async def run_pass_at_k(
	self, db: Session, task: Task, agent: Agent, k: int = 3
	) -> list[Run]:
	"""Run k parallel benchmark runs for pass@k calculation."""
	group_id = str(uuid.uuid4())[:8]
	task_id, agent_id = task.id, agent.id

	async def _single(i):
	run_db = SessionLocal()
	try:
	t = run_db.query(Task).filter(Task.id == task_id).first()
	a = run_db.query(Agent).filter(Agent.id == agent_id).first()
	run = await self.run_single(run_db, t, a)
	run.pass_at_k_group = group_id
	run_db.commit()
	run_db.refresh(run)
	return run
	finally:
	run_db.close()

	results = await asyncio.gather(*[_single(i) for i in range(k)], return_exceptions=True)
	# Log exceptions but don't crash the entire batch
	clean = []
	for i, r in enumerate(results):
	if isinstance(r, Exception):
	logger.error("pass@k run %d/%d failed: %s", i + 1, k, r)
	else:
	clean.append(r)
	return clean

	def get_progress(self, run_id: int) -> dict \| None:
	return self._active_runs.get(run_id)

	def _task_family(self, task: Task) -> str:
	slug = task.slug or ""
	if "_" in slug:
	return slug.split("_", 1)[0]
	if "-" in slug:
	return slug.split("-", 1)[0]
	return slug or "unknown"

	def _next_replicate_index(self, db: Session, task_id: int, agent_id: int) -> int:
	return (
	db.query(Run)
	.filter(Run.task_id == task_id, Run.agent_id == agent_id)
	.count()
	+ 1
	)


	async def _run_llm_judges(
	db: Session, run_id: int, task: Task, agent: Agent,
	full_metrics: dict, diff_text: str, raw_output: str,
	) -> None:
	"""Fire-and-forget LLM judges — updates metrics_json async."""
	try:
	from core.llm_judge import llm_judge
	from core.alignment import llm_alignment_judge

	judge_metrics = full_metrics.get("judge", {})
	run_data = {
	"task_title": task.title,
	"task_description": task.description or "",
	"agent_name": agent.name,
	"adapter_type": agent.adapter_type,
	"status": "success" if judge_metrics.get("functional_score", 0) >= 1.0 else "failure",
	"functional_score": judge_metrics.get("functional_score"),
	"functional_success": judge_metrics.get("functional_success"),
	"regression_score": judge_metrics.get("regression_score"),
	"lint_score": judge_metrics.get("lint_score"),
	"complexity_score": judge_metrics.get("complexity_score"),
	"composite_score": full_metrics.get("composite_score"),
	"trajectory": full_metrics.get("trajectory", {}),
	"diff_text": diff_text or "",
	"judge_notes": "",
	"tests_total": full_metrics.get("eval", {}).get("tests_total", 0),
	"tests_passed": full_metrics.get("eval", {}).get("tests_passed", 0),
	"raw_output": raw_output or "",
	}

	quality_result = await llm_judge(run_data)
	alignment_result = await llm_alignment_judge(run_data)
	if quality_result:
	full_metrics["llm_judge"] = quality_result
	if alignment_result:
	full_metrics["alignment"] = alignment_result
	if quality_result or alignment_result:
	from database import SessionLocal
	judge_db = SessionLocal()
	try:
	from models import Run
	r = judge_db.query(Run).filter(Run.id == run_id).first()
	if r:
	import json
	if alignment_result:
	r.alignment_score_llm = alignment_result.get("alignment_score")
	r.alignment_rubric_version = alignment_result.get("rubric_version")
	r.metrics_json = json.dumps(full_metrics)
	judge_db.commit()
	finally:
	judge_db.close()
	except Exception as e:
	import logging
	logging.getLogger(__name__).warning("LLM judge background task failed: %s", e)