Add deepagents_rl package sources

src/deepagents_rl/data/__init__.py

@@ -1,3 +1,11 @@
 from .mbpp import build_train_eval, load_mbpp_sanitized, save_jsonl, to_hf_dataset
+from .swe_bench_lite import from_hf_record, load_swe_bench_lite
 
-__all__ = ["build_train_eval", "load_mbpp_sanitized", "save_jsonl", "to_hf_dataset"]
+__all__ = [
+    "build_train_eval",
+    "load_mbpp_sanitized",
+    "save_jsonl",
+    "to_hf_dataset",
+    "load_swe_bench_lite",
+    "from_hf_record",
+]

src/deepagents_rl/data/swe_bench_lite.py

@@ -0,0 +1,179 @@
+"""SWE-bench Lite loader for deepagents rollouts + official Docker eval.
+
+Source: https://huggingface.co/datasets/SWE-bench/SWE-bench_Lite
+
+Each row includes the prebuilt instance `image` and `eval_script` used by the
+SWE-bench harness. Agent sandboxes are full git checkouts at `base_commit`
+(see `deepagents_rl.harness.repo`), not sparse file dumps.
+"""
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+DIFF_PATH_RE = re.compile(r"^diff --git a/(.+?) b/(.+)$", re.MULTILINE)
+
+
+def parse_paths_from_patch(patch: str) -> list[str]:
+    paths: list[str] = []
+    for m in DIFF_PATH_RE.finditer(patch or ""):
+        p = m.group(2).strip()
+        if p and p != "/dev/null" and p not in paths:
+            paths.append(p)
+    return paths
+
+
+def _as_str_list(val: Any) -> list[str]:
+    if val is None:
+        return []
+    if isinstance(val, list):
+        return [str(x) for x in val]
+    if isinstance(val, str):
+        s = val.strip()
+        if not s:
+            return []
+        try:
+            parsed = json.loads(s)
+            if isinstance(parsed, list):
+                return [str(x) for x in parsed]
+        except json.JSONDecodeError:
+            pass
+        return [s]
+    return [str(val)]
+
+
+def build_agent_prompt(problem_statement: str, *, instance_id: str) -> str:
+    return (
+        "You are a software engineering agent inside a full git checkout of the repo.\n"
+        "Resolve the GitHub issue by exploring and editing files with your tools "
+        "(ls, glob, grep, read_file, write_file, edit_file, execute).\n"
+        "Prefer minimal correct edits. Do not modify tests unless necessary.\n"
+        "When finished, stop calling tools and reply with DONE plus a short summary.\n\n"
+        f"Instance: {instance_id}\n\n"
+        f"Issue:\n{problem_statement.strip()}\n"
+    )
+
+
+def _row_to_task(ex: dict[str, Any], split: str) -> dict[str, Any]:
+    patch = ex.get("patch") or ""
+    test_patch = ex.get("test_patch") or ""
+    instance_id = str(ex["instance_id"])
+    problem_statement = str(ex.get("problem_statement") or "")
+    return {
+        "task_id": instance_id,
+        "instance_id": instance_id,
+        "split": split,
+        "repo": str(ex["repo"]),
+        "base_commit": str(ex["base_commit"]),
+        "problem_statement": problem_statement,
+        "prompt": build_agent_prompt(problem_statement, instance_id=instance_id),
+        "workspace_paths": parse_paths_from_patch(patch),
+        "gold_patch": patch,
+        "test_patch": test_patch,
+        "fail_to_pass": _as_str_list(ex.get("FAIL_TO_PASS")),
+        "pass_to_pass": _as_str_list(ex.get("PASS_TO_PASS")),
+        "hints_text": str(ex.get("hints_text") or ""),
+        "version": str(ex.get("version") or ""),
+        "difficulty": str(ex.get("difficulty") or ""),
+        "image": str(ex.get("image") or ""),
+        "eval_script": str(ex.get("eval_script") or ""),
+        "eval_type": str(ex.get("eval_type") or "pass_and_fail"),
+        "log_parser": str(ex.get("log_parser") or ""),
+        "environment_setup_commit": str(ex.get("environment_setup_commit") or ""),
+        "source_dataset": "SWE-bench/SWE-bench_Lite",
+    }
+
+
+def load_swe_bench_lite(
+    *,
+    splits: tuple[str, ...] = ("dev", "test"),
+    max_per_split: int | None = None,
+    token: str | None = None,
+) -> dict[str, list[dict[str, Any]]]:
+    from datasets import load_dataset
+
+    kwargs: dict[str, Any] = {}
+    if token:
+        kwargs["token"] = token
+    ds = load_dataset("SWE-bench/SWE-bench_Lite", **kwargs)
+    out: dict[str, list[dict[str, Any]]] = {}
+    for split in splits:
+        if split not in ds:
+            continue
+        rows = []
+        for i, ex in enumerate(ds[split]):
+            if max_per_split is not None and i >= max_per_split:
+                break
+            rows.append(_row_to_task(dict(ex), split=split))
+        out[split] = rows
+    return out
+
+
+def to_hf_records(tasks: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    records = []
+    for t in tasks:
+        records.append(
+            {
+                "task_id": t["task_id"],
+                "instance_id": t["instance_id"],
+                "split": t["split"],
+                "repo": t["repo"],
+                "base_commit": t["base_commit"],
+                "problem_statement": t["problem_statement"],
+                "prompt": t["prompt"],
+                "workspace_paths_json": json.dumps(t["workspace_paths"]),
+                "gold_patch": t["gold_patch"],
+                "test_patch": t["test_patch"],
+                "fail_to_pass_json": json.dumps(t["fail_to_pass"]),
+                "pass_to_pass_json": json.dumps(t["pass_to_pass"]),
+                "hints_text": t.get("hints_text") or "",
+                "version": t.get("version") or "",
+                "difficulty": t.get("difficulty") or "",
+                "image": t.get("image") or "",
+                "eval_script": t.get("eval_script") or "",
+                "eval_type": t.get("eval_type") or "pass_and_fail",
+                "log_parser": t.get("log_parser") or "",
+                "environment_setup_commit": t.get("environment_setup_commit") or "",
+                "source_dataset": t.get("source_dataset") or "SWE-bench/SWE-bench_Lite",
+            }
+        )
+    return records
+
+
+def from_hf_record(rec: dict[str, Any]) -> dict[str, Any]:
+    # Backward compat: older sparse dataset rows may lack image/eval_script —
+    # callers should reload from SWE-bench/SWE-bench_Lite in that case.
+    return {
+        "task_id": rec["task_id"],
+        "instance_id": rec["instance_id"],
+        "split": rec.get("split") or "train",
+        "repo": rec["repo"],
+        "base_commit": rec["base_commit"],
+        "problem_statement": rec["problem_statement"],
+        "prompt": rec["prompt"],
+        "workspace_paths": json.loads(rec.get("workspace_paths_json") or "[]"),
+        "gold_patch": rec.get("gold_patch") or "",
+        "test_patch": rec.get("test_patch") or "",
+        "fail_to_pass": json.loads(rec.get("fail_to_pass_json") or "[]"),
+        "pass_to_pass": json.loads(rec.get("pass_to_pass_json") or "[]"),
+        "hints_text": rec.get("hints_text") or "",
+        "version": rec.get("version") or "",
+        "difficulty": rec.get("difficulty") or "",
+        "image": rec.get("image") or "",
+        "eval_script": rec.get("eval_script") or "",
+        "eval_type": rec.get("eval_type") or "pass_and_fail",
+        "log_parser": rec.get("log_parser") or "",
+        "environment_setup_commit": rec.get("environment_setup_commit") or "",
+        "source_dataset": rec.get("source_dataset") or "SWE-bench/SWE-bench_Lite",
+    }
+
+
+def save_jsonl(tasks: list[dict[str, Any]], path) -> None:
+    from pathlib import Path
+
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as f:
+        for t in to_hf_records(tasks):
+            f.write(json.dumps(t) + "\n")

src/deepagents_rl/harness/__init__.py

@@ -0,0 +1,7 @@
+"""Deepagents harness: multi-turn tool episodes + official SWE-bench Docker grading."""
+
+from deepagents_rl.harness.episode import run_deepagent_episode
+from deepagents_rl.harness.grade import grade_episode
+from deepagents_rl.harness.swe_docker_eval import evaluate_model_patch
+
+__all__ = ["run_deepagent_episode", "grade_episode", "evaluate_model_patch"]

src/deepagents_rl/harness/episode.py

@@ -0,0 +1,863 @@
+"""Run one deepagents episode on a long-horizon coding task and record everything."""
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import time
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from deepagents_rl.harness.grade import grade_episode
+from deepagents_rl.harness.repo import materialize_repo_at_commit
+from deepagents_rl.observability.logging import (
+    call,
+    error as log_error,
+    result,
+    span,
+    step,
+    truncate,
+)
+
+
+SYSTEM_PROMPT = (
+    "You are a software engineering agent with tools.\n"
+    "You MUST use tools to inspect and edit files. Never answer with only prose.\n"
+    "First action: call ls or glob to see the sandbox.\n"
+    "Then read_file the relevant source, edit_file/write_file to fix the bug, "
+    "optionally execute tests.\n"
+    "When finished, stop calling tools and reply DONE plus a short summary.\n"
+    "Do not delete unrelated files."
+)
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _msg_to_dict(msg: Any) -> dict[str, Any]:
+    role = getattr(msg, "type", None) or getattr(msg, "role", None) or msg.__class__.__name__
+    content = getattr(msg, "content", None)
+    if not isinstance(content, str):
+        content = str(content) if content is not None else ""
+    out: dict[str, Any] = {"role": role, "content": content[:8000]}
+    tool_calls = getattr(msg, "tool_calls", None)
+    if tool_calls:
+        out["tool_calls"] = tool_calls
+    name = getattr(msg, "name", None)
+    if name:
+        out["name"] = name
+    tool_call_id = getattr(msg, "tool_call_id", None)
+    if tool_call_id:
+        out["tool_call_id"] = tool_call_id
+    return out
+
+
+def _extract_tool_calls(messages: list[Any]) -> tuple[list[str], list[dict[str, Any]]]:
+    tools_used: list[str] = []
+    tool_calls_log: list[dict[str, Any]] = []
+    pending: dict[str, dict[str, Any]] = {}
+    turn = 0
+    for msg in messages:
+        tname = msg.__class__.__name__
+        if tname == "AIMessage" and getattr(msg, "tool_calls", None):
+            turn += 1
+            for tc in msg.tool_calls:
+                name = tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None)
+                args = tc.get("args") if isinstance(tc, dict) else getattr(tc, "args", {})
+                tc_id = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None)
+                if name:
+                    tools_used.append(name)
+                entry = {
+                    "turn": turn,
+                    "name": name,
+                    "args": args or {},
+                    "tool_call_id": tc_id,
+                    "result": None,
+                    "ok": None,
+                }
+                if tc_id:
+                    pending[tc_id] = entry
+                tool_calls_log.append(entry)
+        elif tname == "ToolMessage":
+            tc_id = getattr(msg, "tool_call_id", None)
+            content = getattr(msg, "content", "")
+            content_s = content if isinstance(content, str) else str(content)
+            entry = pending.get(tc_id) if tc_id else None
+            if entry is None:
+                entry = {
+                    "turn": turn,
+                    "name": getattr(msg, "name", None),
+                    "args": {},
+                    "tool_call_id": tc_id,
+                }
+                tool_calls_log.append(entry)
+            entry["result"] = content_s[:4000]
+            entry["ok"] = not content_s.startswith("Error") and "ERROR" not in content_s[:40]
+    return tools_used, tool_calls_log
+
+
+def _short_prompt(task: dict[str, Any]) -> str:
+    """Compact prompt — long issue bodies break weak GGUF tool-calling templates."""
+    body = (task.get("problem_statement") or "").strip()
+    if len(body) > 1800:
+        body = body[:1800] + "\n...(truncated)..."
+    paths = task.get("workspace_paths") or list((task.get("workspace_files") or {}).keys())
+    path_hint = ", ".join(paths[:8]) if paths else "(explore with ls/glob)"
+    return (
+        f"Fix this GitHub issue in the sandbox using tools.\n"
+        f"Instance: {task['task_id']}\n"
+        f"Likely files: {path_hint}\n\n"
+        f"Issue:\n{body}\n\n"
+        f"Required: call tools (start with ls or read_file). "
+        f"When done, reply DONE."
+    )
+
+
+def _result_to_str(obj: Any) -> str:
+    if obj is None:
+        return ""
+    if hasattr(obj, "model_dump"):
+        try:
+            return json.dumps(obj.model_dump(), default=str)[:4000]
+        except Exception:
+            pass
+    if hasattr(obj, "__dict__"):
+        try:
+            return json.dumps(obj.__dict__, default=str)[:4000]
+        except Exception:
+            pass
+    return str(obj)[:4000]
+
+
+def _build_backend_tools(backend: Any):
+    """LangChain tools wrapping deepagents LocalShellBackend (same FS + execute)."""
+    from langchain_core.tools import tool
+
+    @tool
+    def ls(path: str = ".") -> str:
+        """List files in a directory inside the sandbox."""
+        return _result_to_str(backend.ls(path))
+
+    @tool
+    def glob(pattern: str, path: str = ".") -> str:
+        """Glob for files matching pattern under path."""
+        return _result_to_str(backend.glob(pattern, path))
+
+    @tool
+    def grep(pattern: str, path: str = ".") -> str:
+        """Search file contents for a regex pattern."""
+        return _result_to_str(backend.grep(pattern, path))
+
+    @tool
+    def read_file(file_path: str, offset: int = 0, limit: int = 200) -> str:
+        """Read a text file from the sandbox (line offset/limit)."""
+        return _result_to_str(backend.read(file_path, offset=offset, limit=limit))
+
+    @tool
+    def write_file(file_path: str, content: str) -> str:
+        """Write/create a text file in the sandbox."""
+        return _result_to_str(backend.write(file_path, content))
+
+    @tool
+    def edit_file(file_path: str, old_string: str, new_string: str) -> str:
+        """Replace old_string with new_string in an existing file."""
+        return _result_to_str(backend.edit(file_path, old_string, new_string))
+
+    @tool
+    def execute(command: str) -> str:
+        """Run a shell command in the sandbox (e.g. python -c '...')."""
+        return _result_to_str(backend.execute(command))
+
+    return [ls, glob, grep, read_file, write_file, edit_file, execute]
+
+
+def _run_tool_loop(
+    *,
+    model: Any,
+    backend: Any,
+    prompt: str,
+    max_turns: int,
+    langfuse_client: Any | None = None,
+    callback_handler: Any | None = None,
+    episode_id: str | None = None,
+) -> tuple[list[Any], str, str]:
+    """Reliable multi-turn tool loop on deepagents LocalShellBackend tools.
+
+    When `langfuse_client` is set, each LLM turn is traced via CallbackHandler
+    (generation) and each tool execution is a nested `as_type="tool"` span.
+    """
+    from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
+
+    from deepagents_rl.observability.langfuse_trace import llm_invoke_config
+
+    tools = _build_backend_tools(backend)
+    tool_map = {t.name: t for t in tools}
+    tool_names = list(tool_map.keys())
+    step("agent.bind_tools", f"tools={tool_names}", tools=tool_names)
+    bound = model.bind_tools(tools)
+    messages: list[Any] = [
+        SystemMessage(content=SYSTEM_PROMPT),
+        HumanMessage(content=prompt),
+    ]
+    step("agent.prompt", truncate(prompt, 500), prompt_chars=len(prompt))
+    final_text = ""
+    termination = "max_steps"
+
+    def _llm_call(turn: int) -> AIMessage:
+        cfg = None
+        if callback_handler is not None:
+            cfg = llm_invoke_config(
+                handler=callback_handler,
+                turn=turn,
+                episode_id=episode_id,
+            )
+        if cfg is not None:
+            return bound.invoke(messages, config=cfg)
+        return bound.invoke(messages)
+
+    for turn in range(max_turns):
+        def _one_turn(turn: int = turn) -> str | None:
+            """Run one LLM+tools turn. Returns termination if loop should stop."""
+            nonlocal final_text, termination
+            with span("agent.llm_turn", turn=turn):
+                call("llm.invoke", f"turn={turn} n_messages={len(messages)}")
+                t0 = time.time()
+                ai: AIMessage = _llm_call(turn)
+                dt = int((time.time() - t0) * 1000)
+                messages.append(ai)
+                content_preview = (
+                    ai.content if isinstance(ai.content, str) else str(ai.content)
+                )
+                tcs = getattr(ai, "tool_calls", None) or []
+                result(
+                    "llm.invoke",
+                    f"turn={turn} ({dt}ms)",
+                    duration_ms=dt,
+                    content=truncate(content_preview, 400),
+                    n_tool_calls=len(tcs),
+                    tool_calls=[
+                        {
+                            "name": tc.get("name"),
+                            "args": truncate(tc.get("args") or {}, 300),
+                        }
+                        for tc in tcs
+                    ],
+                )
+
+            if not tcs:
+                final_text = content_preview
+                termination = (
+                    "agent_done"
+                    if "DONE" in (final_text or "").upper()
+                    else "no_tool_final"
+                )
+                step("agent.final", termination, final_text=truncate(final_text, 400))
+                return termination
+
+            tool_results_for_turn: list[dict[str, Any]] = []
+            for tc in tcs:
+                name = tc["name"]
+                args = tc.get("args") or {}
+                tool_out: Any
+                with span("agent.tool", turn=turn, tool=name):
+                    call("tool", name, args=truncate(args, 500))
+
+                    def _exec_tool() -> Any:
+                        try:
+                            return tool_map[name].invoke(args)
+                        except Exception as e:  # noqa: BLE001
+                            err = f"ERROR: {type(e).__name__}: {e}"
+                            log_error("tool", err, tool=name)
+                            return err
+
+                    if langfuse_client is not None:
+                        with langfuse_client.start_as_current_observation(
+                            name=name,
+                            as_type="tool",
+                            input=args if isinstance(args, dict) else {"value": args},
+                            metadata={
+                                "turn": turn,
+                                "tool_call_id": tc.get("id"),
+                                "episode_id": episode_id,
+                            },
+                        ) as tool_span:
+                            tool_out = _exec_tool()
+                            tool_span.update(
+                                output={"result": truncate(str(tool_out), 4000)}
+                            )
+                    else:
+                        tool_out = _exec_tool()
+
+                    result("tool", name, output=truncate(str(tool_out), 600))
+                    tool_results_for_turn.append(
+                        {
+                            "name": name,
+                            "args": truncate(args, 300),
+                            "result": truncate(str(tool_out), 300),
+                        }
+                    )
+                    messages.append(
+                        ToolMessage(
+                            content=str(tool_out)[:4000],
+                            tool_call_id=tc["id"],
+                            name=name,
+                        )
+                    )
+            return None  # continue
+
+        if langfuse_client is not None:
+            with langfuse_client.start_as_current_observation(
+                name=f"turn-{turn}",
+                as_type="span",
+                metadata={"turn": turn, "episode_id": episode_id},
+                input={"n_messages": len(messages)},
+            ) as turn_span:
+                stop = _one_turn()
+                if stop:
+                    turn_span.update(
+                        output={
+                            "termination": stop,
+                            "final_text": truncate(final_text, 1000),
+                            "n_tool_calls": 0,
+                        }
+                    )
+                    break
+                n_tc = 0
+                for m in reversed(messages):
+                    if m.__class__.__name__ == "AIMessage" and getattr(
+                        m, "tool_calls", None
+                    ):
+                        n_tc = len(m.tool_calls)
+                        break
+                turn_span.update(output={"n_tool_calls": n_tc, "continued": True})
+        else:
+            if _one_turn():
+                break
+    else:
+        termination = "max_steps"
+        final_text = "(hit max turns)"
+        step("agent.max_turns", f"hit max_turns={max_turns}")
+
+    return messages, final_text, termination
+
+
+def _run_graph(
+    *,
+    model: Any,
+    backend: Any,
+    prompt: str,
+    recursion_limit: int,
+    langfuse_client: Any | None = None,
+    callback_handler: Any | None = None,
+    episode_id: str | None = None,
+) -> tuple[list[Any], str, str]:
+    from deepagents import create_deep_agent
+
+    step("agent.create_deep_agent", f"recursion_limit={recursion_limit}")
+    agent = create_deep_agent(
+        model=model,
+        backend=backend,
+        system_prompt=SYSTEM_PROMPT,
+        name="deepagents-rl-swe",
+    )
+    step("agent.prompt", truncate(prompt, 500), prompt_chars=len(prompt))
+
+    invoke_cfg: dict[str, Any] = {"recursion_limit": recursion_limit}
+    if callback_handler is not None:
+        invoke_cfg["callbacks"] = [callback_handler]
+        invoke_cfg["run_name"] = "deepagent-graph"
+        invoke_cfg["tags"] = ["deepagents-rl", "graph"]
+        invoke_cfg["metadata"] = {"episode_id": episode_id, "harness": "graph"}
+
+    def _invoke_graph() -> Any:
+        with span("agent.graph_invoke", recursion_limit=recursion_limit):
+            call("graph.invoke", "create_deep_agent.invoke")
+            return agent.invoke(
+                {"messages": [{"role": "user", "content": prompt}]},
+                config=invoke_cfg,
+            )
+
+    if langfuse_client is not None:
+        with langfuse_client.start_as_current_observation(
+            name="agent-graph",
+            as_type="span",
+            metadata={"episode_id": episode_id, "recursion_limit": recursion_limit},
+            input={"prompt": prompt[:4000]},
+        ) as graph_span:
+            graph_out = _invoke_graph()
+            msgs = list(graph_out.get("messages") or [])
+            final_text = ""
+            for m in reversed(msgs):
+                if m.__class__.__name__ == "AIMessage" and not getattr(
+                    m, "tool_calls", None
+                ):
+                    content = getattr(m, "content", "")
+                    final_text = content if isinstance(content, str) else str(content)
+                    break
+            tools_used, tool_calls_log = _extract_tool_calls(msgs)
+            # Replay tool spans under this graph span for UI granularity
+            for tc in tool_calls_log:
+                with langfuse_client.start_as_current_observation(
+                    name=tc.get("name") or "tool",
+                    as_type="tool",
+                    input=tc.get("args") or {},
+                    metadata={
+                        "turn": tc.get("turn"),
+                        "tool_call_id": tc.get("tool_call_id"),
+                        "episode_id": episode_id,
+                        "source": "graph_replay",
+                    },
+                ) as tool_span:
+                    tool_span.update(
+                        output={"result": truncate(tc.get("result") or "", 4000)}
+                    )
+            if tools_used:
+                termination = (
+                    "agent_done"
+                    if "DONE" in (final_text or "").upper()
+                    else "max_steps_or_halt"
+                )
+            else:
+                termination = "no_tools"
+            graph_span.update(
+                output={
+                    "termination": termination,
+                    "n_messages": len(msgs),
+                    "tools_used": tools_used,
+                    "final_text": truncate(final_text, 1000),
+                }
+            )
+    else:
+        graph_out = _invoke_graph()
+        msgs = list(graph_out.get("messages") or [])
+        final_text = ""
+        for m in reversed(msgs):
+            if m.__class__.__name__ == "AIMessage" and not getattr(m, "tool_calls", None):
+                content = getattr(m, "content", "")
+                final_text = content if isinstance(content, str) else str(content)
+                break
+        tools_used, tool_calls_log = _extract_tool_calls(msgs)
+        if tools_used:
+            termination = (
+                "agent_done" if "DONE" in (final_text or "").upper() else "max_steps_or_halt"
+            )
+        else:
+            termination = "no_tools"
+
+    for tc in tool_calls_log:
+        result(
+            "tool",
+            tc.get("name") or "?",
+            turn=tc.get("turn"),
+            args=truncate(tc.get("args") or {}, 400),
+            output=truncate(tc.get("result") or "", 400),
+        )
+    step(
+        "agent.graph_done",
+        termination,
+        n_messages=len(msgs),
+        tools_used=tools_used,
+        final_text=truncate(final_text, 400),
+    )
+    return msgs, final_text, termination
+
+
+def run_deepagent_episode(
+    task: dict[str, Any],
+    *,
+    model: Any | None = None,
+    recursion_limit: int = 80,
+    max_turns: int = 16,
+    mode: str | None = None,
+    langfuse_client: Any | None = None,
+    artifacts_dir: Path | None = None,
+    keep_sandbox: bool = False,
+) -> dict[str, Any]:
+    """
+    Full-repo git checkout -> deepagents LocalShellBackend + agent loop ->
+    git diff model_patch -> official SWE-bench Docker eval -> trajectory.
+    """
+    from deepagents.backends import LocalShellBackend
+    from langchain_openai import ChatOpenAI
+
+    episode_id = str(uuid.uuid4())
+    started = time.time()
+    if not task.get("image") or not task.get("eval_script"):
+        raise ValueError(
+            f"Task {task.get('task_id')} missing image/eval_script — "
+            "reload from SWE-bench/SWE-bench_Lite (not the old sparse dataset)."
+        )
+
+    step(
+        "episode.begin",
+        task["task_id"],
+        episode_id=episode_id,
+        repo=task.get("repo"),
+        base_commit=task.get("base_commit"),
+        image=task.get("image"),
+        mode=mode or os.getenv("HARNESS_MODE") or "tool_loop",
+        n_f2p=len(task.get("fail_to_pass") or []),
+        n_p2p=len(task.get("pass_to_pass") or []),
+    )
+
+    sandbox_parent = Path(tempfile.mkdtemp(prefix=f"darl_swe_{task['task_id'][:40]}_"))
+    sandbox = sandbox_parent / "repo"
+    materialize_repo_at_commit(
+        repo=task["repo"],
+        base_commit=task["base_commit"],
+        dest=sandbox,
+    )
+
+    mode = (mode or os.getenv("HARNESS_MODE") or "tool_loop").strip().lower()
+    # tool_loop: langchain bind_tools over LocalShellBackend (reliable on local GGUF)
+    # graph: create_deep_agent CompiledStateGraph
+
+    if model is None:
+        step(
+            "episode.model",
+            "ChatOpenAI",
+            base_url=os.environ.get("MODEL_BASE_URL"),
+            model=os.environ.get("MODEL_NAME"),
+        )
+        model = ChatOpenAI(
+            base_url=os.environ["MODEL_BASE_URL"],
+            api_key=os.getenv("MODEL_API_KEY", "local"),
+            model=os.environ["MODEL_NAME"],
+            temperature=0,
+            max_tokens=1024,
+        )
+
+    step("episode.backend", f"LocalShellBackend root={sandbox}")
+    backend = LocalShellBackend(root_dir=sandbox, virtual_mode=True, timeout=60)
+    prompt = _short_prompt(task)
+    eval_log_dir = (artifacts_dir / "evals" / episode_id) if artifacts_dir else None
+
+    messages_dump: list[dict[str, Any]] = [{"role": "user", "content": prompt}]
+    final_text = ""
+    termination = "unknown"
+    tools_used: list[str] = []
+    tool_calls_log: list[dict[str, Any]] = []
+    trace_id = ""
+    error: str | None = None
+    msgs: list[Any] = []
+
+    callback_handler = None
+    if langfuse_client is not None:
+        from deepagents_rl.observability.langfuse_trace import make_callback_handler
+
+        callback_handler = make_callback_handler()
+        step("episode.langfuse_handler", "CallbackHandler ready for LLM generations")
+
+    def _invoke() -> None:
+        nonlocal final_text, termination, tools_used, tool_calls_log, messages_dump, msgs
+        step("episode.agent_loop", f"mode={mode}")
+        if mode == "graph":
+            msgs, final_text, termination = _run_graph(
+                model=model,
+                backend=backend,
+                prompt=prompt,
+                recursion_limit=recursion_limit,
+                langfuse_client=langfuse_client,
+                callback_handler=callback_handler,
+                episode_id=episode_id,
+            )
+        else:
+            msgs, final_text, termination = _run_tool_loop(
+                model=model,
+                backend=backend,
+                prompt=prompt,
+                max_turns=max_turns,
+                langfuse_client=langfuse_client,
+                callback_handler=callback_handler,
+                episode_id=episode_id,
+            )
+        messages_dump = [_msg_to_dict(m) for m in msgs]
+        tools_used, tool_calls_log = _extract_tool_calls(msgs)
+        step(
+            "episode.agent_done",
+            termination,
+            n_messages=len(messages_dump),
+            n_tool_calls=len(tool_calls_log),
+            tools_used=tools_used,
+        )
+
+    def _grade() -> dict[str, Any]:
+        from deepagents_rl.harness.repo import git_diff_patch
+
+        step("episode.grade", "starting Docker SWE-bench eval")
+        if langfuse_client is not None:
+            with langfuse_client.start_as_current_observation(
+                name="reward_grader",
+                as_type="evaluator",
+                input={
+                    "fail_to_pass": task.get("fail_to_pass"),
+                    "pass_to_pass": task.get("pass_to_pass"),
+                    "grader": "swebench_docker_official_protocol",
+                    "image": task.get("image"),
+                    "n_tool_calls": len(tool_calls_log),
+                    "tools_used": tools_used,
+                },
+                metadata={
+                    "episode_id": episode_id,
+                    "task_id": task["task_id"],
+                },
+            ) as reward_span:
+                with langfuse_client.start_as_current_observation(
+                    name="git_diff_model_patch",
+                    as_type="span",
+                    metadata={"episode_id": episode_id},
+                ) as diff_span:
+                    model_patch = git_diff_patch(sandbox)
+                    diff_span.update(
+                        output={
+                            "patch_bytes": len(model_patch.encode("utf-8")),
+                            "patch_preview": truncate(model_patch, 1500),
+                            "empty": not bool(model_patch.strip()),
+                        }
+                    )
+                with langfuse_client.start_as_current_observation(
+                    name="docker_swe_eval",
+                    as_type="span",
+                    input={
+                        "image": task.get("image"),
+                        "patch_bytes": len(model_patch.encode("utf-8")),
+                    },
+                    metadata={"episode_id": episode_id},
+                ) as docker_span:
+                    g = grade_episode(
+                        sandbox=sandbox,
+                        task=task,
+                        tools_used=tools_used,
+                        tool_calls=tool_calls_log,
+                        model_patch=model_patch,
+                        eval_log_dir=eval_log_dir,
+                    )
+                    docker_span.update(
+                        output={
+                            "resolved": g.get("resolved"),
+                            "resolution": g.get("resolution"),
+                            "pass_frac": g.get("pass_frac"),
+                            "patch_successfully_applied": g.get(
+                                "patch_successfully_applied"
+                            ),
+                            "fail_to_pass": g.get("fail_to_pass"),
+                            "pass_to_pass": {
+                                "frac": (g.get("pass_to_pass") or {}).get("frac"),
+                                "n_failure": len(
+                                    (g.get("pass_to_pass") or {}).get("failure") or []
+                                ),
+                            },
+                            "status_map_sample": list(
+                                (g.get("status_map") or {}).items()
+                            )[:15],
+                            "eval_error": g.get("eval_error"),
+                            "eval_exit_code": g.get("eval_exit_code"),
+                            "eval_log_dir": g.get("eval_log_dir"),
+                        }
+                    )
+                reward_span.update(
+                    output={
+                        k: g.get(k)
+                        for k in (
+                            "pass_frac",
+                            "reward_total",
+                            "resolved",
+                            "resolution",
+                            "patch_successfully_applied",
+                            "fail_to_pass",
+                            "pass_to_pass",
+                            "eval_error",
+                        )
+                    }
+                )
+        else:
+            g = grade_episode(
+                sandbox=sandbox,
+                task=task,
+                tools_used=tools_used,
+                tool_calls=tool_calls_log,
+                eval_log_dir=eval_log_dir,
+            )
+        step(
+            "episode.grade_done",
+            f"resolved={g.get('resolved')} f2p={g.get('pass_frac')}",
+            resolution=g.get("resolution"),
+            eval_error=g.get("eval_error"),
+        )
+        return g
+
+    try:
+        if langfuse_client is not None:
+            with langfuse_client.start_as_current_observation(
+                name="deepagent-swe-episode",
+                as_type="agent",
+                metadata={
+                    "episode_id": episode_id,
+                    "task_id": task["task_id"],
+                    "repo": task.get("repo"),
+                    "split": task.get("split"),
+                    "source_dataset": task.get("source_dataset"),
+                    "harness": f"deepagents:{mode}",
+                    "model": os.getenv("MODEL_NAME"),
+                    "tags": ["deepagents-rl", "swe-bench", mode],
+                },
+                input={
+                    "task_id": task["task_id"],
+                    "prompt": prompt[:4000],
+                    "workspace_paths": task.get("workspace_paths"),
+                    "mode": mode,
+                    "fail_to_pass": task.get("fail_to_pass"),
+                },
+            ) as root:
+                # Nested agent-loop span so LLM/tool children group cleanly
+                with langfuse_client.start_as_current_observation(
+                    name="agent-loop",
+                    as_type="span",
+                    metadata={"mode": mode, "episode_id": episode_id},
+                    input={"max_turns": max_turns, "recursion_limit": recursion_limit},
+                ) as loop_span:
+                    _invoke()
+                    loop_span.update(
+                        output={
+                            "termination": termination,
+                            "n_tool_calls": len(tool_calls_log),
+                            "tools_used": tools_used,
+                            "final_text": truncate(final_text, 1000),
+                        }
+                    )
+
+                grade = _grade()
+
+                langfuse_client.score_current_trace(
+                    name="pass_frac", value=float(grade["pass_frac"])
+                )
+                langfuse_client.score_current_trace(
+                    name="reward_total", value=float(grade["reward_total"])
+                )
+                langfuse_client.score_current_trace(
+                    name="resolved", value=1.0 if grade.get("resolved") else 0.0
+                )
+                langfuse_client.score_current_trace(
+                    name="n_tool_calls", value=float(len(tool_calls_log))
+                )
+                root.update(
+                    output={
+                        "termination": termination,
+                        "final_text": final_text[:2000],
+                        "tools_used": tools_used,
+                        "n_tool_calls": len(tool_calls_log),
+                        "tool_calls": [
+                            {
+                                "turn": t.get("turn"),
+                                "name": t.get("name"),
+                                "args": t.get("args"),
+                                "result": truncate(t.get("result") or "", 500),
+                            }
+                            for t in tool_calls_log[:40]
+                        ],
+                        "grade": {
+                            k: grade.get(k)
+                            for k in (
+                                "pass_frac",
+                                "reward_total",
+                                "resolved",
+                                "resolution",
+                                "eval_error",
+                            )
+                        },
+                    }
+                )
+                trace_id = langfuse_client.get_current_trace_id() or ""
+            langfuse_client.flush()
+            step("episode.langfuse_flush", trace_id or "(no trace id)")
+        else:
+            _invoke()
+            grade = _grade()
+    except Exception as e:  # noqa: BLE001
+        error = f"{type(e).__name__}: {e}"
+        log_error("episode", error)
+        termination = "error"
+        grade = {
+            "pass_frac": 0.0,
+            "reward_total": 0.0,
+            "resolved": False,
+            "resolution": "NO",
+            "eval_error": error,
+            "grader": "swebench_docker_official_protocol",
+        }
+
+    duration_s = time.time() - started
+    trajectory = {
+        "episode_id": episode_id,
+        "task_id": task["task_id"],
+        "instance_id": task.get("instance_id"),
+        "split": task.get("split"),
+        "repo": task.get("repo"),
+        "base_commit": task.get("base_commit"),
+        "source_dataset": task.get("source_dataset") or "SWE-bench/SWE-bench_Lite",
+        "harness": f"deepagents:{mode}",
+        "model_name": os.getenv("MODEL_NAME"),
+        "model_base_url": os.getenv("MODEL_BASE_URL"),
+        "created_at": _utc_now(),
+        "duration_s": duration_s,
+        "recursion_limit": recursion_limit,
+        "max_turns": max_turns,
+        "sandbox": str(sandbox),
+        "prompt": prompt,
+        "messages": messages_dump,
+        "tool_calls": tool_calls_log,
+        "tools_used": tools_used,
+        "final_text": final_text,
+        "termination": termination,
+        "grade": grade,
+        "reward": {
+            "pass_frac": grade.get("pass_frac", 0.0),
+            "reward_total": grade.get("reward_total", 0.0),
+            "resolved": grade.get("resolved", False),
+            "resolution": grade.get("resolution"),
+        },
+        "model_patch": grade.get("model_patch"),
+        "langfuse_trace_id": trace_id,
+        "error": error,
+        "gold_paths": task.get("workspace_paths"),
+        "image": task.get("image"),
+    }
+
+    if artifacts_dir is not None:
+        artifacts_dir.mkdir(parents=True, exist_ok=True)
+        traj_path = artifacts_dir / f"episode_{episode_id}.json"
+        traj_path.write_text(json.dumps(trajectory, indent=2), encoding="utf-8")
+        jsonl_path = artifacts_dir / "trajectories.jsonl"
+        with jsonl_path.open("a", encoding="utf-8") as f:
+            f.write(json.dumps(trajectory) + "\n")
+        if grade.get("model_patch"):
+            (artifacts_dir / f"patch_{episode_id}.diff").write_text(
+                grade["model_patch"], encoding="utf-8"
+            )
+        trajectory["artifact_json"] = str(traj_path)
+        trajectory["artifact_jsonl"] = str(jsonl_path)
+
+    if not keep_sandbox:
+        try:
+            import shutil
+
+            step("episode.cleanup_sandbox", str(sandbox_parent))
+            shutil.rmtree(sandbox_parent, ignore_errors=True)
+            trajectory["sandbox"] = None
+        except Exception:
+            pass
+
+    step(
+        "episode.end",
+        termination,
+        episode_id=episode_id,
+        duration_s=round(duration_s, 2),
+        n_tool_calls=len(tool_calls_log),
+        resolved=grade.get("resolved"),
+        pass_frac=grade.get("pass_frac"),
+        artifact=trajectory.get("artifact_json"),
+    )
+    return trajectory

src/deepagents_rl/harness/grade.py

@@ -0,0 +1,91 @@
+"""Grade long-horizon coding episodes with official SWE-bench Docker evaluation.
+
+Reward = FAIL_TO_PASS fraction after applying the agent's git patch inside the
+instance image and running the official eval_script (test_patch + pytest).
+
+Resolution follows SWE-bench: FULL only if all F2P pass and all P2P maintained.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from deepagents_rl.harness.repo import git_diff_patch
+from deepagents_rl.harness.swe_docker_eval import evaluate_model_patch
+from deepagents_rl.observability.logging import span, step, truncate
+
+
+def grade_episode(
+    *,
+    sandbox: Path,
+    task: dict[str, Any],
+    tools_used: list[str] | None = None,
+    tool_calls: list[dict[str, Any]] | None = None,
+    model_patch: str | None = None,
+    eval_log_dir: Path | None = None,
+    timeout_s: int = 1800,
+) -> dict[str, Any]:
+    """
+    Produce model_patch from the agent sandbox (if not provided), then run
+    Docker SWE-bench evaluation.
+    """
+    tools_used = tools_used or []
+    tool_calls = tool_calls or []
+
+    with span(
+        "grade.episode",
+        task.get("instance_id") or task.get("task_id"),
+        sandbox=str(sandbox),
+        n_tool_calls=len(tool_calls),
+    ):
+        if model_patch is None:
+            step("grade.diff", "collecting git diff from agent sandbox")
+            model_patch = git_diff_patch(sandbox)
+        else:
+            step("grade.diff", "using provided model_patch")
+
+        step(
+            "grade.patch_ready",
+            f"bytes={len(model_patch.encode('utf-8'))}",
+            empty=not bool(model_patch.strip()),
+            preview=truncate(model_patch, 400),
+        )
+        step(
+            "grade.docker_eval",
+            f"image={task.get('image')}",
+            n_f2p=len(task.get("fail_to_pass") or []),
+            n_p2p=len(task.get("pass_to_pass") or []),
+        )
+        report = evaluate_model_patch(
+            task,
+            model_patch,
+            timeout_s=timeout_s,
+            log_dir=eval_log_dir,
+        )
+
+        out = {
+            "pass_frac": float(report.get("pass_frac") or 0.0),
+            "reward_total": float(report.get("reward_total") or 0.0),
+            "resolved": bool(report.get("resolved")),
+            "resolution": report.get("resolution") or "NO",
+            "patch_successfully_applied": bool(report.get("patch_successfully_applied")),
+            "model_patch": model_patch,
+            "fail_to_pass": report.get("fail_to_pass"),
+            "pass_to_pass": report.get("pass_to_pass"),
+            "status_map": report.get("status_map") or {},
+            "eval_error": report.get("error"),
+            "eval_log_dir": report.get("log_dir"),
+            "eval_exit_code": report.get("eval_exit_code"),
+            "n_tool_calls": len(tool_calls),
+            "tools_used": tools_used,
+            "image": report.get("image") or task.get("image"),
+            "grader": "swebench_docker_official_protocol",
+        }
+        step(
+            "grade.done",
+            f"resolution={out['resolution']}",
+            resolved=out["resolved"],
+            pass_frac=out["pass_frac"],
+            eval_error=out["eval_error"],
+        )
+        return out

src/deepagents_rl/harness/repo.py

@@ -0,0 +1,128 @@
+"""Full git checkout of a SWE-bench instance repo at base_commit."""
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+from deepagents_rl.observability.logging import call, error, result, span, truncate
+
+
+def repo_cache_root() -> Path:
+    root = Path(os.getenv("SWE_REPO_CACHE", "")) if os.getenv("SWE_REPO_CACHE") else None
+    if root is None:
+        root = Path(__file__).resolve().parents[3] / "data" / "repos"
+    root.mkdir(parents=True, exist_ok=True)
+    return root
+
+
+def _run_git(
+    args: list[str],
+    *,
+    cwd: str | Path | None = None,
+    check: bool = True,
+    timeout: int = 600,
+) -> subprocess.CompletedProcess[str]:
+    cmd = ["git", *args]
+    call("git", " ".join(cmd), cwd=str(cwd) if cwd else None, timeout_s=timeout)
+    try:
+        proc = subprocess.run(
+            cmd,
+            cwd=str(cwd) if cwd else None,
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+    except subprocess.TimeoutExpired as e:
+        error("git", f"timeout after {timeout}s", cmd=cmd)
+        raise e
+    out = ((proc.stdout or "") + (proc.stderr or "")).strip()
+    result(
+        "git",
+        f"exit={proc.returncode}",
+        exit_code=proc.returncode,
+        output=truncate(out, 600),
+    )
+    if check and proc.returncode != 0:
+        raise subprocess.CalledProcessError(proc.returncode, cmd, proc.stdout, proc.stderr)
+    return proc
+
+
+def clone_or_fetch(repo: str, cache_dir: Path | None = None) -> Path:
+    """Mirror-clone (once) github.com/{repo} into the local cache."""
+    cache = cache_dir or repo_cache_root()
+    mirror = cache / f"{repo.replace('/', '__')}.git"
+    url = f"https://github.com/{repo}.git"
+    with span("repo.clone_or_fetch", repo=repo, mirror=str(mirror)):
+        if mirror.exists():
+            call("repo.fetch", f"fetch existing mirror {mirror}")
+            _run_git(
+                ["fetch", "--all", "--tags", "--prune"],
+                cwd=mirror,
+                check=False,
+                timeout=600,
+            )
+            return mirror
+        mirror.parent.mkdir(parents=True, exist_ok=True)
+        call("repo.mirror_clone", f"clone --mirror {url}")
+        _run_git(["clone", "--mirror", url, str(mirror)], timeout=1800)
+        return mirror
+
+
+def materialize_repo_at_commit(
+    *,
+    repo: str,
+    base_commit: str,
+    dest: Path,
+    cache_dir: Path | None = None,
+) -> Path:
+    """
+    Create a full working tree at `dest` checked out to `base_commit`.
+
+    This is the agent sandbox — not a sparse file dump.
+    """
+    with span(
+        "repo.materialize",
+        f"{repo}@{base_commit[:12]}",
+        repo=repo,
+        base_commit=base_commit,
+        dest=str(dest),
+    ):
+        if dest.exists():
+            call("repo.rmtree", str(dest))
+            shutil.rmtree(dest, ignore_errors=True)
+        dest.mkdir(parents=True, exist_ok=True)
+
+        mirror = clone_or_fetch(repo, cache_dir=cache_dir)
+        call("repo.workdir_clone", f"from {mirror} -> {dest}")
+        _run_git(["clone", str(mirror), str(dest)], timeout=600)
+        call("repo.checkout", base_commit)
+        _run_git(["checkout", "--force", base_commit], cwd=dest, timeout=120)
+        _run_git(["reset", "--hard", "HEAD"], cwd=dest, timeout=60)
+        _run_git(["clean", "-fdx"], cwd=dest, timeout=60)
+        head = _run_git(["rev-parse", "HEAD"], cwd=dest, timeout=30)
+        result(
+            "repo.materialize",
+            "sandbox ready",
+            head=(head.stdout or "").strip(),
+            dest=str(dest),
+        )
+        return dest
+
+
+def git_diff_patch(repo_dir: Path) -> str:
+    """Unified diff of all agent edits (tracked + untracked as new files)."""
+    with span("repo.git_diff_patch", repo_dir=str(repo_dir)):
+        _run_git(["add", "-A"], cwd=repo_dir, timeout=60)
+        proc = _run_git(["diff", "--cached", "--binary"], cwd=repo_dir, timeout=60)
+        _run_git(["reset", "HEAD"], cwd=repo_dir, check=False, timeout=60)
+        patch = proc.stdout or ""
+        result(
+            "repo.git_diff_patch",
+            f"patch_bytes={len(patch.encode('utf-8'))}",
+            patch_preview=truncate(patch, 500),
+            empty=not bool(patch.strip()),
+        )
+        return patch

src/deepagents_rl/harness/swe_docker_eval.py

@@ -0,0 +1,465 @@
+"""Official-style SWE-bench evaluation via the instance Docker image.
+
+Protocol (matches SWE-bench harness):
+  1. Start the prebuilt instance image (`task['image']`)
+  2. Apply the model's git patch into `/testbed`
+  3. Run the instance `eval_script` (applies test_patch, runs pytest)
+  4. Parse status between `>>>>> Start/End Test Output`
+  5. Resolve = all FAIL_TO_PASS pass AND all PASS_TO_PASS pass
+
+Uses the Docker CLI (not the Python SDK) for reliability on Windows.
+"""
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+import tempfile
+import time
+import uuid
+from pathlib import Path
+from typing import Any
+
+from deepagents_rl.observability.logging import call, error, result, span, step, truncate
+
+START_TEST_OUTPUT = ">>>>> Start Test Output"
+END_TEST_OUTPUT = ">>>>> End Test Output"
+
+GIT_APPLY_CMDS = [
+    "git apply --verbose",
+    "git apply --verbose --3way",
+    "git apply --verbose --reject",
+]
+
+
+def fix_eval_script_for_docker_desktop(script: str) -> str:
+    """
+    Upstream eval_script uses `conda activate testbed`. On Docker Desktop the
+    image's *base* conda entrypoint is often broken (`conda` -> TypeError), while
+    `/opt/miniconda3/envs/testbed/bin/*` works. Prefetch that env onto PATH and
+    stub `conda activate` so the official test_patch + pytest body still runs.
+    """
+    preamble = (
+        "#!/bin/bash\n"
+        "set -uxo pipefail\n"
+        'export PATH="/opt/miniconda3/envs/testbed/bin:/opt/miniconda3/bin:$PATH"\n'
+        "conda() {\n"
+        '  if [ "$1" = "activate" ]; then\n'
+        '    export PATH="/opt/miniconda3/envs/${2:-testbed}/bin:$PATH"\n'
+        "    hash -r\n"
+        "    return 0\n"
+        "  fi\n"
+        "  return 0\n"
+        "}\n"
+    )
+    body: list[str] = []
+    for line in script.replace("\r\n", "\n").splitlines():
+        if line.startswith("#!"):
+            continue
+        if line.strip().startswith("source /opt/miniconda3/bin/activate"):
+            continue
+        if "conda activate" in line:
+            continue
+        body.append(line)
+    return preamble + "\n".join(body) + "\n"
+
+
+def parse_log_pytest(log: str) -> dict[str, str]:
+    """SWE-bench pytest log parser (status token first on the line)."""
+    status_map: dict[str, str] = {}
+    statuses = ("PASSED", "FAILED", "ERROR", "SKIPPED", "XFAIL", "XPASS")
+    for line in log.splitlines():
+        if not any(line.startswith(s) for s in statuses):
+            continue
+        if line.startswith("FAILED"):
+            line = line.replace(" - ", " ")
+        parts = line.split()
+        if len(parts) <= 1:
+            continue
+        status_map[parts[1]] = parts[0]
+    return status_map
+
+
+def _extract_test_region(log: str) -> str | None:
+    if START_TEST_OUTPUT not in log or END_TEST_OUTPUT not in log:
+        return None
+    return log.split(START_TEST_OUTPUT, 1)[1].split(END_TEST_OUTPUT, 1)[0]
+
+
+def _test_passed(case: str, sm: dict[str, str]) -> bool:
+    return case in sm and sm[case] in ("PASSED", "XFAIL")
+
+
+def _test_failed(case: str, sm: dict[str, str]) -> bool:
+    return case not in sm or sm[case] in ("FAILED", "ERROR", "SKIPPED")
+
+
+def _test_maintained(case: str, sm: dict[str, str]) -> bool:
+    return _test_passed(case, sm) or (case in sm and sm[case] == "SKIPPED")
+
+
+def grade_status_map(
+    status_map: dict[str, str],
+    fail_to_pass: list[str],
+    pass_to_pass: list[str],
+) -> dict[str, Any]:
+    """Official resolution: FULL iff all F2P pass and all P2P maintained."""
+    f2p_ok = [t for t in fail_to_pass if _test_passed(t, status_map)]
+    f2p_fail = [t for t in fail_to_pass if _test_failed(t, status_map)]
+    p2p_ok = [t for t in pass_to_pass if _test_maintained(t, status_map)]
+    p2p_fail = [t for t in pass_to_pass if not _test_maintained(t, status_map)]
+
+    f2p_frac = (len(f2p_ok) / len(fail_to_pass)) if fail_to_pass else 1.0
+    p2p_frac = (len(p2p_ok) / len(pass_to_pass)) if pass_to_pass else 1.0
+
+    if f2p_frac == 1.0 and p2p_frac == 1.0:
+        resolved = "FULL"
+    elif 0.0 < f2p_frac < 1.0 and p2p_frac == 1.0:
+        resolved = "PARTIAL"
+    else:
+        resolved = "NO"
+
+    return {
+        "resolved": resolved == "FULL",
+        "resolution": resolved,
+        "fail_to_pass": {"success": f2p_ok, "failure": f2p_fail, "frac": f2p_frac},
+        "pass_to_pass": {"success": p2p_ok, "failure": p2p_fail, "frac": p2p_frac},
+        "pass_frac": float(f2p_frac),
+        # P2P regressions discount reward (cannot claim FULL/PARTIAL)
+        "reward_total": float(f2p_frac if p2p_frac == 1.0 else f2p_frac * 0.5),
+        "status_map": status_map,
+    }
+
+
+def _docker(
+    *args: str,
+    timeout: int | None = None,
+    merge_stderr: bool = False,
+) -> subprocess.CompletedProcess[str]:
+    exe = shutil.which("docker")
+    if not exe:
+        raise RuntimeError("docker CLI not found on PATH")
+    cmdline = "docker " + " ".join(args)
+    call("docker", cmdline, timeout_s=timeout, merge_stderr=merge_stderr)
+    t0 = time.perf_counter()
+    try:
+        if merge_stderr:
+            proc = subprocess.run(
+                [exe, *args],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                text=True,
+                timeout=timeout,
+                encoding="utf-8",
+                errors="replace",
+            )
+        else:
+            proc = subprocess.run(
+                [exe, *args],
+                capture_output=True,
+                text=True,
+                timeout=timeout,
+                encoding="utf-8",
+                errors="replace",
+            )
+    except subprocess.TimeoutExpired:
+        error("docker", f"timeout after {timeout}s", cmd=cmdline)
+        raise
+    dt = int((time.perf_counter() - t0) * 1000)
+    out = (proc.stdout or "") + ("" if merge_stderr else (proc.stderr or ""))
+    result(
+        "docker",
+        f"exit={proc.returncode} ({dt}ms)",
+        exit_code=proc.returncode,
+        duration_ms=dt,
+        output=truncate(out, 800),
+    )
+    return proc
+
+
+def evaluate_model_patch(
+    task: dict[str, Any],
+    model_patch: str,
+    *,
+    timeout_s: int = 1800,
+    log_dir: Path | None = None,
+    keep_container: bool = False,
+) -> dict[str, Any]:
+    """
+    Grade one prediction with the official instance Docker image + eval_script.
+
+    `task` must include: instance_id, image, eval_script, fail_to_pass, pass_to_pass
+    """
+    instance_id = task["instance_id"]
+    image = task.get("image")
+    eval_script = task.get("eval_script") or ""
+    fail_to_pass = list(task.get("fail_to_pass") or [])
+    pass_to_pass = list(task.get("pass_to_pass") or [])
+
+    if not image:
+        raise ValueError(f"{instance_id}: missing Docker image field from SWE-bench Lite")
+    if not eval_script.strip():
+        raise ValueError(f"{instance_id}: missing eval_script")
+
+    if log_dir is None:
+        log_dir = Path(tempfile.mkdtemp(prefix="swe_eval_"))
+    log_dir.mkdir(parents=True, exist_ok=True)
+
+    report: dict[str, Any] = {
+        "instance_id": instance_id,
+        "image": image,
+        "patch_exists": bool(model_patch and model_patch.strip()),
+        "patch_successfully_applied": False,
+        "resolved": False,
+        "resolution": "NO",
+        "pass_frac": 0.0,
+        "reward_total": 0.0,
+        "fail_to_pass": {"success": [], "failure": fail_to_pass, "frac": 0.0},
+        "pass_to_pass": {"success": [], "failure": [], "frac": 0.0},
+        "status_map": {},
+        "error": None,
+        "log_dir": str(log_dir),
+    }
+
+    step(
+        "eval.begin",
+        f"instance={instance_id}",
+        image=image,
+        patch_bytes=len((model_patch or "").encode("utf-8")),
+        n_f2p=len(fail_to_pass),
+        n_p2p=len(pass_to_pass),
+        log_dir=str(log_dir),
+    )
+
+    if not model_patch or not model_patch.strip():
+        step("eval.empty_patch", "no model_patch — unresolved")
+        report["error"] = "empty_model_patch"
+        (log_dir / "report.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
+        return report
+
+    patch_path = log_dir / "model.patch"
+    eval_path = log_dir / "eval.sh"
+    eval_raw_path = log_dir / "eval.upstream.sh"
+    patch_path.write_text(model_patch.replace("\r\n", "\n"), encoding="utf-8", newline="\n")
+    eval_raw_path.write_text(eval_script.replace("\r\n", "\n"), encoding="utf-8", newline="\n")
+    eval_path.write_text(
+        fix_eval_script_for_docker_desktop(eval_script),
+        encoding="utf-8",
+        newline="\n",
+    )
+    step(
+        "eval.write_files",
+        "wrote model.patch + eval.sh",
+        patch_path=str(patch_path),
+        eval_path=str(eval_path),
+        patch_preview=truncate(model_patch, 400),
+    )
+
+    run_tag = uuid.uuid4().hex[:10]
+    container = f"darl-eval-{run_tag}"
+
+    try:
+        with span("eval.docker_lifecycle", container=container, image=image):
+            inspect = _docker("image", "inspect", image, timeout=60)
+            if inspect.returncode != 0:
+                step("eval.docker_pull", f"image missing locally, pulling {image}")
+                pull = _docker("pull", image, timeout=max(timeout_s, 3600))
+                if pull.returncode != 0:
+                    report["error"] = f"docker_pull_failed: {(pull.stderr or pull.stdout)[-2000:]}"
+                    error("eval.docker_pull", report["error"])
+                    (log_dir / "report.json").write_text(
+                        json.dumps(report, indent=2), encoding="utf-8"
+                    )
+                    return report
+            else:
+                step("eval.docker_image", "image already present")
+
+            _docker("rm", "-f", container, timeout=60)
+
+            create = _docker(
+                "create",
+                "--name",
+                container,
+                "--workdir",
+                "/testbed",
+                image,
+                "tail",
+                "-f",
+                "/dev/null",
+                timeout=120,
+            )
+            if create.returncode != 0:
+                report["error"] = (
+                    f"docker_create_failed: {(create.stderr or create.stdout)[-2000:]}"
+                )
+                error("eval.docker_create", report["error"])
+                (log_dir / "report.json").write_text(
+                    json.dumps(report, indent=2), encoding="utf-8"
+                )
+                return report
+
+            start = _docker("start", container, timeout=120)
+            if start.returncode != 0:
+                report["error"] = (
+                    f"docker_start_failed: {(start.stderr or start.stdout)[-2000:]}"
+                )
+                error("eval.docker_start", report["error"])
+                (log_dir / "report.json").write_text(
+                    json.dumps(report, indent=2), encoding="utf-8"
+                )
+                return report
+
+            step("eval.docker_cp", "copy patch + eval.sh into container")
+            cp1 = _docker("cp", str(patch_path), f"{container}:/tmp/model.patch", timeout=120)
+            cp2 = _docker("cp", str(eval_path), f"{container}:/eval.sh", timeout=120)
+            if cp1.returncode != 0 or cp2.returncode != 0:
+                report["error"] = (
+                    f"docker_cp_failed: patch={(cp1.stderr or '')[-500:]} "
+                    f"eval={(cp2.stderr or '')[-500:]}"
+                )
+                error("eval.docker_cp", report["error"])
+                (log_dir / "report.json").write_text(
+                    json.dumps(report, indent=2), encoding="utf-8"
+                )
+                return report
+
+            _docker(
+                "exec",
+                "-w",
+                "/testbed",
+                container,
+                "git",
+                "config",
+                "--global",
+                "--add",
+                "safe.directory",
+                "/testbed",
+                timeout=60,
+            )
+
+            apply_log = []
+            applied = False
+            step("eval.apply_patch", "trying git apply cascade")
+            for cmd in GIT_APPLY_CMDS:
+                parts = cmd.split()
+                resp = _docker(
+                    "exec",
+                    "-w",
+                    "/testbed",
+                    container,
+                    *parts,
+                    "/tmp/model.patch",
+                    timeout=120,
+                )
+                apply_log.append(
+                    {
+                        "cmd": f"{cmd} /tmp/model.patch",
+                        "exit": resp.returncode,
+                        "out": ((resp.stdout or "") + (resp.stderr or ""))[-2000:],
+                    }
+                )
+                if resp.returncode == 0:
+                    applied = True
+                    step("eval.apply_patch_ok", cmd)
+                    break
+                step("eval.apply_patch_retry", f"failed: {cmd}", exit_code=resp.returncode)
+            (log_dir / "apply_patch.json").write_text(
+                json.dumps(apply_log, indent=2), encoding="utf-8"
+            )
+
+            if not applied:
+                report["error"] = "patch_apply_failed"
+                report["apply_log"] = apply_log
+                error("eval.apply_patch", "all git apply strategies failed")
+                (log_dir / "test_output.txt").write_text(
+                    json.dumps(apply_log, indent=2), encoding="utf-8"
+                )
+                (log_dir / "report.json").write_text(
+                    json.dumps(report, indent=2), encoding="utf-8"
+                )
+                return report
+
+            report["patch_successfully_applied"] = True
+
+            step("eval.run_eval_script", "bash /eval.sh (official test_patch + pytest)")
+            t0 = time.time()
+            eval_proc = _docker(
+                "exec",
+                "-w",
+                "/testbed",
+                container,
+                "/bin/bash",
+                "-lc",
+                "/bin/bash /eval.sh 2>&1",
+                timeout=timeout_s,
+                merge_stderr=True,
+            )
+            elapsed = time.time() - t0
+            test_output = eval_proc.stdout or ""
+            (log_dir / "test_output.txt").write_text(test_output, encoding="utf-8")
+            report["eval_exit_code"] = eval_proc.returncode
+            report["eval_seconds"] = elapsed
+            step(
+                "eval.eval_script_done",
+                f"exit={eval_proc.returncode} seconds={elapsed:.1f}",
+                output_chars=len(test_output),
+                output_tail=truncate(test_output, 600),
+            )
+
+            region = _extract_test_region(test_output)
+            status_map = parse_log_pytest(region or "")
+            if not status_map:
+                step("eval.parse_fallback", "marked region empty — parsing full log")
+                status_map = parse_log_pytest(test_output)
+            if not status_map:
+                report["error"] = (
+                    "missing_test_output_markers"
+                    if region is None
+                    else "no_parsed_test_statuses"
+                )
+                error("eval.parse", report["error"])
+                (log_dir / "report.json").write_text(
+                    json.dumps(report, indent=2), encoding="utf-8"
+                )
+                return report
+
+            step(
+                "eval.parse_ok",
+                f"parsed {len(status_map)} test statuses",
+                sample=list(status_map.items())[:8],
+            )
+            graded = grade_status_map(status_map, fail_to_pass, pass_to_pass)
+            report.update(graded)
+            report["error"] = None
+            step(
+                "eval.grade",
+                f"resolution={graded['resolution']} f2p={graded['pass_frac']}",
+                resolved=graded["resolved"],
+                resolution=graded["resolution"],
+                pass_frac=graded["pass_frac"],
+                reward_total=graded["reward_total"],
+                f2p_success=graded["fail_to_pass"]["success"],
+                f2p_failure=graded["fail_to_pass"]["failure"],
+                p2p_frac=graded["pass_to_pass"]["frac"],
+            )
+            (log_dir / "report.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
+            return report
+
+    except subprocess.TimeoutExpired as e:
+        report["error"] = f"timeout: {e}"
+        error("eval.timeout", report["error"])
+        (log_dir / "report.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
+        return report
+    except Exception as e:  # noqa: BLE001
+        report["error"] = f"{type(e).__name__}: {e}"
+        error("eval.exception", report["error"])
+        (log_dir / "report.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
+        return report
+    finally:
+        if not keep_container:
+            try:
+                step("eval.cleanup", f"rm container {container}")
+                _docker("rm", "-f", container, timeout=120)
+            except Exception as e:  # noqa: BLE001
+                error("eval.cleanup", str(e))
+

src/deepagents_rl/observability/__init__.py

@@ -1,3 +1,15 @@
 from .langfuse_eval import init_langfuse, log_checkpoint_eval
+from .langfuse_trace import llm_invoke_config, make_callback_handler
+from .logging import configure_logging, get_logger, log_event, span, step
 
-__all__ = ["init_langfuse", "log_checkpoint_eval"]
+__all__ = [
+    "init_langfuse",
+    "log_checkpoint_eval",
+    "make_callback_handler",
+    "llm_invoke_config",
+    "configure_logging",
+    "get_logger",
+    "log_event",
+    "span",
+    "step",
+]

src/deepagents_rl/observability/langfuse_trace.py

@@ -0,0 +1,31 @@
+"""Helpers for granular Langfuse traces (LLM generations + tool spans)."""
+from __future__ import annotations
+
+from typing import Any
+
+
+def make_callback_handler():
+    """LangChain CallbackHandler that nests generations under the current observation."""
+    from langfuse.langchain import CallbackHandler
+
+    return CallbackHandler()
+
+
+def llm_invoke_config(
+    *,
+    handler: Any,
+    turn: int,
+    episode_id: str | None = None,
+    extra_metadata: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    meta = {"turn": turn}
+    if episode_id:
+        meta["episode_id"] = episode_id
+    if extra_metadata:
+        meta.update(extra_metadata)
+    return {
+        "callbacks": [handler],
+        "run_name": f"llm-turn-{turn}",
+        "tags": ["deepagents-rl", "llm", f"turn-{turn}"],
+        "metadata": meta,
+    }

src/deepagents_rl/observability/logging.py

@@ -0,0 +1,180 @@
+"""Structured step logging for deepagents-rl pipelines.
+
+Emits:
+  - Console (stderr) with timestamps + step tags
+  - Optional JSONL event file for full replay (`events.jsonl`)
+
+Env:
+  DARL_LOG_LEVEL   DEBUG|INFO|WARNING (default INFO)
+  DARL_LOG_FILE    path to text log (optional)
+  DARL_EVENTS_JSONL path to JSONL events (optional; auto under artifacts if set via configure)
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+import sys
+import threading
+import time
+import traceback
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Iterator
+
+_LOCK = threading.Lock()
+_EVENTS_PATH: Path | None = None
+_CONFIGURED = False
+
+LOGGER_NAME = "deepagents_rl"
+
+
+def _utc() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def get_logger(name: str | None = None) -> logging.Logger:
+    return logging.getLogger(name or LOGGER_NAME)
+
+
+def configure_logging(
+    *,
+    level: str | None = None,
+    log_file: str | Path | None = None,
+    events_jsonl: str | Path | None = None,
+    force: bool = False,
+) -> logging.Logger:
+    """Idempotent logging setup for CLI scripts and harness modules."""
+    global _CONFIGURED, _EVENTS_PATH
+    log = get_logger()
+
+    lvl_name = (level or os.getenv("DARL_LOG_LEVEL") or "INFO").upper()
+    lvl = getattr(logging, lvl_name, logging.INFO)
+
+    if _CONFIGURED and not force:
+        log.setLevel(lvl)
+        if events_jsonl:
+            _EVENTS_PATH = Path(events_jsonl)
+            _EVENTS_PATH.parent.mkdir(parents=True, exist_ok=True)
+        return log
+
+    log.setLevel(lvl)
+    log.handlers.clear()
+    log.propagate = False
+
+    fmt = logging.Formatter(
+        fmt="%(asctime)s | %(levelname)-5s | %(name)s | %(message)s",
+        datefmt="%H:%M:%S",
+    )
+    sh = logging.StreamHandler(sys.stderr)
+    sh.setLevel(lvl)
+    sh.setFormatter(fmt)
+    log.addHandler(sh)
+
+    file_path = log_file or os.getenv("DARL_LOG_FILE")
+    if file_path:
+        fp = Path(file_path)
+        fp.parent.mkdir(parents=True, exist_ok=True)
+        fh = logging.FileHandler(fp, encoding="utf-8")
+        fh.setLevel(lvl)
+        fh.setFormatter(fmt)
+        log.addHandler(fh)
+
+    ev = events_jsonl or os.getenv("DARL_EVENTS_JSONL")
+    if ev:
+        _EVENTS_PATH = Path(ev)
+        _EVENTS_PATH.parent.mkdir(parents=True, exist_ok=True)
+    else:
+        _EVENTS_PATH = None
+
+    _CONFIGURED = True
+    log.debug("logging configured level=%s file=%s events=%s", lvl_name, file_path, _EVENTS_PATH)
+    return log
+
+
+def _write_event(event: dict[str, Any]) -> None:
+    if _EVENTS_PATH is None:
+        return
+    with _LOCK:
+        with _EVENTS_PATH.open("a", encoding="utf-8") as f:
+            f.write(json.dumps(event, default=str) + "\n")
+
+
+def log_event(
+    event_type: str,
+    message: str,
+    *,
+    level: int = logging.INFO,
+    **fields: Any,
+) -> None:
+    """Log a structured event to console + JSONL."""
+    log = get_logger()
+    payload = {
+        "ts": _utc(),
+        "event": event_type,
+        "message": message,
+        **fields,
+    }
+    # Compact console line
+    extras = {k: v for k, v in fields.items() if k not in ("traceback",)}
+    extra_s = ""
+    if extras:
+        try:
+            extra_s = " | " + json.dumps(extras, default=str)[:1500]
+        except Exception:
+            extra_s = f" | {extras!r}"[:1500]
+    log.log(level, f"[{event_type}] {message}{extra_s}")
+    _write_event(payload)
+
+
+def step(name: str, message: str = "", **fields: Any) -> None:
+    log_event("step", message or name, step=name, **fields)
+
+
+def call(name: str, message: str = "", **fields: Any) -> None:
+    log_event("call", message or name, call=name, **fields)
+
+
+def result(name: str, message: str = "", **fields: Any) -> None:
+    log_event("result", message or name, call=name, **fields)
+
+
+def error(name: str, message: str, **fields: Any) -> None:
+    log_event("error", message, level=logging.ERROR, call=name, **fields)
+
+
+@contextmanager
+def span(name: str, message: str = "", **fields: Any) -> Iterator[dict[str, Any]]:
+    """Timed span: logs start + end with duration_ms."""
+    t0 = time.perf_counter()
+    ctx: dict[str, Any] = {"span": name, **fields}
+    log_event("span_start", message or name, **ctx)
+    try:
+        yield ctx
+    except Exception as e:
+        dt = int((time.perf_counter() - t0) * 1000)
+        error(
+            name,
+            f"{type(e).__name__}: {e}",
+            duration_ms=dt,
+            traceback=traceback.format_exc()[-3000:],
+            **{k: v for k, v in ctx.items() if k != "span"},
+        )
+        raise
+    else:
+        dt = int((time.perf_counter() - t0) * 1000)
+        log_event(
+            "span_end",
+            message or name,
+            duration_ms=dt,
+            ok=True,
+            **ctx,
+        )
+
+
+def truncate(s: Any, n: int = 800) -> str:
+    text = s if isinstance(s, str) else json.dumps(s, default=str)
+    if len(text) <= n:
+        return text
+    return text[:n] + f"...(+{len(text)-n} chars)"

src/deepagents_rl/train/__init__.py

@@ -1,3 +1,9 @@
 from .resolve_model import resolve_train_model
+from .swe_env import DeepAgentSWEEnv, grade_patch_remote, prepare_swe_grpo_rows
 
-__all__ = ["resolve_train_model"]
+__all__ = [
+    "resolve_train_model",
+    "DeepAgentSWEEnv",
+    "grade_patch_remote",
+    "prepare_swe_grpo_rows",
+]

src/deepagents_rl/train/swe_env.py

@@ -0,0 +1,338 @@
+"""TRL GRPO `environment_factory` for deepagents SWE-bench sandboxes.
+
+Kaggle has no Docker for instance images — `get_reward` POSTs the model patch
+to `SWE_GRADER_URL` (local `scripts/swe_grader_server.py` + tunnel).
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import tempfile
+from pathlib import Path
+from typing import Any
+
+
+def _result_to_str(obj: Any) -> str:
+    if obj is None:
+        return ""
+    if isinstance(obj, str):
+        return obj[:8000]
+    if hasattr(obj, "output") and isinstance(obj.output, str):
+        return obj.output[:8000]
+    if hasattr(obj, "error") and obj.error:
+        return f"ERROR: {obj.error}"[:8000]
+    try:
+        return json.dumps(obj, default=str)[:8000]
+    except Exception:
+        return str(obj)[:8000]
+
+
+def _as_str_list(val: Any) -> list[str]:
+    if val is None:
+        return []
+    if isinstance(val, list):
+        return [str(x) for x in val]
+    if isinstance(val, str):
+        s = val.strip()
+        if not s:
+            return []
+        try:
+            parsed = json.loads(s)
+            if isinstance(parsed, list):
+                return [str(x) for x in parsed]
+        except json.JSONDecodeError:
+            pass
+        return [s]
+    return [str(val)]
+
+
+def grade_patch_remote(
+    *,
+    instance_id: str,
+    docker_image: str,
+    eval_script: str,
+    fail_to_pass: list[str],
+    pass_to_pass: list[str],
+    model_patch: str,
+    timeout_s: int = 1800,
+) -> dict[str, Any]:
+    """POST patch to the local Docker grader service."""
+    import requests
+
+    base = (os.environ.get("SWE_GRADER_URL") or "").rstrip("/")
+    if not base:
+        raise RuntimeError("SWE_GRADER_URL is not set (point at the local grader tunnel)")
+
+    headers = {
+        "Content-Type": "application/json",
+        # ngrok free tier interstitial
+        "ngrok-skip-browser-warning": "1",
+    }
+    token = os.environ.get("SWE_GRADER_TOKEN") or ""
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+
+    payload = {
+        "instance_id": instance_id,
+        "image": docker_image,
+        "eval_script": eval_script,
+        "fail_to_pass": fail_to_pass,
+        "pass_to_pass": pass_to_pass,
+        "model_patch": model_patch or "",
+        "timeout_s": timeout_s,
+    }
+    resp = requests.post(f"{base}/grade", headers=headers, json=payload, timeout=timeout_s + 60)
+    resp.raise_for_status()
+    return resp.json()
+
+
+class DeepAgentSWEEnv:
+    """Per-rollout TRL environment: LocalShellBackend tools + remote Docker reward."""
+
+    def __init__(self) -> None:
+        self._sandbox_parent: Path | None = None
+        self._sandbox: Path | None = None
+        self._backend: Any = None
+        self._task: dict[str, Any] = {}
+        self._last_reward: float = 0.0
+        self._last_report: dict[str, Any] = {}
+
+    def reset(self, **kwargs: Any) -> None:
+        """Materialize repo sandbox from a SWE Lite row. Prompt stays on the dataset."""
+        self._cleanup()
+
+        instance_id = str(kwargs.get("instance_id") or kwargs.get("task_id") or "")
+        repo = str(kwargs.get("repo") or "")
+        base_commit = str(kwargs.get("base_commit") or "")
+        docker_image = str(
+            kwargs.get("docker_image") or kwargs.get("image") or ""
+        )
+        eval_script = str(kwargs.get("eval_script") or "")
+        fail_to_pass = _as_str_list(
+            kwargs.get("fail_to_pass")
+            if kwargs.get("fail_to_pass") is not None
+            else kwargs.get("fail_to_pass_json")
+        )
+        pass_to_pass = _as_str_list(
+            kwargs.get("pass_to_pass")
+            if kwargs.get("pass_to_pass") is not None
+            else kwargs.get("pass_to_pass_json")
+        )
+
+        if not instance_id or not repo or not base_commit:
+            raise ValueError(
+                f"DeepAgentSWEEnv.reset requires instance_id/repo/base_commit; got keys={list(kwargs)}"
+            )
+        if not docker_image or not eval_script.strip():
+            raise ValueError(f"{instance_id}: missing docker_image/eval_script")
+
+        from deepagents.backends import LocalShellBackend
+
+        from deepagents_rl.harness.repo import materialize_repo_at_commit
+
+        self._task = {
+            "instance_id": instance_id,
+            "task_id": instance_id,
+            "repo": repo,
+            "base_commit": base_commit,
+            "image": docker_image,
+            "docker_image": docker_image,
+            "eval_script": eval_script,
+            "fail_to_pass": fail_to_pass,
+            "pass_to_pass": pass_to_pass,
+        }
+        self._last_reward = 0.0
+        self._last_report = {}
+
+        self._sandbox_parent = Path(
+            tempfile.mkdtemp(prefix=f"darl_grpo_{instance_id[:40]}_")
+        )
+        self._sandbox = self._sandbox_parent / "repo"
+        materialize_repo_at_commit(repo=repo, base_commit=base_commit, dest=self._sandbox)
+        self._backend = LocalShellBackend(
+            root_dir=str(self._sandbox),
+            virtual_mode=True,
+            timeout=60,
+        )
+        return None
+
+    def _cleanup(self) -> None:
+        self._backend = None
+        if self._sandbox_parent is not None and self._sandbox_parent.exists():
+            shutil.rmtree(self._sandbox_parent, ignore_errors=True)
+        self._sandbox_parent = None
+        self._sandbox = None
+
+    def _require_backend(self) -> Any:
+        if self._backend is None:
+            raise RuntimeError("DeepAgentSWEEnv tool called before reset()")
+        return self._backend
+
+    def ls(self, path: str = ".") -> str:
+        """List files in a directory inside the sandbox.
+
+        Args:
+            path: Directory path relative to the repo root.
+
+        Returns:
+            Directory listing as text.
+        """
+        return _result_to_str(self._require_backend().ls(path))
+
+    def glob(self, pattern: str, path: str = ".") -> str:
+        """Glob for files matching pattern under path.
+
+        Args:
+            pattern: Glob pattern (e.g. '**/*.py').
+            path: Root path for the search.
+
+        Returns:
+            Matching paths as text.
+        """
+        return _result_to_str(self._require_backend().glob(pattern, path))
+
+    def grep(self, pattern: str, path: str = ".") -> str:
+        """Search file contents for a regex pattern.
+
+        Args:
+            pattern: Regex pattern to search for.
+            path: Root path for the search.
+
+        Returns:
+            Matching lines as text.
+        """
+        return _result_to_str(self._require_backend().grep(pattern, path))
+
+    def read_file(self, file_path: str, offset: int = 0, limit: int = 200) -> str:
+        """Read a text file from the sandbox.
+
+        Args:
+            file_path: Path relative to the repo root.
+            offset: Starting line offset.
+            limit: Max number of lines to return.
+
+        Returns:
+            File contents as text.
+        """
+        return _result_to_str(
+            self._require_backend().read(file_path, offset=offset, limit=limit)
+        )
+
+    def write_file(self, file_path: str, content: str) -> str:
+        """Write or create a text file in the sandbox.
+
+        Args:
+            file_path: Path relative to the repo root.
+            content: Full file contents to write.
+
+        Returns:
+            Write result as text.
+        """
+        return _result_to_str(self._require_backend().write(file_path, content))
+
+    def edit_file(self, file_path: str, old_string: str, new_string: str) -> str:
+        """Replace old_string with new_string in an existing file.
+
+        Args:
+            file_path: Path relative to the repo root.
+            old_string: Exact text to find.
+            new_string: Replacement text.
+
+        Returns:
+            Edit result as text.
+        """
+        return _result_to_str(
+            self._require_backend().edit(file_path, old_string, new_string)
+        )
+
+    def execute(self, command: str) -> str:
+        """Run a shell command in the sandbox.
+
+        Args:
+            command: Shell command (e.g. python -c '...').
+
+        Returns:
+            Command stdout/stderr as text.
+        """
+        return _result_to_str(self._require_backend().execute(command))
+
+    def get_reward(self) -> float:
+        """Export git diff and grade via remote Docker SWE evaluator."""
+        if self._sandbox is None:
+            self._last_reward = 0.0
+            return 0.0
+
+        from deepagents_rl.harness.repo import git_diff_patch
+
+        try:
+            model_patch = git_diff_patch(self._sandbox)
+        except Exception as e:  # noqa: BLE001
+            self._last_report = {"error": f"git_diff_failed: {e}"}
+            self._last_reward = 0.0
+            return 0.0
+
+        if not model_patch.strip():
+            self._last_report = {"error": "empty_model_patch", "reward_total": 0.0}
+            self._last_reward = 0.0
+            return 0.0
+
+        try:
+            report = grade_patch_remote(
+                instance_id=self._task["instance_id"],
+                docker_image=self._task["docker_image"],
+                eval_script=self._task["eval_script"],
+                fail_to_pass=list(self._task["fail_to_pass"]),
+                pass_to_pass=list(self._task["pass_to_pass"]),
+                model_patch=model_patch,
+            )
+        except Exception as e:  # noqa: BLE001
+            self._last_report = {"error": f"grader_failed: {e}"}
+            self._last_reward = 0.0
+            return 0.0
+
+        self._last_report = report
+        self._last_reward = float(report.get("reward_total") or 0.0)
+        return self._last_reward
+
+
+def prepare_swe_grpo_rows(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Normalize HF SWE Lite rows for TRL GRPO + DeepAgentSWEEnv.
+
+    Renames `image` -> `docker_image` so TRL does not treat it as multimodal.
+    Converts `prompt` to conversational chat messages (required for tool calling).
+    """
+    from deepagents_rl.data.swe_bench_lite import from_hf_record
+
+    rows: list[dict[str, Any]] = []
+    for rec in records:
+        # Accept both HF records and already-normalized tasks
+        if "fail_to_pass_json" in rec or (
+            "prompt" in rec and isinstance(rec.get("prompt"), str) and "image" in rec
+        ):
+            task = from_hf_record(rec)
+        else:
+            task = dict(rec)
+
+        prompt = task.get("prompt") or ""
+        if isinstance(prompt, list):
+            messages = prompt
+        else:
+            messages = [{"role": "user", "content": str(prompt)}]
+
+        docker_image = str(task.get("docker_image") or task.get("image") or "")
+        rows.append(
+            {
+                "prompt": messages,
+                "task_id": task["task_id"],
+                "instance_id": task["instance_id"],
+                "repo": task["repo"],
+                "base_commit": task["base_commit"],
+                "docker_image": docker_image,
+                "eval_script": task.get("eval_script") or "",
+                "fail_to_pass": list(task.get("fail_to_pass") or []),
+                "pass_to_pass": list(task.get("pass_to_pass") or []),
+                "problem_statement": task.get("problem_statement") or "",
+            }
+        )
+    return rows