feat(swe_agents): add opt-in unfinished_episodes="fail"

responses_api_agents/swe_agents/README.md

@@ -414,6 +414,7 @@ The full schema lives in `SWEBenchWrapperConfig` (and the per-override `AgentPro
 | `dataset_path`                     | `null`                                            | Optional default dataset JSONL.                                         |
 | `verify_golden_patch`              | `false`                                           | Skip the agent and eval the dataset's own golden patch. Supported for `swe-bench-ext` and the SWE-bench / SWE-bench_Multilingual families. See [Golden-patch validation](#golden-patch-validation). |
 | `skip_eval`                        | `false`                                           | Run the agent normally but skip the eval container entirely; reward is forced to `0.0` since the patch is never graded. Useful for collecting agent trajectories without paying the eval cost. |
+| `unfinished_episodes`              | `mask`                                            | `mask`: turn-cap / context-window episodes are masked. `fail`: they and loop-detector exits stay in the loss with reward `0.0`. See [GRPO masking](#grpo-masking-and-failure-modes). |
 | `agent_prompt_overrides`           | `null`                                            | List of `AgentPromptOverride` entries. See above.                       |
 | `agent_prompt_override_random`     | `false`                                           | `false` = deterministic per `instance_id`; `true` = random per run.     |
 | `opencode_subagents_enabled`       | `false`                                           | (opencode only) Enable opencode's `task` tool so the main agent can spawn subagent sessions. See [Subagents](#opencode-integration). |
@@ -604,11 +605,12 @@ Each `responses` call returns a `NeMoGymResponse` whose `output` is a Responses-
 {
   "responses_create_params": { /* full input incl. system+user prompts the agent saw */ },
   "response":               { /* output messages + tool calls */ },
-  "reward": 1.0,            // 1.0 iff resolved, else 0.0
+  "reward": 1.0,            // 1.0 iff resolved (and, with unfinished_episodes=fail, not unfinished), else 0.0
   "resolved": true,
   "patch_exists": true,
   "model_patch": "diff --git ...",
   "agent_error_kind": null, // "max_iteration" | "context_window" | "stuck_in_loop" | "oom" | "other" | null
+  "unfinished": false,      // agent_error_kind is max_iteration, context_window or stuck_in_loop
   "agent_timed_out": false,
   "eval_timed_out": false,
   "oom_killed": false,       // memory watchdog killed the agent container — see Memory watchdog
@@ -669,12 +671,21 @@ Each `responses` call returns a `NeMoGymResponse` whose `output` is a Responses-
 
 `SWEBenchWrapperInstanceConfig.mask_sample` is set to `True` (so downstream RL drops the gradient for this rollout) when:
 
-1. The patch resolved the tests **but** the agent terminated in a `max_iteration` or `context_window` error — the reward is accidental.
+1. The patch resolved the tests **but** the agent terminated in a `max_iteration` or `context_window` error — the reward is accidental. Only with `unfinished_episodes: mask` (the default); see below.
 2. The eval container hit `swebench_tests_timeout` — reward is unreliable.
 3. The agent hit `swebench_agent_timeout` (wall-clock) regardless of `resolved`.
 4. The memory watchdog killed the **agent** container (OOM).
 5. The memory watchdog killed the **eval** container (OOM).
 
+`unfinished_episodes` decides what happens to episodes the agent did not end itself (`max_iteration`, `context_window`, `stuck_in_loop`; reported as `unfinished: true`):
+
+| `unfinished_episodes` | `max_iteration` / `context_window` | `stuck_in_loop`                   |
+|-----------------------|------------------------------------|-----------------------------------|
+| `mask` (default)      | masked                             | graded on the patch left on disk  |
+| `fail`                | kept in the loss, reward `0.0`     | kept in the loss, reward `0.0`    |
+
+With `fail`, a policy that runs out its turn or context budget gets a failure signal instead of disappearing from the batch. `resolved` still reports the measured test verdict, so you can see how many cut-off episodes had a working patch.
+
 Agent error strings are bucketed by `_classify_agent_error`:
 
 | Substring (case-insensitive)        | Bucket           |

responses_api_agents/swe_agents/app.py

@@ -183,6 +183,20 @@ class SWEBenchWrapperConfig(BaseResponsesAPIAgentConfig):
         ),
     )
 
+    unfinished_episodes: Literal["mask", "fail"] = Field(
+        default="mask",
+        description=(
+            "How to score episodes the agent did not end itself: OpenHands stopped it at agent_max_turns "
+            "(max_iteration), it ran out of context (context_window), or the loop detector stopped it "
+            "(stuck_in_loop).\n"
+            "  'mask' (default): turn-cap and context-window episodes are excluded from the loss (mask_sample); "
+            "loop-detector exits are graded on the patch left on disk.\n"
+            "  'fail': all three stay in the loss with reward 0, so RL sees that running out the budget fails. "
+            "`resolved` still reports the measured test verdict. OOM kills and agent / eval wall-clock timeouts stay "
+            "masked either way."
+        ),
+    )
+
     agent_prompt_overrides: Optional[list[AgentPromptOverride]] = Field(
         default=None,
         description="List of (user_prompt_template, system_prompt_template, agent_cls) overrides. "
@@ -314,6 +328,8 @@ class SWEBenchMetrics(BaseModel):
 
     # Failure-mode signals used to decide mask_sample downstream.
     agent_error_kind: Optional[str] = None
+    # The episode ended without the agent finishing it: turn cap, context window or loop detector.
+    unfinished: Optional[bool] = None
     agent_timed_out: Optional[bool] = None
     eval_timed_out: Optional[bool] = None
     eval_launch_failed: bool = False
@@ -2422,6 +2438,19 @@ def get_run_command(self) -> ExecuteContainerCommandArgs:
 ########################################
 
 
+# Agent error kinds for episodes the agent did not end itself (see SWEBenchWrapperConfig.unfinished_episodes).
+UNFINISHED_AGENT_ERROR_KINDS = ("max_iteration", "context_window", "stuck_in_loop")
+
+
+def _task_reward(metrics: SWEBenchMetrics, unfinished_episodes: str) -> float:
+    """1.0 for a resolved task; with unfinished_episodes="fail", only if the agent also finished the episode."""
+    if not metrics.resolved:
+        return 0.0
+    if unfinished_episodes == "fail" and metrics.unfinished:
+        return 0.0
+    return 1.0
+
+
 def _classify_agent_error(err: Optional[str]) -> Optional[str]:
     if not err:
         return None
@@ -3936,7 +3965,8 @@ async def _inner_responses(
 
         # Decide whether to mask this sample from the GRPO gradient.
         # 1) Patch passed eval but agent did not actually submit (hit max-turns
-        #    or blew the context window) — the reward is accidental.
+        #    or blew the context window) — the reward is accidental. Only with
+        #    unfinished_episodes="mask"; with "fail" these score 0 instead (see run()).
         # 2) Final eval step timed out — reward is unreliable.
         # 3) Agent itself timed out (wall-clock) — mask regardless of resolved.
         # 4) Memory watchdog killed the agent container (OOM).
@@ -3948,8 +3978,9 @@ async def _inner_responses(
         agent_timed_out = bool(persisted_metrics.agent_timed_out)
         oom_killed = bool(persisted_metrics.oom_killed)
         eval_oom_killed = bool(persisted_metrics.eval_oom_killed)
+        metrics_to_update["unfinished"] = agent_error_kind in UNFINISHED_AGENT_ERROR_KINDS
         if (
-            agent_error_kind in ("max_iteration", "context_window")
+            (self.config.unfinished_episodes == "mask" and agent_error_kind in ("max_iteration", "context_window"))
             or eval_timed_out
             or persisted_metrics.eval_launch_failed
             or agent_timed_out
@@ -4096,7 +4127,7 @@ async def run(self, body: BaseRunRequest) -> SWEBenchVerifyResponse:
             return SWEBenchVerifyResponse(
                 responses_create_params=responses_create_params,
                 response=response,
-                reward=1.0 if metrics.resolved else 0.0,
+                reward=_task_reward(metrics, self.config.unfinished_episodes),
                 # Report it on the contract as well; `instance_config.mask_sample` stays
                 # for one release so existing consumers keep working.
                 mask_sample=instance_config.mask_sample,

responses_api_agents/swe_agents/tests/test_app.py

@@ -3644,3 +3644,110 @@ async def test_evaluator_failure_mask(self, tmp_path: Path, monkeypatch, outcome
     def test_legacy_metrics_do_not_imply_launch_failure(self) -> None:
         metrics = SWEBenchMetrics.model_validate({"patch_exists": True, "eval_timed_out": False})
         assert metrics.eval_launch_failed is False
+
+
+########################################
+# unfinished_episodes
+########################################
+
+
+class TestUnfinishedEpisodes:
+    @staticmethod
+    async def _inner_metadata(monkeypatch, tmp_path: Path, persisted_metrics: dict, mode: str) -> tuple[bool, dict]:
+        """Run `_inner_responses` past the masking decision with the agent run stubbed out."""
+        wrapper = _create_wrapper(monkeypatch)
+        wrapper.config.unfinished_episodes = mode
+        params = _make_instance_config(str(tmp_path), unfinished_episodes=mode)
+        params.metrics_fpath.write_text(json.dumps(persisted_metrics))
+        monkeypatch.setattr(swe_app, "runner_ray_remote", MagicMock(remote=AsyncMock(return_value=None)))
+        monkeypatch.setattr(
+            SWEBenchWrapper, "get_openhands_trajectory_from_completions", lambda self, *args: ([], [], 0, None)
+        )
+        response = await wrapper._inner_responses(params, MagicMock())
+        return json.loads(response.metadata["instance_config"])["mask_sample"], json.loads(
+            response.metadata["metrics"]
+        )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "agent_error_kind,masked_by_default",
+        [("max_iteration", True), ("context_window", True), ("stuck_in_loop", False)],
+    )
+    async def test_unfinished_kinds(self, monkeypatch, tmp_path, agent_error_kind, masked_by_default) -> None:
+        persisted = {"agent_error_kind": agent_error_kind}
+        masked, metrics = await self._inner_metadata(monkeypatch, tmp_path / "mask", persisted, "mask")
+        assert masked is masked_by_default
+        assert metrics["unfinished"] is True
+        masked, metrics = await self._inner_metadata(monkeypatch, tmp_path / "fail", persisted, "fail")
+        assert masked is False
+        assert metrics["unfinished"] is True
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("mode", ["mask", "fail"])
+    async def test_infrastructure_failures_stay_masked(self, monkeypatch, tmp_path, mode) -> None:
+        for i, persisted in enumerate(
+            [{"agent_timed_out": True}, {"eval_timed_out": True}, {"oom_killed": True}, {"eval_oom_killed": True}]
+        ):
+            masked, metrics = await self._inner_metadata(monkeypatch, tmp_path / str(i), persisted, mode)
+            assert masked is True
+            assert metrics["unfinished"] is False
+
+    @pytest.mark.asyncio
+    async def test_finished_episode_is_not_unfinished(self, monkeypatch, tmp_path) -> None:
+        masked, metrics = await self._inner_metadata(monkeypatch, tmp_path, {"agent_error_kind": None}, "fail")
+        assert masked is False
+        assert metrics["unfinished"] is False
+
+    def test_task_reward(self) -> None:
+        resolved_unfinished = SWEBenchMetrics(resolved=True, unfinished=True)
+        resolved_finished = SWEBenchMetrics(resolved=True, unfinished=False)
+        assert swe_app._task_reward(resolved_unfinished, "mask") == 1.0
+        assert swe_app._task_reward(resolved_unfinished, "fail") == 0.0
+        assert swe_app._task_reward(resolved_finished, "fail") == 1.0
+        assert swe_app._task_reward(SWEBenchMetrics(resolved=False, unfinished=False), "mask") == 0.0
+        # Metrics written before the field existed.
+        assert swe_app._task_reward(SWEBenchMetrics(resolved=True), "fail") == 1.0
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("mode,expected_reward", [("mask", 1.0), ("fail", 0.0)])
+    async def test_run_reward_for_resolved_unfinished_episode(self, monkeypatch, mode, expected_reward) -> None:
+        wrapper = _create_wrapper(monkeypatch)
+        wrapper.config.unfinished_episodes = mode
+        mock_response = NeMoGymResponse(
+            id="swebench-test",
+            created_at=123,
+            model="test-model",
+            object="response",
+            output=[],
+            parallel_tool_calls=True,
+            tool_choice="auto",
+            tools=[],
+            metadata={
+                "input": "[]",
+                "metrics": json.dumps(
+                    {"resolved": True, "patch_exists": True, "agent_error_kind": "stuck_in_loop", "unfinished": True}
+                ),
+                "instance_config": _make_instance_config(tempfile.mkdtemp()).model_dump_json(),
+            },
+        )
+        with patch.object(SWEBenchWrapper, "responses", new_callable=AsyncMock, return_value=mock_response):
+            from nemo_gym.base_resources_server import BaseRunRequest
+
+            body = BaseRunRequest(
+                responses_create_params=NeMoGymResponseCreateParamsNonStreaming(
+                    model="test-model",
+                    input=[],
+                    metadata={
+                        "problem_statement": "Fix",
+                        "instance_id": "test-1",
+                        "base_commit": "abc",
+                        "dataset_name": "SWE-bench",
+                        "split": "test",
+                        "instance_dict": "{}",
+                    },
+                )
+            )
+            result = await wrapper.run(body)
+        assert result.resolved is True
+        assert result.unfinished is True
+        assert result.reward == expected_reward