From 897556a287c6a9c92186b336d774a92d115a2953 Mon Sep 17 00:00:00 2001 From: Jon Saad-Falcon Date: Sat, 16 May 2026 02:55:08 +0000 Subject: [PATCH] hybrid: fix 7 correctness/reliability bugs surfaced in review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cloud.py: - Skip server_tool_use blocks from tool_calls (they're already in content_blocks; agents would otherwise try to execute Anthropic server-side tools like web_search as if they were local). - Drop the synthetic tool_use_id fallback (f"{btype}_{len(...)}"). Anthropic always returns ids for tool_use blocks; the fallback only masked missing ids with fakes that fail Anthropic's tool_use_id matching on the next assistant turn. Now we log a warning and skip the block. - Streaming/non-streaming parity: emit content_blocks and tool_results at end-of-stream in _stream_full_anthropic so streaming callers see the same rich blocks as generate(). StreamChunk gets two new optional fields. swebench_harness.py: - Log a warning when set_cpu_quota is missing instead of silently no-op'ing — previously a swebench API change would have produced a silent 0-score sweep with no diagnostic. - Delete the prior report JSON before invoking the subprocess so a stale file from a crashed run can't be re-read as fresh. archon.py: - Token tally is now thread-local (threading.local). The runner reuses one ArchonAgent across a ThreadPoolExecutor; the previous module-global dict let concurrent tasks reset and read each other's counters, corrupting per-task cost_usd / tokens_*. toolorchestra.py: - Wrap the orchestration loop in try/finally so shared_workdir is always cleaned up. Previously rmtree only ran on the success path; any exception in the turn loop, worker call, or diff extraction leaked the cloned repo (hundreds of MB at n=500). Matches the pattern conductor.py and mini_swe_agent.py use. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/openjarvis/agents/hybrid/archon.py | 49 ++-- src/openjarvis/agents/hybrid/toolorchestra.py | 241 +++++++++--------- src/openjarvis/engine/_stubs.py | 6 + src/openjarvis/engine/cloud.py | 51 +++- .../evals/scorers/swebench_harness.py | 28 +- 5 files changed, 233 insertions(+), 142 deletions(-) diff --git a/src/openjarvis/agents/hybrid/archon.py b/src/openjarvis/agents/hybrid/archon.py index 51b5da8d..a5d672ec 100644 --- a/src/openjarvis/agents/hybrid/archon.py +++ b/src/openjarvis/agents/hybrid/archon.py @@ -37,6 +37,7 @@ from __future__ import annotations import os import sys +import threading import types from typing import Any, Dict, Optional, Tuple @@ -110,15 +111,31 @@ def _patch_anthropic_for_opus() -> None: # ---------- Per-run token tally + custom generators ---------- -_TOKEN_TALLY = { - "cloud_prompt": 0, "cloud_completion": 0, - "local_prompt": 0, "local_completion": 0, -} +# Token tallies live in thread-local storage because the runner reuses one +# ArchonAgent across a ``ThreadPoolExecutor`` (``concurrency > 1``). A plain +# module-global dict would let concurrent ``_run_paradigm`` calls reset and +# read each other's counters, producing wrong ``cost_usd`` and +# ``tokens_local`` / ``tokens_cloud`` in per-task ``meta``. +_TALLY_LOCAL = threading.local() + + +def _tally() -> Dict[str, int]: + """Return the calling thread's token tally, creating it on first touch.""" + counts = getattr(_TALLY_LOCAL, "counts", None) + if counts is None: + counts = { + "cloud_prompt": 0, "cloud_completion": 0, + "local_prompt": 0, "local_completion": 0, + } + _TALLY_LOCAL.counts = counts + return counts def _reset_tally() -> None: - for k in _TOKEN_TALLY: - _TOKEN_TALLY[k] = 0 + _TALLY_LOCAL.counts = { + "cloud_prompt": 0, "cloud_completion": 0, + "local_prompt": 0, "local_completion": 0, + } def _make_local_generator(local_endpoint: str, local_model: str): @@ -148,8 +165,8 @@ def _make_local_generator(local_endpoint: str, local_model: str): return f"[local-vllm error: {e!r}]" u = resp.usage if u: - _TOKEN_TALLY["local_prompt"] += getattr(u, "prompt_tokens", 0) or 0 - _TOKEN_TALLY["local_completion"] += getattr(u, "completion_tokens", 0) or 0 + _tally()["local_prompt"] += getattr(u, "prompt_tokens", 0) or 0 + _tally()["local_completion"] += getattr(u, "completion_tokens", 0) or 0 text = (resp.choices[0].message.content or "").strip() _record_event({ "kind": "archon_local_gen", @@ -191,8 +208,8 @@ def _wrap_archon_cloud_generators() -> None: resp = client.chat.completions.create(**kwargs) u = resp.usage if u: - _TOKEN_TALLY["cloud_prompt"] += getattr(u, "prompt_tokens", 0) or 0 - _TOKEN_TALLY["cloud_completion"] += getattr(u, "completion_tokens", 0) or 0 + _tally()["cloud_prompt"] += getattr(u, "prompt_tokens", 0) or 0 + _tally()["cloud_completion"] += getattr(u, "completion_tokens", 0) or 0 text = (resp.choices[0].message.content or "").strip() _record_event({ "kind": "archon_cloud_openai", @@ -226,8 +243,8 @@ def _wrap_archon_cloud_generators() -> None: text = "".join(b.text for b in resp.content if hasattr(b, "text")) u = resp.usage if u: - _TOKEN_TALLY["cloud_prompt"] += getattr(u, "input_tokens", 0) or 0 - _TOKEN_TALLY["cloud_completion"] += getattr(u, "output_tokens", 0) or 0 + _tally()["cloud_prompt"] += getattr(u, "input_tokens", 0) or 0 + _tally()["cloud_completion"] += getattr(u, "output_tokens", 0) or 0 _record_event({ "kind": "archon_cloud_anthropic", "model": model, @@ -420,15 +437,15 @@ class ArchonAgent(LocalCloudAgent): answer = answer[-1] if answer else "" answer = str(answer) - cp = _TOKEN_TALLY["cloud_prompt"] - cc = _TOKEN_TALLY["cloud_completion"] + cp = _tally()["cloud_prompt"] + cc = _tally()["cloud_completion"] cost = _cost_cloud(ranker_model, cp, cc) if fuser_model != ranker_model: # Conservative: charge both at the more expensive of the two. cost = max(cost, _cost_cloud(fuser_model, cp, cc)) meta = { - "tokens_local": _TOKEN_TALLY["local_prompt"] + _TOKEN_TALLY["local_completion"], + "tokens_local": _tally()["local_prompt"] + _tally()["local_completion"], "tokens_cloud": cp + cc, "cost_usd": cost, "turns": (K + 2) if arch == "ensemble_rank_fuse" else 1, @@ -438,7 +455,7 @@ class ArchonAgent(LocalCloudAgent): "ranker_model": ranker_model, "fuser_model": fuser_model, "local_model": self._local_model, - "tokens_breakdown": dict(_TOKEN_TALLY), + "tokens_breakdown": dict(_tally()), }, } return answer, meta diff --git a/src/openjarvis/agents/hybrid/toolorchestra.py b/src/openjarvis/agents/hybrid/toolorchestra.py index d3a3c94c..97c1d584 100644 --- a/src/openjarvis/agents/hybrid/toolorchestra.py +++ b/src/openjarvis/agents/hybrid/toolorchestra.py @@ -369,72 +369,109 @@ class ToolOrchestraAgent(LocalCloudAgent): "base_commit": task_meta["base_commit"], }) - history: List[Dict[str, Any]] = [] - tokens_local = 0 - tokens_cloud = 0 - cost = 0.0 - final_answer: Optional[str] = None - forced_final = False - parse_failures = 0 + # try/finally guards ``shared_workdir`` against exceptions raised + # anywhere in the turn loop, the worker calls, the fallback, or + # the diff-extraction step. Without this, at n=500 SWE-bench an + # exception leaves hundreds of MB of cloned repos in tempdir. + try: + history: List[Dict[str, Any]] = [] + tokens_local = 0 + tokens_cloud = 0 + cost = 0.0 + final_answer: Optional[str] = None + forced_final = False + parse_failures = 0 - for turn in range(1, max_turns + 1): - sys_prompt = ORCHESTRATOR_SYS - if turn == max_turns and final_answer is None: - sys_prompt = ORCHESTRATOR_SYS + "\n\n" + FORCE_FINAL_PROMPT - forced_final = True + for turn in range(1, max_turns + 1): + sys_prompt = ORCHESTRATOR_SYS + if turn == max_turns and final_answer is None: + sys_prompt = ORCHESTRATOR_SYS + "\n\n" + FORCE_FINAL_PROMPT + forced_final = True - user = _build_user_prompt(question, workers, history) - text, o_in, o_out = self._call_cloud( - user=user, - system=sys_prompt, - max_tokens=orch_max_tokens, - temperature=0.0, - ) - tokens_cloud += o_in + o_out - cost += self.cost_usd(self._cloud_model, o_in, o_out) + user = _build_user_prompt(question, workers, history) + text, o_in, o_out = self._call_cloud( + user=user, + system=sys_prompt, + max_tokens=orch_max_tokens, + temperature=0.0, + ) + tokens_cloud += o_in + o_out + cost += self.cost_usd(self._cloud_model, o_in, o_out) - action = _parse_action(text) - history.append({ - "role": "orchestrator", "turn": turn, "raw": text, "action": action, - }) - self.record_trace_event({ - "kind": "toolorchestra_action", - "turn": turn, - "action": action, - "raw": text, - }) + action = _parse_action(text) + history.append({ + "role": "orchestrator", "turn": turn, "raw": text, "action": action, + }) + self.record_trace_event({ + "kind": "toolorchestra_action", + "turn": turn, + "action": action, + "raw": text, + }) - if action is None: - parse_failures += 1 - if parse_failures >= 2 or forced_final: - final_answer = _extract_final_answer_text(text) - break - continue - - kind = action.get("action") - if kind == "final_answer": - final_answer = str(action.get("answer", "")).strip() - break - if kind == "call_worker": - wid = action.get("worker_id") - w_input = action.get("input", "") - if not isinstance(wid, int) or not (0 <= wid < len(workers)): + if action is None: parse_failures += 1 if parse_failures >= 2 or forced_final: final_answer = _extract_final_answer_text(text) break continue - worker = workers[wid] - 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, + + kind = action.get("action") + if kind == "final_answer": + final_answer = str(action.get("answer", "")).strip() + break + if kind == "call_worker": + wid = action.get("worker_id") + w_input = action.get("input", "") + if not isinstance(wid, int) or not (0 <= wid < len(workers)): + parse_failures += 1 + if parse_failures >= 2 or forced_final: + final_answer = _extract_final_answer_text(text) + break + continue + worker = workers[wid] + 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: + tokens_cloud += w_in + w_out + cost += self.cost_usd(worker["model"], w_in, w_out) + extra_cost + history.append({ + "role": "worker", + "turn": turn, + "worker_id": wid, + "worker_name": worker["name"], + "worker_model": worker["model"], + "output": w_text, + "tokens_in": w_in, + "tokens_out": w_out, + "n_web_searches": n_searches, + }) + continue + # Unknown action kind — treat as parse failure. + parse_failures += 1 + + if final_answer is None: + # Hard fallback: call the strongest worker (last) directly. + worker = workers[-1] + 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: - w_text, w_in, w_out, is_local, extra_cost, n_searches = ( - _call_worker(worker, str(w_input), cfg) + ans, w_in, w_out, is_local, extra_cost, _ = _call_worker( + worker, question, cfg ) if is_local: tokens_local += w_in + w_out @@ -443,77 +480,47 @@ class ToolOrchestraAgent(LocalCloudAgent): cost += self.cost_usd(worker["model"], w_in, w_out) + extra_cost history.append({ "role": "worker", - "turn": turn, - "worker_id": wid, + "turn": max_turns + 1, + "worker_id": worker["id"], "worker_name": worker["name"], "worker_model": worker["model"], - "output": w_text, + "output": ans, "tokens_in": w_in, "tokens_out": w_out, - "n_web_searches": n_searches, + "fallback": True, }) - continue - # Unknown action kind — treat as parse failure. - parse_failures += 1 + final_answer = ans - if final_answer is None: - # Hard fallback: call the strongest worker (last) directly. - worker = workers[-1] + # 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: - 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: - tokens_cloud += w_in + w_out - cost += self.cost_usd(worker["model"], w_in, w_out) + extra_cost - history.append({ - "role": "worker", - "turn": max_turns + 1, - "worker_id": worker["id"], - "worker_name": worker["name"], - "worker_model": worker["model"], - "output": ans, - "tokens_in": w_in, - "tokens_out": w_out, - "fallback": True, - }) - final_answer = ans + 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}```" + ) - # 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, - "cost_usd": cost, - "turns": len([h for h in history if h["role"] == "orchestrator"]), - "traces": { - "history": history, - "forced_final": forced_final, - "parse_failures": parse_failures, - "workers": workers, - "note": ( - "inference-only port; the RL-trained Nemotron-Orchestrator-8B " - "is NOT in the loop. Results are preliminary." - ), - }, - } - return final_answer, meta + meta = { + "tokens_local": tokens_local, + "tokens_cloud": tokens_cloud, + "cost_usd": cost, + "turns": len([h for h in history if h["role"] == "orchestrator"]), + "traces": { + "history": history, + "forced_final": forced_final, + "parse_failures": parse_failures, + "workers": workers, + "note": ( + "inference-only port; the RL-trained Nemotron-Orchestrator-8B " + "is NOT in the loop. Results are preliminary." + ), + }, + } + return final_answer, meta + finally: + if shared_workdir is not None: + shutil.rmtree(shared_workdir, ignore_errors=True) __all__ = ["ToolOrchestraAgent"] diff --git a/src/openjarvis/engine/_stubs.py b/src/openjarvis/engine/_stubs.py index 4a940274..c0629502 100644 --- a/src/openjarvis/engine/_stubs.py +++ b/src/openjarvis/engine/_stubs.py @@ -21,12 +21,18 @@ class StreamChunk: Used by ``stream_full()`` to yield rich streaming data including tool_calls fragments and finish_reason, unlike ``stream()`` which only yields plain content strings. + + ``content_blocks`` and ``tool_results`` are aggregate fields emitted + once at end-of-stream so streaming callers reach parity with the + non-streaming ``generate()`` return shape. """ content: Optional[str] = None tool_calls: Optional[List[Dict[str, Any]]] = None finish_reason: Optional[str] = None usage: Optional[Dict[str, Any]] = None + content_blocks: Optional[List[Dict[str, Any]]] = None + tool_results: Optional[List[Dict[str, Any]]] = None @dataclass(slots=True) diff --git a/src/openjarvis/engine/cloud.py b/src/openjarvis/engine/cloud.py index 01f73bfd..8a32fc84 100644 --- a/src/openjarvis/engine/cloud.py +++ b/src/openjarvis/engine/cloud.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import logging import os import time from collections.abc import AsyncIterator, Sequence @@ -19,6 +20,8 @@ from openjarvis.engine._base import ( ) from openjarvis.engine._stubs import StreamChunk +logger = logging.getLogger(__name__) + # Pricing per million tokens (input, output) PRICING: Dict[str, tuple[float, float]] = { "gpt-4o": (2.50, 10.00), @@ -613,10 +616,11 @@ class CloudEngine(InferenceEngine): # - web_search_tool_result (server-tool result body) # - tool_result (caller-side tool result echo) # - thinking (Opus reasoning trace) - # We keep everything: a full ``content_blocks`` list for the trace - # collector, plus convenience fields ``tool_calls`` (tool_use + - # server_tool_use) and ``tool_results`` (web_search_tool_result + - # tool_result) for downstream consumers. + # ``content_blocks`` keeps the full serialized list for trace + # observability. ``tool_calls`` is the narrow caller-executable + # surface — only ``tool_use`` blocks (server_tool_use lives in + # content_blocks since Anthropic already ran it server-side). + # ``tool_results`` aggregates both result kinds. content_parts: list[str] = [] tool_calls: list[Dict[str, Any]] = [] tool_results: list[Dict[str, Any]] = [] @@ -625,16 +629,22 @@ class CloudEngine(InferenceEngine): btype = getattr(block, "type", None) or type(block).__name__ serialized = _serialize_anthropic_block(block) content_blocks.append(serialized) - if btype in ("tool_use", "server_tool_use"): + if btype == "tool_use": + block_id = getattr(block, "id", None) + if not block_id: + logger.warning( + "Anthropic tool_use block without an id; skipping. " + "Round-trip into the next assistant turn would fail " + "Anthropic's tool_use_id matching." + ) + continue tool_calls.append( { - "id": getattr(block, "id", None) - or f"{btype}_{len(tool_calls)}", + "id": block_id, "name": getattr(block, "name", ""), "arguments": json.dumps(getattr(block, "input", None)) if isinstance(getattr(block, "input", None), dict) else str(getattr(block, "input", "")), - "server_side": btype == "server_tool_use", } ) elif btype in ("web_search_tool_result", "tool_result"): @@ -1336,6 +1346,31 @@ class CloudEngine(InferenceEngine): finish = "tool_calls" if stop_reason == "tool_use" else "stop" yield StreamChunk(finish_reason=finish) + # End-of-stream parity with ``_generate_anthropic``: emit + # ``content_blocks`` (every block kind, including server_tool_use + # and thinking) and ``tool_results`` (web_search_tool_result + + # tool_result) so streaming traces see what non-streaming traces + # see. + try: + final_msg = stream.get_final_message() + except Exception as exc: # noqa: BLE001 — SDK shape varies + logger.debug("Anthropic stream.get_final_message() failed: %s", exc) + final_msg = None + if final_msg is not None and getattr(final_msg, "content", None): + content_blocks: list[Dict[str, Any]] = [] + tool_results: list[Dict[str, Any]] = [] + for block in final_msg.content: + btype = getattr(block, "type", None) or type(block).__name__ + serialized = _serialize_anthropic_block(block) + content_blocks.append(serialized) + if btype in ("web_search_tool_result", "tool_result"): + tool_results.append(serialized) + if content_blocks or tool_results: + yield StreamChunk( + content_blocks=content_blocks or None, + tool_results=tool_results or None, + ) + async def stream_full( self, messages: Sequence[Message], diff --git a/src/openjarvis/evals/scorers/swebench_harness.py b/src/openjarvis/evals/scorers/swebench_harness.py index 1cd17ca2..1f06ee94 100644 --- a/src/openjarvis/evals/scorers/swebench_harness.py +++ b/src/openjarvis/evals/scorers/swebench_harness.py @@ -34,6 +34,7 @@ called for the first time). from __future__ import annotations import json +import logging import os import re import subprocess @@ -45,6 +46,8 @@ from typing import Any, Dict, Optional, Tuple from openjarvis.evals.core.scorer import Scorer from openjarvis.evals.core.types import EvalRecord +logger = logging.getLogger(__name__) + # ---------- Patch tracking ---------- @@ -67,7 +70,18 @@ def _patch_modal_cgroup_v2() -> None: return orig = getattr(_m, "set_cpu_quota", None) if orig is None: - # API changed — nothing to patch, but mark so we don't retry. + # Upstream API changed: ``set_cpu_quota`` no longer exists. The + # cgroup-v1 write that used to live inside it is either gone (good) + # or now lives somewhere we can't patch (bad — every Modal-v2 + # sandbox will still die). Surface loudly so failures aren't + # silently scored as 0 via the ``no_report`` fallback in + # :func:`_run_harness`. + logger.warning( + "swebench.harness.modal_eval.run_evaluation_modal.set_cpu_quota " + "is missing — cgroup-v2 patch could not be applied. If your " + "Modal sandboxes are scoring 0 with `reason: no_report`, " + "verify upstream swebench's Modal cgroup handling." + ) _m._hybrid_cgroup_patched = True # type: ignore[attr-defined] return @@ -158,6 +172,18 @@ def _run_harness(instance_id: str, patch: str, timeout_s: int) -> Dict[str, Any] cache = _harness_cache_dir() run_id = f"oj-{instance_id}" + # Defend against stale reports: ``run_id`` is deterministic per + # instance, the cache dir is shared across runs, and ``_find_report`` + # globs by filename. If a prior subprocess crashed mid-run (or was + # killed by timeout) and left a stale JSON, we'd silently read that + # old verdict as the current result. Delete it up front. + stale = cache / f"openjarvis-harness.{run_id}.json" + if stale.exists(): + try: + stale.unlink() + except OSError as exc: + logger.warning("Could not remove stale report %s: %s", stale, exc) + with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) preds_path = tmp_path / "predictions.jsonl"