admin/OpenJarvis
1
0
Fork 0
mirror of https://github.com/open-jarvis/OpenJarvis.git synced 2026-07-31 03:12:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

hybrid: fix 7 correctness/reliability bugs surfaced in review

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Jon Saad-Falcon
2026-05-16 02:55:08 +00:00
co-authored by Claude Opus 4.7
parent 7e74758e1c
commit 897556a287
5 changed files with 233 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openjarvis/agents/hybrid/archon.py
+33 -16
View File
@@ -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
src/openjarvis/agents/hybrid/toolorchestra.py
+124 -117
View File
@@ -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"]
src/openjarvis/engine/_stubs.py
+6
View File
@@ -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)
src/openjarvis/engine/cloud.py
+43 -8
View File
@@ -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],
src/openjarvis/evals/scorers/swebench_harness.py
+27 -1
View File
@@ -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"