serve: OpenAI-compatible front door (harness serve)
OpenWebUI becomes the interactive surface by talking to MOPAC like any
OpenAI provider: GET /v1/models lists the servable catalog (one model per
[models.classes] class, named mopac-<class>, routed through the same
tier table `once` uses; unknown model = 400 naming the valid ones) and
POST /v1/chat/completions runs ONE bounded stateless conductor turn over
the sent conversation history — no session storage, tools hard-off,
non-streaming (stream:true gets an explicit 400; OWUI tolerates
non-streaming providers). Bearer vkey auth compares SHA-256 digests in
constant time; missing/wrong keys get one byte-identical 401 body, and
the vkey never reaches logs or responses. temperature/max_tokens are
forwarded upstream; usage is summed across rounds and returned in the
reply. Upstream failures surface as a terse 502. Own port (:8090
default) so it coexists with the events receiver; dev.sh gets a serve
runner publishing 8090 on the LAN. Tests drive a scripted fake OpenAI
upstream through the real HTTP server: auth matrix, catalog + subset,
history assembly (client system message preserved, harness identity
prepended only when missing), multi-round usage accounting, refused
tool-call feedback, knob forwarding, 400/502 paths.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
This commit is contained in:
@@ -0,0 +1,362 @@
|
||||
// Package serve is the `harness serve` OpenAI-compatible front door for
|
||||
// OpenWebUI (DESIGN "OWUI front door"; any OpenAI client works):
|
||||
//
|
||||
// GET /v1/models the servable model catalog
|
||||
// POST /v1/chat/completions ONE bounded stateless conductor turn
|
||||
// GET /healthz liveness
|
||||
//
|
||||
// The class -> tier map in harness.toml is the catalog: every [models.classes]
|
||||
// class is exposed as a model named mopac-<class> and routes through the same
|
||||
// tier table `once` uses. v0 is STATELESS (the client sends the full
|
||||
// conversation history each call; no session storage), NON-streaming, and
|
||||
// tools are OFF — a pure chat path over the shared conductor turn machinery.
|
||||
package serve
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"crypto/subtle"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"ukrrs.com/mopac/harness/internal/config"
|
||||
"ukrrs.com/mopac/harness/internal/llm"
|
||||
"ukrrs.com/mopac/harness/internal/loop"
|
||||
"ukrrs.com/mopac/harness/internal/models"
|
||||
)
|
||||
|
||||
// maxBodyBytes bounds request payloads; OWUI sends the whole conversation
|
||||
// history per call, so this is roomier than the webhook receiver's cap.
|
||||
const maxBodyBytes = 4 << 20
|
||||
|
||||
// ModelPrefix is the servable-model namespace: class "code" is served as
|
||||
// ModelPrefix + "code" = "mopac-code".
|
||||
const ModelPrefix = "mopac-"
|
||||
|
||||
// Turner is the conductor-facing turn sink; *loop.Conductor implements it
|
||||
// (same machinery `once` uses, tools off).
|
||||
type Turner interface {
|
||||
ServeTurn(ctx context.Context, msgs []llm.Message, model string, opts loop.ServeTurnOpts) (*loop.TurnResult, error)
|
||||
}
|
||||
|
||||
// Server is the `harness serve` front door. Every route requires a valid
|
||||
// Bearer vkey (constant-time compare); anything else gets 401 with a single
|
||||
// generic body. The vkey is resolved at startup and never logged.
|
||||
type Server struct {
|
||||
cfg config.ServeConfig
|
||||
catalog map[string]models.Decision // model name -> routing decision
|
||||
names []string // sorted servable model names
|
||||
vkey string
|
||||
turner Turner
|
||||
logger *log.Logger
|
||||
started int64
|
||||
}
|
||||
|
||||
// NewServer resolves the vkey (fail-fast) and builds the servable catalog
|
||||
// from the router's class map — each class becomes ModelPrefix+class and
|
||||
// carries the tier/model it routes to — optionally narrowed to
|
||||
// [serve] enabled_models.
|
||||
func NewServer(cfg config.ServeConfig, router *models.Router, turner Turner, out io.Writer) (*Server, error) {
|
||||
if cfg.VKeyRef == "" {
|
||||
return nil, fmt.Errorf("[serve]: vkey_ref is required (the bearer key OWUI connections present)")
|
||||
}
|
||||
vkey, err := config.ResolveKeyRef(cfg.VKeyRef)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("[serve]: resolve vkey: %w", err)
|
||||
}
|
||||
catalog := make(map[string]models.Decision)
|
||||
for _, class := range router.Classes() {
|
||||
d, err := router.Resolve(class)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("serve catalog: %w", err)
|
||||
}
|
||||
catalog[ModelPrefix+class] = d
|
||||
}
|
||||
names := make([]string, 0, len(catalog))
|
||||
for name := range catalog {
|
||||
names = append(names, name)
|
||||
}
|
||||
sort.Strings(names)
|
||||
if len(cfg.EnabledModels) > 0 {
|
||||
keep := make(map[string]bool, len(cfg.EnabledModels))
|
||||
for _, m := range cfg.EnabledModels {
|
||||
if _, ok := catalog[m]; !ok {
|
||||
return nil, fmt.Errorf("[serve]: enabled_models entry %q is not servable (valid: %s)", m, strings.Join(names, ", "))
|
||||
}
|
||||
keep[m] = true
|
||||
}
|
||||
filtered := make([]string, 0, len(keep))
|
||||
for _, n := range names {
|
||||
if keep[n] {
|
||||
filtered = append(filtered, n)
|
||||
} else {
|
||||
delete(catalog, n)
|
||||
}
|
||||
}
|
||||
names = filtered
|
||||
}
|
||||
if len(names) == 0 {
|
||||
return nil, fmt.Errorf("[serve]: model catalog is empty (define [models.classes] in harness.toml)")
|
||||
}
|
||||
return &Server{
|
||||
cfg: cfg,
|
||||
catalog: catalog,
|
||||
names: names,
|
||||
vkey: vkey,
|
||||
turner: turner,
|
||||
logger: log.New(out, "serve: ", log.LstdFlags|log.Lmsgprefix),
|
||||
started: time.Now().Unix(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Handler builds the front door's HTTP routes.
|
||||
func (s *Server) Handler() http.Handler {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprint(w, `{"status":"ok"}`)
|
||||
})
|
||||
mux.HandleFunc("/v1/models", s.handleModels)
|
||||
mux.HandleFunc("/v1/chat/completions", s.handleChat)
|
||||
return mux
|
||||
}
|
||||
|
||||
// ModelNames returns the sorted servable model names (startup banner, docs).
|
||||
func (s *Server) ModelNames() []string {
|
||||
return append([]string(nil), s.names...)
|
||||
}
|
||||
|
||||
func (s *Server) handleModels(w http.ResponseWriter, r *http.Request) {
|
||||
if !s.authorize(r) {
|
||||
s.logger.Printf("rejected models remote=%s reason=auth", r.RemoteAddr)
|
||||
s.rejectAuth(w)
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodGet {
|
||||
w.Header().Set("Allow", http.MethodGet)
|
||||
s.writeError(w, http.StatusMethodNotAllowed, "GET only")
|
||||
return
|
||||
}
|
||||
data := make([]modelEntry, 0, len(s.names))
|
||||
for _, name := range s.names {
|
||||
data = append(data, modelEntry{ID: name, Object: "model", Created: s.started, OwnedBy: "mopac"})
|
||||
}
|
||||
s.writeJSON(w, http.StatusOK, modelList{Object: "list", Data: data})
|
||||
}
|
||||
|
||||
func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) {
|
||||
start := time.Now()
|
||||
if !s.authorize(r) {
|
||||
s.logger.Printf("rejected chat remote=%s reason=auth", r.RemoteAddr)
|
||||
s.rejectAuth(w)
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodPost {
|
||||
w.Header().Set("Allow", http.MethodPost)
|
||||
s.writeError(w, http.StatusMethodNotAllowed, "POST only")
|
||||
return
|
||||
}
|
||||
body, ok := s.readBody(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var req inboundRequest
|
||||
if err := json.Unmarshal(body, &req); err != nil {
|
||||
s.writeError(w, http.StatusBadRequest, "malformed request body")
|
||||
return
|
||||
}
|
||||
if req.Stream {
|
||||
s.writeError(w, http.StatusBadRequest, "streaming is not supported; set stream to false")
|
||||
return
|
||||
}
|
||||
decision, ok := s.catalog[req.Model]
|
||||
if !ok {
|
||||
s.writeError(w, http.StatusBadRequest,
|
||||
fmt.Sprintf("unknown model %q; valid models: %s", req.Model, strings.Join(s.names, ", ")))
|
||||
return
|
||||
}
|
||||
if len(req.Messages) == 0 {
|
||||
s.writeError(w, http.StatusBadRequest, "messages must not be empty")
|
||||
return
|
||||
}
|
||||
if req.MaxTokens < 0 {
|
||||
s.writeError(w, http.StatusBadRequest, "max_tokens must not be negative")
|
||||
return
|
||||
}
|
||||
msgs := make([]llm.Message, 0, len(req.Messages))
|
||||
for i, m := range req.Messages {
|
||||
switch m.Role {
|
||||
case "system", "user", "assistant":
|
||||
default:
|
||||
s.writeError(w, http.StatusBadRequest, fmt.Sprintf("messages[%d]: unsupported role %q", i, m.Role))
|
||||
return
|
||||
}
|
||||
msgs = append(msgs, llm.Message{Role: m.Role, Content: m.Content})
|
||||
}
|
||||
|
||||
turn, err := s.turner.ServeTurn(r.Context(), msgs, decision.Model, loop.ServeTurnOpts{
|
||||
MaxTokens: req.MaxTokens,
|
||||
Temperature: req.Temperature,
|
||||
})
|
||||
if err != nil {
|
||||
// Log the failure class only; upstream bodies never reach the client.
|
||||
s.logger.Printf("chat model=%s remote=%s error=upstream duration=%s", req.Model, r.RemoteAddr, time.Since(start).Round(time.Millisecond))
|
||||
s.writeError(w, http.StatusBadGateway, "upstream model error")
|
||||
return
|
||||
}
|
||||
// One audit line per request: routing + counters, no message contents.
|
||||
s.logger.Printf("chat model=%s tier=%s concrete=%s rounds=%d tools_refused=%d tokens=%d duration=%s remote=%s",
|
||||
req.Model, decision.Tier, decision.Model, turn.Rounds, turn.Denied, turn.TotalTokens,
|
||||
time.Since(start).Round(time.Millisecond), r.RemoteAddr)
|
||||
s.writeJSON(w, http.StatusOK, chatCompletion{
|
||||
ID: newCompletionID(),
|
||||
Object: "chat.completion",
|
||||
Created: time.Now().Unix(),
|
||||
Model: req.Model,
|
||||
Choices: []outboundChoice{{
|
||||
Index: 0,
|
||||
Message: outboundMessage{Role: "assistant", Content: turn.Content},
|
||||
FinishReason: "stop",
|
||||
}},
|
||||
Usage: usageOut{
|
||||
PromptTokens: turn.PromptTokens,
|
||||
CompletionTokens: turn.CompletionTokens,
|
||||
TotalTokens: turn.TotalTokens,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// authorize checks the Bearer vkey in constant time. The 401 body is byte-
|
||||
// identical for missing, malformed, and wrong keys (no oracle).
|
||||
func (s *Server) authorize(r *http.Request) bool {
|
||||
const prefix = "Bearer "
|
||||
h := r.Header.Get("Authorization")
|
||||
if len(h) <= len(prefix) || !strings.EqualFold(h[:len(prefix)], prefix) {
|
||||
return false
|
||||
}
|
||||
token := h[len(prefix):]
|
||||
// Compare SHA-256 digests so the comparison time does not depend on the
|
||||
// presented key length.
|
||||
got := sha256.Sum256([]byte(token))
|
||||
want := sha256.Sum256([]byte(s.vkey))
|
||||
return subtle.ConstantTimeCompare(got[:], want[:]) == 1
|
||||
}
|
||||
|
||||
func (s *Server) rejectAuth(w http.ResponseWriter) {
|
||||
s.writeError(w, http.StatusUnauthorized, "invalid api key")
|
||||
}
|
||||
|
||||
func (s *Server) readBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) {
|
||||
body, err := io.ReadAll(io.LimitReader(r.Body, maxBodyBytes+1))
|
||||
if err != nil {
|
||||
s.writeError(w, http.StatusBadRequest, "unreadable body")
|
||||
return nil, false
|
||||
}
|
||||
if len(body) > maxBodyBytes {
|
||||
s.writeError(w, http.StatusRequestEntityTooLarge, "payload too large")
|
||||
return nil, false
|
||||
}
|
||||
return body, true
|
||||
}
|
||||
|
||||
func (s *Server) writeJSON(w http.ResponseWriter, code int, v any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(code)
|
||||
_ = json.NewEncoder(w).Encode(v)
|
||||
}
|
||||
|
||||
// writeError emits an OpenAI-shaped error body.
|
||||
func (s *Server) writeError(w http.ResponseWriter, code int, msg string) {
|
||||
s.writeJSON(w, code, errorBody{Error: errorDetail{Message: msg, Type: errorType(code)}})
|
||||
}
|
||||
|
||||
func errorType(code int) string {
|
||||
switch {
|
||||
case code == http.StatusUnauthorized:
|
||||
return "authentication_error"
|
||||
case code >= 500:
|
||||
return "server_error"
|
||||
default:
|
||||
return "invalid_request_error"
|
||||
}
|
||||
}
|
||||
|
||||
// inboundRequest is the accepted OpenAI chat-completions request subset:
|
||||
// model, messages, temperature, max_tokens. Streaming is rejected; tools
|
||||
// are ignored (v0 serve turns are pure chat).
|
||||
type inboundRequest struct {
|
||||
Model string `json:"model"`
|
||||
Messages []inboundMessage `json:"messages"`
|
||||
Temperature *float64 `json:"temperature"`
|
||||
MaxTokens int `json:"max_tokens"`
|
||||
Stream bool `json:"stream"`
|
||||
}
|
||||
|
||||
type inboundMessage struct {
|
||||
Role string `json:"role"`
|
||||
Content string `json:"content"`
|
||||
}
|
||||
|
||||
type modelEntry struct {
|
||||
ID string `json:"id"`
|
||||
Object string `json:"object"`
|
||||
Created int64 `json:"created"`
|
||||
OwnedBy string `json:"owned_by"`
|
||||
}
|
||||
|
||||
type modelList struct {
|
||||
Object string `json:"object"`
|
||||
Data []modelEntry `json:"data"`
|
||||
}
|
||||
|
||||
type outboundMessage struct {
|
||||
Role string `json:"role"`
|
||||
Content string `json:"content"`
|
||||
}
|
||||
|
||||
type outboundChoice struct {
|
||||
Index int `json:"index"`
|
||||
Message outboundMessage `json:"message"`
|
||||
FinishReason string `json:"finish_reason"`
|
||||
}
|
||||
|
||||
type usageOut struct {
|
||||
PromptTokens int `json:"prompt_tokens"`
|
||||
CompletionTokens int `json:"completion_tokens"`
|
||||
TotalTokens int `json:"total_tokens"`
|
||||
}
|
||||
|
||||
type chatCompletion struct {
|
||||
ID string `json:"id"`
|
||||
Object string `json:"object"`
|
||||
Created int64 `json:"created"`
|
||||
Model string `json:"model"`
|
||||
Choices []outboundChoice `json:"choices"`
|
||||
Usage usageOut `json:"usage"`
|
||||
}
|
||||
|
||||
type errorDetail struct {
|
||||
Message string `json:"message"`
|
||||
Type string `json:"type"`
|
||||
}
|
||||
|
||||
type errorBody struct {
|
||||
Error errorDetail `json:"error"`
|
||||
}
|
||||
|
||||
// newCompletionID mints an OpenAI-shaped id (chatcmpl-<hex>).
|
||||
func newCompletionID() string {
|
||||
var b [12]byte
|
||||
if _, err := rand.Read(b[:]); err != nil {
|
||||
return fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano())
|
||||
}
|
||||
return "chatcmpl-" + hex.EncodeToString(b[:])
|
||||
}
|
||||
@@ -0,0 +1,474 @@
|
||||
package serve
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"ukrrs.com/mopac/harness/internal/config"
|
||||
"ukrrs.com/mopac/harness/internal/llm"
|
||||
"ukrrs.com/mopac/harness/internal/loop"
|
||||
)
|
||||
|
||||
const testVKey = "owui-test-vkey"
|
||||
|
||||
// fakeUpstream is a scripted OpenAI-compatible upstream that records every
|
||||
// request (the "fake OpenAI client against the real server" pattern).
|
||||
type fakeUpstream struct {
|
||||
mu sync.Mutex
|
||||
requests []llm.ChatRequest
|
||||
srv *httptest.Server
|
||||
}
|
||||
|
||||
func (f *fakeUpstream) handler(t *testing.T, bodies ...string) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
var req llm.ChatRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
t.Errorf("decode upstream request: %v", err)
|
||||
}
|
||||
f.mu.Lock()
|
||||
f.requests = append(f.requests, req)
|
||||
n := len(f.requests)
|
||||
f.mu.Unlock()
|
||||
body := bodies[len(bodies)-1]
|
||||
if n <= len(bodies) {
|
||||
body = bodies[n-1]
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprint(w, body)
|
||||
})
|
||||
}
|
||||
|
||||
func textBody(content string, prompt, completion int) string {
|
||||
return fmt.Sprintf(`{"choices":[{"message":{"role":"assistant","content":%q},"finish_reason":"stop"}],"usage":{"prompt_tokens":%d,"completion_tokens":%d,"total_tokens":%d}}`,
|
||||
content, prompt, completion, prompt+completion)
|
||||
}
|
||||
|
||||
func (f *fakeUpstream) request(i int) llm.ChatRequest {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.requests[i]
|
||||
}
|
||||
|
||||
func (f *fakeUpstream) count() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return len(f.requests)
|
||||
}
|
||||
|
||||
// newStack wires fake upstream + conductor + serve server, the same way
|
||||
// `harness serve` does in main.
|
||||
func newStack(t *testing.T, bodies ...string) (*Server, *fakeUpstream, *httptest.Server) {
|
||||
t.Helper()
|
||||
up := &fakeUpstream{}
|
||||
up.srv = httptest.NewServer(up.handler(t, bodies...))
|
||||
t.Cleanup(up.srv.Close)
|
||||
|
||||
dir := t.TempDir()
|
||||
cfg := config.Default()
|
||||
cfg.Vertical = "teststack"
|
||||
cfg.WorkRoot = dir
|
||||
cfg.ReportDir = filepath.Join(dir, "reports")
|
||||
cfg.LiteLLM.BaseURL = up.srv.URL
|
||||
cfg.LiteLLM.KeyRef = "literal:upstream-key"
|
||||
cfg.LiteLLM.MaxRetries = 0
|
||||
cfg.Models.Tiers = map[string]string{
|
||||
"mopac-study": "glm-4.7-flash",
|
||||
"mopac-code": "glm-5.2",
|
||||
"mopac-review": "glm-5-turbo",
|
||||
"mopac-primary": "glm-5.3",
|
||||
}
|
||||
cfg.Models.Classes = map[string]string{
|
||||
"study": "mopac-study",
|
||||
"read": "mopac-study",
|
||||
"code": "mopac-code",
|
||||
"primary": "mopac-primary",
|
||||
}
|
||||
cfg.Serve.VKeyRef = "literal:" + testVKey
|
||||
|
||||
cond, err := loop.New(cfg, io.Discard)
|
||||
if err != nil {
|
||||
t.Fatalf("loop.New: %v", err)
|
||||
}
|
||||
srv, err := NewServer(cfg.Serve, cond.Router(), cond, io.Discard)
|
||||
if err != nil {
|
||||
t.Fatalf("NewServer: %v", err)
|
||||
}
|
||||
ts := httptest.NewServer(srv.Handler())
|
||||
t.Cleanup(ts.Close)
|
||||
return srv, up, ts
|
||||
}
|
||||
|
||||
func do(t *testing.T, method, url, auth string, body string) (int, string) {
|
||||
t.Helper()
|
||||
var rd io.Reader
|
||||
if body != "" {
|
||||
rd = strings.NewReader(body)
|
||||
}
|
||||
req, err := http.NewRequest(method, url, rd)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if auth != "" {
|
||||
req.Header.Set("Authorization", auth)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
b, _ := io.ReadAll(resp.Body)
|
||||
return resp.StatusCode, string(b)
|
||||
}
|
||||
|
||||
func bearer(key string) string { return "Bearer " + key }
|
||||
|
||||
func TestAuth(t *testing.T) {
|
||||
_, _, ts := newStack(t, textBody("hi", 5, 5))
|
||||
chatURL := ts.URL + "/v1/chat/completions"
|
||||
body := `{"model":"mopac-code","messages":[{"role":"user","content":"hi"}]}`
|
||||
modelsURL := ts.URL + "/v1/models"
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
url string
|
||||
method string
|
||||
auth string
|
||||
body string
|
||||
wantCode int
|
||||
}{
|
||||
{"chat no auth header", chatURL, http.MethodPost, "", body, 401},
|
||||
{"chat malformed scheme", chatURL, http.MethodPost, "Basic " + testVKey, body, 401},
|
||||
{"chat wrong key", chatURL, http.MethodPost, bearer("wrong-key"), body, 401},
|
||||
{"chat right key", chatURL, http.MethodPost, bearer(testVKey), body, 200},
|
||||
{"models no auth", modelsURL, http.MethodGet, "", "", 401},
|
||||
{"models wrong key", modelsURL, http.MethodGet, bearer("nope"), "", 401},
|
||||
{"models right key", modelsURL, http.MethodGet, bearer(testVKey), "", 200},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
code, respBody := do(t, tc.method, tc.url, tc.auth, tc.body)
|
||||
if code != tc.wantCode {
|
||||
t.Errorf("code = %d, want %d (body %s)", code, tc.wantCode, respBody)
|
||||
}
|
||||
if code == 401 {
|
||||
if !strings.Contains(respBody, "invalid api key") {
|
||||
t.Errorf("401 body = %q, want the generic message", respBody)
|
||||
}
|
||||
if strings.Contains(respBody, testVKey) {
|
||||
t.Errorf("401 body leaks the vkey: %s", respBody)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestModelsCatalog(t *testing.T) {
|
||||
srv, _, ts := newStack(t)
|
||||
code, body := do(t, http.MethodGet, ts.URL+"/v1/models", bearer(testVKey), "")
|
||||
if code != 200 {
|
||||
t.Fatalf("code = %d body %s", code, body)
|
||||
}
|
||||
var list modelList
|
||||
if err := json.Unmarshal([]byte(body), &list); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
var ids []string
|
||||
for _, m := range list.Data {
|
||||
ids = append(ids, m.ID)
|
||||
if m.Object != "model" || m.OwnedBy != "mopac" || m.Created == 0 {
|
||||
t.Errorf("entry shape wrong: %+v", m)
|
||||
}
|
||||
}
|
||||
want := "mopac-code mopac-primary mopac-read mopac-study"
|
||||
if strings.Join(ids, " ") != want {
|
||||
t.Errorf("catalog = %v, want %v", ids, want)
|
||||
}
|
||||
if srv.ModelNames()[0] != "mopac-code" {
|
||||
t.Errorf("ModelNames() = %v", srv.ModelNames())
|
||||
}
|
||||
}
|
||||
|
||||
func TestModelsCatalogEnabledSubset(t *testing.T) {
|
||||
// Rebuild the stack with enabled_models narrowed to two classes.
|
||||
up := &fakeUpstream{}
|
||||
up.srv = httptest.NewServer(up.handler(t, textBody("x", 1, 1)))
|
||||
defer up.srv.Close()
|
||||
cfg := config.Default()
|
||||
cfg.Vertical = "teststack"
|
||||
cfg.LiteLLM.BaseURL = up.srv.URL
|
||||
cfg.LiteLLM.KeyRef = "literal:k"
|
||||
cfg.Models.Tiers = map[string]string{"mopac-study": "glm-4.7-flash", "mopac-code": "glm-5.2", "mopac-primary": "glm-5.3"}
|
||||
cfg.Models.Classes = map[string]string{"study": "mopac-study", "code": "mopac-code", "primary": "mopac-primary"}
|
||||
cfg.Serve.VKeyRef = "literal:" + testVKey
|
||||
cfg.Serve.EnabledModels = []string{"mopac-code"}
|
||||
cond, err := loop.New(cfg, io.Discard)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
srv, err := NewServer(cfg.Serve, cond.Router(), cond, io.Discard)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ts := httptest.NewServer(srv.Handler())
|
||||
defer ts.Close()
|
||||
|
||||
code, body := do(t, http.MethodGet, ts.URL+"/v1/models", bearer(testVKey), "")
|
||||
if code != 200 {
|
||||
t.Fatalf("code = %d body %s", code, body)
|
||||
}
|
||||
var list modelList
|
||||
if err := json.Unmarshal([]byte(body), &list); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(list.Data) != 1 || list.Data[0].ID != "mopac-code" {
|
||||
t.Errorf("catalog = %+v, want only mopac-code", list.Data)
|
||||
}
|
||||
// A disabled model must now be unknown.
|
||||
code, body = do(t, http.MethodPost, ts.URL+"/v1/chat/completions", bearer(testVKey),
|
||||
`{"model":"mopac-study","messages":[{"role":"user","content":"x"}]}`)
|
||||
if code != 400 || !strings.Contains(body, "unknown model") {
|
||||
t.Errorf("disabled model: code=%d body=%s, want 400 unknown model", code, body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChatEndToEnd(t *testing.T) {
|
||||
_, up, ts := newStack(t, textBody("the serve reply", 20, 40))
|
||||
code, body := do(t, http.MethodPost, ts.URL+"/v1/chat/completions", bearer(testVKey),
|
||||
`{"model":"mopac-code","messages":[{"role":"user","content":"write a function"}]}`)
|
||||
if code != 200 {
|
||||
t.Fatalf("code = %d body %s", code, body)
|
||||
}
|
||||
var out chatCompletion
|
||||
if err := json.Unmarshal([]byte(body), &out); err != nil {
|
||||
t.Fatalf("decode: %v (%s)", err, body)
|
||||
}
|
||||
if !strings.HasPrefix(out.ID, "chatcmpl-") || out.Object != "chat.completion" {
|
||||
t.Errorf("id/object wrong: %q %q", out.ID, out.Object)
|
||||
}
|
||||
if out.Model != "mopac-code" {
|
||||
t.Errorf("response model = %q, want the requested name echoed", out.Model)
|
||||
}
|
||||
if len(out.Choices) != 1 || out.Choices[0].Message.Role != "assistant" ||
|
||||
out.Choices[0].Message.Content != "the serve reply" || out.Choices[0].FinishReason != "stop" {
|
||||
t.Errorf("choices wrong: %+v", out.Choices)
|
||||
}
|
||||
if out.Usage.PromptTokens != 20 || out.Usage.CompletionTokens != 40 || out.Usage.TotalTokens != 60 {
|
||||
t.Errorf("usage wrong: %+v", out.Usage)
|
||||
}
|
||||
// Routing: the request the upstream saw must carry the CONCRETE model
|
||||
// resolved through the tier map, and no tools.
|
||||
req := up.request(0)
|
||||
if req.Model != "glm-5.2" {
|
||||
t.Errorf("upstream model = %q, want glm-5.2 (mopac-code tier)", req.Model)
|
||||
}
|
||||
if len(req.Tools) != 0 {
|
||||
t.Errorf("upstream tools = %+v, want none (v0 serve turns are pure chat)", req.Tools)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChatHistoryAssembly(t *testing.T) {
|
||||
// OWUI sends the whole conversation each call. The system message is
|
||||
// the client's own (no harness prepend) and the history order is
|
||||
// preserved verbatim; only when no system message leads does the
|
||||
// conductor inject the vertical identity prompt.
|
||||
_, up, ts := newStack(t, textBody("reply", 1, 1))
|
||||
payload := `{"model":"mopac-primary","messages":[
|
||||
{"role":"system","content":"You are the test persona."},
|
||||
{"role":"user","content":"first question"},
|
||||
{"role":"assistant","content":"first answer"},
|
||||
{"role":"user","content":"second question"}
|
||||
]}`
|
||||
if code, body := do(t, http.MethodPost, ts.URL+"/v1/chat/completions", bearer(testVKey), payload); code != 200 {
|
||||
t.Fatalf("code = %d body %s", code, body)
|
||||
}
|
||||
got := up.request(0).Messages
|
||||
if len(got) != 4 {
|
||||
t.Fatalf("upstream history length = %d, want 4 (client system wins)", len(got))
|
||||
}
|
||||
wantRoles := "system user assistant user"
|
||||
var roles []string
|
||||
for _, m := range got {
|
||||
roles = append(roles, m.Role)
|
||||
}
|
||||
if strings.Join(roles, " ") != wantRoles {
|
||||
t.Errorf("roles = %v, want %v", roles, wantRoles)
|
||||
}
|
||||
if got[0].Content != "You are the test persona." || got[3].Content != "second question" {
|
||||
t.Errorf("history contents wrong: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChatPrependsSystemWhenMissing(t *testing.T) {
|
||||
_, up, ts := newStack(t, textBody("r", 1, 1))
|
||||
if code, body := do(t, http.MethodPost, ts.URL+"/v1/chat/completions", bearer(testVKey),
|
||||
`{"model":"mopac-study","messages":[{"role":"user","content":"hi"}]}`); code != 200 {
|
||||
t.Fatalf("code = %d body %s", code, body)
|
||||
}
|
||||
got := up.request(0).Messages
|
||||
if len(got) != 2 || got[0].Role != "system" || got[1].Role != "user" {
|
||||
t.Fatalf("messages = %+v, want [system, user]", got)
|
||||
}
|
||||
if !strings.Contains(got[0].Content, "teststack") {
|
||||
t.Errorf("prepended system prompt lacks vertical identity: %q", got[0].Content)
|
||||
}
|
||||
// flash tier routes through the same class map
|
||||
if up.request(0).Model != "glm-4.7-flash" {
|
||||
t.Errorf("upstream model = %q, want glm-4.7-flash (mopac-study tier)", up.request(0).Model)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChatUsageAccountingAcrossRounds(t *testing.T) {
|
||||
// A model that hallucinates a tool call first (tools are off) then
|
||||
// answers: usage must be the SUM over both rounds.
|
||||
toolCall := `{"choices":[{"message":{"role":"assistant","content":"","tool_calls":[{"id":"c1","type":"function","function":{"name":"bash","arguments":"{\"command\":\"pwd\"}"}}]},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":10,"completion_tokens":5,"total_tokens":15}}`
|
||||
_, up, ts := newStack(t, toolCall, textBody("final text", 30, 25))
|
||||
code, body := do(t, http.MethodPost, ts.URL+"/v1/chat/completions", bearer(testVKey),
|
||||
`{"model":"mopac-primary","messages":[{"role":"user","content":"go"}]}`)
|
||||
if code != 200 {
|
||||
t.Fatalf("code = %d body %s", code, body)
|
||||
}
|
||||
if up.count() != 2 {
|
||||
t.Fatalf("upstream calls = %d, want 2", up.count())
|
||||
}
|
||||
// The refused tool call must come back as a tool result, never executed.
|
||||
second := up.request(1).Messages
|
||||
last := second[len(second)-1]
|
||||
if last.Role != "tool" || !strings.Contains(last.Content, "tools are disabled") {
|
||||
t.Errorf("tool result = %+v, want the disabled refusal", last)
|
||||
}
|
||||
var out chatCompletion
|
||||
if err := json.Unmarshal([]byte(body), &out); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if out.Usage.PromptTokens != 40 || out.Usage.CompletionTokens != 30 || out.Usage.TotalTokens != 70 {
|
||||
t.Errorf("usage = %+v, want sums over both rounds (40/30/70)", out.Usage)
|
||||
}
|
||||
if out.Choices[0].Message.Content != "final text" {
|
||||
t.Errorf("content = %q", out.Choices[0].Message.Content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChatForwardsKnobs(t *testing.T) {
|
||||
_, up, ts := newStack(t, textBody("r", 1, 1))
|
||||
if code, body := do(t, http.MethodPost, ts.URL+"/v1/chat/completions", bearer(testVKey),
|
||||
`{"model":"mopac-code","messages":[{"role":"user","content":"hi"}],"temperature":0.3,"max_tokens":256}`); code != 200 {
|
||||
t.Fatalf("code = %d body %s", code, body)
|
||||
}
|
||||
req := up.request(0)
|
||||
if req.Temperature == nil || *req.Temperature != 0.3 {
|
||||
t.Errorf("temperature = %v, want 0.3 forwarded", req.Temperature)
|
||||
}
|
||||
if req.MaxTokens != 256 {
|
||||
t.Errorf("max_tokens = %d, want 256 forwarded", req.MaxTokens)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChatBadRequestTable(t *testing.T) {
|
||||
_, _, ts := newStack(t, textBody("unused", 1, 1))
|
||||
url := ts.URL + "/v1/chat/completions"
|
||||
cases := []struct {
|
||||
name string
|
||||
body string
|
||||
wantCode int
|
||||
wantBody string
|
||||
}{
|
||||
{
|
||||
"unknown model names valid ones",
|
||||
`{"model":"gpt-4","messages":[{"role":"user","content":"x"}]}`,
|
||||
400, `valid models: mopac-code, mopac-primary, mopac-read, mopac-study`,
|
||||
},
|
||||
{"stream requested", `{"model":"mopac-code","stream":true,"messages":[{"role":"user","content":"x"}]}`, 400, "streaming is not supported"},
|
||||
{"empty messages", `{"model":"mopac-code","messages":[]}`, 400, "messages must not be empty"},
|
||||
{"bad role", `{"model":"mopac-code","messages":[{"role":"tool","content":"x"}]}`, 400, "unsupported role"},
|
||||
{"malformed json", `{not json`, 400, "malformed request body"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
code, body := do(t, http.MethodPost, url, bearer(testVKey), tc.body)
|
||||
if code != tc.wantCode {
|
||||
t.Errorf("code = %d, want %d (body %s)", code, tc.wantCode, body)
|
||||
}
|
||||
if !strings.Contains(body, tc.wantBody) {
|
||||
t.Errorf("body %q does not contain %q", body, tc.wantBody)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestChatUpstreamErrorIs502(t *testing.T) {
|
||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "upstream exploded", http.StatusInternalServerError)
|
||||
}))
|
||||
defer up.Close()
|
||||
cfg := config.Default()
|
||||
cfg.Vertical = "teststack"
|
||||
cfg.LiteLLM.BaseURL = up.URL
|
||||
cfg.LiteLLM.KeyRef = "literal:k"
|
||||
cfg.LiteLLM.MaxRetries = 0
|
||||
cfg.Models.Tiers = map[string]string{"mopac-primary": "glm-5.3"}
|
||||
cfg.Models.Classes = map[string]string{"primary": "mopac-primary"}
|
||||
cfg.Serve.VKeyRef = "literal:" + testVKey
|
||||
cond, err := loop.New(cfg, io.Discard)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
srv, err := NewServer(cfg.Serve, cond.Router(), cond, io.Discard)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ts := httptest.NewServer(srv.Handler())
|
||||
defer ts.Close()
|
||||
|
||||
code, body := do(t, http.MethodPost, ts.URL+"/v1/chat/completions", bearer(testVKey),
|
||||
`{"model":"mopac-primary","messages":[{"role":"user","content":"x"}]}`)
|
||||
if code != http.StatusBadGateway {
|
||||
t.Fatalf("code = %d, want 502 (body %s)", code, body)
|
||||
}
|
||||
if strings.Contains(body, "exploded") {
|
||||
t.Errorf("502 body must not echo upstream details: %s", body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewServerFailFast(t *testing.T) {
|
||||
cfg := config.Default()
|
||||
cfg.Vertical = "teststack"
|
||||
cfg.LiteLLM.BaseURL = "http://unused"
|
||||
cfg.LiteLLM.KeyRef = "literal:k"
|
||||
cfg.Models.Tiers = map[string]string{"mopac-primary": "glm-5.3"}
|
||||
cfg.Models.Classes = map[string]string{"primary": "mopac-primary"}
|
||||
cond, err := loop.New(cfg, os.Stdout)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if _, err := NewServer(cfg.Serve, cond.Router(), cond, io.Discard); err == nil || !strings.Contains(err.Error(), "vkey_ref is required") {
|
||||
t.Errorf("missing vkey_ref: err = %v", err)
|
||||
}
|
||||
|
||||
cfg.Serve.VKeyRef = "literal:" + testVKey
|
||||
cfg.Serve.EnabledModels = []string{"mopac-nope"}
|
||||
if _, err := NewServer(cfg.Serve, cond.Router(), cond, io.Discard); err == nil || !strings.Contains(err.Error(), "not servable") {
|
||||
t.Errorf("bad enabled_models: err = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHealthz(t *testing.T) {
|
||||
_, _, ts := newStack(t)
|
||||
// No auth on healthz (liveness probe, same as the events receiver).
|
||||
code, body := do(t, http.MethodGet, ts.URL+"/healthz", "", "")
|
||||
if code != 200 || !strings.Contains(body, `"status":"ok"`) {
|
||||
t.Errorf("healthz: code=%d body=%s", code, body)
|
||||
}
|
||||
}
|
||||
|
||||
// Compile-time check that the conductor satisfies the Turner contract.
|
||||
var _ Turner = (*loop.Conductor)(nil)
|
||||
Reference in New Issue
Block a user