From d83ae02c6bfa76dbe52c64b784e227f31ad9ca09 Mon Sep 17 00:00:00 2001 From: Andrew Park Date: Fri, 15 May 2026 13:42:10 -0700 Subject: [PATCH] hybrid: wire Advisors + ToolOrchestra workers through mini-SWE-agent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same 'swe_use_agent_loop' opt-in flag as Conductor/SkillOrchestra. Advisors (swe mode): - Initial executor pass = full mini-SWE-agent run on a fresh workdir. - Advisor (local) critiques the produced patch (sees summary + diff). - Final executor pass = SECOND mini-SWE-agent run on ANOTHER fresh workdir, with the initial summary + advisor feedback folded into the prompt. Don't reuse the initial workdir — that would smuggle the bad-fix into the new attempt; the final pass should incorporate only the parts of the advisor feedback that hold up. ToolOrchestra (swe mode): - One SHARED workdir created at start of run, cloned from the SWE-bench repo at base_commit. - Each call_worker action whose worker is a solver (vllm/anthropic) runs as a mini-SWE-agent subloop on the shared workdir. Web-search workers stay one-shot (search isn't an agent loop). OpenAI workers fall back to one-shot too (loop tool format isn't wired for OpenAI). - After loop ends, the framed final answer wraps the working-tree git diff so the swebench scorer picks it up. --- src/openjarvis/agents/hybrid/advisors.py | 124 ++++++++++++++++++ src/openjarvis/agents/hybrid/toolorchestra.py | 116 +++++++++++++++- 2 files changed, 234 insertions(+), 6 deletions(-) diff --git a/src/openjarvis/agents/hybrid/advisors.py b/src/openjarvis/agents/hybrid/advisors.py index f3979685..d7fd3523 100644 --- a/src/openjarvis/agents/hybrid/advisors.py +++ b/src/openjarvis/agents/hybrid/advisors.py @@ -27,8 +27,17 @@ import json import urllib.request from typing import Any, Dict, Optional, Tuple +import shutil +import tempfile +from pathlib import Path + from openjarvis.agents._stubs import AgentContext from openjarvis.agents.hybrid._base import LocalCloudAgent +from openjarvis.agents.hybrid.mini_swe_agent import ( + _clone_repo, + _extract_diff, + run_swe_agent_loop, +) from openjarvis.core.registry import AgentRegistry @@ -96,6 +105,16 @@ class AdvisorsAgent(LocalCloudAgent): ) -> Tuple[str, Dict[str, Any]]: question = input cfg = self._cfg + task_meta = (context.metadata.get("task") if context is not None else {}) or {} + swe_mode = ( + bool(cfg.get("swe_use_agent_loop")) + and bool(task_meta.get("problem_statement")) + and bool(task_meta.get("repo")) + and bool(task_meta.get("base_commit")) + ) + if swe_mode: + return self._run_swe(question, task_meta, cfg) + executor_max_tokens = int(cfg.get("executor_max_tokens", 4096)) advisor_max_tokens = int(cfg.get("advisor_max_tokens", 1024)) advisor_temperature = float(cfg.get("advisor_temperature", 0.2)) @@ -159,5 +178,110 @@ class AdvisorsAgent(LocalCloudAgent): } return final_answer, meta + # ------------------------------------------------------------------ + # SWE-bench variant: each executor pass is a full mini-SWE-agent run. + # ------------------------------------------------------------------ + + def _run_swe( + self, + question: str, + task: Dict[str, Any], + cfg: Dict[str, Any], + ) -> Tuple[str, Dict[str, Any]]: + if not self._local_endpoint or not self._local_model: + raise ValueError( + "AdvisorsAgent (swe mode) still needs local_model + local_endpoint " + "for the advisor critique step." + ) + + # Each executor pass is its own mini-SWE-agent run with a FRESH + # workdir — the advisor doesn't get a workdir, just the patch text. + max_turns = int(cfg.get("swe_max_turns", 30)) + bash_timeout = int(cfg.get("swe_bash_timeout_s", 120)) + output_cap = int(cfg.get("swe_output_cap", 10_000)) + turn_max_tokens = int(cfg.get("swe_turn_max_tokens", 4096)) + + # 1. Initial executor pass + initial_out = run_swe_agent_loop( + task, + backbone="cloud", + backbone_model=self._cloud_model, + cloud_endpoint=self._cloud_endpoint, + initial_prompt=question, + max_turns=max_turns, + bash_timeout=bash_timeout, + output_cap=output_cap, + turn_max_tokens=turn_max_tokens, + trace_prefix="advisors_executor1", + ) + + # 2. Advisor pass — local model critiques the produced patch + local_model = _resolve_local_model(self._local_endpoint, self._local_model) + advisor_prompt = ADVISOR_TEMPLATE.format( + question=question, + initial_response=( + f"Summary: {initial_out['final_summary']}\n\n" + f"Patch produced:\n```diff\n{initial_out['patch']}```" + ), + ) + advisor_text, adv_in, adv_out = self._call_vllm( + local_model, + self._local_endpoint, + user=advisor_prompt, + max_tokens=int(cfg.get("advisor_max_tokens", 2048)), + temperature=float(cfg.get("advisor_temperature", 0.2)), + enable_thinking=False, + ) + + # 3. Final executor pass — FRESH workdir, advisor feedback folded + # into the initial prompt. (Don't reuse initial workdir — that + # would smuggle the bad-fix into the new attempt; we want the + # final pass to incorporate ONLY the parts of the advisor's + # feedback that hold up.) + final_prompt = ( + f"{question}\n\n" + f"-----\n" + f"You previously attempted a fix. Your initial summary was:\n" + f"{initial_out['final_summary']}\n\n" + f"An advisor reviewed your attempt and wrote this feedback:\n" + f"{advisor_text}\n\n" + f"Incorporate the advisor's feedback where it improves correctness; " + f"ignore where it is wrong. Produce your best fix now." + ) + final_out = run_swe_agent_loop( + task, + backbone="cloud", + backbone_model=self._cloud_model, + cloud_endpoint=self._cloud_endpoint, + initial_prompt=final_prompt, + max_turns=max_turns, + bash_timeout=bash_timeout, + output_cap=output_cap, + turn_max_tokens=turn_max_tokens, + trace_prefix="advisors_executor2", + ) + + tokens_local = adv_in + adv_out + tokens_cloud = ( + initial_out["tokens_in"] + initial_out["tokens_out"] + + final_out["tokens_in"] + final_out["tokens_out"] + ) + cost = initial_out["cost_usd"] + final_out["cost_usd"] + meta: Dict[str, Any] = { + "tokens_local": tokens_local, + "tokens_cloud": tokens_cloud, + "cost_usd": cost, + "turns": initial_out["turns"] + 1 + final_out["turns"], + "traces": { + "swe_mode": True, + "initial_summary": initial_out["final_summary"], + "initial_patch_chars": len(initial_out["patch"]), + "advisor_feedback": advisor_text, + "final_summary": final_out["final_summary"], + "final_patch_chars": len(final_out["patch"]), + }, + } + return final_out["answer"], meta + __all__ = ["AdvisorsAgent"] diff --git a/src/openjarvis/agents/hybrid/toolorchestra.py b/src/openjarvis/agents/hybrid/toolorchestra.py index 20feb9ec..d3a3c94c 100644 --- a/src/openjarvis/agents/hybrid/toolorchestra.py +++ b/src/openjarvis/agents/hybrid/toolorchestra.py @@ -43,6 +43,10 @@ import json import re from typing import Any, Dict, List, Optional, Tuple +import shutil +import tempfile +from pathlib import Path + from openjarvis.agents._stubs import AgentContext from openjarvis.agents.hybrid._base import ( ANTHROPIC_WEB_SEARCH_TOOL, @@ -53,6 +57,11 @@ from openjarvis.agents.hybrid._prices import ( is_gpt5_family, supports_temperature, ) +from openjarvis.agents.hybrid.mini_swe_agent import ( + _clone_repo, + _extract_diff, + run_swe_agent_loop, +) from openjarvis.core.registry import AgentRegistry @@ -263,6 +272,52 @@ def _call_worker( raise ValueError(f"unsupported worker type: {wtype!r}") +def _swe_call_worker( + worker: Dict[str, Any], + prompt: str, + cfg: Dict[str, Any], + task: Dict[str, Any], + workdir: Path, + turn: int, +) -> Tuple[str, int, int, bool, float, int]: + """SWE-bench worker dispatch: route solver workers through + run_swe_agent_loop on a shared workdir. Web-search workers fall back + to the regular one-shot dispatch (search isn't an agent loop).""" + wtype = worker.get("type", "openai") + if wtype == "anthropic-web-search": + # Search workers stay one-shot. + return _call_worker(worker, prompt, cfg) + if wtype == "vllm": + backbone = "local" + endpoint = worker.get("base_url") + elif wtype == "anthropic": + backbone = "cloud" + endpoint = None + else: + # OpenAI workers fall back to one-shot. + return _call_worker(worker, prompt, cfg) + out = run_swe_agent_loop( + task, + backbone=backbone, + backbone_model=worker["model"], + cloud_endpoint="anthropic", + local_endpoint=endpoint, + initial_prompt=prompt, + max_turns=int(cfg.get("swe_max_turns", 30)), + bash_timeout=int(cfg.get("swe_bash_timeout_s", 120)), + output_cap=int(cfg.get("swe_output_cap", 10_000)), + turn_max_tokens=int(cfg.get("swe_turn_max_tokens", 4096)), + trace_prefix=f"toolorch_turn{turn}", + workdir=workdir, + ) + is_local = backbone == "local" + return ( + out["final_summary"] or out["answer"], + out["tokens_in"], out["tokens_out"], + is_local, 0.0, 0, + ) + + @AgentRegistry.register("toolorchestra") class ToolOrchestraAgent(LocalCloudAgent): """Prompted multi-turn dispatcher over a mixed worker pool. @@ -290,6 +345,30 @@ class ToolOrchestraAgent(LocalCloudAgent): max_turns = int(cfg.get("max_turns", 6)) orch_max_tokens = int(cfg.get("orchestrator_max_tokens", 1024)) + task_meta = (context.metadata.get("task") if context is not None else {}) or {} + swe_mode = ( + bool(cfg.get("swe_use_agent_loop")) + and bool(task_meta.get("problem_statement")) + and bool(task_meta.get("repo")) + and bool(task_meta.get("base_commit")) + ) + shared_workdir: Optional[Path] = None + if swe_mode: + shared_workdir = Path(tempfile.mkdtemp( + prefix=f"toolorch-swe-{task_meta.get('task_id','x')}-" + )) + try: + _clone_repo(task_meta["repo"], task_meta["base_commit"], shared_workdir) + except Exception: + shutil.rmtree(shared_workdir, ignore_errors=True) + raise + self.record_trace_event({ + "kind": "toolorchestra_swe_workdir", + "workdir": str(shared_workdir), + "repo": task_meta["repo"], + "base_commit": task_meta["base_commit"], + }) + history: List[Dict[str, Any]] = [] tokens_local = 0 tokens_cloud = 0 @@ -346,9 +425,17 @@ class ToolOrchestraAgent(LocalCloudAgent): break continue worker = workers[wid] - w_text, w_in, w_out, is_local, extra_cost, n_searches = _call_worker( - worker, str(w_input), cfg - ) + if swe_mode and shared_workdir is not None: + w_text, w_in, w_out, is_local, extra_cost, n_searches = ( + _swe_call_worker( + worker, str(w_input), cfg, task_meta, + shared_workdir, turn, + ) + ) + else: + w_text, w_in, w_out, is_local, extra_cost, n_searches = ( + _call_worker(worker, str(w_input), cfg) + ) if is_local: tokens_local += w_in + w_out else: @@ -372,9 +459,15 @@ class ToolOrchestraAgent(LocalCloudAgent): if final_answer is None: # Hard fallback: call the strongest worker (last) directly. worker = workers[-1] - ans, w_in, w_out, is_local, extra_cost, _ = _call_worker( - worker, question, cfg - ) + if swe_mode and shared_workdir is not None: + ans, w_in, w_out, is_local, extra_cost, _ = _swe_call_worker( + worker, question, cfg, task_meta, + shared_workdir, max_turns + 1, + ) + else: + ans, w_in, w_out, is_local, extra_cost, _ = _call_worker( + worker, question, cfg + ) if is_local: tokens_local += w_in + w_out else: @@ -393,6 +486,17 @@ class ToolOrchestraAgent(LocalCloudAgent): }) final_answer = ans + # In SWE mode, the authoritative output is the working-tree diff — + # frame it (the runner extracts it via the scorer's ```diff fence). + if swe_mode and shared_workdir is not None: + patch = _extract_diff(shared_workdir) + if patch.strip(): + final_answer = ( + f"{final_answer}\n\n```diff\n{patch}```" + if final_answer else f"```diff\n{patch}```" + ) + shutil.rmtree(shared_workdir, ignore_errors=True) + meta = { "tokens_local": tokens_local, "tokens_cloud": tokens_cloud,