#!/usr/bin/env bash # run-lane-gate.sh — brain-side driver for the OpenAI-dialect gate. # # Drives the REAL soul binary against stub-openai.py for every class and every # phrasing in scenarios-openai.json, plus the three hostile provider modes, and # asserts the brain's claims against the stub's ground-truth JSONL (truth, not # narration) and against files on disk. # # SAFETY (hard rules, enforced below): # - never binds 7770 / 7779 / 17779 - only 7891-7894 # - never reads or writes ~/.neuron - HOME is redirected to a scratch dir # - every process started here is killed on exit (trap) and proven with lsof # # The soul runs under `script -q /dev/null` so its stdout is a pty: El's # println() uses puts(), which is FULLY buffered to a file, and the process is # killed without flushing — the DRIFT lines would be invisible otherwise. # # Usage: ./run-lane-gate.sh [all|bridge|local|toolsoff|hostile] # bridge = consent round-trip config (no workspace root -> write_file is # "escalate" -> the loop suspends and the CLIENT executes the tool) # local = workspace-root config (write_file is "reversible" + builtin -> # the loop executes the tool in-process and runs to completion) # toolsoff = supplementary: non-agentic lane against a base URL WITHOUT the # /v1 suffix (the el-runtime provider chain appends # /v1/chat/completions itself, unlike chat.el which appends only # /chat/completions) # hostile = black-hole / mid-body-drop / tool-pending-forever # # Env overrides: SOUL_BIN, STUB_PORT, SOUL_PORT, SOUL_PORT_B, RUN_ROOT set -uo pipefail HERE="$(cd "$(dirname "$0")" && pwd)" PHASES="${1:-all}" SOUL_BIN="${SOUL_BIN:-/tmp/soul-oai2/soul-openai-tools}" STUB_PORT="${STUB_PORT:-7891}" SOUL_PORT="${SOUL_PORT:-7892}" SOUL_PORT_B="${SOUL_PORT_B:-7893}" RUN_ROOT="${RUN_ROOT:-/tmp/oa-lane-gate}" STAMP="$(date +%Y%m%d-%H%M%S)" RUN="$RUN_ROOT/$STAMP" for p in "$STUB_PORT" "$SOUL_PORT" "$SOUL_PORT_B"; do case "$p" in 7770|7779|17779) echo "FATAL: refusing production Neuron port $p"; exit 2;; 789[1-4]) ;; *) echo "FATAL: port $p outside the allowed 7891-7894 range"; exit 2;; esac done [ -x "$SOUL_BIN" ] || { echo "FATAL: soul binary not found/executable: $SOUL_BIN"; exit 2; } mkdir -p "$RUN/home" "$RUN/ws-bridge" "$RUN/ws-local" "$RUN/ws-off" "$RUN/engram" echo '{"nodes":[],"edges":[]}' > "$RUN/engram/snapshot.json" DRV="$RUN/drv.py" STUB_PID=""; SOUL_PID="" cleanup() { [ -n "$SOUL_PID" ] && kill "$SOUL_PID" 2>/dev/null pkill -f "$SOUL_BIN" 2>/dev/null [ -n "$STUB_PID" ] && kill "$STUB_PID" 2>/dev/null sleep 0.4 [ -n "$SOUL_PID" ] && kill -9 "$SOUL_PID" 2>/dev/null [ -n "$STUB_PID" ] && kill -9 "$STUB_PID" 2>/dev/null return 0 } trap cleanup EXIT INT TERM start_stub() { # $1 = mode, $2 = log path local mode="$1" log="$2" args="" [ "$mode" = "normal" ] && args="--scenarios $HERE/scenarios-openai.json" # shellcheck disable=SC2086 python3 "$HERE/stub-openai.py" --port "$STUB_PORT" --mode "$mode" --log "$log" $args \ > "$RUN/stub-$mode.out" 2>&1 & STUB_PID=$! for _ in $(seq 1 50); do curl -sf "http://127.0.0.1:$STUB_PORT/gate/health" >/dev/null 2>&1 && return 0 sleep 0.2 done echo "FATAL: stub did not come up on $STUB_PORT"; cat "$RUN/stub-$mode.out"; exit 3 } stop_stub() { [ -n "$STUB_PID" ] && kill "$STUB_PID" 2>/dev/null; sleep 0.3; STUB_PID=""; } start_soul() { # $1 = port, $2 = base url, $3 = soul log, $4 = agent root ("" = none) local port="$1" base="$2" log="$3" root="$4" script -q /dev/null \ env -u ANTHROPIC_API_KEY -u SOUL_API_KEY -u ENGRAM_URL -u ENGRAM_API_KEY \ -u NEURON_API_URL -u NEURON_TOKEN -u SOUL_LLM_PROVIDER -u SOUL_LLM_BASE_URL \ -u NEURON_LLM_1_URL -u NEURON_LLM_1_KEY -u SOUL_IDENTITY \ HOME="$RUN/home" PATH="$PATH" \ NEURON_PORT="$port" EL_HTTP_BIND_HOST=127.0.0.1 \ SOUL_ENGRAM_PATH="$RUN/engram/snapshot.json" \ SOUL_CGI_ID=ntn-test SOUL_PERSONA_NAME=Neuron \ NEURON_LLM_0_URL="$base" NEURON_LLM_0_FORMAT=openai NEURON_LLM_0_KEY=gate-test-key \ ${root:+NEURON_AGENT_ROOT="$root"} \ "$SOUL_BIN" > "$log" 2>&1 & SOUL_PID=$! for _ in $(seq 1 100); do curl -sf "http://127.0.0.1:$port/health" >/dev/null 2>&1 && return 0 sleep 0.2 done echo "FATAL: soul did not come up on $port"; tail -20 "$log"; exit 3 } stop_soul() { [ -n "$SOUL_PID" ] && kill "$SOUL_PID" 2>/dev/null pkill -f "$SOUL_BIN" 2>/dev/null sleep 0.6; SOUL_PID="" } # ------------------------------------------------------------------ driver ---- cat > "$DRV" <<'PYEOF' import json, os, sys, time, threading, urllib.request, urllib.error CFG = json.load(open(sys.argv[1])) SOUL = "http://127.0.0.1:%d" % CFG["soul_port"] STUB = "http://127.0.0.1:%d" % CFG["stub_port"] WS = CFG["workspace"] MODE = CFG["mode"] # bridge | local | toolsoff SCEN = json.load(open(CFG["scenarios"])) STUBLOG = CFG["stub_log"] SOULLOG = CFG["soul_log"] ONLY = CFG.get("classes") or list(SCEN["classes"].keys()) MAXHOPS = CFG.get("max_hops", 15) OUT = CFG["out"] # the chat-only class must be driven on the NON-agentic door: the agentic door # always advertises tools, which is a 400 on that scenario by contract. NON_AGENTIC = {"oa-tools-off"} def http(method, url, obj=None, timeout=300): data = None if obj is None else json.dumps(obj).encode() req = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=timeout) as r: body = r.read().decode("utf-8", "replace") st = r.status except urllib.error.HTTPError as e: body = e.read().decode("utf-8", "replace"); st = e.code except Exception as e: return -1, "TRANSPORT-ERROR: %r" % (e,), None try: return st, body, json.loads(body) except ValueError: return st, body, None def fsize(p): return os.path.getsize(p) if os.path.exists(p) else 0 def tail_from(path, off): if not os.path.exists(path): return "", off with open(path, "rb") as f: f.seek(off); chunk = f.read(); return chunk.decode("utf-8", "replace"), f.tell() def stub_since(off): """Exact correlation: only the JSONL bytes appended during this phrasing.""" txt, noff = tail_from(STUBLOG, off) recs = [] for line in txt.splitlines(): line = line.strip() if line: try: recs.append(json.loads(line)) except ValueError: pass return recs, noff def perform(name, ti): """Execute the bridged tool for real, like the desktop client would.""" if name in ("write_file", "edit_file"): p = ti.get("path", "") dest = p if os.path.isabs(p) else os.path.join(WS, p) os.makedirs(os.path.dirname(dest) or WS, exist_ok=True) body = ti.get("content", "") with open(dest, "w") as f: f.write(body) return "wrote %s (%d bytes)" % (p, len(body.encode())) return "ok" class Poller(threading.Thread): def __init__(self, sid): super().__init__(daemon=True); self.sid = sid; self.snaps = []; self.stop = False def run(self): while not self.stop: st, body, js = http("GET", SOUL + "/api/run-progress/" + self.sid, timeout=60) if js and js.get("progress"): if not self.snaps or self.snaps[-1] != js["progress"]: self.snaps.append(js["progress"]) time.sleep(0.1) def progress(sid): _, _, pj = http("GET", SOUL + "/api/run-progress/" + sid, timeout=30) return (pj or {}).get("progress") def run_phrasing(cname, ph): st, body, js = http("POST", SOUL + "/api/sessions", {"title": ph["id"]}, timeout=60) sid = (js or {}).get("id", "") rec = {"class": cname, "phrasing": ph["id"], "session_id": sid, "legs": [], "pendings": [], "progress_during": [], "progress_per_leg": [], "progress_final": None, "soul_log": "", "stub": [], "http": [], "agentic": cname not in NON_AGENTIC} if not sid: rec["fatal"] = "session create failed: %s %s" % (st, body[:300]); return rec soff = fsize(SOULLOG); loff = fsize(STUBLOG) t0 = time.time() pol = Poller(sid); pol.start() payload = {"message": ph["prompt"], "session_id": sid, "workspace_root": WS, "agentic": rec["agentic"]} if MODE == "local": payload["agent_workspace_root"] = WS st, body, js = http("POST", SOUL + "/api/chat", payload, timeout=CFG.get("chat_timeout", 240)) rec["http"].append(st) rec["legs"].append(js if js is not None else body[:600]) rec["progress_per_leg"].append(progress(sid)) hops = 0 while isinstance(js, dict) and js.get("tool_pending") and hops < MAXHOPS: rec["pendings"].append({"call_id": js.get("call_id"), "tool_name": js.get("tool_name"), "tool_input": js.get("tool_input"), "risk_tier": js.get("risk_tier"), "narration": js.get("narration"), "tools_used": js.get("tools_used")}) try: eff = perform(js.get("tool_name", ""), js.get("tool_input") or {}) except Exception as e: eff = "client error: %r" % (e,) st, body, js = http("POST", SOUL + "/api/sessions/%s/tool_result" % sid, {"call_id": js.get("call_id"), "content": eff}, timeout=CFG.get("chat_timeout", 240)) rec["http"].append(st) rec["legs"].append(js if js is not None else body[:600]) rec["progress_per_leg"].append(progress(sid)) hops += 1 pol.stop = True; time.sleep(0.3) t1 = time.time() rec["elapsed"] = round(t1 - t0, 2) rec["progress_during"] = pol.snaps rec["progress_final"] = progress(sid) rec["soul_log"], _ = tail_from(SOULLOG, soff) rec["stub"], _ = stub_since(loff) rec["hops"] = hops return rec # ------------------------------------------------------------- assertions ---- def expected_calls(cname, first_only=False): out = [] for step in SCEN["classes"][cname]["script"]: for k, call in enumerate(step.get("tool_calls") or []): if first_only and k > 0: continue out.append((call["name"], call["arguments"])) return out def final_text(cname): for step in reversed(SCEN["classes"][cname]["script"]): if step.get("text") and not step.get("tool_calls"): return step["text"] return None def judge(rec): cname = rec["class"]; ok = []; bad = [] last = rec["legs"][-1] if rec["legs"] else None reply = last.get("reply") if isinstance(last, dict) else None err = last.get("error") if isinstance(last, dict) else None tools_used = last.get("tools_used") if isinstance(last, dict) else None stub = rec["stub"] scen_recs = [r for r in stub if r.get("kind") == "scenario"] rejects = [r for r in stub if r.get("validation") != "ok"] bg = [r for r in stub if r.get("kind") in ("wrong_path", "background")] def wire_clean(): if rejects: for r in rejects: bad.append("stub REJECTED a request: [%s] %s" % (r.get("validation"), r.get("validation_detail"))) else: ok.append("stub ground truth: validation \"ok\" on all %d scenario leg(s), no " "gate_echo_mismatch / gate_tool_call_shape / dialect-leak 400s" % len(scen_recs)) if bg: ok.append("NOTE background non-scenario request(s) in this window: %s" % [(r.get("kind"), r.get("path"), r.get("http_status")) for r in bg]) if cname == "oa-plain": wire_clean() want = final_text(cname) if reply == want: ok.append("final reply == scripted final text (byte-exact)") else: bad.append("final reply mismatch:\n WANT: %r\n GOT : %r" % (want, reply)) if tools_used == []: ok.append("tools_used == [] (no tool ran)") else: bad.append("tools_used expected [] got %r" % (tools_used,)) if reply and ('"tool_calls"' in reply or '"function"' in reply or '"tool_use"' in reply): bad.append("tool-call JSON leaked into the reply text") else: ok.append("no tool-call JSON anywhere in the reply") elif cname in ("oa-single-tool", "oa-torture", "oa-mission"): wire_clean() want = final_text(cname) if reply == want: ok.append("final reply == scripted final text (byte-exact)") else: bad.append("final reply mismatch:\n WANT: %r\n GOT : %r" % (want, reply)) exp = expected_calls(cname) wantnames = [n for n, _ in exp] if tools_used == wantnames: ok.append("tools_used == %r (carried across %d suspension(s))" % (wantnames, rec["hops"])) else: bad.append("tools_used expected %r got %r" % (wantnames, tools_used)) for name, args in exp: p = args.get("path"); c = args.get("content") dest = os.path.join(WS, p) if not os.path.exists(dest): bad.append("expected file missing on disk: %s" % dest); continue got = open(dest, "rb").read() if got == c.encode(): ok.append("%s on disk is byte-for-byte the issued payload (%d bytes)" % (p, len(got))) else: bad.append("%s content differs\n WANT %r\n GOT %r" % (p, c[:300], got[:300].decode("utf-8", "replace"))) if MODE == "bridge": for pend, (name, args) in zip(rec["pendings"], exp): if pend["tool_input"] == args: ok.append("tool_input for %s survived exactly ONE decode (deep-equal to the " "issued arguments; no double-escaping)" % name) else: bad.append("tool_input != issued arguments for %s\n WANT %r\n GOT %r" % (name, args, pend["tool_input"])) if rec["pendings"] and all(p["risk_tier"] == "escalate" for p in rec["pendings"]): ok.append("every write_file classified \"escalate\" and bridged for consent") elif cname == "oa-parallel": drift = [l.strip() for l in rec["soul_log"].splitlines() if "DRIFT: provider returned" in l] if drift: ok.append("soul log: " + drift[0]) else: bad.append("no 'DRIFT: provider returned N parallel tool_calls' line in the soul log") delivered = [r for r in stub if r.get("delivered", {}).get("tool_calls")] if delivered and len(delivered[0]["delivered"]["tool_calls"]) == 2: ok.append("stub delivered 2 parallel tool_calls in one response (ground truth)") if MODE == "bridge": if len(rec["pendings"]) == 1: ok.append("exactly ONE call honored: %s" % rec["pendings"][0]["call_id"]) else: bad.append("expected exactly 1 honored call, got %d" % len(rec["pendings"])) pairing = [r for r in rejects if "gate_pairing" in str(r.get("validation_detail")) or "tool_calls at end of thread" in str(r.get("validation_detail")) or "not fully answered" in str(r.get("validation_detail"))] for r in pairing: ok.append("EXPECTED-BY-CONTRACT stub 400 on the unpaired echo: %s" % r.get("validation_detail")) other = [r for r in rejects if r not in pairing] for r in other: bad.append("unexpected stub rejection: [%s] %s" % (r.get("validation"), r.get("validation_detail"))) if err and not reply: ok.append("honest error envelope after the 400 (no fabricated answer): %r" % err) elif reply == final_text(cname): ok.append("final reply == scripted final text (both calls paired)") else: bad.append("neither an honest error nor the scripted final text: %r" % (last,)) elif cname == "oa-api-error": if err and not reply: ok.append("honest error envelope: error=%r reply=%r" % (err, reply)) else: bad.append("expected an error envelope with an empty reply, got %r" % (last,)) delivered = [r["delivered"].get("api_error") for r in stub if r.get("delivered")] ok.append("stub delivered api_error status(es): %r" % [d for d in delivered if d]) n = len([r for r in stub if r.get("kind") == "scenario"]) ok.append("provider hit %d time(s) - no retry storm" % n) if reply: bad.append("FABRICATED ANSWER: reply non-empty on a provider error") elif cname == "oa-tools-off": ok.append("stub records for this phrasing: %r" % [{k: r.get(k) for k in ("kind", "path", "validation", "http_status")} for r in stub]) wrong = [r for r in stub if r.get("kind") == "wrong_path"] matched = [r for r in stub if r.get("scenario_class") == cname] if matched and not rejects: ok.append("chat-only request reached /v1/chat/completions with NO tools offered") want = final_text(cname) if reply == want: ok.append("final reply == scripted final text (byte-exact)") else: bad.append("final reply mismatch:\n WANT: %r\n GOT : %r" % (want, reply)) if reply and ('"tool_calls"' in reply or '"function"' in reply): bad.append("tool-call JSON leaked into the reply text") else: ok.append("no tool-call JSON in the reply") elif wrong: bad.append("the non-agentic lane never reached the provider endpoint: stub saw " "%s -> %s (the el-runtime provider chain appends /v1/chat/completions " "to NEURON_LLM_0_URL, chat.el appends only /chat/completions)" % (wrong[0]["path"], wrong[0]["http_status"])) elif not stub: bad.append("no request reached the stub at all") else: for r in rejects: bad.append("stub REJECTED: [%s] %s" % (r.get("validation"), r.get("validation_detail"))) return ok, bad def main(): results = [] for cname in ONLY: for ph in SCEN["classes"][cname]["phrasings"]: rec = run_phrasing(cname, ph) ok, bad = judge(rec) rec["ok"] = ok; rec["bad"] = bad rec["verdict"] = "FAIL" if bad else "PASS" results.append(rec) print("=" * 78) print("[%s] %s / %s (%.2fs, %d bridge hop(s), agentic=%s, mode=%s)" % (rec["verdict"], cname, ph["id"], rec.get("elapsed", 0), rec.get("hops", 0), rec["agentic"], MODE)) for l in ok: print(" ok " + l.replace("\n", "\n ")) for l in bad: print(" FAIL " + l.replace("\n", "\n ")) for i, leg in enumerate(rec["legs"]): print(" leg%d envelope: %s" % (i, json.dumps(leg)[:430])) for i, pr in enumerate(rec["progress_per_leg"]): print(" run-progress after leg%d: %s" % (i, json.dumps(pr)[:380])) if rec["progress_during"]: print(" run-progress polled DURING (%d distinct snapshot(s)), last: %s" % (len(rec["progress_during"]), json.dumps(rec["progress_during"][-1])[:300])) for r in rec["stub"]: print(" stub: kind=%s class=%s phrasing=%s step=%s validation=%s%s delivered=%s http=%s" % (r.get("kind"), r.get("scenario_class"), r.get("phrasing"), r.get("step"), r.get("validation"), ("(" + str(r.get("validation_detail")) + ")") if r.get("validation_detail") else "", json.dumps(r.get("delivered")), r.get("http_status"))) if rec["soul_log"].strip(): for l in rec["soul_log"].splitlines(): if l.strip(): print(" soul: " + l.strip()) json.dump(results, open(OUT, "w"), indent=1) npass = sum(1 for r in results if r["verdict"] == "PASS") print("=" * 78) print("PHASE %s: %d/%d PASS" % (MODE, npass, len(results))) for r in results: print(" %-6s %-16s %s" % (r["verdict"], r["class"], r["phrasing"])) return 0 if npass == len(results) else 1 sys.exit(main()) PYEOF # ------------------------------------------------------------- hostile drv --- cat > "$RUN/hostile.py" <<'PYEOF' import json, os, sys, time, urllib.request, urllib.error CFG = json.load(open(sys.argv[1])) SOUL = "http://127.0.0.1:%d" % CFG["soul_port"] STUB = "http://127.0.0.1:%d" % CFG["stub_port"] def http(method, url, obj=None, timeout=400): data = None if obj is None else json.dumps(obj).encode() req = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=timeout) as r: b = r.read().decode("utf-8", "replace"); st = r.status except urllib.error.HTTPError as e: b = e.read().decode("utf-8", "replace"); st = e.code except Exception as e: return -1, "TRANSPORT-ERROR: %r" % (e,), None try: return st, b, json.loads(b) except ValueError: return st, b, None mode = CFG["mode"]; wsmode = CFG["ws_mode"]; WS = CFG["workspace"] _, _, js = http("POST", SOUL + "/api/sessions", {"title": "hostile-" + mode}, timeout=60) sid = (js or {}).get("id", "") payload = {"message": "oa-gate plain probe: hostile mode %s" % mode, "agentic": True, "session_id": sid, "workspace_root": WS} if wsmode == "local": payload["agent_workspace_root"] = WS t0 = time.time() st, body, js = http("POST", SOUL + "/api/chat", payload, timeout=CFG.get("timeout", 400)) t_first = time.time() - t0 legs = [js if js is not None else body[:500]] hops = 0 while isinstance(js, dict) and js.get("tool_pending") and hops < CFG.get("max_hops", 14): ti = js.get("tool_input") or {} p = ti.get("path", "x.md") dest = p if os.path.isabs(p) else os.path.join(WS, p) try: open(dest, "w").write(ti.get("content", "")) except Exception: pass st, body, js = http("POST", SOUL + "/api/sessions/%s/tool_result" % sid, {"call_id": js.get("call_id"), "content": "ok"}, timeout=CFG.get("timeout", 400)) legs.append(js if js is not None else body[:500]); hops += 1 el = time.time() - t0 _, _, stats = http("GET", STUB + "/gate/stats", timeout=30) _, _, prog = http("GET", SOUL + "/api/run-progress/" + sid, timeout=30) fab = [l for l in legs if isinstance(l, dict) and l.get("reply")] print("HOSTILE %s (ws_mode=%s)" % (mode, wsmode)) print(" first /api/chat POST returned after %.2fs; whole chain %.2fs; client bridge hops=%d; " "stub chat_hits=%s" % (t_first, el, hops, (stats or {}).get("chat_hits"))) print(" first envelope : " + json.dumps(legs[0])[:430]) print(" final envelope : " + json.dumps(legs[-1])[:430]) print(" non-empty replies anywhere in the chain (fabrication check): %d" % len(fab)) print(" run-progress : " + json.dumps(prog)[:300]) json.dump({"mode": mode, "ws_mode": wsmode, "t_first": t_first, "elapsed": el, "hops": hops, "chat_hits": (stats or {}).get("chat_hits"), "legs": legs, "progress": prog}, open(CFG["out"], "w"), indent=1) PYEOF # ------------------------------------------------------------------ phases --- RC_BRIDGE=0; RC_LOCAL=0; RC_OFF=0 run_normal_phase() { # $1 = label, $2 = soul port, $3 = agent root, $4 = ws, $5 = base, $6 = classes json local m="$1" port="$2" root="$3" ws="$4" base="$5" classes="$6" echo; echo "############ PHASE: $m (soul :$port, NEURON_LLM_0_URL=$base) ############" start_stub normal "$RUN/stub-$m.jsonl" start_soul "$port" "$base" "$RUN/soul-$m.log" "$root" cat > "$RUN/cfg-$m.json" < "$RUN/cfg-$mode-$wsm.json" </dev/null || echo " (no listeners - ports free)" echo echo "############ SUMMARY ############" echo "run dir: $RUN" echo "bridge rc=$RC_BRIDGE local rc=$RC_LOCAL toolsoff rc=$RC_OFF (0 = every class PASS)" exit $(( RC_BRIDGE + RC_LOCAL + RC_OFF ))