#!/usr/bin/env python3 """stub-openai.py - deterministic local stand-in for an OpenAI-format /v1/chat/completions provider, for the soul-openai-tools-v2 gate (docs/specs/SPEC-soul-openai-tools-v2-2026-08-06.md). No API key, no network, no model. Sibling of gate9's stub-llm.py (Anthropic dialect, _wt-beta-round9/scripts/ gate9/): same scenario mechanism (marker matching, assistant-count step indexing, ground-truth JSONL log, prod-port refusal), different wire. Staging home is tests/gate-openai/ in _wt-openai-tools; folds into scripts/gate9/ after round 9 merges (see README.md). WHAT IT DOES * Serves POST /v1/chat/completions on 127.0.0.1 only (OpenAI dialect). * VALIDATES every request - this is the gate's discriminator, built BEFORE the brain-side El code exists so dialect leakage fails loudly: - Anthropic tells are 400 code=gate_dialect_leak: `anthropic-version` header; top-level `system` / `stop_sequences` / `max_tokens_to_sample` / `anthropic_version`; `input_schema` inside a tool entry; Anthropic content blocks (tool_use / tool_result / server_tool_use / ...). - tools[] must be OpenAI-shaped {type:"function", function:{name, description, parameters}} with unique names -> 400 gate_tools_shape. - assistant tool_calls echoes must be {id, type:"function", function:{name, arguments:}}; a decoded-object `arguments` is a wire bug -> 400 gate_tool_call_shape. - every assistant tool_calls turn must be answered by role:"tool" messages covering EVERY tool_call_id, immediately following; unknown / duplicate / missing ids -> 400 gate_pairing. - echoed `arguments` for gate-issued call ids (call_gate_*) are recomputed from the script and compared after ONE json decode -> 400 gate_echo_mismatch. This is the two-escaper-trap discriminator named in the spec's security model (section 6). - scenario-level request expectations from scenarios-openai.json (tools offered, OpenAI-shaped tool_choice, parallel_tool_calls pinned false per ADR-0005) -> 400 gate_expect. * Answers with SCRIPTED responses: plain text (finish_reason "stop"), tool calls (finish_reason "tool_calls", arguments JSON-encoded, incl. a nested-quote/escaping torture payload and a parallel two-call case), and API-error injection (OpenAI error envelope). Scenario is selected by scanning user-message text (newest first) for a registered marker substring; the step index is the number of assistant messages already in the request (stateless replay - resumes index correctly by construction). * Writes a ground-truth JSONL log (--log): one record per request with the validation verdict, matched scenario/step, and exactly which tool calls were delivered. Gate assertions compare the brain's claims against THIS log - truth, not narration. * Unmatched requests (boot probes, awareness chatter) get a benign "ok" text response, logged kind=background, never counted as ground truth. * HOSTILE MODES (--mode) on the same file: black-hole accept + read the request, never respond; mid-body-drop send half a JSON body, then abort the socket; tool-pending-forever every request gets a FRESH tool_call (finish_reason "tool_calls"), forever - tests the agentic loop's iteration cap; count the brain's round-trips via GET /gate/stats. usage: stub-openai.py --port P --scenarios scenarios-openai.json \ --log requests.jsonl [--mode MODE] Listens on 127.0.0.1 only. Refuses production ports 7770/7779/17779. """ import argparse import itertools import json import socket import struct import threading import time import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer STATE = {"scenarios": None, "log_path": None, "lock": threading.Lock(), "seq": 0, "mode": "normal", "chat_hits": 0} _PENDING_SEQ = itertools.count(1) ANTHROPIC_TOP_KEYS = ("system", "stop_sequences", "max_tokens_to_sample", "anthropic_version") ANTHROPIC_BLOCK_TYPES = {"tool_use", "tool_result", "server_tool_use", "web_search_tool_result", "thinking", "redacted_thinking"} DEFAULT_EXPECT = {"require_tools": True, "require_tool_choice": True, "parallel_tool_calls": False, "forbid_tools": False} # ---------------------------------------------------------------- loading ---- def load_scenarios(path): cfg = json.load(open(path)) defaults = dict(DEFAULT_EXPECT) defaults.update(cfg.get("defaults", {}).get("expect_request", {})) marker_map = [] # (marker_lower, cname, pid) scripts = {} # cname or cname/pid -> expanded script pid_map = {} # pid -> cname (for call_gate_* id -> script lookup) expects = {} # cname -> merged expect_request for cname, cls in cfg["classes"].items(): scripts[cname] = expand_script(cls.get("script", [])) exp = dict(defaults) exp.update(cls.get("expect_request", {})) expects[cname] = exp for ph in cls["phrasings"]: if ph.get("script") is not None: scripts[cname + "/" + ph["id"]] = expand_script(ph["script"]) marker_map.append((ph["marker"].lower(), cname, ph["id"])) pid_map[ph["id"]] = cname return {"cfg": cfg, "marker_map": marker_map, "scripts": scripts, "pid_map": pid_map, "expects": expects} def expand_script(script): """Same repeat-expansion contract as gate9's stub-llm.py ({N}/{NN}).""" out = [] for step in script: if "repeat" in step: for n in range(1, step["repeat"] + 1): t = {k: v for k, v in step.items() if k != "repeat"} out.append(json.loads(json.dumps(t) .replace("{NN}", "%02d" % n) .replace("{N}", str(n)))) else: out.append(step) return out # ------------------------------------------------------------- validation ---- def _rej(message, code): return {"status": 400, "message": message, "code": code} def validate_dialect(headers, req): """Universal checks - run on EVERY request, scenario-matched or not. Anything Anthropic-shaped on this lane means the brain's translator leaked; the whole point is that it fails loudly, here, with a reason.""" if headers.get("anthropic-version"): return _rej("anthropic-version header on the OpenAI lane: this " "request was built by the Anthropic dialect path", "gate_dialect_leak") for k in ANTHROPIC_TOP_KEYS: if k in req: return _rej("top-level `%s` is Anthropic dialect; the OpenAI " "dialect has no such field (system prompt goes in " "messages[0])" % k, "gate_dialect_leak") tools = req.get("tools") if tools is not None: if not isinstance(tools, list): return _rej("`tools` must be an array", "gate_tools_shape") names = [] for i, t in enumerate(tools): if not isinstance(t, dict): return _rej("tools[%d] is not an object" % i, "gate_tools_shape") if "input_schema" in t or (isinstance(t.get("function"), dict) and "input_schema" in t["function"]): return _rej("tools[%d] carries `input_schema` (Anthropic " "dialect); OpenAI dialect wants " "function.parameters" % i, "gate_dialect_leak") if t.get("type") != "function": return _rej("tools[%d].type must be \"function\", got %r" % (i, t.get("type")), "gate_tools_shape") fn = t.get("function") if not isinstance(fn, dict): return _rej("tools[%d].function missing" % i, "gate_tools_shape") if not isinstance(fn.get("name"), str) or not fn["name"]: return _rej("tools[%d].function.name missing/empty" % i, "gate_tools_shape") if not isinstance(fn.get("description"), str) or not fn["description"]: return _rej("tools[%d].function.description missing/empty" % i, "gate_tools_shape") if not isinstance(fn.get("parameters"), dict): return _rej("tools[%d].function.parameters missing (JSON " "Schema object expected)" % i, "gate_tools_shape") names.append(fn["name"]) if len(names) != len(set(names)): return _rej("tools: tool names must be unique", "gate_tools_shape") msgs = req.get("messages") if not isinstance(msgs, list) or not msgs: return _rej("`messages` must be a non-empty array", "gate_messages_shape") for i, m in enumerate(msgs): if not isinstance(m, dict): return _rej("messages[%d] is not an object" % i, "gate_messages_shape") c = m.get("content") if isinstance(c, list): for j, b in enumerate(c): if isinstance(b, dict) and b.get("type") in ANTHROPIC_BLOCK_TYPES: return _rej("messages[%d].content[%d] is an Anthropic " "`%s` block; the OpenAI dialect uses " "tool_calls / role:\"tool\" messages" % (i, j, b.get("type")), "gate_dialect_leak") if m.get("role") == "tool": if not isinstance(m.get("tool_call_id"), str) or not m["tool_call_id"]: return _rej("messages[%d]: role \"tool\" requires a " "`tool_call_id`" % i, "gate_messages_shape") if "content" not in m: return _rej("messages[%d]: role \"tool\" requires `content`" % i, "gate_messages_shape") if m.get("role") == "assistant" and m.get("tool_calls") is not None: tcs = m["tool_calls"] if not isinstance(tcs, list) or not tcs: return _rej("messages[%d].tool_calls must be a non-empty " "array" % i, "gate_tool_call_shape") for j, tc in enumerate(tcs): if not isinstance(tc, dict) or tc.get("type") != "function": return _rej("messages[%d].tool_calls[%d].type must be " "\"function\"" % (i, j), "gate_tool_call_shape") if not isinstance(tc.get("id"), str) or not tc["id"]: return _rej("messages[%d].tool_calls[%d].id missing" % (i, j), "gate_tool_call_shape") fn = tc.get("function") if not isinstance(fn, dict) or not isinstance(fn.get("name"), str): return _rej("messages[%d].tool_calls[%d].function.name " "missing" % (i, j), "gate_tool_call_shape") if not isinstance(fn.get("arguments"), str): return _rej("messages[%d].tool_calls[%d].function." "arguments must be a JSON-encoded STRING, " "got %s" % (i, j, type(fn.get("arguments")).__name__), "gate_tool_call_shape") return None def validate_pairing(msgs): """OpenAI pairing rule: every assistant tool_calls turn must be followed immediately by role:"tool" messages answering every tool_call_id.""" open_ids, open_at = set(), None for i, m in enumerate(msgs): role = m.get("role") if role == "tool": tid = m.get("tool_call_id") if open_at is None: return _rej("messages[%d]: role \"tool\" message with no " "preceding assistant tool_calls turn " "(tool_call_id=%s)" % (i, tid), "gate_pairing") if tid not in open_ids: return _rej("messages[%d]: tool message answers unknown or " "already-answered tool_call_id %s" % (i, tid), "gate_pairing") open_ids.discard(tid) continue if open_ids: return _rej("messages[%d]: assistant tool_calls not fully " "answered before messages[%d]; missing tool " "responses for: %s" % (open_at, i, sorted(open_ids)), "gate_pairing") open_ids, open_at = set(), None if role == "assistant" and m.get("tool_calls"): ids = [tc.get("id") for tc in m["tool_calls"]] open_ids, open_at = set(ids), i if open_ids: return _rej("messages[%d]: assistant tool_calls at end of thread " "without tool responses for: %s" % (open_at, sorted(open_ids)), "gate_pairing") return None def validate_echo_args(msgs, loaded): """Ground-truth round-trip check: for every echoed gate-issued call id, recompute the arguments this stub originally sent from the script and require one json decode to reproduce them exactly. Catches the two-escaper trap (spec section 6) deterministically.""" if not loaded: return None for i, m in enumerate(msgs): if m.get("role") != "assistant": continue for tc in m.get("tool_calls") or []: tid = tc.get("id", "") if not tid.startswith("call_gate_"): continue rest = tid[len("call_gate_"):] try: pid, s_part, k_part = rest.rsplit("_", 2) step_idx, k = int(s_part[1:]), int(k_part) except (ValueError, IndexError): continue cname = loaded["pid_map"].get(pid) if cname is None: continue script = (loaded["scripts"].get(cname + "/" + pid) or loaded["scripts"].get(cname) or []) if step_idx >= len(script): continue calls = script[step_idx].get("tool_calls") or [] if k >= len(calls): continue expected = calls[k] fn = tc.get("function") or {} if fn.get("name") != expected["name"]: return _rej("messages[%d]: echoed tool name %r != issued %r " "for %s" % (i, fn.get("name"), expected["name"], tid), "gate_echo_mismatch") try: got = json.loads(fn.get("arguments", "")) except ValueError: return _rej("messages[%d]: echoed arguments for %s are not " "valid JSON after one decode (truncated or " "half-escaped?)" % (i, tid), "gate_echo_mismatch") if got != expected["arguments"]: hint = (" (decoded to a string, not an object: " "double-encoded - the two-escaper trap)" if isinstance(got, str) else "") return _rej("messages[%d]: echoed arguments for %s do not " "round-trip to the issued payload%s" % (i, tid, hint), "gate_echo_mismatch") return None def validate_expect(req, exp): """Scenario-level request expectations (scenarios-openai.json).""" tools = req.get("tools") or [] if exp.get("forbid_tools") and tools: return _rej("this scenario is chat-only: no `tools` may be offered " "on it", "gate_expect") if exp.get("require_tools") and not tools: return _rej("scenario expects a `tools` array to be offered (the " "agentic lane must advertise its tools)", "gate_expect") if exp.get("require_tool_choice"): tc = req.get("tool_choice") ok = tc in ("auto", "none", "required") or ( isinstance(tc, dict) and tc.get("type") == "function" and isinstance(tc.get("function"), dict) and tc["function"].get("name")) if not ok: return _rej("scenario expects an OpenAI-shaped `tool_choice`, " "got %r" % (tc,), "gate_expect") want_ptc = exp.get("parallel_tool_calls", None) if want_ptc is not None: if "parallel_tool_calls" not in req: return _rej("scenario expects explicit `parallel_tool_calls` " "(ADR-0005: must be pinned false on the wire)", "gate_expect") if req["parallel_tool_calls"] != want_ptc: return _rej("scenario expects parallel_tool_calls=%s, got %s" % (json.dumps(want_ptc), json.dumps(req["parallel_tool_calls"])), "gate_expect") return None # --------------------------------------------------------- scenario match ---- def extract_user_texts_newest_first(msgs): texts = [] for m in reversed(msgs): if not isinstance(m, dict) or m.get("role") != "user": continue c = m.get("content") if isinstance(c, str): texts.append(c) elif isinstance(c, list): for b in c: if isinstance(b, dict) and b.get("type") == "text": texts.append(b.get("text", "")) return texts def match_scenario(loaded, msgs): for text in extract_user_texts_newest_first(msgs): tl = text.lower() for marker, cname, pid in loaded["marker_map"]: if marker in tl: return cname, pid return None, None # ------------------------------------------------------------- rendering ---- def completion_envelope(msg, finish, model, usage=(100, 100)): return {"id": "chatcmpl-gate-" + uuid.uuid4().hex[:12], "object": "chat.completion", "created": int(time.time()), "model": model, "choices": [{"index": 0, "message": msg, "finish_reason": finish, "logprobs": None}], "usage": {"prompt_tokens": usage[0], "completion_tokens": usage[1], "total_tokens": usage[0] + usage[1]}} def text_completion(text, model): return completion_envelope({"role": "assistant", "content": text}, "stop", model, usage=(1, 1)) def pending_body(seq, model): args = json.dumps({"path": "never-%04d.md" % seq, "content": "this run never completes"}) msg = {"role": "assistant", "content": None, "tool_calls": [{"id": "call_hostile_pending_%04d" % seq, "type": "function", "function": {"name": "write_file", "arguments": args}}]} return completion_envelope(msg, "tool_calls", model, usage=(1, 1)) def render_step(step, cname, pid, step_idx, model): """Returns (http_status, body_dict, delivered) - delivered is ground truth for the JSONL log.""" delivered = {"tool_calls": [], "finish_reason": None, "api_error": None} if "api_error" in step: e = step["api_error"] delivered["api_error"] = e["status"] return (e["status"], {"error": {"message": e["message"], "type": e.get("type", "server_error"), "param": None, "code": e.get("code")}}, delivered) msg = {"role": "assistant"} finish = "stop" if step.get("tool_calls"): tcs = [] for k, call in enumerate(step["tool_calls"]): tid = "call_gate_%s_s%d_%d" % (pid, step_idx, k) tcs.append({"id": tid, "type": "function", "function": {"name": call["name"], "arguments": json.dumps( call["arguments"], ensure_ascii=False)}}) delivered["tool_calls"].append(call["name"]) msg["tool_calls"] = tcs msg["content"] = step.get("text") # null when no narration, like real finish = "tool_calls" else: msg["content"] = step["text"] delivered["finish_reason"] = finish return 200, completion_envelope(msg, finish, model), delivered # ------------------------------------------------------------------ log ------ def log_record(rec): with STATE["lock"]: STATE["seq"] += 1 rec["seq"] = STATE["seq"] with open(STATE["log_path"], "a") as f: f.write(json.dumps(rec, ensure_ascii=False) + "\n") # ---------------------------------------------------------------- server ----- class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def _send_json(self, status, obj): body = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _send_error(self, verdict): self._send_json(verdict["status"], {"error": {"message": verdict["message"], "type": "invalid_request_error", "param": None, "code": verdict["code"]}}) def _drop_mid_body(self): """Valid 200 headers, half the promised body, then a socket abort (same SO_LINGER teardown as gate9's mid-body-drop-brain.py).""" full = json.dumps(text_completion( "This reply will never finish arriving because the connection " "dies in the middle of the body, which is exactly the point of " "this hostile fixture.", "hostile-mid-drop")).encode("utf-8") half = full[: len(full) // 2] self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(full))) # promises more self.end_headers() self.wfile.write(half) self.wfile.flush() try: self.connection.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0)) self.connection.shutdown(socket.SHUT_RDWR) except OSError: pass self.close_connection = True def do_GET(self): path = self.path.split("?")[0] if path == "/gate/health": self._send_json(200, {"ok": True, "mode": STATE["mode"]}) elif path == "/gate/stats": with STATE["lock"]: self._send_json(200, {"mode": STATE["mode"], "chat_hits": STATE["chat_hits"]}) else: self._send_json(404, {"error": {"message": "not found", "type": "invalid_request_error", "param": None, "code": "unknown_route"}}) def do_POST(self): n = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(n) mode = STATE["mode"] rec = {"ts": time.time(), "path": self.path, "mode": mode, "kind": "background", "scenario_class": None, "phrasing": None, "step": None, "n_messages": 0, "n_assistant": 0, "validation": "ok", "validation_detail": None, "delivered": {"tool_calls": [], "finish_reason": None, "api_error": None}, "http_status": 200} if self.path.split("?")[0] != "/v1/chat/completions": rec.update(kind="wrong_path", http_status=404) log_record(rec) self._send_json(404, {"error": { "message": "no such route: %s" % self.path, "type": "invalid_request_error", "param": None, "code": "unknown_route"}}) return with STATE["lock"]: STATE["chat_hits"] += 1 # ---- hostile modes: behavior first, no validation ---------------- if mode == "black-hole": rec.update(kind="hostile", http_status=None) log_record(rec) threading.Event().wait() # hold the socket open forever return if mode == "mid-body-drop": rec.update(kind="hostile", http_status=200) log_record(rec) self._drop_mid_body() return if mode == "tool-pending-forever": seq = next(_PENDING_SEQ) rec.update(kind="hostile", delivered={"tool_calls": ["write_file"], "finish_reason": "tool_calls", "api_error": None}) log_record(rec) self._send_json(200, pending_body(seq, "gate-openai-model")) return # ---- normal mode ------------------------------------------------- try: req = json.loads(raw) except ValueError as exc: # DIAGNOSTIC CAPTURE (2026-08-06): an unparseable body used to be recorded as # a bare "bad_json" with the bytes thrown away, which made an intermittent # failure impossible to root-cause — you cannot fix what you did not keep. # Dump the raw body next to the log, and record exactly where the parser gave # up plus the offending byte, so one occurrence is enough to diagnose. dump_path = "%s.badbody.%s" % (STATE.get("log_path", "/tmp/stub-openai"), rec.get("seq", "x")) try: data = raw if isinstance(raw, (bytes, bytearray)) else str(raw).encode() with open(dump_path, "wb") as fh: fh.write(data) except Exception as dump_exc: dump_path = "(dump failed: %s)" % dump_exc pos = getattr(exc, "pos", None) near = "" byte_repr = "" if isinstance(pos, int): blob = raw if isinstance(raw, (bytes, bytearray)) else str(raw).encode() near = blob[max(0, pos - 60):pos + 60].decode("utf-8", "replace") if 0 <= pos < len(blob): byte_repr = "0x%02x" % blob[pos] rec.update(kind="bad_json", validation="rejected", validation_detail="request body is not valid JSON: %s" % exc, http_status=400, raw_len=len(raw), raw_dump=dump_path, err_pos=pos, err_byte=byte_repr, err_near=near) log_record(rec) self._send_error(_rej("request body is not valid JSON", "bad_json")) return msgs = req.get("messages") or [] rec["n_messages"] = len(msgs) rec["n_assistant"] = sum(1 for m in msgs if isinstance(m, dict) and m.get("role") == "assistant") loaded = STATE["scenarios"] cname, pid = match_scenario(loaded, msgs) if cname: rec.update(kind="scenario", scenario_class=cname, phrasing=pid) # Wire-level validation runs for EVERY request, scenario or not. verdict = (validate_dialect(self.headers, req) or validate_pairing([m for m in msgs if isinstance(m, dict)]) or validate_echo_args(msgs, loaded)) if verdict: rec.update(validation="rejected", validation_detail=verdict["message"], http_status=verdict["status"]) log_record(rec) self._send_error(verdict) return model = req.get("model", "gate-openai-model") if not cname: log_record(rec) self._send_json(200, text_completion("ok", model)) return script = (loaded["scripts"].get(cname + "/" + pid) or loaded["scripts"][cname]) step_idx = rec["n_assistant"] if step_idx >= len(script): rec.update(kind="overrun", step=step_idx) log_record(rec) self._send_json(200, text_completion( "GATE-SCRIPT-EXHAUSTED %s step %d" % (pid, step_idx), model)) return step = script[step_idx] exp = dict(loaded["expects"][cname]) exp.update(step.get("expect_request", {})) verdict = validate_expect(req, exp) if verdict: rec.update(step=step_idx, validation="rejected", validation_detail=verdict["message"], http_status=verdict["status"]) log_record(rec) self._send_error(verdict) return status, body, delivered = render_step(step, cname, pid, step_idx, model) rec.update(step=step_idx, delivered=delivered, http_status=status) log_record(rec) self._send_json(status, body) def log_message(self, *a): pass def main(): ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, required=True) ap.add_argument("--scenarios", help="scenarios-openai.json (required in normal mode)") ap.add_argument("--log", required=True) ap.add_argument("--mode", default="normal", choices=["normal", "black-hole", "mid-body-drop", "tool-pending-forever"]) args = ap.parse_args() if args.port in (7770, 7779, 17779): raise SystemExit("stub-openai: refusing production Neuron port") if args.mode == "normal" and not args.scenarios: raise SystemExit("stub-openai: --scenarios is required in normal mode") STATE["mode"] = args.mode STATE["scenarios"] = (load_scenarios(args.scenarios) if args.scenarios else None) STATE["log_path"] = args.log open(args.log, "w").close() n_markers = (len(STATE["scenarios"]["marker_map"]) if STATE["scenarios"] else 0) print("stub-openai [%s]: 127.0.0.1:%d /v1/chat/completions " "(%d markers registered, log=%s)" % (args.mode, args.port, n_markers, args.log), flush=True) ThreadingHTTPServer(("127.0.0.1", args.port), Handler).serve_forever() if __name__ == "__main__": main()