From 22a71bafe35a3de9e9949ff462ee3d7307f3aeaf Mon Sep 17 00:00:00 2001 From: Multica Eve Date: Thu, 9 Jul 2026 12:42:59 +0800 Subject: [PATCH] feat(server): add basic LLM API layer with OpenAI-compatible endpoints (#5138) Integrate the official openai-go SDK (v3) as a thin, reusable LLM layer (pkg/llm) backing lightweight utility calls that do not need the agent runtime (chat titles, quick-create drafts, ...). Expose two user-authenticated, OpenAI-compatible chat-completions endpoints: - POST /api/llm/v1/chat/completions (JSON response) - POST /api/llm/v1/chat/completions/stream (SSE stream) Requests decode directly into the SDK's ChatCompletionNewParams and responses are relayed via RawJSON() for byte-exact OpenAI-format compatibility. Base URL and API key are configurable (MULTICA_LLM_*), and the model is taken from the request with a configurable default fallback (MULTICA_LLM_DEFAULT_MODEL, else gpt-4o-mini). When unconfigured the endpoints return 503. Co-authored-by: Eve Co-authored-by: multica-agent --- .env.example | 14 ++ server/cmd/server/router.go | 12 ++ server/go.mod | 5 + server/go.sum | 12 ++ server/internal/handler/handler.go | 24 +++- server/internal/handler/llm.go | 210 ++++++++++++++++++++++++++++ server/internal/handler/llm_test.go | 150 ++++++++++++++++++++ server/pkg/llm/client.go | 204 +++++++++++++++++++++++++++ server/pkg/llm/client_test.go | 174 +++++++++++++++++++++++ 9 files changed, 804 insertions(+), 1 deletion(-) create mode 100644 server/internal/handler/llm.go create mode 100644 server/internal/handler/llm_test.go create mode 100644 server/pkg/llm/client.go create mode 100644 server/pkg/llm/client_test.go diff --git a/.env.example b/.env.example index 9179d2b390..45525016ed 100644 --- a/.env.example +++ b/.env.example @@ -60,6 +60,20 @@ MULTICA_PUBLIC_URL= # e.g. "127.0.0.1/32" for a same-host reverse proxy, or the CDN's # announced ranges for cloud deployments. MULTICA_TRUSTED_PROXIES= +# Basic LLM API layer (MUL-4238). Backs the OpenAI-compatible chat +# completions endpoints (POST /api/llm/v1/chat/completions and +# .../stream) and internal one-shot LLM helpers. Both endpoints require a +# logged-in user. When both API key and base URL are empty the layer is +# disabled and the endpoints return 503. +# - API key for the upstream (OpenAI or any OpenAI-compatible gateway). +MULTICA_LLM_API_KEY= +# - Base URL of the upstream. Leave unset to use OpenAI's default +# (https://api.openai.com/v1). Set this to point at a gateway or a +# self-hosted, OpenAI-compatible model server. +MULTICA_LLM_BASE_URL= +# - Default model used when a request omits `model`. When unset, falls +# back to a small built-in default (gpt-4o-mini). +MULTICA_LLM_DEFAULT_MODEL= MULTICA_DAEMON_CONFIG= MULTICA_WORKSPACE_ID= MULTICA_DAEMON_ID= diff --git a/server/cmd/server/router.go b/server/cmd/server/router.go index 7a45ba5354..8037c51f73 100644 --- a/server/cmd/server/router.go +++ b/server/cmd/server/router.go @@ -176,6 +176,9 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus AttachmentDownloadMode: os.Getenv("ATTACHMENT_DOWNLOAD_MODE"), AttachmentDownloadURLTTL: envDuration("ATTACHMENT_DOWNLOAD_URL_TTL", 30*time.Minute), AttachmentFrameAncestors: origins, + LLMAPIKey: strings.TrimSpace(os.Getenv("MULTICA_LLM_API_KEY")), + LLMBaseURL: strings.TrimSpace(os.Getenv("MULTICA_LLM_BASE_URL")), + LLMDefaultModel: strings.TrimSpace(os.Getenv("MULTICA_LLM_DEFAULT_MODEL")), } h := handler.New(queries, pool, hub, bus, emailSvc, store, cfSigner, analyticsClient, signupConfig, daemonHub) h.Metrics = opts.BusinessMetrics @@ -781,6 +784,15 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus r.Post("/api/upload-file", h.UploadFile) r.Post("/api/feedback", h.CreateFeedback) + // OpenAI-compatible LLM endpoints (MUL-4238). User-scoped (auth-only, + // no workspace context needed): the basic LLM layer is a shared + // utility, not a per-workspace resource. Both accept the standard + // OpenAI chat-completions request body; model is taken from the request + // with a configured default fallback. Return 503 when the LLM layer is + // unconfigured (no MULTICA_LLM_API_KEY / MULTICA_LLM_BASE_URL). + r.Post("/api/llm/v1/chat/completions", h.LLMChatCompletions) + r.Post("/api/llm/v1/chat/completions/stream", h.LLMChatCompletionsStream) + // Attachment download — user-scoped (auth-only), NOT // workspace-scoped. The handler self-resolves the workspace // from the attachment row and enforces membership inside, so diff --git a/server/go.mod b/server/go.mod index 997d8f8d75..082461c5fd 100644 --- a/server/go.mod +++ b/server/go.mod @@ -19,6 +19,7 @@ require ( github.com/lmittmann/tint v1.1.3 github.com/mattn/go-shellwords v1.0.13 github.com/oklog/ulid/v2 v2.1.1 + github.com/openai/openai-go/v3 v3.41.1 github.com/pelletier/go-toml/v2 v2.3.0 github.com/prometheus/client_golang v1.23.2 github.com/prometheus/client_model v0.6.2 @@ -61,6 +62,10 @@ require ( github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/prometheus/common v0.66.1 // indirect github.com/prometheus/procfs v0.16.1 // indirect + github.com/tidwall/gjson v1.18.0 // indirect + github.com/tidwall/match v1.1.1 // indirect + github.com/tidwall/pretty v1.2.1 // indirect + github.com/tidwall/sjson v1.2.5 // indirect go.uber.org/atomic v1.11.0 // indirect go.yaml.in/yaml/v2 v2.4.2 // indirect golang.org/x/net v0.43.0 // indirect diff --git a/server/go.sum b/server/go.sum index 2d40a2ebc5..b30a6629d7 100644 --- a/server/go.sum +++ b/server/go.sum @@ -107,6 +107,8 @@ github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE= github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU= github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y= github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM= +github.com/openai/openai-go/v3 v3.41.1 h1:c7vLxUL4I1GjHcHoQvb/5qpBHEt3m7ZJIb8lFWIRkfc= +github.com/openai/openai-go/v3 v3.41.1/go.mod h1:cdufnVK14cWcT9qA1rRtrXx4FTRsgbDPW7Ia7SS5cZo= github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o= github.com/pelletier/go-toml/v2 v2.3.0 h1:k59bC/lIZREW0/iVaQR8nDHxVq8OVlIzYCOJf421CaM= github.com/pelletier/go-toml/v2 v2.3.0/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= @@ -140,6 +142,16 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= +github.com/tidwall/gjson v1.18.0 h1:FIDeeyB800efLX89e5a8Y0BNH+LOngJyGrIWxG2FKQY= +github.com/tidwall/gjson v1.18.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= +github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA= +github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM= +github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4= +github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY= +github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28= github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0= github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= diff --git a/server/internal/handler/handler.go b/server/internal/handler/handler.go index 7dca3a0536..8a39ad2a0f 100644 --- a/server/internal/handler/handler.go +++ b/server/internal/handler/handler.go @@ -34,6 +34,7 @@ import ( "github.com/multica-ai/multica/server/internal/util" db "github.com/multica-ai/multica/server/pkg/db/generated" "github.com/multica-ai/multica/server/pkg/featureflag" + "github.com/multica-ai/multica/server/pkg/llm" ) // randomID returns a random 16-byte hex string used as a request ID for @@ -95,6 +96,16 @@ type Config struct { // frontend/CORS origin allowlist so split app/api self-hosted deployments // can frame API-hosted PDFs without allowing arbitrary third-party frames. AttachmentFrameAncestors []string + // LLM* configure the basic LLM API layer (MUL-4238). They back the + // OpenAI-compatible chat-completions endpoints and any internal one-shot + // LLM helpers. When both LLMAPIKey and LLMBaseURL are empty the layer is + // disabled and the endpoints return 503. + // - LLMAPIKey -> MULTICA_LLM_API_KEY + // - LLMBaseURL -> MULTICA_LLM_BASE_URL (OpenAI or any compatible gateway) + // - LLMDefaultModel -> MULTICA_LLM_DEFAULT_MODEL (used when a request omits `model`) + LLMAPIKey string + LLMBaseURL string + LLMDefaultModel string } type cloudRuntimeProxy interface { @@ -198,7 +209,13 @@ type Handler struct { // unless Slack is configured; GetChatChannelHistory then reports "no channel // integration". A future platform satisfies the same reader interface. SlackHistory ChatChannelHistoryReader - cfg Config + // LLM is the basic LLM API layer (MUL-4238): a thin wrapper over the + // OpenAI Go SDK backing the OpenAI-compatible chat-completions endpoints + // and any internal one-shot LLM helpers. Always non-nil (New builds it + // from Config); when unconfigured its Enabled() reports false and the + // handlers return 503. + LLM *llm.Client + cfg Config } func New(queries *db.Queries, txStarter txStarter, hub *realtime.Hub, bus *events.Bus, emailService *service.EmailService, store storage.Storage, cfSigner *auth.CloudFrontSigner, analyticsClient analytics.Client, cfg Config, daemonHubs ...*daemonws.Hub) *Handler { @@ -258,6 +275,11 @@ func New(queries *db.Queries, txStarter txStarter, hub *realtime.Hub, bus *event BaseURL: cfg.CloudRuntimeFleetURL, Timeout: cfg.CloudRuntimeFleetTimeout, }), + LLM: llm.New(llm.Config{ + APIKey: cfg.LLMAPIKey, + BaseURL: cfg.LLMBaseURL, + DefaultModel: cfg.LLMDefaultModel, + }), cfg: cfg, } } diff --git a/server/internal/handler/llm.go b/server/internal/handler/llm.go new file mode 100644 index 0000000000..12d9bdce92 --- /dev/null +++ b/server/internal/handler/llm.go @@ -0,0 +1,210 @@ +package handler + +import ( + "errors" + "io" + "log/slog" + "net/http" + + openai "github.com/openai/openai-go/v3" + + "github.com/multica-ai/multica/server/internal/logger" + "github.com/multica-ai/multica/server/pkg/llm" +) + +// llmRequestBodyLimit caps the request body for the chat-completions endpoints. +// 1 MiB comfortably fits large multi-message conversations (including small +// inline image data URLs) while preventing an authenticated client from +// streaming unbounded bytes into the JSON decoder. +const llmRequestBodyLimit = 1 << 20 // 1 MiB + +// decodeLLMChatParams reads and validates the OpenAI chat-completions request +// body. It decodes directly into the SDK's ChatCompletionNewParams (which +// implements UnmarshalJSON with full OpenAI fidelity) so every supported field +// — messages, tools, response_format, temperature, etc. — passes through +// untouched. On any failure it writes the appropriate 4xx and returns ok=false. +func decodeLLMChatParams(w http.ResponseWriter, r *http.Request) (openai.ChatCompletionNewParams, bool) { + var params openai.ChatCompletionNewParams + + r.Body = http.MaxBytesReader(w, r.Body, llmRequestBodyLimit) + body, err := io.ReadAll(r.Body) + if err != nil { + // MaxBytesReader surfaces an oversize body as an error here. + writeError(w, http.StatusRequestEntityTooLarge, "request body too large") + return params, false + } + if len(body) == 0 { + writeError(w, http.StatusBadRequest, "request body is required") + return params, false + } + if err := params.UnmarshalJSON(body); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body") + return params, false + } + if len(params.Messages) == 0 { + writeError(w, http.StatusBadRequest, "messages is required") + return params, false + } + return params, true +} + +// writeUpstreamError maps an error from the LLM layer to an HTTP response. +// Upstream OpenAI errors carry their own status code, which we preserve so the +// caller sees e.g. a 401 (bad key) or 429 (rate limited) rather than a blanket +// 500. A missing configuration is a 503; anything else is a 502 (bad gateway). +func writeUpstreamError(w http.ResponseWriter, r *http.Request, err error) { + if errors.Is(err, llm.ErrNotConfigured) { + writeError(w, http.StatusServiceUnavailable, "LLM API is not configured") + return + } + + var apiErr *openai.Error + if errors.As(err, &apiErr) && apiErr.StatusCode >= 400 { + slog.Warn("llm upstream error", + append(logger.RequestAttrs(r), "status", apiErr.StatusCode, "error", err)...) + // Preserve the upstream OpenAI-shaped error body so OpenAI clients can + // parse it as they would a direct call. Fall back to a plain message if + // the SDK gave us no body. + if raw := apiErr.RawJSON(); raw != "" { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(apiErr.StatusCode) + _, _ = io.WriteString(w, raw) + return + } + writeError(w, apiErr.StatusCode, "upstream error") + return + } + + slog.Warn("llm request failed", append(logger.RequestAttrs(r), "error", err)...) + writeError(w, http.StatusBadGateway, "failed to reach LLM upstream") +} + +// LLMChatCompletions is an OpenAI-compatible, non-streaming chat-completions +// endpoint (POST /api/llm/v1/chat/completions). It accepts the standard OpenAI +// request body and returns the byte-exact upstream ChatCompletion object. +// +// Auth: user-scoped (the route sits inside the authenticated group). Model is +// taken from the request; when omitted, the configured default is used. +func (h *Handler) LLMChatCompletions(w http.ResponseWriter, r *http.Request) { + if _, ok := requireUserID(w, r); !ok { + return + } + if h.LLM == nil || !h.LLM.Enabled() { + writeError(w, http.StatusServiceUnavailable, "LLM API is not configured") + return + } + + params, ok := decodeLLMChatParams(w, r) + if !ok { + return + } + + completion, err := h.LLM.Chat(r.Context(), params) + if err != nil { + writeUpstreamError(w, r, err) + return + } + + // Relay the upstream response verbatim to preserve full OpenAI-format + // compatibility (fields the SDK struct doesn't model are still present in + // RawJSON). Fall back to marshaling the typed object if RawJSON is empty. + if raw := completion.RawJSON(); raw != "" { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, raw) + return + } + writeJSON(w, http.StatusOK, completion) +} + +// LLMChatCompletionsStream is an OpenAI-compatible streaming chat-completions +// endpoint (POST /api/llm/v1/chat/completions/stream). It emits server-sent +// events whose `data:` payloads are the byte-exact upstream +// chat.completion.chunk objects, terminated by the OpenAI `data: [DONE]` +// sentinel — identical to calling OpenAI with stream=true. +// +// Auth: user-scoped (the route sits inside the authenticated group). Model is +// taken from the request; when omitted, the configured default is used. +func (h *Handler) LLMChatCompletionsStream(w http.ResponseWriter, r *http.Request) { + if _, ok := requireUserID(w, r); !ok { + return + } + if h.LLM == nil || !h.LLM.Enabled() { + writeError(w, http.StatusServiceUnavailable, "LLM API is not configured") + return + } + + params, ok := decodeLLMChatParams(w, r) + if !ok { + return + } + + stream, err := h.LLM.ChatStream(r.Context(), params) + if err != nil { + writeUpstreamError(w, r, err) + return + } + defer stream.Close() + + // SSE headers. Disable proxy buffering so chunks flush promptly through + // nginx-style reverse proxies. + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + + rc := http.NewResponseController(w) + + // Peek the first chunk before committing a 200 so a fast upstream failure + // (bad key, unknown model) can still be reported as a JSON error with the + // correct status instead of a half-open 200 stream. + if !stream.Next() { + if err := stream.Err(); err != nil { + writeUpstreamError(w, r, err) + return + } + // No content at all: emit a well-formed, empty SSE stream. + w.WriteHeader(http.StatusOK) + writeSSEData(w, rc, "[DONE]") + return + } + + w.WriteHeader(http.StatusOK) + + // First chunk (already fetched), then the rest. + if chunk := stream.Current(); chunk.RawJSON() != "" { + if !writeSSEData(w, rc, chunk.RawJSON()) { + return + } + } + for stream.Next() { + chunk := stream.Current() + if chunk.RawJSON() == "" { + continue + } + if !writeSSEData(w, rc, chunk.RawJSON()) { + return + } + } + if err := stream.Err(); err != nil { + // The connection is already a 200 SSE stream; we cannot change the + // status. Emit an OpenAI-style error event so the client can detect the + // mid-stream failure, then stop. + slog.Warn("llm stream error mid-flight", append(logger.RequestAttrs(r), "error", err)...) + writeSSEData(w, rc, `{"error":{"message":"upstream stream error","type":"upstream_error"}}`) + return + } + + writeSSEData(w, rc, "[DONE]") +} + +// writeSSEData writes a single SSE `data:` event and flushes it. It returns +// false if the write failed (client disconnected), signaling the caller to +// stop. Flush errors are non-fatal (some ResponseWriters don't support it). +func writeSSEData(w http.ResponseWriter, rc *http.ResponseController, payload string) bool { + if _, err := io.WriteString(w, "data: "+payload+"\n\n"); err != nil { + return false + } + _ = rc.Flush() + return true +} diff --git a/server/internal/handler/llm_test.go b/server/internal/handler/llm_test.go new file mode 100644 index 0000000000..17f8db4a0f --- /dev/null +++ b/server/internal/handler/llm_test.go @@ -0,0 +1,150 @@ +package handler + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/multica-ai/multica/server/pkg/llm" +) + +// newLLMUpstream returns a stub OpenAI upstream and a Handler wired to it. +func newLLMUpstream(t *testing.T, upstream http.HandlerFunc) *Handler { + t.Helper() + srv := httptest.NewServer(upstream) + t.Cleanup(srv.Close) + return &Handler{ + LLM: llm.New(llm.Config{APIKey: "test-key", BaseURL: srv.URL, DefaultModel: "test-model"}), + } +} + +func postLLM(t *testing.T, h http.HandlerFunc, body, userID string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodPost, "/api/llm/v1/chat/completions", strings.NewReader(body)) + if userID != "" { + req.Header.Set("X-User-ID", userID) + } + rec := httptest.NewRecorder() + h(rec, req) + return rec +} + +func TestLLMChatCompletions_Unauthenticated(t *testing.T) { + h := &Handler{LLM: llm.New(llm.Config{APIKey: "k"})} + rec := postLLM(t, h.LLMChatCompletions, `{"messages":[{"role":"user","content":"hi"}]}`, "") + if rec.Code != http.StatusUnauthorized { + t.Fatalf("expected 401, got %d", rec.Code) + } +} + +func TestLLMChatCompletions_NotConfigured(t *testing.T) { + h := &Handler{LLM: llm.New(llm.Config{})} // disabled + rec := postLLM(t, h.LLMChatCompletions, `{"messages":[{"role":"user","content":"hi"}]}`, "user-1") + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 503, got %d", rec.Code) + } +} + +func TestLLMChatCompletions_BadBody(t *testing.T) { + h := &Handler{LLM: llm.New(llm.Config{APIKey: "k"})} + rec := postLLM(t, h.LLMChatCompletions, `not json`, "user-1") + if rec.Code != http.StatusBadRequest { + t.Fatalf("expected 400 for malformed body, got %d", rec.Code) + } +} + +func TestLLMChatCompletions_MissingMessages(t *testing.T) { + h := &Handler{LLM: llm.New(llm.Config{APIKey: "k"})} + rec := postLLM(t, h.LLMChatCompletions, `{"model":"x"}`, "user-1") + if rec.Code != http.StatusBadRequest { + t.Fatalf("expected 400 for missing messages, got %d", rec.Code) + } +} + +func TestLLMChatCompletions_Success(t *testing.T) { + h := newLLMUpstream(t, func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Authorization"); got != "Bearer test-key" { + t.Errorf("expected upstream auth header, got %q", got) + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"id":"cmpl-42","object":"chat.completion","model":"test-model","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}]}`) + }) + + rec := postLLM(t, h.LLMChatCompletions, `{"messages":[{"role":"user","content":"ping"}]}`, "user-1") + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d (body=%s)", rec.Code, rec.Body.String()) + } + if ct := rec.Header().Get("Content-Type"); ct != "application/json" { + t.Fatalf("expected application/json, got %q", ct) + } + if !strings.Contains(rec.Body.String(), `"cmpl-42"`) || !strings.Contains(rec.Body.String(), `"pong"`) { + t.Fatalf("expected upstream body relayed verbatim, got %s", rec.Body.String()) + } +} + +func TestLLMChatCompletions_UpstreamErrorStatusPreserved(t *testing.T) { + h := newLLMUpstream(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusUnauthorized) + _, _ = io.WriteString(w, `{"error":{"message":"bad key","type":"invalid_request_error"}}`) + }) + + rec := postLLM(t, h.LLMChatCompletions, `{"messages":[{"role":"user","content":"ping"}]}`, "user-1") + if rec.Code != http.StatusUnauthorized { + t.Fatalf("expected upstream 401 preserved, got %d (body=%s)", rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "bad key") { + t.Fatalf("expected upstream error body relayed, got %s", rec.Body.String()) + } +} + +func TestLLMChatCompletionsStream_Success(t *testing.T) { + h := newLLMUpstream(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + flusher, _ := w.(http.Flusher) + for _, ch := range []string{ + `{"id":"c1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"Hel"}}]}`, + `{"id":"c1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"lo"}}]}`, + } { + _, _ = io.WriteString(w, "data: "+ch+"\n\n") + if flusher != nil { + flusher.Flush() + } + } + _, _ = io.WriteString(w, "data: [DONE]\n\n") + }) + + req := httptest.NewRequest(http.MethodPost, "/api/llm/v1/chat/completions/stream", + strings.NewReader(`{"messages":[{"role":"user","content":"hi"}]}`)) + req.Header.Set("X-User-ID", "user-1") + rec := httptest.NewRecorder() + h.LLMChatCompletionsStream(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + if ct := rec.Header().Get("Content-Type"); ct != "text/event-stream" { + t.Fatalf("expected text/event-stream, got %q", ct) + } + out := rec.Body.String() + if !strings.Contains(out, `"Hel"`) || !strings.Contains(out, `"lo"`) { + t.Fatalf("expected relayed chunks, got %s", out) + } + if !strings.Contains(out, "data: [DONE]") { + t.Fatalf("expected terminating [DONE] sentinel, got %s", out) + } +} + +func TestLLMChatCompletionsStream_NotConfigured(t *testing.T) { + h := &Handler{LLM: llm.New(llm.Config{})} + req := httptest.NewRequest(http.MethodPost, "/api/llm/v1/chat/completions/stream", + strings.NewReader(`{"messages":[{"role":"user","content":"hi"}]}`)) + req.Header.Set("X-User-ID", "user-1") + rec := httptest.NewRecorder() + h.LLMChatCompletionsStream(rec, req) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 503, got %d", rec.Code) + } +} diff --git a/server/pkg/llm/client.go b/server/pkg/llm/client.go new file mode 100644 index 0000000000..668849d423 --- /dev/null +++ b/server/pkg/llm/client.go @@ -0,0 +1,204 @@ +// Package llm is a thin, reusable wrapper around the official OpenAI Go SDK +// (github.com/openai/openai-go). It exists so the rest of the server has a +// single, well-typed entry point for "just call an LLM" needs that do NOT +// require the full agent runtime — e.g. generating a chat title or drafting a +// quick-create issue (MUL-4238). +// +// The wrapper is intentionally small: +// +// - It owns the SDK client construction (base URL + API key + retry/timeout +// defaults) so callers never touch option.RequestOption directly. +// - It exposes both the raw Chat Completions surface (Chat / ChatStream), +// used by the OpenAI-compatible HTTP proxy handlers, and a convenience +// GenerateText helper for simple internal one-shot completions. +// - The default model is configurable; when a request omits the model we +// fall back to it, and when it too is empty we fall back to a sane +// built-in default so a misconfigured deployment still returns a clear +// upstream error rather than a 400 from our own layer. +// +// Base URL and API key are configurable so the same layer can target OpenAI, +// an OpenAI-compatible gateway, or a self-hosted model server. +package llm + +import ( + "context" + "errors" + "net/http" + "strings" + "time" + + openai "github.com/openai/openai-go/v3" + "github.com/openai/openai-go/v3/option" + "github.com/openai/openai-go/v3/packages/ssestream" + "github.com/openai/openai-go/v3/shared" +) + +// FallbackModel is the last-resort model used when neither the request nor the +// configured default supplies one. It is deliberately a small, inexpensive +// model since this layer backs lightweight utility calls. +const FallbackModel = "gpt-4o-mini" + +// defaultTimeout bounds the full request lifecycle (including SDK retries) when +// the caller's context has no deadline of its own. Streaming requests are not +// subject to this because the handler owns the connection lifetime. +const defaultRequestTimeout = 60 * time.Second + +// ErrNotConfigured is returned by Chat/ChatStream/GenerateText when the client +// was constructed without any credentials or base URL. Callers (e.g. the HTTP +// handlers) should translate this into a 503 so a misconfigured self-hosted +// deployment surfaces a clear error instead of dialing OpenAI with no key. +var ErrNotConfigured = errors.New("llm: no API key or base URL configured") + +// Config holds the tunables for the LLM layer. All fields are optional; an +// empty Config yields a disabled client (see Client.Enabled). +type Config struct { + // APIKey authenticates against the upstream. Maps to MULTICA_LLM_API_KEY. + APIKey string + // BaseURL points at OpenAI or any OpenAI-compatible gateway. When empty the + // SDK's default (https://api.openai.com/v1) is used. Maps to + // MULTICA_LLM_BASE_URL. + BaseURL string + // DefaultModel is used when a request omits the model. Maps to + // MULTICA_LLM_DEFAULT_MODEL. When empty, FallbackModel is used. + DefaultModel string + // MaxRetries overrides the SDK default (2). A negative value is treated as + // zero (no retries). + MaxRetries int + // HTTPClient, when set, replaces the SDK's default transport. Primarily a + // test seam. + HTTPClient option.HTTPClient +} + +// Client is a configured, reusable LLM caller. It is safe for concurrent use; +// the underlying SDK client holds no per-request state. +type Client struct { + sdk openai.Client + defaultModel string + enabled bool +} + +// New builds a Client from cfg. It never returns an error: an unconfigured +// Config produces a disabled client whose calls return ErrNotConfigured, which +// keeps wiring in main/router simple (no boot-time failure when the LLM layer +// is simply not set up on a given deployment). +func New(cfg Config) *Client { + opts := make([]option.RequestOption, 0, 4) + if key := strings.TrimSpace(cfg.APIKey); key != "" { + opts = append(opts, option.WithAPIKey(key)) + } + if base := strings.TrimSpace(cfg.BaseURL); base != "" { + opts = append(opts, option.WithBaseURL(base)) + } + if cfg.MaxRetries != 0 { + retries := cfg.MaxRetries + if retries < 0 { + retries = 0 + } + opts = append(opts, option.WithMaxRetries(retries)) + } + if cfg.HTTPClient != nil { + opts = append(opts, option.WithHTTPClient(cfg.HTTPClient)) + } + + defaultModel := strings.TrimSpace(cfg.DefaultModel) + if defaultModel == "" { + defaultModel = FallbackModel + } + + return &Client{ + sdk: openai.NewClient(opts...), + defaultModel: defaultModel, + // A deployment is "configured" if it gave us either a key or a base + // URL. A bare base URL (no key) is valid for keyless local gateways. + enabled: strings.TrimSpace(cfg.APIKey) != "" || strings.TrimSpace(cfg.BaseURL) != "", + } +} + +// Enabled reports whether the client was given any credentials or base URL. +// Handlers use this to short-circuit with a 503 before doing any work. +func (c *Client) Enabled() bool { return c != nil && c.enabled } + +// DefaultModel returns the effective default model (never empty). +func (c *Client) DefaultModel() string { return c.defaultModel } + +// applyDefaultModel fills in the default model when the caller left it blank. +func (c *Client) applyDefaultModel(params *openai.ChatCompletionNewParams) { + if strings.TrimSpace(string(params.Model)) == "" { + params.Model = shared.ChatModel(c.defaultModel) + } +} + +// Chat performs a non-streaming chat completion. The params are passed through +// to the SDK verbatim (so tools, response_format, temperature, etc. are all +// honored); only the model default is applied. The returned *ChatCompletion +// exposes RawJSON() for byte-exact OpenAI-compatible responses. +func (c *Client) Chat(ctx context.Context, params openai.ChatCompletionNewParams) (*openai.ChatCompletion, error) { + if !c.Enabled() { + return nil, ErrNotConfigured + } + c.applyDefaultModel(¶ms) + + // Give the request a bounded lifetime when the caller supplied none, so a + // hung upstream cannot pin a goroutine indefinitely. + ctx, cancel := withDefaultTimeout(ctx) + defer cancel() + + return c.sdk.Chat.Completions.New(ctx, params) +} + +// ChatStream performs a streaming chat completion, returning the SDK stream so +// the caller can relay chunks (each chunk exposes RawJSON() for byte-exact +// OpenAI-compatible SSE). The caller MUST call Close on the returned stream. +// +// Unlike Chat, no default timeout is imposed: the stream's lifetime is owned by +// the caller (typically an HTTP handler bound to the client connection). +func (c *Client) ChatStream(ctx context.Context, params openai.ChatCompletionNewParams) (*ssestream.Stream[openai.ChatCompletionChunk], error) { + if !c.Enabled() { + return nil, ErrNotConfigured + } + c.applyDefaultModel(¶ms) + return c.sdk.Chat.Completions.NewStreaming(ctx, params), nil +} + +// GenerateText is a convenience for simple internal one-shot completions (chat +// titles, quick-create drafts, ...). It sends an optional system prompt plus a +// single user prompt and returns the assistant's text content. Model empty -> +// the configured default. +func (c *Client) GenerateText(ctx context.Context, model, systemPrompt, userPrompt string) (string, error) { + if !c.Enabled() { + return "", ErrNotConfigured + } + + messages := make([]openai.ChatCompletionMessageParamUnion, 0, 2) + if strings.TrimSpace(systemPrompt) != "" { + messages = append(messages, openai.SystemMessage(systemPrompt)) + } + messages = append(messages, openai.UserMessage(userPrompt)) + + params := openai.ChatCompletionNewParams{ + Messages: messages, + Model: shared.ChatModel(strings.TrimSpace(model)), + } + + completion, err := c.Chat(ctx, params) + if err != nil { + return "", err + } + if len(completion.Choices) == 0 { + return "", errors.New("llm: upstream returned no choices") + } + return completion.Choices[0].Message.Content, nil +} + +// withDefaultTimeout returns ctx unchanged (with a no-op cancel) when it already +// has a deadline, otherwise a child context bounded by defaultRequestTimeout. +func withDefaultTimeout(ctx context.Context) (context.Context, context.CancelFunc) { + if _, ok := ctx.Deadline(); ok { + return ctx, func() {} + } + return context.WithTimeout(ctx, defaultRequestTimeout) +} + +// compile-time assertion that option.HTTPClient is satisfied by *http.Client so +// callers can pass a plain *http.Client as the test seam. +var _ option.HTTPClient = (*http.Client)(nil) diff --git a/server/pkg/llm/client_test.go b/server/pkg/llm/client_test.go new file mode 100644 index 0000000000..c8abbc41db --- /dev/null +++ b/server/pkg/llm/client_test.go @@ -0,0 +1,174 @@ +package llm + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + openai "github.com/openai/openai-go/v3" + "github.com/openai/openai-go/v3/shared" +) + +// stubUpstream returns an httptest server that mimics the OpenAI +// chat-completions endpoint. handler receives the decoded request body. +func stubUpstream(t *testing.T, handler func(w http.ResponseWriter, body map[string]any)) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw, _ := io.ReadAll(r.Body) + var body map[string]any + _ = json.Unmarshal(raw, &body) + handler(w, body) + })) + t.Cleanup(srv.Close) + return srv +} + +func TestNewDisabledClient(t *testing.T) { + c := New(Config{}) + if c.Enabled() { + t.Fatal("expected disabled client with empty config") + } + if c.DefaultModel() != FallbackModel { + t.Fatalf("expected fallback model %q, got %q", FallbackModel, c.DefaultModel()) + } + if _, err := c.Chat(context.Background(), openai.ChatCompletionNewParams{}); err != ErrNotConfigured { + t.Fatalf("expected ErrNotConfigured, got %v", err) + } + if _, err := c.GenerateText(context.Background(), "", "", "hi"); err != ErrNotConfigured { + t.Fatalf("expected ErrNotConfigured from GenerateText, got %v", err) + } +} + +func TestEnabledWithBaseURLOnly(t *testing.T) { + c := New(Config{BaseURL: "http://localhost:1234"}) + if !c.Enabled() { + t.Fatal("expected enabled client when only base URL is set (keyless gateway)") + } +} + +func TestConfiguredDefaultModel(t *testing.T) { + c := New(Config{APIKey: "k", DefaultModel: "my-model"}) + if c.DefaultModel() != "my-model" { + t.Fatalf("expected configured default model, got %q", c.DefaultModel()) + } +} + +func TestChatAppliesDefaultModel(t *testing.T) { + var gotModel string + srv := stubUpstream(t, func(w http.ResponseWriter, body map[string]any) { + gotModel, _ = body["model"].(string) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"id":"cmpl-1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hello"},"finish_reason":"stop"}]}`) + }) + + c := New(Config{APIKey: "test-key", BaseURL: srv.URL, DefaultModel: "default-x"}) + // Request omits the model -> the configured default must be applied. + completion, err := c.Chat(context.Background(), openai.ChatCompletionNewParams{ + Messages: []openai.ChatCompletionMessageParamUnion{openai.UserMessage("hi")}, + }) + if err != nil { + t.Fatalf("Chat failed: %v", err) + } + if gotModel != "default-x" { + t.Fatalf("expected default model forwarded upstream, got %q", gotModel) + } + if len(completion.Choices) != 1 || completion.Choices[0].Message.Content != "hello" { + t.Fatalf("unexpected completion: %+v", completion.Choices) + } + if completion.RawJSON() == "" { + t.Fatal("expected non-empty RawJSON for passthrough") + } +} + +func TestChatRespectsRequestModel(t *testing.T) { + var gotModel string + srv := stubUpstream(t, func(w http.ResponseWriter, body map[string]any) { + gotModel, _ = body["model"].(string) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"id":"cmpl-1","object":"chat.completion","choices":[]}`) + }) + + c := New(Config{APIKey: "test-key", BaseURL: srv.URL, DefaultModel: "default-x"}) + _, err := c.Chat(context.Background(), openai.ChatCompletionNewParams{ + Model: shared.ChatModel("caller-model"), + Messages: []openai.ChatCompletionMessageParamUnion{openai.UserMessage("hi")}, + }) + if err != nil { + t.Fatalf("Chat failed: %v", err) + } + if gotModel != "caller-model" { + t.Fatalf("expected caller model preserved, got %q", gotModel) + } +} + +func TestGenerateText(t *testing.T) { + var sawSystem bool + srv := stubUpstream(t, func(w http.ResponseWriter, body map[string]any) { + if msgs, ok := body["messages"].([]any); ok { + for _, m := range msgs { + if mm, ok := m.(map[string]any); ok && mm["role"] == "system" { + sawSystem = true + } + } + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"id":"cmpl-1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"a title"},"finish_reason":"stop"}]}`) + }) + + c := New(Config{APIKey: "k", BaseURL: srv.URL}) + out, err := c.GenerateText(context.Background(), "", "you are helpful", "make a title") + if err != nil { + t.Fatalf("GenerateText failed: %v", err) + } + if out != "a title" { + t.Fatalf("expected %q, got %q", "a title", out) + } + if !sawSystem { + t.Fatal("expected system message to be sent") + } +} + +func TestChatStream(t *testing.T) { + srv := stubUpstream(t, func(w http.ResponseWriter, _ map[string]any) { + w.Header().Set("Content-Type", "text/event-stream") + flusher, _ := w.(http.Flusher) + chunks := []string{ + `{"id":"c1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"Hel"}}]}`, + `{"id":"c1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"lo"}}]}`, + } + for _, ch := range chunks { + _, _ = io.WriteString(w, "data: "+ch+"\n\n") + if flusher != nil { + flusher.Flush() + } + } + _, _ = io.WriteString(w, "data: [DONE]\n\n") + }) + + c := New(Config{APIKey: "k", BaseURL: srv.URL}) + stream, err := c.ChatStream(context.Background(), openai.ChatCompletionNewParams{ + Messages: []openai.ChatCompletionMessageParamUnion{openai.UserMessage("hi")}, + }) + if err != nil { + t.Fatalf("ChatStream failed: %v", err) + } + defer stream.Close() + + var content strings.Builder + for stream.Next() { + chunk := stream.Current() + if len(chunk.Choices) > 0 { + content.WriteString(chunk.Choices[0].Delta.Content) + } + } + if err := stream.Err(); err != nil { + t.Fatalf("stream error: %v", err) + } + if content.String() != "Hello" { + t.Fatalf("expected assembled content %q, got %q", "Hello", content.String()) + } +}