hybrid: ship CLI runner + registry + new_experiment.sh + README

Final piece — make the ported paradigms runnable end-to-end through the
same flow as the hybrid harness:

- `runner.py`: `python -m openjarvis.agents.hybrid.runner --cell <name>`.
  Loads cells from packaged registry TOMLs, constructs the registered
  agent via AgentRegistry, iterates tasks from OpenJarvis's existing
  GAIADataset / SWEBenchDataset, scores via gaia exact-match or the new
  Modal-backed SWE-bench harness scorer, writes
  `results.jsonl`+`summary.json` in the same shape as the hybrid harness
  so existing rescore / dashboard scripts work unmodified. Supports
  resume (drops errored rows), per-cell flock, ThreadPoolExecutor
  concurrency, and the same per-row heartbeat format.

- `registry/{advisors,archon,conductor,minions,skillorchestra,toolorchestra}.toml`:
  35 cells ported verbatim from the hybrid harness — same models, same N,
  same `method_cfg` — so OpenJarvis runs should reproduce the numbers in
  `hybrid-local-cloud-compute/docs/results.md`. ToolOrchestra cells
  rewritten from the original NotImplementedError stubs to point at the
  prompted port (cloud-as-orchestrator, no Nemotron-Orchestrator-8B).

- `scripts/new_experiment.sh`: mirror of hybrid's scaffolder. Maps
  shortnames (`qwen3.5-27b`, `claude-opus-4-7`, …) to full HF ids +
  endpoints, picks method-appropriate `method_cfg` defaults, appends a
  `[cells.<name>]` block to the right registry TOML.

- `README.md`: quickstart, paradigm taxonomy, reproducibility table
  pointing at the hybrid results for headline numbers.

End-to-end smoke: `python -m openjarvis.agents.hybrid.runner --help` works,
all 6 agents register, all 35 cells load.

src/openjarvis/agents/hybrid/README.md

@@ -0,0 +1,103 @@
+# Hybrid local+cloud paradigm agents
+
+Six paradigms ported from
+[`/matx/u/aspark/hybrid-local-cloud-compute`](../../../../../) — each is
+registered as a standard OpenJarvis agent so the rest of the platform
+(SDK, CLI, distillation, evals) can use them like any other agent.
+
+| Agent             | Plan shape      | Trains what?         | Workers                   |
+|-------------------|-----------------|----------------------|---------------------------|
+| `minions`         | reactive loop   | nothing              | 1 local + 1 cloud         |
+| `conductor`       | static DAG      | (paper: 7B planner)  | up to 5 frontier+open     |
+| `archon`          | static recipe   | nothing (search)     | K local + cloud rank/fuse |
+| `advisors`        | reactive loop   | (paper: local model) | 1 local + 1 cloud         |
+| `skillorchestra`  | per-query pick  | (paper: profiler)    | 1 local + 1 cloud         |
+| `toolorchestra`   | reactive loop   | (paper: 8B RL)       | local + tools + LLM pool  |
+
+Items in parentheses are what the *paper* trains. These OpenJarvis ports
+are **inference-only** — none modify weights. The trained variants (advisor
+RL, Orchestrator-8B, SkillOrchestra learn-phase) stay TODOs; the prompted
+lower-bounds get you 80-90% of the headline accuracy at zero training cost.
+
+## What's where
+
+```
+src/openjarvis/agents/hybrid/
+├── _base.py          LocalCloudAgent ABC + SDK helpers
+├── _prices.py        cloud-model pricing + temp-strip quirks
+├── _prompts.py       GAIA / SWE-bench answer-format instructions
+├── advisors.py       executor ↔ advisor ↔ executor (3-step)
+├── conductor.py      static DAG planner
+├── minions.py        HazyResearch Minions wrapper
+├── archon.py         Archon (generator → ranker → fuser)
+├── skillorchestra.py skill-aware router
+├── toolorchestra.py  prompted multi-turn tool dispatcher
+├── runner.py         CLI: python -m ...hybrid.runner --cell NAME
+├── registry/         <method>.toml — one cell per (bench, local, cloud, N)
+└── scripts/
+    └── new_experiment.sh   scaffold a new cell, run instructions
+```
+
+The Modal-backed SWE-bench-Verified scorer is in
+`src/openjarvis/evals/scorers/swebench_harness.py` (next to the existing
+structural scorer).
+
+## Quickstart
+
+```bash
+cd /matx/u/aspark/OpenJarvis
+source .env                                           # API keys
+
+# 1. Start vLLM in another shell (see CLAUDE.md for the full recipe)
+#    CUDA_VISIBLE_DEVICES=0 .venv/bin/python -m vllm.entrypoints.openai.api_server \
+#       --model Qwen/Qwen3.5-27B-FP8 --port 8001 ...
+
+# 2. (Optional) for Minions: install the upstream library
+.venv/bin/uv pip install -e /matx/u/aspark/hybrid-local-cloud-compute/external/minions
+
+# 3. Run a smoke cell
+.venv/bin/python -m openjarvis.agents.hybrid.runner \
+    --cell minions-gaia-qwen27b-opus-3
+```
+
+Outputs land in
+`$OPENJARVIS_HYBRID_EXPERIMENTS_DIR/<cell>/{results.jsonl,summary.json,config.json,logs/}`
+(defaults to `~/.openjarvis-hybrid/experiments/`). The schema matches the
+hybrid harness so the existing rescore / dashboard scripts work
+unmodified.
+
+## Adding a cell
+
+```bash
+src/openjarvis/agents/hybrid/scripts/new_experiment.sh \
+    --method conductor --bench gaia \
+    --local qwen3.5-27b --cloud claude-opus-4-7 --n 30
+```
+
+That appends a `[cells.<name>]` block to
+`registry/conductor.toml` and prints the runner invocation.
+
+## Reproducing the hybrid harness numbers
+
+The cell configs in `registry/` are copies of the hybrid harness's
+`experiments/registry/` — same models, same N, same `method_cfg`. The
+agent code is a faithful port of each adapter (same prompts, same JSON
+schemas, same patches), so running these cells in OpenJarvis should
+reproduce the numbers in
+`/matx/u/aspark/hybrid-local-cloud-compute/docs/results.md`:
+
+| paradigm        | bench    | N   | acc    | $/task |
+|-----------------|----------|-----|--------|--------|
+| minions         | swebench | 500 | 0.274  | $0.09  |
+| minions         | gaia     | 165 | 0.576  | $0.67  |
+| conductor       | swebench | 30  | 0.367  | $0.22  |
+| skillorchestra  | gaia     | 30  | 0.500  | $0.02  |
+| advisors        | gaia     | 30  | 0.533  | $0.02  |
+
+Baselines for comparison:
+- `baseline-cloud-gaia-opus` = 0.570 / $1.09 per task
+- `baseline-cloud-swebench-opus` = 0.236 / $0.95 per task
+
+The hybrid harness stays the authoritative reference for n=500 numbers
+while we validate the port. Once OpenJarvis cells reproduce the headline
+accuracies within noise, the harness can be deprecated in favor of these.

src/openjarvis/agents/hybrid/registry/advisors.toml

@@ -0,0 +1,41 @@
+# Advisors cells (arXiv:2510.02453).
+#
+# Inference-only repro of the advisor paradigm: cloud drafts → local advisor
+# critiques → cloud rewrites. The paper's RL-trained advisor isn't reproduced
+# (no training in this harness) — these are the untrained-advisor lower bound.
+# Adapter auto-resolves the local model name against /v1/models.
+
+[cells.advisors-gaia-qwen9b-opus-3]
+method = "advisors"
+bench  = "gaia"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-9B", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = {}
+
+[cells.advisors-swebenchverified-qwen9b-opus-3]
+method = "advisors"
+bench  = "swebench-verified"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = {}
+concurrency = 3
+
+[cells.advisors-gaia-qwen9b-opus-30]
+method = "advisors"
+bench  = "gaia"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = {}
+concurrency = 4
+
+[cells.advisors-swebenchverified-qwen9b-opus-30]
+method = "advisors"
+bench  = "swebench-verified"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = {}
+concurrency = 3

src/openjarvis/agents/hybrid/registry/archon.toml

@@ -0,0 +1,94 @@
+# Archon cells. Inference-time architecture search (arXiv:2409.15254).
+#
+# Hybrid wiring: K local proposers (vLLM on 8001/8004) → 1 cloud ranker →
+# 1 cloud fuser. method_cfg.architecture = "ensemble_rank_fuse" is the
+# only non-debug preset; "single_local" exists for adapter smoke tests
+# (no cloud calls).
+#
+# Naming: archon-<bench>-<local-short>-<cloud-short>-<arch>-<K>-<N>
+
+# ---- Stage 1 smokes (n=3) ----
+
+[cells.archon-gaia-qwen27b-opus-ensemble-K3-3]
+method = "archon"
+bench  = "gaia"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 3, max_tokens = 1024, temperature = 0.7 }
+
+# Cheap sanity smoke: local-only generator, zero cloud spend. Use to verify
+# vLLM + Archon plumbing without paying for Opus.
+[cells.archon-gaia-qwen27b-singlelocal-3]
+method = "archon"
+bench  = "gaia"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { architecture = "single_local", max_tokens = 1024 }
+
+[cells.archon-swebenchverified-qwen27b-opus-ensemble-K3-3]
+method = "archon"
+bench  = "swebench-verified"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 3, max_tokens = 1024, temperature = 0.7 }
+concurrency = 3
+
+
+# ---- Stage 1.5 mid (n=30) ----
+
+[cells.archon-gaia-qwen27b-opus-ensemble-K5-30]
+method = "archon"
+bench  = "gaia"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 5, max_tokens = 2048, temperature = 0.7 }
+concurrency = 4
+
+[cells.archon-gaia-qwen27b-singlelocal-30]
+method = "archon"
+bench  = "gaia"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { architecture = "single_local", max_tokens = 2048 }
+concurrency = 4
+
+[cells.archon-gaia-gemma31b-opus-ensemble-K5-30]
+method = "archon"
+bench  = "gaia"
+n      = 30
+local  = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" }
+cloud  = { model = "claude-opus-4-7",        endpoint = "anthropic" }
+method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 5, max_tokens = 2048, temperature = 0.7 }
+
+[cells.archon-swebenchverified-qwen27b-opus-ensemble-K5-30]
+method = "archon"
+bench  = "swebench-verified"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 5, max_tokens = 2048, temperature = 0.7 }
+concurrency = 3
+
+
+# ---- Stage 2 full ----
+
+[cells.archon-gaia-qwen27b-opus-ensemble-K5-165]
+method = "archon"
+bench  = "gaia"
+n      = 165
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 5, max_tokens = 2048, temperature = 0.7 }
+
+[cells.archon-gaia-gemma31b-opus-ensemble-K5-165]
+method = "archon"
+bench  = "gaia"
+n      = 165
+local  = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" }
+cloud  = { model = "claude-opus-4-7",        endpoint = "anthropic" }
+method_cfg = { architecture = "ensemble_rank_fuse", n_samples = 5, max_tokens = 2048, temperature = 0.7 }

src/openjarvis/agents/hybrid/registry/conductor.toml

@@ -0,0 +1,54 @@
+# Conductor cells. Inference-only repro of arXiv 2512.04388 (Sakana, 2025).
+#
+# Stage-1 substitutes the (untrained) Qwen2.5-7B conductor with a zero-shot
+# frontier planner (Opus). Workers default to local Qwen3.5-27B + Opus +
+# gpt-5-mini; the adapter auto-drops the local worker if vLLM is down.
+# Override `method_cfg.workers` to customize the pool.
+#
+# Naming: conductor-<bench>-<conductor-short>-<N>
+
+[cells.conductor-gaia-opusplan-5]
+method = "conductor"
+bench  = "gaia"
+n      = 5
+# vLLM is the default local worker; auto-skipped if /v1/models is unreachable.
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { worker_max_tokens = 4096, worker_temperature = 0.2, conductor_max_tokens = 2048 }
+
+[cells.conductor-gaia-gpt5miniplan-5]
+method = "conductor"
+bench  = "gaia"
+n      = 5
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "gpt-5-mini", endpoint = "openai" }
+method_cfg = { worker_max_tokens = 4096, worker_temperature = 0.2, conductor_max_tokens = 2048 }
+
+[cells.conductor-swebenchverified-opusplan-5]
+method = "conductor"
+bench  = "swebench-verified"
+n      = 5
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { worker_max_tokens = 8192, worker_temperature = 0.2, conductor_max_tokens = 4096 }
+
+
+# ---- n=30 (scaled-up cells for paradigms that worked at smoke tier) ----
+
+[cells.conductor-gaia-opusplan-30]
+method = "conductor"
+bench  = "gaia"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { worker_max_tokens = 4096, worker_temperature = 0.2, conductor_max_tokens = 2048 }
+concurrency = 8
+
+[cells.conductor-swebenchverified-opusplan-30]
+method = "conductor"
+bench  = "swebench-verified"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { worker_max_tokens = 8192, worker_temperature = 0.2, conductor_max_tokens = 4096 }
+concurrency = 8

src/openjarvis/agents/hybrid/registry/minions.toml

@@ -0,0 +1,96 @@
+# Minions cells. One [cells.<name>] block per (bench × local × cloud × N).
+#
+# Naming: minions-<bench>-<local-short>-<cloud-short>-<N>
+# HF model IDs verified Feb 2026 / Gemma 4 release:
+#   Qwen/Qwen3.5-27B          (dense, 262K native context)
+#   google/gemma-4-31B-it     (dense instruct, 256K context — 26B is the MoE variant)
+
+# ---- Stage 1 smokes (n=3) ----
+
+[cells.minions-gaia-qwen27b-opus-3]
+method = "minions"
+bench  = "gaia"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7",  endpoint = "anthropic" }
+method_cfg = { mode = "minion", max_rounds = 3, worker_max_tokens = 4096 }
+
+[cells.minions-swebenchverified-qwen27b-opus-3]
+method = "minions"
+bench  = "swebench-verified"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7",  endpoint = "anthropic" }
+method_cfg = { mode = "minion", max_rounds = 3, worker_max_tokens = 8192 }
+
+
+# ---- Stage 1.5 mid (n=30) — catches cells that smoke-pass but full-fail ----
+
+[cells.minions-gaia-qwen27b-opus-30]
+method = "minions"
+bench  = "gaia"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7",  endpoint = "anthropic" }
+method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 4096 }
+
+[cells.minions-gaia-gemma31b-opus-30]
+method = "minions"
+bench  = "gaia"
+n      = 30
+local  = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" }
+cloud  = { model = "claude-opus-4-7",        endpoint = "anthropic" }
+method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 4096 }
+
+[cells.minions-swebenchverified-qwen27b-opus-30]
+method = "minions"
+bench  = "swebench-verified"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7",  endpoint = "anthropic" }
+method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 8192 }
+
+[cells.minions-swebenchverified-gemma31b-opus-30]
+method = "minions"
+bench  = "swebench-verified"
+n      = 30
+local  = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" }
+cloud  = { model = "claude-opus-4-7",        endpoint = "anthropic" }
+method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 8192 }
+
+
+# ---- Stage 2 full ----
+
+[cells.minions-gaia-qwen27b-opus-165]
+method = "minions"
+bench  = "gaia"
+n      = 165
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7",  endpoint = "anthropic" }
+method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 4096 }
+concurrency = 4
+
+[cells.minions-gaia-gemma31b-opus-165]
+method = "minions"
+bench  = "gaia"
+n      = 165
+local  = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" }
+cloud  = { model = "claude-opus-4-7",        endpoint = "anthropic" }
+method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 4096 }
+
+[cells.minions-swebenchverified-qwen27b-opus-500]
+method = "minions"
+bench  = "swebench-verified"
+n      = 500
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7",  endpoint = "anthropic" }
+method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 8192 }
+concurrency = 4
+
+[cells.minions-swebenchverified-gemma31b-opus-500]
+method = "minions"
+bench  = "swebench-verified"
+n      = 500
+local  = { model = "google/gemma-4-31B-it", endpoint = "http://localhost:8004/v1" }
+cloud  = { model = "claude-opus-4-7",        endpoint = "anthropic" }
+method_cfg = { mode = "minion", max_rounds = 5, worker_max_tokens = 8192 }

src/openjarvis/agents/hybrid/registry/skillorchestra.toml

@@ -0,0 +1,47 @@
+# SkillOrchestra cells (arXiv:2602.19672).
+#
+# Deployment-time skill-aware routing only. The paper's offline pipeline
+# (explore → learn → select over an SGLang model pool + FAISS wiki index)
+# isn't reproduced here — we don't have the SGLang serving stack, the
+# retriever index, or a training split to learn a routing policy on. What
+# we DO reproduce is the inference-time orchestrator: an Opus router
+# analyzes the question into skill weights, picks one of two agents
+# (local Qwen-27B vs cloud Opus) under an explicit cost trade-off, then
+# the chosen agent answers. Mirrors the eval_orchestrator paradigm in
+# external/SkillOrchestra/skillorchestra/prompts/eval_orchestrator.py.
+# See adapters/skillorchestra_adapter.py for full notes.
+
+[cells.skillorchestra-gaia-qwen27b-opus-3]
+method = "skillorchestra"
+bench  = "gaia"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7",      endpoint = "anthropic" }
+method_cfg = {}
+
+[cells.skillorchestra-swebenchverified-qwen27b-opus-3]
+method = "skillorchestra"
+bench  = "swebench-verified"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7",      endpoint = "anthropic" }
+method_cfg = {}
+concurrency = 3
+
+[cells.skillorchestra-gaia-qwen27b-opus-30]
+method = "skillorchestra"
+bench  = "gaia"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7",      endpoint = "anthropic" }
+method_cfg = {}
+concurrency = 4
+
+[cells.skillorchestra-swebenchverified-qwen27b-opus-30]
+method = "skillorchestra"
+bench  = "swebench-verified"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7",      endpoint = "anthropic" }
+method_cfg = {}
+concurrency = 3

src/openjarvis/agents/hybrid/registry/toolorchestra.toml

@@ -0,0 +1,30 @@
+# ToolOrchestra cells. Prompted port — uses a cloud model (Opus) as the
+# orchestrator, NOT the RL-trained Nemotron-Orchestrator-8B from the paper
+# (arXiv:2511.21689). Treat results as preliminary until a real
+# Orchestrator-8B deployment is wired up.
+#
+# Naming: toolorchestra-<bench>-<local-short>-<cloud-short>-<N>
+
+[cells.toolorchestra-gaia-qwen27b-opus-3]
+method = "toolorchestra"
+bench  = "gaia"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { max_turns = 6, orchestrator_max_tokens = 1024, worker_max_tokens = 4096 }
+
+[cells.toolorchestra-swebenchverified-qwen27b-opus-3]
+method = "toolorchestra"
+bench  = "swebench-verified"
+n      = 3
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { max_turns = 6, orchestrator_max_tokens = 1024, worker_max_tokens = 8192 }
+
+[cells.toolorchestra-gaia-qwen27b-opus-30]
+method = "toolorchestra"
+bench  = "gaia"
+n      = 30
+local  = { model = "Qwen/Qwen3.5-27B-FP8", endpoint = "http://localhost:8001/v1" }
+cloud  = { model = "claude-opus-4-7", endpoint = "anthropic" }
+method_cfg = { max_turns = 8, orchestrator_max_tokens = 1024, worker_max_tokens = 4096 }

src/openjarvis/agents/hybrid/runner.py

@@ -0,0 +1,495 @@
+"""CLI runner for hybrid paradigm experiments.
+
+::
+
+    python -m openjarvis.agents.hybrid.runner --cell minions-gaia-qwen27b-opus-3
+
+Reads a cell definition from ``registry/<method>.toml`` (bundled with this
+package or pointed at by ``OPENJARVIS_HYBRID_REGISTRY_DIR``), constructs
+the registered agent, loads bench tasks via OpenJarvis's existing dataset
+providers, runs every task, scores it, and writes
+``<EXPERIMENTS_DIR>/<cell>/results.jsonl`` + ``summary.json``.
+
+The output schema matches ``hybrid-local-cloud-compute/runner.py`` so the
+existing rescore / dashboard scripts can read OpenJarvis cells without
+modification.
+"""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import json
+import os
+import sys
+import threading
+import time
+import traceback
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any, Dict, Iterable, List, Optional, Tuple
+
+try:
+    import tomllib  # py3.11+
+except ModuleNotFoundError:
+    import tomli as tomllib  # type: ignore[import-not-found,no-redef]
+
+from openjarvis.agents._stubs import AgentContext, AgentResult
+from openjarvis.agents.hybrid._prompts import format_prompt as _format_prompt
+
+
+PACKAGE_DIR = Path(__file__).parent
+DEFAULT_REGISTRY_DIR = PACKAGE_DIR / "registry"
+DEFAULT_EXPERIMENTS_DIR = Path(
+    os.environ.get(
+        "OPENJARVIS_HYBRID_EXPERIMENTS_DIR",
+        str(Path.home() / ".openjarvis-hybrid" / "experiments"),
+    )
+)
+
+
+# ---------- Registry ----------
+
+def load_registry(registry_dir: Optional[Path] = None) -> Dict[str, Dict[str, Any]]:
+    """Merge every ``<registry_dir>/*.toml``. Cell names must be unique."""
+    base = registry_dir or DEFAULT_REGISTRY_DIR
+    env_override = os.environ.get("OPENJARVIS_HYBRID_REGISTRY_DIR")
+    if env_override:
+        base = Path(env_override)
+    if not base.is_dir():
+        return {}
+    cells: Dict[str, Dict[str, Any]] = {}
+    for p in sorted(base.glob("*.toml")):
+        data = tomllib.loads(p.read_text())
+        for name, cell in data.get("cells", {}).items():
+            if name in cells:
+                raise ValueError(
+                    f"duplicate cell {name!r} (already defined before {p.name})"
+                )
+            cells[name] = cell
+    return cells
+
+
+# ---------- Bench dispatch ----------
+
+def _load_gaia_tasks(n: Optional[int]) -> List[Dict[str, Any]]:
+    """GAIA validation. Each task is a dict with `task_id` + `question`."""
+    from openjarvis.evals.datasets.gaia import GAIADataset
+
+    ds = GAIADataset()
+    ds.load(max_samples=n)
+    out: List[Dict[str, Any]] = []
+    for rec in ds.iter_records():
+        # rec.problem is the formatted question prompt; rec.metadata carries
+        # the GAIA-specific fields including any reference answer.
+        out.append({
+            "task_id": rec.record_id,
+            "question": rec.metadata.get("question", rec.problem),
+            "reference": rec.reference,
+            "metadata": dict(rec.metadata),
+        })
+    return out
+
+
+def _load_swebench_tasks(n: Optional[int]) -> List[Dict[str, Any]]:
+    """SWE-bench-Verified test. Each task carries patch-evaluation fields."""
+    from openjarvis.evals.datasets.swebench import SWEBenchDataset
+
+    ds = SWEBenchDataset()
+    ds.load(max_samples=n)
+    out: List[Dict[str, Any]] = []
+    for rec in ds.iter_records():
+        md = rec.metadata or {}
+        out.append({
+            "task_id": md.get("instance_id", rec.record_id),
+            "repo": md.get("repo", ""),
+            "base_commit": md.get("base_commit", ""),
+            "problem_statement": md.get("problem_statement", rec.problem),
+            "hints_text": md.get("hints_text", ""),
+            "test_patch": md.get("test_patch", ""),
+            "FAIL_TO_PASS": md.get("FAIL_TO_PASS", []),
+            "PASS_TO_PASS": md.get("PASS_TO_PASS", []),
+            "version": md.get("version"),
+            "reference": rec.reference,
+            "metadata": dict(md),
+        })
+    return out
+
+
+def load_tasks(bench: str, n: Optional[int]) -> List[Dict[str, Any]]:
+    if bench == "gaia":
+        return _load_gaia_tasks(n)
+    if bench in ("swebench-verified", "swebench_verified", "swebench"):
+        return _load_swebench_tasks(n)
+    raise ValueError(f"unknown bench: {bench!r}")
+
+
+# ---------- Scoring ----------
+
+def _score_gaia(task: Dict[str, Any], answer: str) -> Dict[str, Any]:
+    """Exact-match-with-format-normalization GAIA scorer.
+
+    Lightweight version: extracts the final-answer line and string-compares
+    against the reference. Use the OpenJarvis gaia_exact scorer for the
+    judge-tiebreaker path.
+    """
+    import re
+
+    ref = (task.get("reference") or "").strip()
+    if not ref:
+        return {"success": False, "score": 0.0, "details": {"reason": "no_reference"}}
+    m = re.search(
+        r"FINAL\s*ANSWER\s*:\s*(.+?)\s*$",
+        answer,
+        re.IGNORECASE | re.MULTILINE,
+    )
+    pred = (m.group(1).strip() if m else answer.strip()).rstrip(".").strip()
+    success = pred.lower() == ref.lower()
+    return {
+        "success": success,
+        "score": 1.0 if success else 0.0,
+        "details": {"prediction": pred, "reference": ref},
+    }
+
+
+def _score_swebench(task: Dict[str, Any], answer: str) -> Dict[str, Any]:
+    """Modal-backed SWE-bench Verified harness scorer."""
+    from openjarvis.evals.scorers.swebench_harness import (
+        SWEBenchHarnessScorer,
+        extract_patch,
+    )
+    from openjarvis.evals.core.types import EvalRecord
+
+    patch = extract_patch(answer)
+    if patch is None:
+        return {"success": False, "score": 0.0, "details": {"reason": "no_patch_extracted"}}
+
+    record = EvalRecord(
+        record_id=task["task_id"],
+        problem=task.get("problem_statement", ""),
+        reference="",
+        category="agentic",
+        metadata={"instance_id": task["task_id"]},
+    )
+    scorer = SWEBenchHarnessScorer(timeout_s=int(os.environ.get("SWEBENCH_TIMEOUT_S", "1800")))
+    is_correct, details = scorer.score(record, answer)
+    return {
+        "success": bool(is_correct),
+        "score": 1.0 if is_correct else 0.0,
+        "details": details,
+    }
+
+
+def score(bench: str, task: Dict[str, Any], answer: str) -> Dict[str, Any]:
+    if bench == "gaia":
+        return _score_gaia(task, answer)
+    if bench in ("swebench-verified", "swebench_verified", "swebench"):
+        return _score_swebench(task, answer)
+    raise ValueError(f"unknown bench: {bench!r}")
+
+
+# ---------- Cell run ----------
+
+def _cell_dir(cell_name: str, root: Path) -> Path:
+    d = root / cell_name
+    d.mkdir(parents=True, exist_ok=True)
+    (d / "logs").mkdir(exist_ok=True)
+    return d
+
+
+@contextmanager
+def _cell_lock(out_dir: Path, cell_name: str):
+    """Exclusive flock on ``<cell>/.lock`` to prevent concurrent runner stomps."""
+    lock_path = out_dir / ".lock"
+    f = lock_path.open("a+")
+    try:
+        fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except BlockingIOError:
+        f.seek(0)
+        prev = (f.read() or "?").strip() or "?"
+        f.close()
+        raise SystemExit(
+            f"[lock] another runner is already running cell {cell_name!r} "
+            f"(holder pid: {prev}). refusing to start a second instance."
+        )
+    f.seek(0); f.truncate(); f.write(str(os.getpid())); f.flush()
+    try:
+        yield
+    finally:
+        try:
+            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+        except Exception:
+            pass
+        f.close()
+        try:
+            lock_path.unlink()
+        except FileNotFoundError:
+            pass
+
+
+def _build_agent(cell: Dict[str, Any]):
+    """Construct the registered agent for this cell."""
+    import openjarvis.agents  # noqa: F401 — populate registry
+    from openjarvis.core.registry import AgentRegistry
+
+    method = cell["method"]
+    if not AgentRegistry.contains(method):
+        raise ValueError(
+            f"agent {method!r} not registered. Available: "
+            f"{', '.join(sorted(AgentRegistry.keys()))}"
+        )
+    agent_cls = AgentRegistry.get(method)
+    local = cell.get("local") or {}
+    cloud = cell.get("cloud") or {}
+    method_cfg = dict(cell.get("method_cfg") or {})
+
+    return agent_cls(
+        engine=None,  # raw SDK calls — engine unused
+        model=cloud.get("model", ""),
+        local_model=local.get("model"),
+        local_endpoint=local.get("endpoint"),
+        cloud_endpoint=(cloud.get("endpoint") or "anthropic").lower(),
+        cfg=method_cfg,
+    )
+
+
+def _run_one(agent, bench: str, task: Dict[str, Any], log_dir: str) -> Dict[str, Any]:
+    """Run the agent on one task. Returns a hybrid-shape row."""
+    prompt = _format_prompt(task)
+    ctx = AgentContext(metadata={"task": task, "task_id": task["task_id"]})
+    t0 = time.time()
+    try:
+        result: AgentResult = agent.run(prompt, ctx)
+        meta = dict(result.metadata or {})
+        out = {
+            "task_id": task["task_id"],
+            "answer": result.content or "",
+            "tokens_local": int(meta.get("tokens_local", 0)),
+            "tokens_cloud": int(meta.get("tokens_cloud", 0)),
+            "cost_usd": float(meta.get("cost_usd", 0.0)),
+            "latency_s": float(meta.get("latency_s", time.time() - t0)),
+            "traces": meta.get("traces", {}),
+        }
+        if "soft_error" in meta:
+            out["soft_error"] = meta["soft_error"]
+        return {**out, "error": None}
+    except Exception as e:
+        return {
+            "task_id": task["task_id"],
+            "answer": "",
+            "tokens_local": 0, "tokens_cloud": 0,
+            "cost_usd": 0.0, "latency_s": time.time() - t0,
+            "traces": {},
+            "error": f"{type(e).__name__}: {e}\n{traceback.format_exc()}",
+        }
+
+
+def _heartbeat(done: int, total: int, row: Dict[str, Any], t_start: float) -> None:
+    elapsed = time.time() - t_start
+    eta = (total - done) * (elapsed / max(done, 1))
+    ok = "OK" if not row.get("error") else "ERR"
+    s = row.get("score") or {}
+    sc = s.get("score")
+    sc_str = f"{sc:.2f}" if isinstance(sc, (int, float)) else "—"
+    print(
+        f"[{done}/{total}] {ok} task={row['task_id']} score={sc_str} "
+        f"local={row['tokens_local']} cloud={row['tokens_cloud']} "
+        f"${row['cost_usd']:.3f} {row['latency_s']:.1f}s eta={eta/60:.1f}m",
+        flush=True,
+    )
+
+
+def _write_summary(
+    out_dir: Path,
+    cell_name: str,
+    cell: Dict[str, Any],
+    tasks: List[Dict[str, Any]],
+    t_start: float,
+) -> None:
+    results_path = out_dir / "results.jsonl"
+    rows = [
+        json.loads(l)
+        for l in results_path.read_text().splitlines()
+        if l.strip()
+    ]
+    n_done = len(rows)
+    n_err = sum(1 for r in rows if r.get("error"))
+    successes = [r for r in rows if r.get("score") and r["score"].get("success")]
+    acc = (len(successes) / n_done) if n_done else 0.0
+    total_cost = sum(r.get("cost_usd", 0.0) for r in rows)
+    total_local = sum(r.get("tokens_local", 0) for r in rows)
+    total_cloud = sum(r.get("tokens_cloud", 0) for r in rows)
+    elapsed = time.time() - t_start
+
+    summary = {
+        "cell": cell_name,
+        "method": cell["method"],
+        "bench": cell["bench"],
+        "n_target": cell["n"],
+        "n_done": n_done,
+        "n_err": n_err,
+        "accuracy": acc,
+        "tokens_local_total": total_local,
+        "tokens_cloud_total": total_cloud,
+        "cost_usd_total": total_cost,
+        "wall_time_s": elapsed,
+        "task_count": len(tasks),
+    }
+    (out_dir / "summary.json").write_text(json.dumps(summary, indent=2))
+    print(
+        f"[summary] {cell_name}: n={n_done}/{cell['n']} err={n_err} "
+        f"acc={acc:.3f} cost=${total_cost:.2f} time={elapsed/60:.1f}m",
+        flush=True,
+    )
+
+
+def run_cell(
+    cell_name: str,
+    cell: Dict[str, Any],
+    *,
+    do_score: bool = True,
+    resume: bool = True,
+    root: Optional[Path] = None,
+) -> None:
+    out_root = root or DEFAULT_EXPERIMENTS_DIR
+    out_dir = _cell_dir(cell_name, out_root)
+    with _cell_lock(out_dir, cell_name):
+        _run_cell_locked(
+            cell_name, cell, out_dir,
+            do_score=do_score, resume=resume,
+        )
+
+
+def _run_cell_locked(
+    cell_name: str,
+    cell: Dict[str, Any],
+    out_dir: Path,
+    *,
+    do_score: bool,
+    resume: bool,
+) -> None:
+    (out_dir / "config.json").write_text(
+        json.dumps({"name": cell_name, **cell}, indent=2)
+    )
+
+    results_path = out_dir / "results.jsonl"
+    done_ids: set = set()
+    if resume and results_path.exists():
+        # Keep only successful rows; drop errored rows so they retry.
+        kept: List[str] = []
+        for line in results_path.read_text().splitlines():
+            try:
+                row = json.loads(line)
+            except Exception:
+                continue
+            if not row.get("error"):
+                kept.append(line)
+                done_ids.add(row["task_id"])
+        results_path.write_text("\n".join(kept) + ("\n" if kept else ""))
+        print(
+            f"[resume] {len(done_ids)} tasks already done (errored rows dropped)",
+            flush=True,
+        )
+
+    tasks = load_tasks(cell["bench"], n=cell["n"])
+    print(f"[load] {cell['bench']} → {len(tasks)} tasks", flush=True)
+
+    pending = [t for t in tasks if t["task_id"] not in done_ids]
+    concurrency = max(1, int(cell.get("concurrency", 1)))
+    if concurrency > 1:
+        print(f"[concurrency] {concurrency} workers", flush=True)
+
+    agent = _build_agent(cell)
+
+    t_start = time.time()
+    write_lock = threading.Lock()
+    completed = [0]
+    log_dir = str(out_dir / "logs")
+
+    def _process(task: Dict[str, Any]) -> None:
+        row = _run_one(agent, cell["bench"], task, log_dir)
+        scored: Optional[Dict[str, Any]] = None
+        if do_score and row.get("error") is None:
+            try:
+                scored = score(cell["bench"], task, row["answer"])
+            except Exception as e:
+                scored = {
+                    "success": False, "score": 0.0,
+                    "details": {"score_error": str(e)},
+                }
+        full_row = {**row, "score": scored}
+        with write_lock, results_path.open("a") as f:
+            f.write(json.dumps(full_row) + "\n")
+            f.flush()
+            completed[0] += 1
+            _heartbeat(completed[0], len(tasks), full_row, t_start)
+
+    if concurrency == 1:
+        for task in pending:
+            _process(task)
+    else:
+        with ThreadPoolExecutor(max_workers=concurrency) as ex:
+            futures = [ex.submit(_process, t) for t in pending]
+            for fut in as_completed(futures):
+                fut.result()
+
+    _write_summary(out_dir, cell_name, cell, tasks, t_start)
+
+
+# ---------- CLI ----------
+
+def main(argv: Optional[List[str]] = None) -> int:
+    p = argparse.ArgumentParser(
+        prog="python -m openjarvis.agents.hybrid.runner",
+        description="Run a hybrid paradigm experiment cell.",
+    )
+    p.add_argument("--cell", required=True, help="Cell name from the registry TOMLs.")
+    p.add_argument(
+        "--registry-dir",
+        default=None,
+        help="Override registry dir (defaults to package registry/).",
+    )
+    p.add_argument(
+        "--root",
+        default=None,
+        help="Override experiments output root.",
+    )
+    p.add_argument("--no-score", action="store_true", help="Skip scoring.")
+    p.add_argument("--no-resume", action="store_true", help="Don't resume from results.jsonl.")
+    args = p.parse_args(argv)
+
+    reg_dir = Path(args.registry_dir) if args.registry_dir else None
+    cells = load_registry(reg_dir)
+    if not cells:
+        print(f"[error] no cells found in {reg_dir or DEFAULT_REGISTRY_DIR}", file=sys.stderr)
+        return 2
+    if args.cell not in cells:
+        print(
+            f"[error] unknown cell {args.cell!r}. Known: {', '.join(sorted(cells))}",
+            file=sys.stderr,
+        )
+        return 2
+    root = Path(args.root) if args.root else None
+    run_cell(
+        args.cell, cells[args.cell],
+        do_score=not args.no_score,
+        resume=not args.no_resume,
+        root=root,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+
+
+__all__ = [
+    "DEFAULT_EXPERIMENTS_DIR",
+    "DEFAULT_REGISTRY_DIR",
+    "load_registry",
+    "load_tasks",
+    "main",
+    "run_cell",
+    "score",
+]

src/openjarvis/agents/hybrid/scripts/new_experiment.sh

@@ -0,0 +1,108 @@
+#!/usr/bin/env bash
+# Scaffold a new hybrid paradigm experiment cell.
+#
+# Appends a [cells.<name>] block to
+# src/openjarvis/agents/hybrid/registry/<method>.toml
+# (registry is split by method — minions.toml, conductor.toml, etc.).
+#
+# Usage:
+#   src/openjarvis/agents/hybrid/scripts/new_experiment.sh \
+#       --method minions --bench gaia \
+#       --local qwen3.5-27b --cloud claude-opus-4-7 --n 50 \
+#       [--mode minion|minions] [--max-rounds 3]
+#
+# Then run:
+#   python -m openjarvis.agents.hybrid.runner --cell <printed name>
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+REGISTRY_DIR="$(cd "$SCRIPT_DIR/.." && pwd)/registry"
+
+method=""; bench=""; local_model=""; cloud_model=""; n=""
+mode="minion"; max_rounds=3
+
+while [[ $# -gt 0 ]]; do
+    case "$1" in
+        --method)      method="$2";       shift 2;;
+        --bench)       bench="$2";        shift 2;;
+        --local)       local_model="$2";  shift 2;;
+        --cloud)       cloud_model="$2";  shift 2;;
+        --n)           n="$2";            shift 2;;
+        --mode)        mode="$2";         shift 2;;
+        --max-rounds)  max_rounds="$2";   shift 2;;
+        *) echo "unknown arg: $1" >&2; exit 2;;
+    esac
+done
+
+for v in method bench local_model cloud_model n; do
+    if [[ -z "${!v}" ]]; then
+        echo "missing required --${v//_/-}" >&2
+        exit 2
+    fi
+done
+
+# Map shortnames → full HF id + endpoint
+case "$local_model" in
+    qwen3.5-27b|qwen-27b|qwen27b)   local_full="Qwen/Qwen3.5-27B-FP8";       local_ep="http://localhost:8001/v1";;
+    qwen3.5-9b|qwen-9b|qwen9b)      local_full="Qwen/Qwen3.5-9B";            local_ep="http://localhost:8001/v1";;
+    gemma-4-31b|gemma-31b|gemma31b) local_full="google/gemma-4-31B-it";      local_ep="http://localhost:8004/v1";;
+    gemma-4-26b|gemma-26b|gemma26b) local_full="google/gemma-4-26B-A4B-it";  local_ep="http://localhost:8004/v1";;
+    *)                              local_full="$local_model";               local_ep="${LOCAL_ENDPOINT:-http://localhost:8001/v1}";;
+esac
+
+case "$cloud_model" in
+    gpt-5|gpt-5-mini|gpt-4o)               cloud_ep="openai";;
+    claude-opus-4-7|claude-sonnet-4-6)     cloud_ep="anthropic";;
+    *)                                     cloud_ep="${CLOUD_ENDPOINT:-anthropic}";;
+esac
+
+local_short="$(echo "$local_model" | tr '/' '_')"
+cloud_short="$cloud_model"
+name="${method}-${bench}-${local_short}-${cloud_short}-${n}"
+
+registry_file="$REGISTRY_DIR/${method}.toml"
+if [[ ! -f "$registry_file" ]]; then
+    echo "no registry file for method '${method}' — expected $registry_file" >&2
+    echo "known: $(ls "$REGISTRY_DIR" | tr '\n' ' ')" >&2
+    exit 2
+fi
+
+# Method-specific method_cfg defaults — keep simple, override later in TOML.
+case "$method" in
+    minions)
+        cfg_block="method_cfg = { mode = \"${mode}\", max_rounds = ${max_rounds}, worker_max_tokens = 4096 }"
+        ;;
+    conductor)
+        cfg_block="method_cfg = { conductor_max_tokens = 2048, worker_max_tokens = 4096 }"
+        ;;
+    advisors)
+        cfg_block="method_cfg = { executor_max_tokens = 4096, advisor_max_tokens = 1024 }"
+        ;;
+    archon)
+        cfg_block="method_cfg = { architecture = \"ensemble_rank_fuse\", n_samples = 5, max_tokens = 2048 }"
+        ;;
+    skillorchestra)
+        cfg_block="method_cfg = { router_max_tokens = 1024, local_max_tokens = 4096, cloud_max_tokens = 4096 }"
+        ;;
+    toolorchestra)
+        cfg_block="method_cfg = { max_turns = 6, orchestrator_max_tokens = 1024, worker_max_tokens = 4096 }"
+        ;;
+    *)
+        cfg_block="method_cfg = {}"
+        ;;
+esac
+
+cat >> "$registry_file" <<EOF
+
+[cells.${name}]
+method = "${method}"
+bench  = "${bench}"
+n      = ${n}
+local  = { model = "${local_full}", endpoint = "${local_ep}" }
+cloud  = { model = "${cloud_model}", endpoint = "${cloud_ep}" }
+${cfg_block}
+EOF
+
+echo "Added cell: ${name}  →  $(realpath --relative-to="$(pwd)" "$registry_file")"
+echo "Run with: python -m openjarvis.agents.hybrid.runner --cell ${name}"