EvalHarness/evalharness/agent/envs/swe_agentic.py

"""swe_bench_verified_agentic: multi-turn SWE agent (mini-swe-agent protocol).

Per sample: start a LONG-RUNNING per-instance Docker container (the official
swebench image, /testbed workdir), give the model a single `bash` tool whose
execs run inside it, loop until the sentinel submission or max_turns, then
recover the patch (sentinel payload, else `git diff` in /testbed).

Ports es's swe_bench_agentic_adapter (which itself mirrors mini-swe-agent's
swebench.yaml). Scoring stays in the recipe (official swebench harness).
"""

import asyncio
import json
import re
import subprocess
import uuid
from typing import Any, Dict

from ...data.sample import ChatMessage, Sample
from ..loop import Environment, register_env

SENTINEL = 'COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT'

# mini-swe-agent swebench.yaml contract (verbatim sections that matter)
INSTANCE_TEMPLATE = """<pr_description>
Consider the following PR description:
{problem_statement}
</pr_description>

<instructions>
# Task Instructions

## Overview

You're a software engineer interacting continuously with a computer by submitting commands.
You'll be helping implement necessary changes to meet requirements in the PR description.
Your task is specifically to make changes to non-test files in the current directory in order to fix the issue described in the PR description in a way that is general and consistent with the codebase.
<IMPORTANT>This is an interactive process where you will think and issue AT LEAST ONE command, see the result, then think and issue your next command(s).</important>

For each response:
1. Include a THOUGHT section explaining your reasoning and what you're trying to accomplish
2. Provide one or more bash tool calls to execute

## Important Boundaries

- MODIFY: Regular source code files in /testbed (this is the working directory for all your subsequent commands)
- DO NOT MODIFY: Tests, configuration files (pyproject.toml, setup.cfg, etc.)

## Recommended Workflow

1. Analyze the codebase by finding and reading relevant files
2. Create a script to reproduce the issue
3. Edit the source code to resolve the issue
4. Verify your fix works by running your script again
5. Test edge cases to ensure your fix is robust

## Command Execution Rules

- Directory or environment variable changes are not persistent; every action runs in a new subshell
- Prefix actions with `cd /testbed && ...` when needed
- Always use non-interactive flags (-y, -f); avoid vi/nano

## Submission

When you've completed your work, you MUST submit your changes as a git patch:

Step 1: Create the patch file
Run `git diff -- path/to/file1 path/to/file2 > patch.txt` listing only the source files you modified.
Do NOT commit your changes.

Step 2: Verify the patch
Inspect patch.txt to confirm it only contains your intended changes and headers show `--- a/` and `+++ b/` paths.

Step 3: Submit (EXACT command required)
You MUST use this EXACT command to submit:

```bash
echo {sentinel} && cat patch.txt
```

If the command fails (nonzero exit status), it will not submit.

<CRITICAL>
- Creating/viewing the patch and submitting MUST be separate commands (not combined with &&).
- You CANNOT continue working after submitting.
</CRITICAL>
</instructions>"""

# OpenAI function-calling bash tool (mini-swe-agent mainline protocol)
BASH_TOOL = {
    'type': 'function',
    'function': {
        'name': 'bash',
        'description': 'Execute a bash command in /testbed (persistent working '
                       'tree, fresh subshell per call). Use for exploring, '
                       'editing, testing.',
        'parameters': {'type': 'object',
                       'properties': {'command': {'type': 'string',
                                                  'description': 'the bash command to run'}},
                       'required': ['command']},
    },
}


def _docker(args, timeout=300):
    return subprocess.run(['docker'] + args, capture_output=True, text=True,
                          timeout=timeout)


@register_env('swe_agentic')
class SWEAgenticEnvironment(Environment):
    """Self-running env: one persistent container per sample."""

    def __init__(self, adapter=None, **_):
        # adapter arg: get_env uniform signature (unused here; the adapter
        # is passed to run_task per sample)
        self.container = ''

    # ---- container lifecycle ----
    def _start(self, image: str) -> str:
        # per-instance image: ensure (local check -> mirror-chain pull with
        # progress) BEFORE docker run -- `docker run` auto-pulls with zero
        # output and its 120s timeout kills runs on slow mirrors
        from ...sandbox.docker import ensure_image

        try:
            ensure_image(image)
        except RuntimeError as e:
            raise RuntimeError(f'image {image} unavailable: {str(e)[:150]}')
        name = f'eh-swe-{uuid.uuid4().hex[:10]}'
        r = _docker(['run', '-d', '--name', name,
                     '-w', '/testbed',
                     '-e', 'PAGER=cat', '-e', 'MANPAGER=cat',
                     '-e', 'LESS=-R', '-e', 'PIP_PROGRESS_BAR=off',
                     '-e', 'TQDM_DISABLE=1',
                     image, 'tail', '-f', '/dev/null'], timeout=120)
        if r.returncode != 0:
            raise RuntimeError(f'start container failed: {r.stderr[:200]}')
        return name

    def _exec(self, cmd: str, timeout: int = 600) -> Dict[str, Any]:
        # bash -lc: swebench images activate the per-instance testbed via
        # shell startup files (es parity: _SWE_BENCH_INTERPRETER)
        r = _docker(['exec', self.container, 'bash', '-lc', cmd], timeout=timeout)
        out = (r.stdout or '') + (r.stderr or '')
        # cap observation: agents choke on 100k-char dumps
        if len(out) > 30000:
            out = out[:15000] + '\n...[truncated]...\n' + out[-15000:]
        return {'exit': r.returncode, 'out': out}

    def _stop(self):
        if self.container:
            _docker(['rm', '-f', self.container], timeout=60)
            self.container = ''

    # ---- the agent loop ----
    async def run_task(self, adapter, sample: Sample, max_turns: int = 250,
                       system: str = '', user_adapter=None, gen_kwargs=None,
                       **kw) -> Dict[str, Any]:
        image = (sample.sandbox and sample.sandbox.image) or ''
        if not image:
            raise RuntimeError('swe_agentic: sample has no sandbox image '
                               '(dataset plugin must declare it)')
        self.container = self._start(image)
        ps = (sample.metadata or {}).get('problem_statement') or sample.input_text
        messages = [ChatMessage(role='user', content=INSTANCE_TEMPLATE.format(
            problem_statement=ps, sentinel=SENTINEL))]
        tools = [{'name': BASH_TOOL['function']['name'],
                  'description': BASH_TOOL['function']['description'],
                  'parameters': BASH_TOOL['function']['parameters']['properties']}]

        from ...model.output import Usage
        total_usage = Usage()
        patch = ''
        turns = 0
        try:
            for turns in range(1, max_turns + 1):
                out = await adapter.generate(messages, tools=tools,
                                             **(gen_kwargs or {}))
                total_usage = total_usage + out.usage
                text = out.text or ''
                if SENTINEL in text:
                    # payload after the sentinel echo = `cat patch.txt` output.
                    # A unified diff must end with a newline; a stripped
                    # payload fails git apply with 'corrupt patch at line N'
                    patch = text.split(SENTINEL, 1)[1].strip() + '\n'
                    break
                if out.tool_calls:
                    messages.append(ChatMessage(role='assistant', content=text))
                    for tc in out.tool_calls:
                        cmd = (tc.arguments_dict or {}).get('command', '')
                        res = await asyncio.to_thread(self._exec, cmd)
                        messages.append(ChatMessage(
                            role='user',
                            content=f"exit={res['exit']}\n{res['out']}"))
                        if SENTINEL in res['out']:
                            # sentinel appeared in command output: capture patch
                            patch = res['out'].split(SENTINEL, 1)[1].strip() + '\n'
                else:
                    # no tool call: nudge per protocol
                    messages.append(ChatMessage(role='assistant', content=text))
                    messages.append(ChatMessage(
                        role='user',
                        content='Continue: issue a bash tool call (or submit '
                                f'via `echo {SENTINEL} && cat patch.txt`).'))
            if not patch:
                # fallback (es parity): recover from the working tree
                res = await asyncio.to_thread(
                    self._exec, 'cd /testbed && git diff')
                patch = res['out'].strip()
        finally:
            self._stop()

        md = sample.metadata or {}
        return {
            'raw': patch or '(no patch produced)',
            'usage': total_usage.model_dump(),
            'env_state': {
                'patch': patch,
                'instance_id': md.get('instance_id'),
                'repo': md.get('repo'),
                'base_commit': md.get('base_commit'),
                'test_patch': md.get('test_patch'),
                'FAIL_TO_PASS': md.get('FAIL_TO_PASS'),
                'PASS_TO_PASS': md.get('PASS_TO_PASS'),
                'image': image,
                'turns_used': turns,
            },
            'trajectory': [{'role': m.role, 'content': (m.content or '')[:2000]}
                           for m in messages],
            'group_key': str(md.get('instance_id') or ''),
        }