| """Visible verifier: the immediate reward signal the Repair Agent sees.
|
|
|
| 4 weighted components, summed to a scalar. This is what drives the Repair
|
| Agent's GRPO updates each rollout. Multiple independent components were
|
| chosen on purpose, per the reward-engineering survey (arxiv 2408.10215)
|
| and software-tasks survey (arxiv 2601.19100): a single scalar is far
|
| easier to game than a composable rubric.
|
| """
|
| from __future__ import annotations
|
|
|
| import re
|
|
|
| from forgeenv.tasks.models import ExecutionResult, Task
|
|
|
| WEIGHTS: dict[str, float] = {
|
| "script_executes": 1.0,
|
| "loss_decreased": 0.5,
|
| "checkpoint_appeared": 0.3,
|
| "diff_size_penalty": 0.2,
|
| }
|
|
|
|
|
| def compute_visible_reward(
|
| result: ExecutionResult, task: Task
|
| ) -> tuple[float, dict[str, float]]:
|
| """Compute scalar visible reward and per-component breakdown."""
|
|
|
| components: dict[str, float] = {}
|
|
|
| components["script_executes"] = 1.0 if result.exit_code == 0 else 0.0
|
| components["loss_decreased"] = _check_loss_trend(result.stdout)
|
| components["checkpoint_appeared"] = 1.0 if result.checkpoint_exists else 0.0
|
|
|
| original_lines = max(len(task.script_content.splitlines()), 1)
|
| current_lines = (
|
| len(result.script_content.splitlines()) if result.script_content else original_lines
|
| )
|
| diff_ratio = abs(current_lines - original_lines) / original_lines
|
| components["diff_size_penalty"] = -1.0 * diff_ratio if diff_ratio > 0.5 else 0.0
|
|
|
| total = sum(components[k] * WEIGHTS[k] for k in components)
|
| return total, components
|
|
|
|
|
| def _check_loss_trend(stdout: str) -> float:
|
| """Parse stdout for `loss=...` patterns and return the fraction of
|
| consecutive steps where loss strictly decreased."""
|
|
|
| losses: list[float] = []
|
| for line in stdout.splitlines():
|
| match = re.search(r"loss[=:\s]+([\d.]+)", line, re.IGNORECASE)
|
| if match:
|
| try:
|
| losses.append(float(match.group(1)))
|
| except ValueError:
|
| continue
|
|
|
| if len(losses) < 2:
|
| return 0.0
|
|
|
| decreasing_steps = sum(
|
| 1 for i in range(1, len(losses)) if losses[i] < losses[i - 1]
|
| )
|
| return decreasing_steps / (len(losses) - 1)
|
|
|
