Score reward against targeted tests — FAIL_TO_PASS/PASS_TO_PASS (#2b)

README.md

@@ -384,13 +384,14 @@ Keep `.env` local. It is already ignored by the repository template and should n
 - Applies the gold patch.
 - Runs the configured test command in Docker after the patch.
 - Runs the post-patch command a second time as a deterministic rerun check.
+- For pytest commands, collects a per-test JUnit report from the before/after runs and derives the targeted test sets `fail_to_pass` (failed before, pass after) and `pass_to_pass` (pass both) — the SWE-bench-style scoping spec.
 - Writes `.forge/tasks/<task_id>/verification.json`.
 
 ### `forge package <task_id>`
 
 Creates `taskpacks/<task_id>/` with:
 
-- `task.json`
+- `task.json` (task metadata plus an `eval` block carrying the `fail_to_pass`/`pass_to_pass` test sets)
 - `prompt.md`
 - `gold.patch`
 - `reward.py`
@@ -412,10 +413,10 @@ Runs the taskpack's standalone reward script and returns JSON:
 }
 ```
 
-For v1, the reward is binary:
+The reward is binary:
 
-- `1.0` if the configured test command exits with code `0`
-- `0.0` otherwise
+- If the task has a scoping spec (`eval.fail_to_pass` from a pytest task), the score is `1.0` only when every targeted test passes — all `fail_to_pass` tests pass **and** no `pass_to_pass` test regresses — so unrelated failing or flaky tests in the repo don't mask the signal. The reward re-collects a per-test JUnit report and scores against exactly those node ids.
+- Otherwise (non-pytest tasks, or no spec) it falls back to whole-suite scoring: `1.0` if the configured test command exits with code `0`, else `0.0`.
 
 `error` is `null` on success. On a plain test failure it carries `"Test command exited with code <n>"`; on an infrastructure failure (missing artifacts, Docker build failure, or timeout) it carries the corresponding message.
 
@@ -493,6 +494,7 @@ The configured test command runs inside the container with a timeout, with no ne
 ## Current Limitations
 
 - Python-only Dockerfile generation.
+- Test-scoped reward (`fail_to_pass`/`pass_to_pass`) is derived for pytest commands only; other tasks fall back to whole-suite exit-code scoring. The scoping is computed from the combined gold patch (test files are not yet separated from the fix).
 - No hidden tests or leakage detection.
 - No agent rollout orchestration.
 - No model training, GRPO, LoRA, or fine-tuning pipeline.

forge/docker_runner.py

@@ -97,8 +97,14 @@ def run_tests_in_docker(
     timeout_seconds: int,
     image_tag_prefix: str,
     dockerfile_path: Path | None = None,
+    exec_command: str | None = None,
 ) -> DockerRunResult:
-    """Build a Docker image from repo_path and run the configured test command."""
+    """Build a Docker image from repo_path and run the configured test command.
+
+    ``exec_command`` overrides the shell command actually run in the container
+    (e.g. a pytest command wrapped to emit a JUnit report); the recorded
+    ``command`` stays ``test_command`` for display.
+    """
 
     started_at = time.monotonic()
     if shutil.which("docker") is None:
@@ -171,7 +177,7 @@ def run_tests_in_docker(
                 duration_seconds=time.monotonic() - started_at,
             )
 
-        run_command = ["docker", "run", "--rm", "--name", container_name, *CONTAINER_RUN_FLAGS, image_tag, "sh", "-lc", test_command]
+        run_command = ["docker", "run", "--rm", "--name", container_name, *CONTAINER_RUN_FLAGS, image_tag, "sh", "-lc", exec_command or test_command]
         try:
             run = subprocess.run(
                 run_command,

forge/task_builder.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import shutil
 from pathlib import Path
 from typing import Callable
@@ -10,6 +11,7 @@
 from forge.github_pr import fetch_pr_diff, fetch_pull_request
 from forge.patch_utils import apply_patch_file, check_patch
 from forge.task_schema import CommandRun, TaskMetadata, VerificationResult, detect_test_infrastructure_failure
+from forge.test_report import derive_test_deltas, extract_report, parse_junit_xml, wrap_pytest_command
 
 
 FORGE_DIR = Path(".forge")
@@ -145,6 +147,12 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo
     before_patch: CommandRun | None = None
     after_patch: CommandRun | None = None
     deterministic_rerun: CommandRun | None = None
+    before_results: dict[str, str] = {}
+    after_results: dict[str, str] = {}
+
+    # Wrap pytest commands to emit a per-test JUnit report; None for non-pytest
+    # commands, which fall back to whole-suite exit-code scoring.
+    exec_command = wrap_pytest_command(metadata.test_command)
 
     try:
         ensure_commit_available(repo_dir, metadata.base_commit)
@@ -174,8 +182,10 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo
                 metadata.test_command,
                 timeout_seconds=metadata.timeout_seconds,
                 image_tag_prefix=f"forge-{task_id}-before",
+                exec_command=exec_command,
             )
             before_patch = _command_run("before_patch", before)
+            before_results = parse_junit_xml(extract_report(before.stdout))
 
         if base_commit_found and patch_applies:
             log("Applying the gold patch")
@@ -189,8 +199,10 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo
                     metadata.test_command,
                     timeout_seconds=metadata.timeout_seconds,
                     image_tag_prefix=f"forge-{task_id}-after",
+                    exec_command=exec_command,
                 )
                 after_patch = _command_run("after_patch", after)
+                after_results = parse_junit_xml(extract_report(after.stdout))
 
                 # Attests post-patch idempotence (same patched tree, freshly rebuilt
                 # image, run twice). It does NOT re-check the pre-patch baseline, so a
@@ -201,6 +213,7 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo
                     metadata.test_command,
                     timeout_seconds=metadata.timeout_seconds,
                     image_tag_prefix=f"forge-{task_id}-rerun",
+                    exec_command=exec_command,
                 )
                 deterministic_rerun = _command_run("deterministic_rerun", rerun)
     finally:
@@ -229,6 +242,13 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo
     tests_pass_after_patch = after_patch is not None and test_environment_success and after_patch.docker_build_success and after_patch.exit_code == 0
     deterministic_rerun_success = deterministic_rerun is not None and test_environment_success and deterministic_rerun.docker_build_success and deterministic_rerun.exit_code == 0
 
+    # Derive the targeted test sets only when both runs produced a parseable report;
+    # otherwise leave them empty so reward falls back to whole-suite scoring.
+    if before_results and after_results and test_environment_success:
+        fail_to_pass, pass_to_pass = derive_test_deltas(before_results, after_results)
+    else:
+        fail_to_pass, pass_to_pass = [], []
+
     verification = VerificationResult(
         task_id=task_id,
         base_commit_found=base_commit_found,
@@ -238,6 +258,8 @@ def verify_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_lo
         docker_build_success=docker_build_success,
         test_environment_success=test_environment_success,
         deterministic_rerun_success=deterministic_rerun_success,
+        fail_to_pass=fail_to_pass,
+        pass_to_pass=pass_to_pass,
         before_patch=before_patch,
         after_patch=after_patch,
         deterministic_rerun=deterministic_rerun,
@@ -292,8 +314,17 @@ def _reward_script() -> str:
 import tempfile
 import time
 import uuid
+import xml.etree.ElementTree as ET
 from pathlib import Path
 
+# Sentinels for embedding a JUnit report in container stdout (kept in sync with
+# forge.test_report). Targeted scoring uses the report; otherwise we fall back to
+# the test command's exit code.
+REPORT_PATH = "/tmp/forge-report.xml"
+REPORT_START = "<<<FORGE_JUNIT_START>>>"
+REPORT_END = "<<<FORGE_JUNIT_END>>>"
+_PYTEST_RE = re.compile(r"(^|\s)(pytest|py\.test)(\s|$)|python[0-9.]*\s+-m\s+pytest")
+
 
 def emit(score: float, tests_passed: bool, error: str | None) -> int:
     print(json.dumps({"score": score, "tests_passed": tests_passed, "error": error}))
@@ -305,6 +336,55 @@ def slug(value: str) -> str:
     return normalized or "task"
 
 
+def is_pytest_command(test_command: str) -> bool:
+    return bool(_PYTEST_RE.search(test_command or ""))
+
+
+def wrap_pytest_command(test_command: str) -> str:
+    inner = f"{test_command} --junitxml={REPORT_PATH}"
+    return (
+        f"{inner}; __forge_code=$?; "
+        f"echo '{REPORT_START}'; cat {REPORT_PATH} 2>/dev/null; echo '{REPORT_END}'; "
+        f"exit $__forge_code"
+    )
+
+
+def extract_report(stdout: str):
+    if not stdout or REPORT_START not in stdout or REPORT_END not in stdout:
+        return None
+    start = stdout.index(REPORT_START) + len(REPORT_START)
+    end = stdout.index(REPORT_END, start)
+    xml = stdout[start:end].strip()
+    return xml or None
+
+
+def parse_junit_xml(xml_text):
+    results = {}
+    if not xml_text:
+        return results
+    try:
+        root = ET.fromstring(xml_text)
+    except ET.ParseError:
+        return results
+    for case in root.iter("testcase"):
+        classname = case.get("classname") or ""
+        name = case.get("name") or ""
+        nodeid = f"{classname}::{name}" if classname else name
+        status = "passed"
+        for child in case:
+            tag = child.tag.lower()
+            if tag in ("failure", "error", "skipped"):
+                status = "failed" if tag == "failure" else tag
+                break
+        results[nodeid] = status
+    return results
+
+
+def score_targeted(results, fail_to_pass, pass_to_pass):
+    ok = {nodeid for nodeid, status in results.items() if status == "passed"}
+    return all(t in ok for t in fail_to_pass) and all(t in ok for t in pass_to_pass)
+
+
 def stage_repo_with_patch(repo_dir: Path, patch_file: Path) -> tuple[Path | None, str | None]:
     """Copy repo_dir into a temp dir and apply the candidate patch. Returns (work_dir, error)."""
 
@@ -367,6 +447,12 @@ def main() -> int:
         image_tag = f"forge-reward-{slug(task['id'])}-{uuid.uuid4().hex[:12]}"
         container_name = f"{image_tag}-run"
 
+        eval_spec = task.get("eval") or {}
+        fail_to_pass = list(eval_spec.get("fail_to_pass") or [])
+        pass_to_pass = list(eval_spec.get("pass_to_pass") or [])
+        scoped = bool(fail_to_pass) and is_pytest_command(test_command)
+        exec_command = wrap_pytest_command(test_command) if scoped else test_command
+
         started_at = time.monotonic()
         try:
             build = subprocess.run(
@@ -390,14 +476,23 @@ def main() -> int:
                     docker_bin, "run", "--rm", "--name", container_name,
                     "--network", "none", "--memory", "4g", "--cpus", "2",
                     "--pids-limit", "512", "--security-opt", "no-new-privileges",
-                    image_tag, "sh", "-lc", test_command,
+                    image_tag, "sh", "-lc", exec_command,
                 ],
                 text=True,
                 encoding="utf-8",
                 errors="replace",
                 capture_output=True,
                 timeout=timeout_seconds,
             )
+            if scoped:
+                results = parse_junit_xml(extract_report(run.stdout))
+                if results:
+                    if score_targeted(results, fail_to_pass, pass_to_pass):
+                        return emit(1.0, True, None)
+                    missing = [t for t in fail_to_pass if results.get(t) != "passed"]
+                    regressed = [t for t in pass_to_pass if results.get(t) != "passed"]
+                    return emit(0.0, False, f"Targeted tests not satisfied (fail_to_pass missing: {len(missing)}, pass_to_pass regressed: {len(regressed)})")
+                # No parseable report (e.g. collection error): fall back to exit code.
             if run.returncode == 0:
                 return emit(1.0, True, None)
             return emit(0.0, False, f"Test command exited with code {run.returncode}")
@@ -417,6 +512,19 @@ def main() -> int:
 '''
 
 
+def _eval_spec(verification_file: Path) -> dict[str, list[str]]:
+    """Read the targeted test sets from verification.json, robust to partial files."""
+
+    try:
+        data = json.loads(verification_file.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        data = {}
+    return {
+        "fail_to_pass": list(data.get("fail_to_pass") or []),
+        "pass_to_pass": list(data.get("pass_to_pass") or []),
+    }
+
+
 def package_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_log) -> Path:
     root = root or Path.cwd()
     metadata = load_metadata(root, task_id)
@@ -437,7 +545,11 @@ def package_task(task_id: str, *, root: Path | None = None, log: LogFn = _noop_l
     log(f"Checking out base commit {metadata.base_commit} for package snapshot")
     checkout_clean(repo_dir, metadata.base_commit)
 
-    (package_dir / "task.json").write_text(metadata.model_dump_json(indent=2), encoding="utf-8")
+    # The packed task.json is a superset of the metadata: it also carries the eval
+    # spec (targeted test sets) so the standalone reward.py can score without forge.
+    task_payload = json.loads(metadata.model_dump_json())
+    task_payload["eval"] = _eval_spec(verification_file)
+    (package_dir / "task.json").write_text(json.dumps(task_payload, indent=2), encoding="utf-8")
     (package_dir / "prompt.md").write_text(_prompt(metadata), encoding="utf-8")
     (package_dir / "Dockerfile").write_text(generate_python_dockerfile(), encoding="utf-8")
     (package_dir / "reward.py").write_text(_reward_script(), encoding="utf-8")

forge/task_schema.py

@@ -93,6 +93,8 @@ class VerificationResult(BaseModel):
     docker_build_success: bool = False
     test_environment_success: bool = True
     deterministic_rerun_success: bool = False
+    fail_to_pass: list[str] = Field(default_factory=list)
+    pass_to_pass: list[str] = Field(default_factory=list)
     before_patch: CommandRun | None = None
     after_patch: CommandRun | None = None
     deterministic_rerun: CommandRun | None = None