feat: add scaleswe v1 taskset + per-task setup/workdir hooks

examples/tasksets/scaleswe_v1/pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "scaleswe-v1"
+version = "0.1.0"
+description = "scaleswe-v1 — Scale-SWE issue-resolving tasks (agentic; sandbox pytest reward)."
+requires-python = ">=3.10"
+dependencies = ["datasets"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build]
+include = ["scaleswe_v1.py", "score.py", "pyproject.toml"]

examples/tasksets/scaleswe_v1/scaleswe_v1.py

@@ -0,0 +1,154 @@
+"""scaleswe-v1 — Scale-SWE (AweAI-Team/Scale-SWE) as a v1 taskset.
+
+Each row ships a per-task Docker image with the repo checked out, `pre_commands` that
+reset it to the base commit on a clean `scaleswe` branch, F2P/P2P pytest ids, and an
+optional `f2p_patch` / `f2p_script` carrying the failing test. `setup` runs the row's
+`pre_commands` in the live runtime before the agent (the runtime's workdir is the row's
+repo). The `solved` reward restores the test files to base (the agent only fixes the
+source), applies the f2p test, then runs the merged F2P+P2P ids through `score.py` —
+which scores 1.0 iff every expected id passed. A v1 port of the v0 ComposableEnv
+`ScaleSWETaskSet`.
+"""
+
+import json
+from pathlib import Path
+
+import verifiers.v1 as vf
+
+DATASET = "AweAI-Team/Scale-SWE"
+
+# The testbed conda env (with the project + pytest installed) and quiet, non-interactive
+# tooling — exported for every command the taskset runs in the sandbox.
+ENV = {
+    "PATH": (
+        "/opt/miniconda3/envs/testbed/bin:/opt/miniconda3/bin:"
+        "/opt/conda/envs/testbed/bin:/opt/conda/bin:"
+        "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
+    ),
+    "PAGER": "cat",
+    "MANPAGER": "cat",
+    "LESS": "-R",
+    "PIP_PROGRESS_BAR": "off",
+    "TQDM_DISABLE": "1",
+    "CI": "1",
+}
+
+# Restore the test files to the base commit so the agent's edits to the source are scored
+# against the original tests (and tests it added are dropped). `$base` comes from the env.
+RESTORE = r"""
+git checkout "$base" -- tests/ test/ Test/ Tests/ 2>/dev/null || true
+git ls-tree -r --name-only "$base" 2>/dev/null | while IFS= read -r path; do
+  case "$path" in
+    test_*.py|*/test_*.py|*_test.py|*/*_test.py|conftest.py|*/conftest.py)
+      git checkout "$base" -- "$path" 2>/dev/null || true ;;
+  esac
+done
+git ls-files 2>/dev/null | while IFS= read -r path; do
+  case "$path" in
+    tests/*|test/*|Test/*|Tests/*|test_*.py|*/test_*.py|*_test.py|*/*_test.py|conftest.py|*/conftest.py)
+      if ! git cat-file -e "$base:$path" 2>/dev/null; then
+        rm -f -- "$path"; git rm -q --cached -- "$path" 2>/dev/null || true
+      fi ;;
+  esac
+done
+"""
+
+PATCH = "/tmp/scaleswe_f2p.patch"
+SCORER = "/tmp/scaleswe_scorer.py"
+SCORER_SRC = (Path(__file__).parent / "score.py").read_bytes()
+
+
+def _ids(raw: str | list[str] | None) -> list[str]:
+    """F2P/P2P ids arrive as a list or a JSON-encoded string; normalize to list[str]."""
+    if not raw:
+        return []
+    if isinstance(raw, list):
+        return [str(t).strip() for t in raw if t]
+    raw = raw.strip()
+    try:
+        parsed = json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        return [raw] if raw else []
+    if isinstance(parsed, list):
+        return [str(t).strip() for t in parsed if t]
+    return [parsed] if isinstance(parsed, str) and parsed else []
+
+
+class ScaleSWETask(vf.Task):
+    base_commit: str
+    """Commit the repo is reset to before the agent runs and tests are scored against."""
+    pre_commands: str
+    """Shell run in the repo before the agent — resets to `base_commit` on branch `scaleswe`."""
+    f2p_patch: str = ""
+    """Optional patch adding the fail-to-pass test (applied before scoring)."""
+    f2p_script: str = ""
+    """Optional test file uploaded as `test_fail_to_pass.py` before scoring."""
+    fail_to_pass: list[str] = []
+    pass_to_pass: list[str] = []
+
+
+class ScaleSWETaskset(vf.Taskset[ScaleSWETask, vf.TasksetConfig]):
+    NEEDS_CONTAINER = True
+
+    def load_tasks(self) -> list[ScaleSWETask]:
+        from datasets import load_dataset
+
+        rows = load_dataset(DATASET, split="train")
+        return [
+            ScaleSWETask(
+                idx=i,
+                name=row["instance_id"],
+                instruction=row["problem_statement"],
+                image=row["image_url"],
+                workdir=row["workdir"],
+                base_commit=row.get("parent_commit") or row.get("base_commit") or "",
+                pre_commands=(row.get("pre_commands") or "")
+                .strip()
+                .removesuffix("\\n"),
+                f2p_patch=row.get("f2p_patch") or "",
+                f2p_script=row.get("f2p_script") or "",
+                fail_to_pass=_ids(row.get("FAIL_TO_PASS")),
+                pass_to_pass=_ids(row.get("PASS_TO_PASS")),
+            )
+            for i, row in enumerate(rows)
+        ]
+
+    async def setup(self, task: ScaleSWETask, runtime: vf.Runtime) -> None:
+        result = await runtime.run(["sh", "-c", task.pre_commands], ENV)
+        if result.exit_code != 0:
+            raise vf.ProgramError(
+                f"scaleswe setup failed ({task.name}): {result.stderr.strip()[-500:]}"
+            )
+
+    @vf.reward(weight=1.0)
+    async def solved(self, task: ScaleSWETask, runtime: vf.Runtime) -> float:
+        test_ids = task.fail_to_pass + task.pass_to_pass
+        if not test_ids:
+            return 0.0
+        await runtime.run(["sh", "-c", RESTORE], {**ENV, "base": task.base_commit})
+        if task.f2p_patch.strip():
+            await self._apply_f2p_patch(runtime, task.f2p_patch)
+        if task.f2p_script:
+            await runtime.write("test_fail_to_pass.py", task.f2p_script.encode())
+        await runtime.write(SCORER, SCORER_SRC)
+        result = await runtime.run(["python", SCORER, json.dumps(test_ids)], ENV)
+        lines = result.stdout.strip().splitlines()
+        return float(lines[-1]) if lines else 0.0
+
+    async def _apply_f2p_patch(self, runtime: vf.Runtime, patch: str) -> None:
+        # Try strict, then whitespace-tolerant, then a fuzzy `patch`; tolerate a partial
+        # `--reject` apply last (mirrors the v0 multi-strategy helper). Scoring catches a
+        # patch that didn't take — a missing test id fails the reward.
+        await runtime.write(PATCH, patch.encode())
+        for cmd in (
+            f"git apply --verbose {PATCH}",
+            f"git apply --verbose --ignore-space-change --ignore-whitespace {PATCH}",
+            f"patch --batch --fuzz=5 -p1 -i {PATCH}",
+            f"git apply --verbose --reject --ignore-whitespace {PATCH} || true",
+        ):
+            if (await runtime.run(["sh", "-c", cmd], ENV)).exit_code == 0:
+                return
+
+
+def load_taskset(config: vf.TasksetConfig) -> ScaleSWETaskset:
+    return ScaleSWETaskset(config)

examples/tasksets/scaleswe_v1/score.py

@@ -0,0 +1,86 @@
+"""Run the merged F2P+P2P pytest ids and print 1.0 iff every expected id passed.
+
+Run inside the task's repo (cwd) by the testbed python (which has pytest + the project
+installed) — NOT a uv script, since the tests need the project's own environment. argv[1]
+is a JSON list of pytest node ids. We write JUnit XML and match each expected id against
+it (a pytest node id and the JUnit classname/name don't line up, so we try a few forms),
+printing the score on the last line. Any failure prints 0.0.
+"""
+
+import json
+import re
+import sys
+import xml.etree.ElementTree as ET
+
+import pytest
+
+XML = "/tmp/scaleswe_results.xml"
+
+
+def _normalize(value: str) -> str:
+    parts = value.strip().split("::")
+    if parts and parts[0].endswith(".py"):
+        parts[0] = parts[0][:-3]
+    return ".".join(parts).replace("/", ".").strip(".")
+
+
+def _all_passed(xml_content: str, expected: list[str]) -> bool:
+    try:
+        root = ET.fromstring(xml_content)
+    except ET.ParseError:
+        return False
+    exact = set(expected)
+    norm = {_normalize(t): t for t in expected}
+    fp = {re.sub(r"\s+", "", _normalize(t)): t for t in expected}
+    matched: dict[str, str] = {}
+    found: set[str] = set()
+    for tc in root.iter("testcase"):
+        if tc.find("skipped") is not None:
+            continue
+        name, classname = tc.get("name", ""), tc.get("classname", "")
+        file_attr = tc.get("file", "")
+        status = (
+            "failed"
+            if tc.find("failure") is not None or tc.find("error") is not None
+            else "passed"
+        )
+        for candidate in (
+            f"{file_attr}::{name}" if file_attr else "",
+            _normalize(f"{classname}.{name}"),
+            re.sub(r"\s+", "", _normalize(f"{classname}.{name}")),
+            f"{classname.replace('.', '/')}.py::{name}",
+        ):
+            original = (
+                candidate
+                if candidate in exact
+                else norm.get(candidate) or fp.get(candidate)
+            )
+            if original:
+                matched[original] = status
+                found.add(original)
+                break
+    return (
+        bool(found)
+        and all(status == "passed" for status in matched.values())
+        and not [t for t in expected if t not in found]
+    )
+
+
+def main() -> None:
+    expected = json.loads(sys.argv[1])
+    if not expected:
+        print(0.0)
+        return
+    pytest.main(
+        ["-vv", f"--junitxml={XML}", "-o", "addopts=", "--rootdir=.", *expected]
+    )
+    try:
+        xml_content = open(XML).read()
+    except OSError:
+        print(0.0)
+        return
+    print(1.0 if _all_passed(xml_content, expected) else 0.0)
+
+
+if __name__ == "__main__":
+    main()

verifiers/v1/env.py

@@ -142,6 +142,14 @@ def __init__(self, config: EnvConfig) -> None:
                 f"{self.taskset.config.id!r} exposes tool servers (MCP). Run it with a harness "
                 f"that supports task tools (e.g. --harness.id default), or use a taskset without tools."
             )
+        if self.taskset.NEEDS_CONTAINER and isinstance(
+            self.harness.config.runtime, SubprocessConfig
+        ):
+            raise ValueError(
+                f"Taskset {self.taskset.config.id!r} needs a container runtime "
+                "(NEEDS_CONTAINER), but the harness runs on the subprocess runtime; "
+                "use --harness.runtime.type docker or prime."
+            )
         self.harness_timeout = config.timeout.rollout
         self.scoring_timeout = config.timeout.scoring
         self.limits = RolloutLimits(
@@ -166,6 +174,13 @@ def runtime_for(self, task: Task) -> RuntimeConfig:
                     "runtime has no container; use the docker or prime runtime"
                 )
             updates["image"] = task.image
+        workdir_spec = type(config).model_fields.get("workdir")
+        if (
+            task.workdir is not None
+            and workdir_spec is not None
+            and getattr(config, "workdir") == workdir_spec.default
+        ):
+            updates["workdir"] = task.workdir
         for field, value in task.resources.model_dump(exclude_none=True).items():
             spec = type(config).model_fields.get(field)
             if spec is None:

verifiers/v1/rollout.py

@@ -123,6 +123,7 @@ async def run(
         try:
             session = RolloutSession(self.ctx, trace, stops, self.limits)
             await runtime.start()
+            await self.taskset.setup(self.task, runtime)
             async with self._serve_interception(interception, runtime, session) as (
                 endpoint,
                 secret,

verifiers/v1/task.py

@@ -52,6 +52,11 @@ class Task(StrictBaseModel):
     """Container image this task needs (e.g. its harbor environment). When set, the
     runtime must be a container (docker/prime): the Environment injects it into the
     runtime config and refuses the subprocess runtime, which has no container."""
+    workdir: str | None = None
+    """Working directory the harness and scoring run in — the Environment injects it into
+    the runtime config's `workdir` (where the runtime supports one). For a containerized
+    task whose image puts the working tree at a non-default path (e.g. a SWE row's
+    `/workspace/<repo>`)."""
     harness_timeout: float | None = None
     """Optional per-task harness timeout (seconds). Merges with the eval's
     `harness_timeout`: cli/toml > this > default (no limit)."""

verifiers/v1/taskset.py

@@ -17,7 +17,7 @@
 
 import asyncio
 from collections.abc import Mapping
-from typing import Generic, TypeVar
+from typing import ClassVar, Generic, TypeVar
 
 from pydantic import model_validator
 from pydantic_config import BaseConfig
@@ -90,6 +90,11 @@ class Taskset(Generic[TaskT, ConfigT]):
     """Generic over its task and config types, so `self.config` and `load_tasks`
     are fully typed. Subclass: implement `load_tasks`, add @reward/@metric."""
 
+    NEEDS_CONTAINER: ClassVar[bool] = False
+    """Whether this taskset only runs in a container runtime (docker/prime). When True the
+    Environment refuses the subprocess runtime — for tasksets whose work only makes sense
+    inside a per-task image (e.g. a SWE repo sandbox)."""
+
     def __init__(self, config: ConfigT) -> None:
         self.config = config
 
@@ -109,6 +114,12 @@ def user(self, task: TaskT) -> User | None:
         multi-turn conversation (e.g. a TextArena game)."""
         return None
 
+    async def setup(self, task: TaskT, runtime: Runtime) -> None:
+        """Prepare the live runtime for this task, after `runtime.start()` and before the
+        harness runs. No-op by default; override to run per-task setup in the runtime (e.g.
+        a SWE row checking out its base commit). Errors propagate and fail the rollout."""
+        return None
+
     async def score(self, trace: Trace, runtime: Runtime) -> None:
         """Score one rollout: run all `@metric` then `@reward` over its trace,
         concurrently within each phase. Each metric is recorded in `trace.metrics`