From c9e86eefa887cf00b3380dbcbe6a4f3e0aff39f1 Mon Sep 17 00:00:00 2001 From: reachableceo Date: Sat, 29 Aug 2026 01:19:59 -0500 Subject: [PATCH] serve: OpenAI-compatible front door (`harness serve`) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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-, 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 --- cmd/harness/main.go | 79 +++++- dev.sh | 16 +- internal/serve/server.go | 362 ++++++++++++++++++++++++++ internal/serve/server_test.go | 474 ++++++++++++++++++++++++++++++++++ 4 files changed, 928 insertions(+), 3 deletions(-) create mode 100644 internal/serve/server.go create mode 100644 internal/serve/server_test.go diff --git a/cmd/harness/main.go b/cmd/harness/main.go index d346b96..6f61714 100644 --- a/cmd/harness/main.go +++ b/cmd/harness/main.go @@ -10,12 +10,14 @@ import ( "net/http" "os" "os/signal" + "strings" "syscall" "time" "ukrrs.com/mopac/harness/internal/config" "ukrrs.com/mopac/harness/internal/events" "ukrrs.com/mopac/harness/internal/loop" + "ukrrs.com/mopac/harness/internal/serve" ) const usage = `MOPAC harness (v0) @@ -24,6 +26,7 @@ Usage: harness once [-config PATH] [-dry-run] [-demo] [-task-id ID] harness loop [-config PATH] [-interval DUR] [-once] [-dry-run] harness events [-config PATH] [-listen ADDR] + harness serve [-config PATH] [-listen ADDR] once runs ONE conductor iteration and exits (chain by re-invoking; no daemon): intake (Redmine scope, or the [demo] issue) -> plan/model routing -> @@ -41,6 +44,14 @@ events runs the webhook receiver until SIGINT/SIGTERM: Redmine/Discourse/ provider event id) and handed to the conductor (dispatch stub for now). Routes: POST /hooks/{redmine,discourse,gitea}, GET /healthz. +serve runs the OpenAI-compatible front door until SIGINT/SIGTERM (the + OpenWebUI connection): GET /v1/models lists the servable models (one per + [models.classes] class, named mopac-), POST /v1/chat/completions + runs ONE bounded stateless conductor turn over the sent history and + returns the final text + usage. Bearer vkey auth ([serve] vkey_ref); + non-streaming v0; tools off. Own port - coexists with events. + Routes: POST /v1/chat/completions, GET /v1/models, GET /healthz. + Flags: -config PATH config file (default $HARNESS_CONFIG or ./harness.toml) -dry-run once/loop: intake + plan only; no LLM call, no REPORT, @@ -49,7 +60,7 @@ Flags: -task-id ID once: run only the task/issue with this id -interval DUR loop: poll interval (overrides [loop] poll_interval_secs) -once loop: single scan then exit (cron-able) - -listen ADDR events: bind address (overrides [events] listen) + -listen ADDR events/serve: bind address (overrides [events]/[serve] listen) Exit codes: 0 ok (including "no tasks in scope"; loop: clean SIGINT stop) @@ -77,6 +88,8 @@ func run(args []string) int { return runLoop(args[1:]) case "events": return runEvents(args[1:]) + case "serve": + return runServe(args[1:]) default: fmt.Fprintf(os.Stderr, "harness: unknown command %q\n\n%s", args[0], usage) return 1 @@ -249,3 +262,67 @@ func runEvents(args []string) int { fmt.Printf("harness: events receiver stopped\n") return 0 } + +func runServe(args []string) int { + fs := flag.NewFlagSet("serve", flag.ContinueOnError) + cfgPath := fs.String("config", "", "config file path") + listen := fs.String("listen", "", "bind address (overrides [serve] listen)") + if err := fs.Parse(args); err != nil { + return 1 + } + if fs.NArg() > 0 { + fmt.Fprintf(os.Stderr, "harness: unexpected argument %q\n", fs.Arg(0)) + return 1 + } + + if *cfgPath == "" { + *cfgPath = os.Getenv("HARNESS_CONFIG") + } + if *cfgPath == "" { + *cfgPath = "harness.toml" + } + cfg, err := config.Load(*cfgPath) + if err != nil { + fmt.Fprintf(os.Stderr, "harness: %v\n", err) + return 1 + } + // The conductor owns the turn machinery and the model router the + // catalog is built from. + conductor, err := loop.New(cfg, os.Stdout) + if err != nil { + fmt.Fprintf(os.Stderr, "harness: %v\n", err) + return 1 + } + srv, err := serve.NewServer(cfg.Serve, conductor.Router(), conductor, os.Stdout) + if err != nil { + fmt.Fprintf(os.Stderr, "harness: %v\n", err) + return 1 + } + if *listen == "" { + *listen = cfg.Serve.Listen + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + httpSrv := &http.Server{ + Addr: *listen, + Handler: srv.Handler(), + ReadHeaderTimeout: 10 * time.Second, + } + go func() { + <-ctx.Done() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = httpSrv.Shutdown(shutdownCtx) + }() + + fmt.Printf("harness: serve (OpenAI-compatible) on %s (models: %s)\n", *listen, strings.Join(srv.ModelNames(), ", ")) + err = httpSrv.ListenAndServe() + if err != nil && !errors.Is(err, http.ErrServerClosed) { + fmt.Fprintf(os.Stderr, "harness: %v\n", err) + return 1 + } + fmt.Printf("harness: serve stopped\n") + return 0 +} diff --git a/dev.sh b/dev.sh index 29bac0d..65f37ad 100755 --- a/dev.sh +++ b/dev.sh @@ -3,13 +3,16 @@ # digest-pinned Docker builder (DESIGN "ALL dev work in Docker" — big rule); # the host runs containers, never toolchains. # -# Usage: ./dev.sh {build|vet|test|check|events|loop|smoke|shell} [args...] +# Usage: ./dev.sh {build|vet|test|check|events|serve|loop|smoke|shell} [args...] # # build compile ./cmd/harness into bin/harness # vet go vet ./... # test go test ./... # check build + vet + test (the pre-commit gate) # events run `harness events` with host port 4100 published (LAN smoke) +# serve run `harness serve` (OpenAI-compatible front door for OWUI) +# with host port 8090 published; vkey + litellm key env passed +# through (the vkey value is what the OWUI connection config gets) # loop run `harness loop` (the self-host daemon; repo bind-mounted so # REPORTs + state/loop/loop.jsonl land in the repo; key env vars # passed through; args go to the loop, e.g. `./dev.sh loop --once`) @@ -51,6 +54,15 @@ events) -e HARNESS_GITEA_WEBHOOK_SECRET \ "$IMAGE" /h/bin/harness events -listen ":4100" "$@" ;; +serve) + # OpenAI-compatible front door for OpenWebUI: port 8090 on the host LAN + # (OWUI base URL http://:8090/v1, api key = HARNESS_SERVE_VKEY). + # Stop with SIGINT or `docker stop mopac-serve`. + exec docker run --rm -i --name mopac-serve -p 8090:8090 \ + -v "$PWD:/h" -w /h -u "$(id -u):$(id -g)" -e HOME=/tmp \ + -e HARNESS_SERVE_VKEY -e HARNESS_LITELLM_KEY -e HARNESS_KEYPROXY_TOKEN \ + "$IMAGE" /h/bin/harness serve -listen ":8090" "$@" + ;; loop) # Self-host daemon: repo bind-mounted (REPORTs + loop.jsonl land in the # repo); key refs' env vars passed through. Stop with SIGINT/double Ctrl-C @@ -68,7 +80,7 @@ shell) run sh ;; *) - echo "dev.sh: unknown command $cmd (build|vet|test|check|events|loop|smoke|shell)" >&2 + echo "dev.sh: unknown command $cmd (build|vet|test|check|events|serve|loop|smoke|shell)" >&2 exit 1 ;; esac diff --git a/internal/serve/server.go b/internal/serve/server.go new file mode 100644 index 0000000..089e83c --- /dev/null +++ b/internal/serve/server.go @@ -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- 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-). +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[:]) +} diff --git a/internal/serve/server_test.go b/internal/serve/server_test.go new file mode 100644 index 0000000..99898bc --- /dev/null +++ b/internal/serve/server_test.go @@ -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)