feat(runtime): gen-6 architecture consolidation — one recursive agent tree, observable + deep-cleaned

.gitignore

@@ -4,3 +4,8 @@ node_modules/
 
 # meta-harness / evolve loop runtime state
 .evolve/
+bench/data/
+bench/experiments/
+**/__pycache__/
+.claude/
+bench/scripts/__pycache__/

bench/.gitignore

@@ -4,3 +4,4 @@ node_modules/
 logs/
 run-artifacts/
 corpus/
+.eops-task-cache/

bench/HARNESS.md

@@ -7,17 +7,34 @@ Verified against source 2026-06-03 · agent-eval pinned `^0.76.0` (the optimizeP
 heldoutSignificance API is version-coupled).
 
 ## What this harness answers
-The decision gate (docs/roadmap-rsi.md): **does any non-blind topology beat blind compute
-at EQUAL k, under a DEPLOYABLE (non-oracle) selector, at significant n?**
+**The success criterion is Gate B** (docs/learning-flywheel.md, docs/architecture.md §2): across
+repeated runs on a persistent, checkable task family, the deployed policy's verifier-graded
+**multi-objective** score (correct · fast · secure · cheap, each its own deployable checker)
+improves **run-over-run** at matched per-run compute, surviving a frozen-policy control, significant
+at adequate n. That across-run slope is RSI. **The harness has NOT yet run Gate B** — see the durable
+gap below.
+
+What the harness measures **today is Gate A** (docs/roadmap-rsi.md — the inner GO/NO-GO for the
+within-run adaptive-driver layer): **does any non-blind topology beat blind compute at EQUAL COMPUTE
+(Σ rollouts × turns — `k` counts rollouts, each may be multi-turn/stateful), under a DEPLOYABLE
+(non-oracle) selector, at significant n?** Gate A is a **narrow diagnostic** — the cost-justification
+for parallel/adaptive topology, **NOT** the product verdict. A failed Gate A deletes within-run
+steering only; it never touches the corpus+policy product (Gate B). The invariant is equal-COMPUTE,
+not equal-k-on-stateless-samples. Two things to keep straight: today's judges grade a single
+*correctness* scalar (the multi-objective vector is the open contract, architecture.md §6), and every
+number below is single-objective + within-run — read them as Gate-A diagnostics, not Gate-B results.
 - Within-run STEER (verify-and-revise family) **LOSES** (rung-0, n=40: blind 37.5% →
   random@3 60.0% → refineGepa@3 45.0%; the earlier +20pp was confounded compute).
 - On the COMMITTED finsearch corpus, the self-consistency selector also **loses**:
   selector@k − random@k = **−8.2pp** (n=51). So "pick the consensus among k identical-ish
   attempts" does not beat a random draw here.
-- **UNTESTED**: parallel **DIVERSE strategies** (different reasoning paths, `directives.ts`
-  → `DIVERSE_STRATEGY_LENSES` / `composeStrategies`) @k vs blind sample(n=k). A distinct
-  family from what rung-0 falsified — this is the open gate, and what runProgram's
+- **UNTESTED (still Gate A):** parallel **DIVERSE strategies** (different reasoning paths,
+  `directives.ts` → `DIVERSE_STRATEGY_LENSES` / `composeStrategies`) @k vs blind sample(n=k). A
+  distinct family from what rung-0 falsified — the open *within-run* question, and what runProgram's
   `parallel` is built to deploy.
+- **UNBUILT (Gate B):** the across-run policy-improvement curve on a multi-objective task stream.
+  No harness runs it yet; it is the durable next step, not a corpus-replay over the existing
+  single-objective records.
 
 ## Data flow (the whole experiment in one line)
 `rollout (worker → answer) → adapter.judge (valid?) → CORPUS RunRecord (k attempts, output+valid each) → corpus-replay --selector (pick WITHOUT the judge) → corpus-report CI → gate verdict`

bench/scripts/appworld_driver.py

@@ -9,14 +9,173 @@
 
 import argparse
 import json
+import os
+import re
 import sys
+import time
 
 
 def fail(msg: str) -> None:
     print(json.dumps({"error": msg}))
     sys.exit(1)
 
 
+_CODE_RE = re.compile(r"```(?:python|py)?\s*\n(.*?)```", re.DOTALL)
+
+
+def _extract_code(text: str) -> str:
+    blocks = _CODE_RE.findall(text or "")
+    return (blocks[-1] if blocks else "").strip()
+
+
+def _router_chat(base: str, key: str, model: str, messages: list, timeout: float = 180.0):
+    """One router chat-completion with retry on transient/429/5xx. Returns
+    (content, input_tokens, output_tokens). Raises on exhausted retries."""
+    import httpx
+
+    url = base.rstrip("/") + "/chat/completions"
+    last = None
+    for attempt in range(4):
+        try:
+            r = httpx.post(
+                url,
+                headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
+                json={"model": model, "messages": messages},
+                timeout=timeout,
+            )
+            if r.status_code in (429, 500, 502, 503, 504):
+                last = f"{r.status_code}: {r.text[:160]}"
+                time.sleep(2**attempt)
+                continue
+            r.raise_for_status()
+            d = r.json()
+            content = (d["choices"][0]["message"].get("content") or "")
+            usage = d.get("usage") or {}
+            return content, int(usage.get("prompt_tokens", 0) or 0), int(usage.get("completion_tokens", 0) or 0)
+        except Exception as e:  # noqa: BLE001
+            last = str(e)
+            if attempt < 3:
+                time.sleep(2**attempt)
+                continue
+            raise RuntimeError(f"router_chat failed after retries: {last}")
+    raise RuntimeError(f"router_chat exhausted: {last}")
+
+
+def _build_system(directive: str, world) -> str:
+    sup = world.task.supervisor
+    apps = list(getattr(world.task, "allowed_apps", []) or [])
+    descs = getattr(world.task, "app_descriptions", "")
+    desc_str = json.dumps(descs) if isinstance(descs, (dict, list)) else str(descs)
+    return (
+        f"You are an AI agent completing a digital task for your supervisor "
+        f"{getattr(sup, 'first_name', '')} {getattr(sup, 'last_name', '')} "
+        f"(email {getattr(sup, 'email', '')}, phone {getattr(sup, 'phone_number', '')}) "
+        "by WRITING PYTHON that calls app APIs (the apis.<app>.<function>(...) surface).\n\n"
+        f"Available apps: {', '.join(apps)}.\n"
+        f"App descriptions: {desc_str[:1500]}\n\n"
+        "How to work, one step per turn:\n"
+        "- Discover APIs with apis.api_docs.show_api_descriptions(app_name='<app>') and "
+        "apis.api_docs.show_api_doc(app_name='<app>', api_name='<api>') BEFORE calling them.\n"
+        "- Get the supervisor's app passwords with apis.supervisor.show_account_passwords(), then log in "
+        "to each app you use to obtain its access_token.\n"
+        "- Write ONE short Python code block per turn. After it runs you SEE its OUTPUT (or error "
+        "traceback) — use that to decide the next step. Print intermediate values you need.\n"
+        "- Iterate: inspect -> authenticate -> act -> verify. Do not guess API names or arguments.\n"
+        "- When the task is fully done call apis.supervisor.complete_task(answer=<answer>) (include the "
+        "answer if the task asks a question, otherwise apis.supervisor.complete_task()).\n"
+        "- Reply with EXACTLY ONE fenced ```python block per turn and nothing else.\n\n"
+        f"{directive}"
+    )
+
+
+def cmd_react(args) -> None:
+    """Multi-turn REPL agent: the model writes a python block, the engine executes
+    it in the PERSISTENT world, the output is fed back, and it iterates until it
+    completes the task or hits max-turns. Then AppWorld's own evaluator scores it.
+    Config (directive, model, router creds, max_turns) arrives as JSON on stdin so
+    the candidate directive can be arbitrarily long. The directive is the optimized
+    surface; the loop + contract are fixed."""
+    cfg = {}
+    raw = sys.stdin.read()
+    if raw.strip():
+        try:
+            cfg = json.loads(raw)
+        except Exception as e:  # noqa: BLE001
+            fail(f"react config JSON parse failed: {e}")
+    directive = str(cfg.get("directive", ""))
+    model = str(cfg.get("model", "gpt-4o"))
+    max_turns = int(cfg.get("max_turns", 8))
+    router_base = str(cfg.get("router_base", "https://router.tangle.tools/v1"))
+    router_key = str(cfg.get("router_key") or os.environ.get("TANGLE_API_KEY", ""))
+    if not router_key:
+        fail("react: router_key/TANGLE_API_KEY required")
+
+    try:
+        from appworld import AppWorld
+    except Exception as e:  # noqa: BLE001
+        fail(f"appworld import failed: {e}")
+
+    in_tok = 0
+    out_tok = 0
+    turns = 0
+    turns_log: list = []
+    try:
+        with AppWorld(
+            task_id=args.task_id,
+            experiment_name="bench-react",
+            raise_on_failure=False,
+        ) as world:
+            messages = [
+                {"role": "system", "content": _build_system(directive, world)},
+                {"role": "user", "content": f"Task: {world.task.instruction}"},
+            ]
+            for turn in range(max_turns):
+                turns = turn + 1
+                content, ui, uo = _router_chat(router_base, router_key, model, messages)
+                in_tok += ui
+                out_tok += uo
+                code = _extract_code(content)
+                messages.append({"role": "assistant", "content": content})
+                if not code:
+                    messages.append({
+                        "role": "user",
+                        "content": "Reply with exactly one ```python block that makes progress, "
+                        "or call apis.supervisor.complete_task().",
+                    })
+                    continue
+                output = world.execute(code)
+                turns_log.append({"code": code[:600], "output": str(output)[:600]})
+                messages.append({"role": "user", "content": "OUTPUT:\n" + str(output)[:4000]})
+                if world.task_completed():
+                    break
+            evaluation = world.evaluate().to_dict()
+    except Exception as e:  # noqa: BLE001
+        fail(f"react of {args.task_id} failed: {e}")
+
+    if "success" not in evaluation or "num_tests" not in evaluation:
+        fail(f"evaluation dict missing success/num_tests keys: {sorted(evaluation.keys())}")
+    passes = evaluation.get("passes", [])
+    failures = evaluation.get("failures", [])
+    n_pass = len(passes) if isinstance(passes, list) else int(passes or 0)
+    n_fail = len(failures) if isinstance(failures, list) else int(failures or 0)
+    print(
+        json.dumps(
+            {
+                "success": bool(evaluation["success"]),
+                "passes": n_pass,
+                "fails": n_fail,
+                "num_tests": int(evaluation["num_tests"]),
+                "turns": turns,
+                "input_tokens": in_tok,
+                "output_tokens": out_tok,
+                "transcript": "\n---\n".join(
+                    f"CODE:\n{t['code']}\nOUTPUT:\n{t['output']}" for t in turns_log[-3:]
+                )[:1600],
+            }
+        )
+    )
+
+
 def cmd_load(args) -> None:
     try:
         from appworld import load_task_ids
@@ -103,11 +262,17 @@ def main() -> None:
     p_eval.add_argument("--task-id", required=True)
     p_eval.add_argument("--split", required=True)
 
+    p_react = sub.add_parser("react")
+    p_react.add_argument("--task-id", required=True)
+    p_react.add_argument("--split", required=True)
+
     args = ap.parse_args()
     if args.cmd == "load":
         cmd_load(args)
     elif args.cmd == "evaluate":
         cmd_evaluate(args)
+    elif args.cmd == "react":
+        cmd_react(args)
 
 
 if __name__ == "__main__":