Tim Lingo
de65991807
Teaches the OpenAI-format lane (Groq/OpenAI/Grok/Gemini/Ollama) to offer tools,
execute them, and loop — the capability that until now existed only on the
Anthropic wire. The tool-execution, consent, bridge and run-progress machinery is
reused unchanged; only the wire dialect is new.
Two pre-existing defects were found while proving it, and are fixed here because
both silently break chat:
1. PROVIDER WIRING NEVER CONNECTED. The launcher exports SOUL_LLM_PROVIDER /
SOUL_LLM_BASE_URL and puts the provider key in ANTHROPIC_API_KEY + SOUL_API_KEY;
the engine's provider fork read only NEURON_LLM_0_*, which nothing sets in a
customer build. So use_openai was ALWAYS false: every non-Anthropic user's turns
went to api.anthropic.com carrying, say, a Groq key, and came back
"llm unavailable". Proven side-by-side against the pinned round-9 brain
(sha256 15cf7d1b…): identical env, shipped brain = "llm unavailable" both chat
modes with ZERO calls to the configured endpoint; this build = a real answer,
with the probe logging POST /v1/chat/completions and Bearer <provider key>.
Fixed brain-side only (env fallbacks) — no app or launcher change needed.
2. TRUNCATION SPLITS UTF-8 CHARACTERS. The session preload cuts recalled memory at
fixed BYTE lengths (continuity snippet 350; session_preload_bullets per bullet).
A cut landing inside a multi-byte character leaves a dangling lead byte in the
SYSTEM PROMPT, making the whole request body invalid UTF-8 — providers reject it
and the user sees an unexplained failure. Captured from a real body: 18,710 bytes,
decode fails at 18,248 on 'e2', a box-drawing rule (U+2500 = E2 94 80) sliced in
half. Trigger is ordinary content — em dash, curly quote, accented name, emoji,
table border — and it gets MORE likely as memory grows. Shared code: this hit the
Anthropic wire too. Fixed with utf8_safe_slice() applied at BOTH cut sites.
WHAT IS IN THE PORT
- llm_base_url / llm_wire_format / agentic_api_key: fall back to the launcher's own
SOUL_LLM_* names; anthropic deliberately still returns "" so its native path is
untouched (endpoint configurability remains neuron#62).
- openai_tools_json(): Anthropic tool schema -> OpenAI function schema; entries with
no input_schema (Anthropic's server-side web_search) are skipped — they cannot
execute on this wire.
- agentic_tools_no_web(): the standard set minus that server tool.
- openai_agentic_loop(): forked rather than parameterised, so agentic_loop — which
carries every round-7/8/9 fix — is provably untouched. Same envelopes, same state
keys, same consent policy (ask_all / escalate / builtin / always-allow), same
client-bridge contract, same run-progress ledger, same 12-iteration cap.
- ADR-0005 mirrored on this wire: parallel_tool_calls:false is sent explicitly, and
if a provider ignores it we honour the FIRST call and echo only that one, so the
conversation we send is never self-contradictory. The drop is logged loudly.
- The assistant turn echoes the provider's own content bytes (json_get_raw), so a
JSON null stays null and nothing is lost to a decode/re-encode round trip.
- Tool results are embedded already-escaped (dispatch_tool json_safe's them);
truncation trims a dangling escape so a cut can't invalidate the body.
- bridge_save() gains a "wire" scalar and agentic_resume branches on it, so a
suspended turn resumes on the wire it suspended on. Legacy blobs (no field) resume
as anthropic. The field is read from the blob's SCALAR HEAD only — an unbounded
first-match scan would run on into messages_raw, which is model-controlled, and
that is exactly the round-9 resume defect. Pinned by a test.
- Three fork sites: handle_chat_agentic, handle_dharma_room_turn_agentic,
agentic_resume. Tool assembly is computed once per lane at both entry points
(it makes an HTTP call to the connector bridge; it was being paid for twice).
TOOLING THAT DID NOT EXIST
- tests/run-el-test.sh — engine tests were never runnable: elc is a compiler, it
emits C and exits. This emits the test to C, compiles soul.c with main renamed
away, links the rest + the repo-pinned runtime, and runs it. It also COMPUTES THE
VERDICT, because every counted test file's "N passed, M failed" summary is a
permanent 0/0 — the counters increment inside if BLOCKS, which El scoping
discards (9 files; real fix filed as neuron#116). Proven to discriminate with a
deliberately-broken assertion.
- tests/gate-openai/ — deterministic OpenAI-dialect provider stub + scenarios +
driver + hostile modes, and a strict request validator that rejects any
Anthropic-shaped field so dialect leakage fails loudly.
VERIFICATION (rungs named)
- E2E-VERIFIED against a LIVE provider (Anthropic's OpenAI-compatible endpoint,
confirmed live): real answer; a tool call whose out-of-root path was DENIED by the
guard, after which the model refused to claim success ("I won't tell you I did it,
because I didn't"); then a valid path -> file physically on disk with exact content,
honest reply, ledger with per-round entries + {done:true}.
- Deterministic lane gate: 11/12 in both consent configurations (bridge + local);
hostile providers produce no hang and no fabricated answer; the 12-iteration cap
trips with its honest message. The one FAIL is oa-tools-off and is NOT this port —
see "Known, not fixed here".
- ANTHROPIC LANE UNCHANGED: gate9 32/32 on this build and on the pinned round-9
brain; request bytes differ only within the noise band that two runs of the
UNMODIFIED brain also produce (proven with a baseline-vs-baseline control), and
the preload sections — the shared code touched here — are byte-identical.
The rig discriminates: the round-8 brain scores 24/32 on it.
- verify-soul-contract.sh: PASS (27/27 routes, no hard-deletes).
- Unit: test_bridge_serialization 36/36 (incl. 8 new wire/field-order assertions),
test_utf8_slice 18/18, test_agentic_tools 18 PASS / 0 FAIL / 3 documented skips.
KNOWN, NOT FIXED HERE (deliberate)
- Tools:Off on an OpenAI provider still fails: the non-agentic path goes through the
el-runtime provider chain, which appends /v1/chat/completions to a base URL that
already ends in /v1 -> /v1/v1/... 404. Runtime/plain-chat territory, untouched
mid-beta. Note openai_chat_complete() has zero callers — that lane is served
entirely by the runtime chain.
- The 12-iteration cap does not bound a chain of BRIDGED tools (iteration is
per-invocation and resume starts fresh). Parity with the Anthropic lane.
- run_progress resets on each resume, so a client rendering cumulative steps across a
consent pause sees earlier legs vanish. Parity with the Anthropic lane.
- verify-soul-contract.sh needs bash >= 4; under macOS's stock bash 3.2 it dies
instantly with a FALSE red ("local: -n: invalid option").
- Groq-specific live E2E not run: no Groq key exists on this machine.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
653 lines
30 KiB
Python
Executable File
653 lines
30 KiB
Python
Executable File
#!/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:<JSON-encoded STRING>}}; 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()
|