"""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 = """ Consider the following PR description: {problem_statement} # 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. 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). 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. - Creating/viewing the patch and submitting MUST be separate commands (not combined with &&). - You CANNOT continue working after submitting. """ # 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: 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 patch = text.split(SENTINEL, 1)[1].strip() 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() 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 ''), }