CrossRelay stage 2: Codex /v1/responses relay, config.toml manager, Codex page

This commit is contained in:
kate522 committed 2026-10-18 18:35:01 +08:00
1 parent 404704171d
commit 5112f7ca06
30 files changed
+3539 -72

No files matched your search

+257
View File
@@ -62,6 +62,7 @@ type testEnv struct {
upstream *fakeUpstream
store *config.Store
settings string
codex string
}
func newTestEnv(t *testing.T) *testEnv {
@@ -133,6 +134,262 @@ func messagesBody(model string, stream bool) map[string]any {
}
}
// newCodexTestEnv builds the standard environment with a realistic
// config.toml in place and the shared provider activated for Codex.
func newCodexTestEnv(t *testing.T) *testEnv {
t.Helper()
env := newTestEnv(t)
codexHome := t.TempDir()
t.Setenv("CODEX_HOME", codexHome)
env.codex = filepath.Join(codexHome, "config.toml")
if err := os.WriteFile(env.codex, []byte("model_provider = \"custom\"\n\n[model_providers.custom]\nname = \"custom\"\nwire_api = \"responses\"\nbase_url = \"https://old.example.com/v1\"\n"), 0o600); err != nil {
t.Fatalf("write config.toml: %v", err)
}
if _, err := env.store.ActivateCodex(env.store.Config().Providers[0].ID); err != nil {
t.Fatalf("ActivateCodex: %v", err)
}
return env
}
func (e *testEnv) get(t *testing.T, path string) map[string]any {
t.Helper()
resp, err := http.Get(e.gateway.URL + path)
if err != nil {
t.Fatalf("get %s: %v", path, err)
}
defer resp.Body.Close()
var out map[string]any
_ = json.NewDecoder(resp.Body).Decode(&out)
return out
}
func responsesBody(model string, stream bool) map[string]any {
return map[string]any{
"model": model,
"instructions": "be brief",
"input": []any{map[string]any{
"type": "message", "role": "user",
"content": []any{map[string]any{"type": "input_text", "text": "hello"}},
}},
"stream": stream,
}
}
func TestResponsesNonStream(t *testing.T) {
env := newCodexTestEnv(t)
resp := env.post(t, "/v1/responses", responsesBody("gpt-5-codex", false), env.store.Config().RelayToken)
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("status = %d", resp.StatusCode)
}
var out map[string]any
if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
t.Fatalf("decode: %v", err)
}
if out["object"] != "response" || out["status"] != "completed" {
t.Errorf("envelope = %v", out)
}
if out["model"] != "gpt-5-codex" {
t.Errorf("model = %v, want the requested model echoed", out["model"])
}
if out["output_text"] != "Hi there" {
t.Errorf("output_text = %v", out["output_text"])
}
usage := out["usage"].(map[string]any)
if usage["input_tokens"].(float64) != 9 || usage["output_tokens"].(float64) != 2 {
t.Errorf("usage = %v", usage)
}
if env.upstream.lastModel != "ds-chat" {
t.Errorf("upstream model = %q, want the resolved model", env.upstream.lastModel)
}
if env.upstream.lastAuth != "Bearer up-key" {
t.Errorf("upstream auth = %q", env.upstream.lastAuth)
}
msgs := env.upstream.lastRequest["messages"].([]any)
system := msgs[0].(map[string]any)
if system["role"] != "system" || system["content"] != "be brief" {
t.Errorf("system message = %v", system)
}
user := msgs[1].(map[string]any)
if user["role"] != "user" || user["content"] != "hello" {
t.Errorf("user message = %v", user)
}
}
func TestResponsesStream(t *testing.T) {
env := newCodexTestEnv(t)
resp := env.post(t, "/v1/responses", responsesBody("gpt-5-codex", true), env.store.Config().RelayToken)
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("status = %d", resp.StatusCode)
}
if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
t.Errorf("content type = %q", ct)
}
body, err := io.ReadAll(resp.Body)
if err != nil {
t.Fatalf("read stream: %v", err)
}
var names []string
var text strings.Builder
for _, block := range strings.Split(strings.TrimSpace(string(body)), "\n\n") {
if block == "" {
continue
}
for _, line := range strings.Split(block, "\n") {
switch {
case strings.HasPrefix(line, "event: "):
names = append(names, strings.TrimPrefix(line, "event: "))
case strings.HasPrefix(line, "data: "):
var payload map[string]any
if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &payload); err == nil {
if payload["type"] == "response.output_text.delta" {
text.WriteString(payload["delta"].(string))
}
}
}
}
}
want := []string{
"response.created", "response.in_progress",
"response.output_item.added", "response.content_part.added",
"response.output_text.delta", "response.output_text.delta",
"response.output_text.done", "response.content_part.done",
"response.output_item.done", "response.completed",
}
if len(names) != len(want) {
t.Fatalf("events = %v, want %v", names, want)
}
for i := range want {
if names[i] != want[i] {
t.Fatalf("events = %v, want %v", names, want)
}
}
if text.String() != "Hi there" {
t.Errorf("streamed text = %q", text.String())
}
}
func TestResponsesRejectsBadToken(t *testing.T) {
env := newCodexTestEnv(t)
resp := env.post(t, "/v1/responses", responsesBody("gpt-5-codex", false), "crl_wrong")
defer resp.Body.Close()
if resp.StatusCode != http.StatusUnauthorized {
t.Errorf("status = %d, want 401", resp.StatusCode)
}
}
func TestResponsesWithoutActiveProvider(t *testing.T) {
env := newTestEnv(t)
t.Setenv("CODEX_HOME", t.TempDir())
resp := env.post(t, "/v1/responses", responsesBody("gpt-5-codex", false), env.store.Config().RelayToken)
defer resp.Body.Close()
if resp.StatusCode != http.StatusBadRequest {
t.Fatalf("status = %d, want 400", resp.StatusCode)
}
var out map[string]any
_ = json.NewDecoder(resp.Body).Decode(&out)
detail, ok := out["error"].(map[string]any)
if !ok || !strings.Contains(detail["message"].(string), "no active provider") {
t.Errorf("error = %v", out["error"])
}
}
func TestApplyAndRestoreCodex(t *testing.T) {
env := newCodexTestEnv(t)
id := env.store.Config().Providers[0].ID
resp := env.post(t, "/api/providers/"+id+"/apply", map[string]any{"target": "codex"}, "")
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("apply status = %d", resp.StatusCode)
}
var out map[string]any
_ = json.NewDecoder(resp.Body).Decode(&out)
injection, ok := out["injection"].(map[string]any)
if !ok || injection["model_provider"] != "crossrelay" {
t.Fatalf("injection = %v", out["injection"])
}
table := injection["table"].(map[string]any)
if table["wire_api"] != "responses" || table["base_url"] != "http://127.0.0.1:18787" {
t.Errorf("injected table = %v", table)
}
if table["experimental_bearer_token"] != env.store.Config().RelayToken {
t.Errorf("injected token = %v", table["experimental_bearer_token"])
}
data, err := os.ReadFile(env.codex)
if err != nil {
t.Fatalf("read config.toml: %v", err)
}
got := string(data)
for _, want := range []string{
`model_provider = "crossrelay"`,
"[model_providers.crossrelay]",
"requires_openai_auth = false",
"[model_providers.custom]",
`base_url = "https://old.example.com/v1"`,
} {
if !strings.Contains(got, want) {
t.Errorf("config.toml missing %q:\n%s", want, got)
}
}
if env.store.Config().CodexPriorProvider != "custom" {
t.Errorf("prior provider = %q, want custom", env.store.Config().CodexPriorProvider)
}
if status := env.get(t, "/api/status"); status["active_codex_provider"] == nil {
t.Errorf("status lacks active_codex_provider: %v", status)
}
resp = env.post(t, "/api/restore", map[string]any{"target": "codex"}, "")
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("restore status = %d", resp.StatusCode)
}
data, _ = os.ReadFile(env.codex)
got = string(data)
if !strings.Contains(got, `model_provider = "custom"`) {
t.Errorf("prior provider not restored:\n%s", got)
}
if strings.Contains(got, "crossrelay") {
t.Errorf("managed keys survived restore:\n%s", got)
}
if !strings.Contains(got, `base_url = "https://old.example.com/v1"`) {
t.Errorf("user's provider table damaged:\n%s", got)
}
if env.store.ActiveCodex() != nil {
t.Error("active codex provider survived restore")
}
if env.store.Config().CodexPriorProvider != "" {
t.Error("prior provider record survived restore")
}
}
func TestBackupsTargetDimension(t *testing.T) {
env := newCodexTestEnv(t)
id := env.store.Config().Providers[0].ID
resp := env.post(t, "/api/providers/"+id+"/apply", map[string]any{"target": "claude"}, "")
resp.Body.Close()
resp = env.post(t, "/api/providers/"+id+"/apply", map[string]any{"target": "codex"}, "")
resp.Body.Close()
codexBackups := env.get(t, "/api/backups?target=codex")["items"].([]any)
if len(codexBackups) != 1 {
t.Fatalf("codex backups = %d, want 1", len(codexBackups))
}
if name := codexBackups[0].(map[string]any)["name"].(string); !strings.HasPrefix(name, "config-") {
t.Errorf("codex backup name = %q", name)
}
claudeBackups := env.get(t, "/api/backups")["items"].([]any)
if len(claudeBackups) != 1 {
t.Fatalf("claude backups = %d, want 1", len(claudeBackups))
}
if name := claudeBackups[0].(map[string]any)["name"].(string); !strings.HasPrefix(name, "settings-") {
t.Errorf("claude backup name = %q", name)
}
}
func TestMessagesNonStream(t *testing.T) {
env := newTestEnv(t)
resp := env.post(t, "/v1/messages", messagesBody("claude-sonnet-4-5", false), env.store.Config().RelayToken)
+162 -17
View File
@@ -6,6 +6,7 @@ import (
"time"
"gitea.rutua.cn/kate522/CrossRelay/internal/claudecfg"
"gitea.rutua.cn/kate522/CrossRelay/internal/codexcfg"
"gitea.rutua.cn/kate522/CrossRelay/internal/config"
)
@@ -29,12 +30,22 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
"models": p.Models,
}
}
var activeCodex any
if p := s.store.ActiveCodex(); p != nil {
activeCodex = map[string]any{
"id": p.ID,
"name": p.Name,
"base_url": p.BaseURL,
"models": p.Models,
}
}
writeJSON(w, http.StatusOK, map[string]any{
"running": true,
"version": Version,
"port": s.port,
"uptime_sec": int(time.Since(s.started).Seconds()),
"active_claude_provider": active,
"active_codex_provider": activeCodex,
"stats": s.snapshotStats(),
"language": cfg.Language,
"port_mode": cfg.Gateway.PortMode,
@@ -84,10 +95,33 @@ func (s *Server) handleDeleteProvider(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]any{"deleted": id})
}
// handleApply activates a provider and writes it into Claude Code's
// settings.json.
// applyInput selects which client configuration receives the provider.
type applyInput struct {
Target string `json:"target"`
}
// handleApply activates a provider and writes it into the target
// client's configuration. The body is optional; an absent or empty
// target means Claude Code, keeping older clients working.
func (s *Server) handleApply(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var in applyInput
if r.Body != nil {
_ = decodeJSON(r, &in, 1<<20) // an empty body leaves the zero value
}
switch in.Target {
case "", "claude":
s.applyClaude(w, id)
case "codex":
s.applyCodex(w, id)
default:
writeAPIError(w, http.StatusBadRequest, `target must be "claude" or "codex"`)
}
}
// applyClaude activates a provider and writes it into Claude Code's
// settings.json.
func (s *Server) applyClaude(w http.ResponseWriter, id string) {
p, err := s.store.ActivateClaude(id)
if err != nil {
writeAPIError(w, http.StatusNotFound, err.Error())
@@ -109,18 +143,84 @@ func (s *Server) handleApply(w http.ResponseWriter, r *http.Request) {
}
writeJSON(w, http.StatusOK, map[string]any{
"provider": p,
"target": "claude",
"env": settings.Env,
})
}
// handleRestore puts back the most recent backup of settings.json.
// applyCodex activates a provider and writes it into Codex's
// config.toml.
func (s *Server) applyCodex(w http.ResponseWriter, id string) {
p, err := s.store.ActivateCodex(id)
if err != nil {
writeAPIError(w, http.StatusNotFound, err.Error())
return
}
mgr, err := s.codexManager()
if err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
injection, err := mgr.Apply(codexcfg.ApplyOptions{
BaseURL: fmt.Sprintf("http://127.0.0.1:%d", s.port),
Token: s.store.Config().RelayToken,
})
if err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
// Remember the provider the user had before the first apply so
// restore can put it back instead of dropping the key.
if injection.PriorProvider != "" && s.store.Config().CodexPriorProvider == "" {
if err := s.store.SetCodexPriorProvider(injection.PriorProvider); err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
}
writeJSON(w, http.StatusOK, map[string]any{
"provider": p,
"target": "codex",
"injection": injection,
})
}
// restoreInput selects the client and, optionally, one specific backup
// file to roll back to.
type restoreInput struct {
Target string `json:"target"`
Backup string `json:"backup,omitempty"`
}
// handleRestore puts a client's configuration back. Claude restores
// the newest backup (or the named one); Codex removes the managed
// keys, or rolls back to the named backup.
func (s *Server) handleRestore(w http.ResponseWriter, r *http.Request) {
var in restoreInput
if r.Body != nil {
_ = decodeJSON(r, &in, 1<<20)
}
switch in.Target {
case "", "claude":
s.restoreClaude(w, in.Backup)
case "codex":
s.restoreCodex(w, in.Backup)
default:
writeAPIError(w, http.StatusBadRequest, `target must be "claude" or "codex"`)
}
}
func (s *Server) restoreClaude(w http.ResponseWriter, backup string) {
mgr, err := s.claudeManager()
if err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
name, err := mgr.Restore()
var name string
if backup != "" {
name, err = mgr.RestoreFrom(backup)
} else {
name, err = mgr.Restore()
}
if err != nil {
writeAPIError(w, http.StatusNotFound, err.Error())
return
@@ -129,21 +229,66 @@ func (s *Server) handleRestore(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]any{"restored_from": name})
}
func (s *Server) restoreCodex(w http.ResponseWriter, backup string) {
mgr, err := s.codexManager()
if err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
if backup != "" {
name, err := mgr.RestoreBackup(backup)
if err != nil {
writeAPIError(w, http.StatusNotFound, err.Error())
return
}
_ = s.store.ClearActiveCodex()
writeJSON(w, http.StatusOK, map[string]any{"restored_from": name})
return
}
if err := mgr.RemoveManaged(s.store.Config().CodexPriorProvider); err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
_ = s.store.ClearActiveCodex()
_ = s.store.SetCodexPriorProvider("")
writeJSON(w, http.StatusOK, map[string]any{"restored_from": ""})
}
func (s *Server) handleBackups(w http.ResponseWriter, r *http.Request) {
mgr, err := s.claudeManager()
if err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
switch target := r.URL.Query().Get("target"); target {
case "", "claude":
mgr, err := s.claudeManager()
if err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
items, err := mgr.Backups()
if err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
if items == nil {
items = []claudecfg.BackupInfo{}
}
writeJSON(w, http.StatusOK, map[string]any{"items": items})
case "codex":
mgr, err := s.codexManager()
if err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
items, err := mgr.Backups()
if err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
if items == nil {
items = []codexcfg.BackupInfo{}
}
writeJSON(w, http.StatusOK, map[string]any{"items": items})
default:
writeAPIError(w, http.StatusBadRequest, `target must be "claude" or "codex"`)
}
items, err := mgr.Backups()
if err != nil {
writeAPIError(w, http.StatusInternalServerError, err.Error())
return
}
if items == nil {
items = []claudecfg.BackupInfo{}
}
writeJSON(w, http.StatusOK, map[string]any{"items": items})
}
func (s *Server) handleGetSettings(w http.ResponseWriter, r *http.Request) {
+118
View File
@@ -0,0 +1,118 @@
package gateway
import (
"errors"
"fmt"
"io"
"net/http"
"gitea.rutua.cn/kate522/CrossRelay/internal/chatcmpl"
"gitea.rutua.cn/kate522/CrossRelay/internal/responses"
)
// handleResponses serves POST /v1/responses in both streaming and
// non-streaming modes.
func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) {
var req responses.Request
if err := decodeJSON(r, &req, 64<<20); err != nil {
writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", "invalid request body: "+err.Error())
return
}
provider := s.store.ActiveCodex()
if provider == nil {
writeResponsesError(w, http.StatusBadRequest, "invalid_request_error",
"no active provider: activate one in the CrossRelay app")
return
}
model := provider.ResolveModel(req.Model)
chatReq, err := responses.ConvertRequest(&req, model)
if err != nil {
writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
return
}
rec := recordFrom(r)
rec.update(func(e *LogEntry) {
e.Model = model
e.InputTokens = responses.EstimateTokens(chatReq)
})
client := chatcmpl.NewClient(provider.BaseURL, provider.APIKey)
if req.Stream {
s.handleResponsesStream(w, r, client, chatReq, &req, model, rec)
return
}
resp, err := client.Create(r.Context(), chatReq)
if err != nil {
s.failUpstreamResponses(w, rec, err)
return
}
out := responses.ConvertResponse(resp, req.Model)
rec.update(func(e *LogEntry) {
if out.Usage != nil {
e.OutputTokens = out.Usage.OutputTokens
}
})
writeJSON(w, http.StatusOK, out)
}
// handleResponsesStream relays a Chat Completions SSE stream as
// Responses events.
func (s *Server) handleResponsesStream(w http.ResponseWriter, r *http.Request, client *chatcmpl.Client,
chatReq *chatcmpl.Request, req *responses.Request, model string, rec *record) {
resp, err := client.CreateStream(r.Context(), chatReq)
if err != nil {
s.failUpstreamResponses(w, rec, err)
return
}
defer resp.Body.Close()
conv := responses.NewStreamConverter(w, req.Model, responses.EstimateTokens(chatReq))
if err := conv.WriteStart(); err != nil {
return
}
stream := chatcmpl.NewStream(resp.Body)
for {
chunk, err := stream.Next()
if errors.Is(err, io.EOF) {
break
}
if err != nil {
rec.update(func(e *LogEntry) { e.Error = err.Error() })
conv.WriteError("api_error", err.Error())
return
}
if err := conv.OnChunk(chunk); err != nil {
rec.update(func(e *LogEntry) { e.Error = err.Error() })
conv.WriteError("api_error", err.Error())
return
}
}
if err := conv.Finish(); err != nil {
rec.update(func(e *LogEntry) { e.Error = err.Error() })
}
}
// failUpstreamResponses maps an upstream failure onto a Responses
// error envelope.
func (s *Server) failUpstreamResponses(w http.ResponseWriter, rec *record, err error) {
rec.update(func(e *LogEntry) { e.Error = err.Error() })
var ue *chatcmpl.UpstreamError
if errors.As(err, &ue) {
switch {
case ue.StatusCode == http.StatusUnauthorized || ue.StatusCode == http.StatusForbidden:
writeResponsesError(w, http.StatusUnauthorized, "authentication_error", ue.Message)
case ue.StatusCode == http.StatusTooManyRequests:
writeResponsesError(w, http.StatusTooManyRequests, "rate_limit_error", ue.Message)
case ue.StatusCode == http.StatusBadRequest:
writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", ue.Message)
default:
writeResponsesError(w, http.StatusBadGateway, "api_error", ue.Message)
}
return
}
writeResponsesError(w, http.StatusBadGateway, "api_error", fmt.Sprintf("upstream request failed: %v", err))
}
+19 -1
View File
@@ -1,5 +1,6 @@
// Package gateway serves the Anthropic Messages API to Claude Code and
// the local management API to the CrossRelay UI.
// the Responses API to Codex, plus the local management API to the
// CrossRelay UI.
package gateway
import (
@@ -13,7 +14,9 @@ import (
"gitea.rutua.cn/kate522/CrossRelay/internal/anthropic"
"gitea.rutua.cn/kate522/CrossRelay/internal/claudecfg"
"gitea.rutua.cn/kate522/CrossRelay/internal/codexcfg"
"gitea.rutua.cn/kate522/CrossRelay/internal/config"
"gitea.rutua.cn/kate522/CrossRelay/internal/responses"
)
// Version is reported by /api/status.
@@ -70,6 +73,7 @@ func (s *Server) Handler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("POST /v1/messages", s.withAuth(s.handleMessages))
mux.HandleFunc("POST /v1/messages/count_tokens", s.withAuth(s.handleCountTokens))
mux.HandleFunc("POST /v1/responses", s.withAuth(s.handleResponses))
mux.HandleFunc("GET /api/status", s.handleStatus)
mux.HandleFunc("GET /api/providers", s.handleListProviders)
mux.HandleFunc("POST /api/providers", s.handleCreateProvider)
@@ -199,6 +203,14 @@ func (s *Server) claudeManager() (*claudecfg.Manager, error) {
return claudecfg.New(path, filepath.Join(config.Dir(), "backups")), nil
}
func (s *Server) codexManager() (*codexcfg.Manager, error) {
path, err := codexcfg.DefaultPath()
if err != nil {
return nil, err
}
return codexcfg.New(path, filepath.Join(config.Dir(), "backups")), nil
}
func (s *Server) snapshotStats() Stats {
s.statsMu.Lock()
defer s.statsMu.Unlock()
@@ -222,6 +234,12 @@ func writeAPIError(w http.ResponseWriter, status int, message string) {
writeJSON(w, status, map[string]any{"error": message})
}
func writeResponsesError(w http.ResponseWriter, status int, errType, message string) {
writeJSON(w, status, responses.ErrorResponse{
Error: responses.Detail{Type: errType, Message: message},
})
}
func decodeJSON(r *http.Request, v any, limit int64) error {
dec := json.NewDecoder(io.LimitReader(r.Body, limit))
return dec.Decode(v)