hybrid: port MinionsAgent (HazyResearch Minions protocol)

Cloud supervisor decomposes + reads back; local worker(s) do bulk
reading. Two modes: `minion` (single worker, default) / `minions`
(parallel workers + aggregator).

Carries forward every compatibility patch from the hybrid adapter so the
n=500 numbers transfer:
- Strip `temperature` for Opus 4.7+ (rejected with 400).
- Inject server-side `output_config` JSON schema per supervisor turn
  (first-turn `{reasoning, message}` vs conversation-turn anyOf
  `{decision, message|answer}`), picked by sniffing the prompt for
  `"decision": "provide_final_answer"`.
- Replace `minions._extract_json` with a JSON-first wrapper.
- Inject `timeout=600`/`max_retries=5` into `anthropic.Anthropic()` so
  Minions's bare-client construction doesn't 60s-timeout under SWE-bench
  concurrency=8.
- GAIA-only `web_search` prefetch so the worker has a real doc to read.

Soft-fail row on JSONDecodeError/BadRequestError/prompt-too-long — same
deterministic-failure handling that produced the n=165 GAIA `err=6` rows
in the hybrid harness without crashing the cell.

`minions` library import is lazy; install via
`uv pip install -e /matx/u/aspark/hybrid-local-cloud-compute/external/minions`.

Hybrid harness results being reproduced:
- `minions-swebenchverified-qwen27b-opus-500` = 0.274 acc, $0.09/task
  (+3.8pp vs baseline-cloud at 10× cheaper — cleanest paradigm-validation
  result in the harness).
- `minions-gaia-qwen27b-opus-165` = 0.576 acc, $0.67/task
  (~tied with baseline-cloud at 1.6× cheaper).

Registered as `minions`. Ported from
`hybrid-local-cloud-compute/adapters/minions_adapter.py`.

src/openjarvis/agents/hybrid/minions.py

@@ -0,0 +1,430 @@
+"""MinionsAgent — port of HazyResearch Minions protocol.
+
+Cloud supervisor decomposes the task and reads back local-worker output;
+local worker(s) do the bulk reading/extraction. Multi-turn loop until the
+supervisor commits to a final answer.
+
+Two modes (``cfg["mode"]``):
+
+- ``"minion"`` — single local worker, one cloud supervisor (cheaper).
+  Default.
+- ``"minions"`` — parallel local workers, cloud aggregator.
+
+Hybrid harness result: ``minions-swebenchverified-qwen27b-opus-500`` =
+0.274 acc / $0.09 per task — beats baseline-cloud's 0.236 / $0.95 on
+**both** accuracy and cost. GAIA at n=165 ties baseline-cloud at 0.576
+acc / $0.67 (vs $1.09).
+
+Requires the ``minions`` library from
+https://github.com/HazyResearch/minions installed in the same env (e.g.
+``uv pip install -e /matx/u/aspark/hybrid-local-cloud-compute/external/minions``).
+Import is lazy — the agent class registers without ``minions`` available,
+and the import error only fires on ``run()``.
+
+Compatibility patches applied at first ``run()`` (idempotent):
+
+- Strip ``temperature`` for Opus 4.7+ (rejected with 400).
+- Inject server-side ``output_config`` JSON schema on supervisor turns
+  so Opus replies in the shape Minions's parser expects (per-turn schema
+  picked by sniffing the prompt for ``"decision": "provide_final_answer"``).
+- Replace Minions's ``_extract_json`` with a wrapper that short-circuits
+  when the response is already valid JSON.
+- Inject ``timeout=600``/``max_retries=5`` defaults into
+  ``anthropic.Anthropic()`` — Minions builds bare clients which 60s-timeout
+  under SWE-bench concurrency=8.
+
+Ported from ``hybrid-local-cloud-compute/adapters/minions_adapter.py``.
+"""
+
+from __future__ import annotations
+
+import json as _json
+import sys
+import types
+from typing import Any, Dict, List, Optional, Tuple
+
+from openjarvis.agents._stubs import AgentContext
+from openjarvis.agents.hybrid._base import (
+    ANTHROPIC_WEB_SEARCH_TOOL,
+    LocalCloudAgent,
+    WEB_SEARCH_COST_PER_CALL,
+)
+from openjarvis.agents.hybrid._prices import NO_TEMP_PREFIXES, supports_temperature
+from openjarvis.core.registry import AgentRegistry
+
+
+# ---------- Per-turn JSON schemas (server-side enforcement) ----------
+#
+# Minions's supervisor produces different JSON shapes per turn:
+#   turn 1 (decompose):     {reasoning, message}
+#   turn 2+ (continue):     {decision="request_additional_info", message}
+#   turn 2+ (final answer): {decision="provide_final_answer", answer}
+#
+# Anthropic strict mode requires additionalProperties:false + all-props
+# required, so we pick the schema PER TURN by sniffing the prompt for
+# Minions's turn-2 template marker.
+
+MINIONS_FIRST_TURN_SCHEMA = {
+    "format": {
+        "type": "json_schema",
+        "schema": {
+            "type": "object",
+            "properties": {
+                "reasoning": {"type": "string"},
+                "message":   {"type": "string"},
+            },
+            "required": ["reasoning", "message"],
+            "additionalProperties": False,
+        },
+    }
+}
+
+MINIONS_CONVERSATION_SCHEMA = {
+    "format": {
+        "type": "json_schema",
+        "schema": {
+            "anyOf": [
+                {
+                    "type": "object",
+                    "properties": {
+                        "decision": {"const": "request_additional_info"},
+                        "message":  {"type": "string"},
+                    },
+                    "required": ["decision", "message"],
+                    "additionalProperties": False,
+                },
+                {
+                    "type": "object",
+                    "properties": {
+                        "decision": {"const": "provide_final_answer"},
+                        "answer":   {"type": "string"},
+                    },
+                    "required": ["decision", "answer"],
+                    "additionalProperties": False,
+                },
+            ],
+        },
+    }
+}
+
+# Markers from Minions's supervisor prompts (prompts/minion.py). Any one
+# being present in the call's messages/system is a strong Minions signal.
+MINIONS_PROMPT_MARKERS = (
+    "small language model that has read",   # SUPERVISOR_INITIAL_PROMPT
+    "provide_final_answer",                  # SUPERVISOR_CONVERSATION_PROMPT
+    "request_additional_info",
+)
+
+
+def _looks_like_minions_call(kwargs: Dict[str, Any]) -> bool:
+    blob_parts = [str(kwargs.get("system", ""))]
+    for msg in kwargs.get("messages", []) or []:
+        content = msg.get("content", "") if isinstance(msg, dict) else ""
+        blob_parts.append(str(content))
+    blob = "\n".join(blob_parts)
+    return any(m in blob for m in MINIONS_PROMPT_MARKERS)
+
+
+def _minions_turn_schema(kwargs: Dict[str, Any]) -> Dict[str, Any]:
+    """Pick the schema by sniffing the prompt — see module docstring."""
+    blob_parts = [str(kwargs.get("system", ""))]
+    for msg in kwargs.get("messages", []) or []:
+        content = msg.get("content", "") if isinstance(msg, dict) else ""
+        blob_parts.append(str(content))
+    blob = "\n".join(blob_parts)
+    if (
+        '"decision": "provide_final_answer"' in blob
+        or '"decision":"provide_final_answer"' in blob
+    ):
+        return MINIONS_CONVERSATION_SCHEMA
+    return MINIONS_FIRST_TURN_SCHEMA
+
+
+# ---------- Compatibility patches (idempotent) ----------
+
+_PATCHES_APPLIED = False
+
+
+def _stub_missing_imports() -> None:
+    """Minions's clients/__init__.py eager-imports every provider client.
+
+    Two annoyances:
+    1. ``mistralai`` 2.x dropped top-level ``Mistral`` → ImportError.
+    2. The secure-chat path imports ``nv_attestation_sdk`` which writes a log
+       file to CWD at import-time. We don't use secure chat.
+    """
+    try:
+        import mistralai
+        if not hasattr(mistralai, "Mistral"):
+            mistralai.Mistral = type("Mistral", (), {})  # type: ignore[attr-defined]
+    except ImportError:
+        sys.modules["mistralai"] = types.ModuleType("mistralai")
+        sys.modules["mistralai"].Mistral = type("Mistral", (), {})  # type: ignore[attr-defined]
+    sys.modules.setdefault("nv_attestation_sdk", None)  # type: ignore[arg-type]
+
+
+def _patch_anthropic_globally() -> None:
+    import anthropic as _anth_mod
+    from anthropic.resources.messages import messages as _msgs_mod
+    from anthropic.resources.beta.messages import messages as _beta_msgs_mod
+
+    # External Minions builds bare anthropic.Anthropic() clients (no timeout
+    # / max_retries). Under concurrency=8 SWE-bench load those default to
+    # ~60s and timeout in droves. Inject sane defaults at the constructor.
+    if not getattr(_anth_mod.Anthropic.__init__, "_hybrid_patched", False):
+        _orig_init = _anth_mod.Anthropic.__init__
+
+        def _patched_init(self, *args, **kwargs):  # type: ignore[no-untyped-def]
+            kwargs.setdefault("timeout", 600.0)
+            kwargs.setdefault("max_retries", 5)
+            return _orig_init(self, *args, **kwargs)
+
+        _patched_init._hybrid_patched = True  # type: ignore[attr-defined]
+        _anth_mod.Anthropic.__init__ = _patched_init  # type: ignore[assignment]
+
+    for cls in (_msgs_mod.Messages, _beta_msgs_mod.Messages):
+        if getattr(cls.create, "_hybrid_patched", False):
+            continue
+        orig = cls.create
+
+        def make_patched(orig):  # type: ignore[no-untyped-def]
+            def patched(self, **kwargs):  # type: ignore[no-untyped-def]
+                model = kwargs.get("model", "")
+                if model.startswith(NO_TEMP_PREFIXES):
+                    kwargs.pop("temperature", None)
+                    if (
+                        "output_config" not in kwargs
+                        and _looks_like_minions_call(kwargs)
+                    ):
+                        kwargs["output_config"] = _minions_turn_schema(kwargs)
+                return orig(self, **kwargs)
+            patched._hybrid_patched = True  # type: ignore[attr-defined]
+            return patched
+
+        cls.create = make_patched(orig)  # type: ignore[assignment]
+
+
+def _patch_minions_extract_json() -> None:
+    """Minions's ``_extract_json`` uses a non-greedy regex that grabs the
+    first short bracket pair and prefers ```json``` fences. With structured
+    outputs the entire response IS valid JSON, so short-circuit on that.
+    """
+    from minions import minion as _minion_mod  # type: ignore[import-not-found]
+
+    if getattr(_minion_mod._extract_json, "_hybrid_patched", False):
+        return
+    _orig = _minion_mod._extract_json
+
+    def patched(text):  # type: ignore[no-untyped-def]
+        s = (text or "").strip()
+        if s.startswith("{") and s.endswith("}"):
+            try:
+                return _json.loads(s)
+            except _json.JSONDecodeError:
+                pass
+        return _orig(text)
+
+    patched._hybrid_patched = True  # type: ignore[attr-defined]
+    _minion_mod._extract_json = patched  # type: ignore[assignment]
+
+
+def _apply_patches_once() -> None:
+    global _PATCHES_APPLIED
+    if _PATCHES_APPLIED:
+        return
+    _stub_missing_imports()
+    _patch_anthropic_globally()
+    _patch_minions_extract_json()
+    _PATCHES_APPLIED = True
+
+
+# ---------- Pre-fetch helper (GAIA only) ----------
+
+def _prefetch_context(question: str, cloud_endpoint: str, cloud_model: str) -> Dict[str, Any]:
+    """Use Anthropic web_search to fetch real source material the worker can read.
+
+    Minions's premise is "worker reads a doc, asks cloud for help" — but GAIA
+    tasks ship with no doc, so we synthesize one by having Opus do an actual
+    web search first and dump the results back as the worker's context.
+
+    Returns {text, tokens, cost_usd, n_searches}. On any failure: empty text
+    and zeros — the protocol still runs.
+    """
+    out: Dict[str, Any] = {
+        "text": "", "tokens": 0, "cost_usd": 0.0, "n_searches": 0,
+    }
+    if cloud_endpoint != "anthropic" or not (question or "").strip():
+        return out
+    try:
+        prompt = (
+            "Research the following question using web_search. Do NOT answer it. "
+            "Instead, gather all relevant facts, numbers, names, dates, sources, "
+            "and direct quotes you find, and report them as a dense reference "
+            "document with URLs. The downstream reader is a small LLM that "
+            "needs raw material to reason over.\n\nQUESTION:\n" + question
+        )
+        text, p, c, n_searches = LocalCloudAgent._call_anthropic(
+            cloud_model,
+            user=prompt,
+            max_tokens=8192,
+            tools=[ANTHROPIC_WEB_SEARCH_TOOL],
+            tool_choice={"type": "any"},
+        )
+        from openjarvis.agents.hybrid._prices import cost as _cost_usd
+        out.update(
+            text=text,
+            tokens=p + c,
+            cost_usd=_cost_usd(cloud_model, p, c) + n_searches * WEB_SEARCH_COST_PER_CALL,
+            n_searches=n_searches,
+        )
+    except Exception as e:
+        out["error"] = f"{type(e).__name__}: {e}"
+    return out
+
+
+def _context_for(
+    task: Optional[Dict[str, Any]], prefetched: str = ""
+) -> List[str]:
+    """Minions wants a context list."""
+    bits: List[str] = []
+    task = task or {}
+    if task.get("hints_text"):
+        bits.append(task["hints_text"])
+    if task.get("problem_statement") and not task.get("question"):
+        bits.append(task["problem_statement"])
+    if prefetched:
+        bits.append(prefetched)
+    return bits or [""]
+
+
+# ---------- Main agent ----------
+
+@AgentRegistry.register("minions")
+class MinionsAgent(LocalCloudAgent):
+    """HazyResearch Minions supervisor/worker protocol. See module docstring."""
+
+    agent_id = "minions"
+
+    def _is_soft_failure(self, exc: BaseException) -> Optional[str]:
+        # Known soft-failure modes: Qwen worker JSON malformed, Anthropic
+        # 400/529, KeyError on missing schema fields.
+        try:
+            import anthropic
+            if isinstance(exc, anthropic.BadRequestError):
+                return f"{type(exc).__name__}: {str(exc)[:120]}"
+        except Exception:
+            pass
+        if isinstance(exc, (_json.JSONDecodeError, ValueError, KeyError)):
+            return f"{type(exc).__name__}: {str(exc)[:120]}"
+        if "JSONDecodeError" in type(exc).__name__:
+            return f"{type(exc).__name__}: {str(exc)[:120]}"
+        if "prompt is too long" in str(exc):
+            return f"{type(exc).__name__}: {str(exc)[:120]}"
+        return None
+
+    def _run_paradigm(
+        self,
+        input: str,
+        context: Optional[AgentContext],
+        **kwargs: Any,
+    ) -> Tuple[str, Dict[str, Any]]:
+        _apply_patches_once()
+        from minions.clients.openai import OpenAIClient  # type: ignore[import-not-found]
+        from minions.clients.anthropic import AnthropicClient  # type: ignore[import-not-found]
+        from minions.minion import Minion  # type: ignore[import-not-found]
+        from minions.minions import Minions  # type: ignore[import-not-found]
+
+        cfg = self._cfg
+        mode = cfg.get("mode", "minion")
+        task_meta: Dict[str, Any] = {}
+        if context is not None:
+            task_meta = context.metadata.get("task", {}) or {}
+
+        if not self._local_endpoint or not self._local_model:
+            raise ValueError(
+                "MinionsAgent needs local_model + local_endpoint; got "
+                f"model={self._local_model!r} endpoint={self._local_endpoint!r}"
+            )
+
+        local_client = OpenAIClient(
+            model_name=self._local_model,
+            base_url=self._local_endpoint,
+            api_key="EMPTY",
+            temperature=cfg.get("local_temperature", 0.0),
+            max_tokens=cfg.get("worker_max_tokens", 4096),
+            local=True,
+        )
+        if self._cloud_endpoint == "openai":
+            cloud_client = OpenAIClient(
+                model_name=self._cloud_model,
+                temperature=0.0,
+                max_tokens=4096,
+            )
+        elif self._cloud_endpoint == "anthropic":
+            # Temperature stripping is handled by the global patch above for Opus 4.7+.
+            cloud_client = AnthropicClient(
+                model_name=self._cloud_model,
+                temperature=0.0,
+                max_tokens=4096,
+            )
+        else:
+            raise ValueError(f"unsupported cloud endpoint: {self._cloud_endpoint!r}")
+
+        cls = Minions if mode == "minions" else Minion
+        log_dir = cfg.get("log_dir") or "/tmp/minions_logs"
+        protocol = cls(
+            local_client=local_client,
+            remote_client=cloud_client,
+            max_rounds=cfg.get("max_rounds", 3),
+            log_dir=log_dir,
+        )
+
+        # GAIA-shape only: prefetch a web_search digest so the worker has
+        # something real to read. SWE-bench (problem_statement only) already
+        # ships its own doc.
+        prefetch: Dict[str, Any] = {
+            "text": "", "tokens": 0, "cost_usd": 0.0, "n_searches": 0,
+        }
+        if task_meta.get("question"):
+            prefetch = _prefetch_context(
+                task_meta["question"], self._cloud_endpoint, self._cloud_model
+            )
+
+        out = protocol(
+            task=input,  # full formatted prompt (with bench instruction)
+            context=_context_for(task_meta, prefetched=prefetch["text"]),
+            doc_metadata=cfg.get("doc_metadata", "task"),
+            max_rounds=cfg.get("max_rounds", 3),
+        )
+
+        local_usage = out.get("local_usage")
+        remote_usage = out.get("remote_usage")
+        lp = getattr(local_usage, "prompt_tokens", 0)
+        lc = getattr(local_usage, "completion_tokens", 0)
+        rp = getattr(remote_usage, "prompt_tokens", 0)
+        rc = getattr(remote_usage, "completion_tokens", 0)
+
+        meta = {
+            "tokens_local": lp + lc,
+            "tokens_cloud": (rp + rc) + prefetch["tokens"],
+            "cost_usd": self.cost_usd(self._cloud_model, rp, rc) + prefetch["cost_usd"],
+            "turns": cfg.get("max_rounds", 3),
+            "traces": {
+                "mode": mode,
+                "supervisor_messages": out.get("supervisor_messages"),
+                "worker_messages": out.get("worker_messages"),
+                "timing": out.get("timing"),
+                "log_file": out.get("log_file"),
+                "prefetch": {
+                    "n_searches": prefetch["n_searches"],
+                    "tokens": prefetch["tokens"],
+                    "cost_usd": prefetch["cost_usd"],
+                    "chars": len(prefetch["text"]),
+                    "error": prefetch.get("error"),
+                },
+            },
+        }
+        return out.get("final_answer", ""), meta
+
+
+__all__ = ["MinionsAgent"]