DBT/clawgo
1
0
Fork 0
mirror of https://github.com/YspCoder/clawgo.git synced 2026-05-04 17:57:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

Release v1.0.2

Browse Source
This commit is contained in:
lpf
2026-04-08 15:25:28 +08:00
parent ce2263ac8c
commit a9169c66ff
15 changed files with 1670 additions and 450 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pkg/providers/codex_provider.go
View File

@@ -495,9 +495,7 @@ func (p *CodexProvider) doStreamAttempt(req *http.Request, attempt authAttempt,
if typ := strings.TrimSpace(fmt.Sprintf("%v", obj["type"])); typ == "response.completed" {
completed = true
if respObj, ok := obj["response"]; ok {
if b, err := json.Marshal(respObj); err == nil {
finalJSON = b
}
finalJSON = mergeStreamFinalJSON(finalJSON, respObj)
}
}
}

21
pkg/providers/codex_provider_test.go
View File

@@ -214,6 +214,27 @@ func TestCodexProviderChatFallsBackToHTTPStreamResponse(t *testing.T) {
}
}
func TestCodexProviderChatMergesLateUsageFromStreamingCompletion(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
_, _ = fmt.Fprint(w, "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"output_text\":\"hello\"}}\n\n")
_, _ = fmt.Fprint(w, "data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":1,\"output_tokens\":2,\"total_tokens\":3}}}\n\n")
}))
defer server.Close()
provider := NewCodexProvider("codex", "test-api-key", server.URL, "gpt-5.4", false, "", 5*time.Second, nil)
resp, err := provider.Chat(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "gpt-5.4", nil)
if err != nil {
t.Fatalf("Chat error: %v", err)
}
if resp.Content != "hello" {
t.Fatalf("unexpected response content: %q", resp.Content)
}
if resp.Usage == nil || resp.Usage.PromptTokens != 1 || resp.Usage.CompletionTokens != 2 || resp.Usage.TotalTokens != 3 {
t.Fatalf("unexpected usage: %#v", resp.Usage)
}
}
func TestCodexHandleAttemptFailureMarksAPIKeyCooldown(t *testing.T) {
provider := NewCodexProvider("codex-websocket-failure", "test-api-key", "", "gpt-5.4", false, "", 5*time.Second, nil)
provider.handleAttemptFailure(authAttempt{kind: "api_key", token: "test-api-key"}, http.StatusTooManyRequests, []byte(`{"error":{"message":"rate limit exceeded"}}`))

60
pkg/providers/http_provider.go
View File

@@ -1022,15 +1022,13 @@ func (p *HTTPProvider) doStreamAttempt(req *http.Request, attempt authAttempt, o
if err := json.Unmarshal([]byte(payload), &obj); err == nil {
if typ := strings.TrimSpace(fmt.Sprintf("%v", obj["type"])); typ == "response.completed" {
if respObj, ok := obj["response"]; ok {
if b, err := json.Marshal(respObj); err == nil {
finalJSON = b
}
finalJSON = mergeStreamFinalJSON(finalJSON, respObj)
}
}
if choices, ok := obj["choices"]; ok {
if b, err := json.Marshal(map[string]interface{}{"choices": choices, "usage": obj["usage"]}); err == nil {
finalJSON = b
}
finalJSON = mergeStreamFinalJSON(finalJSON, map[string]interface{}{"choices": choices, "usage": obj["usage"]})
} else if _, ok := obj["usage"]; ok && len(finalJSON) > 0 {
finalJSON = mergeStreamFinalJSON(finalJSON, map[string]interface{}{"usage": obj["usage"]})
}
}
}
@@ -1049,6 +1047,56 @@ func (p *HTTPProvider) doStreamAttempt(req *http.Request, attempt authAttempt, o
return finalJSON, resp.StatusCode, ctype, false, nil
}
func mergeStreamFinalJSON(existing []byte, incoming interface{}) []byte {
if incoming == nil {
return existing
}
incomingMap, ok := incoming.(map[string]interface{})
if !ok {
data, err := json.Marshal(incoming)
if err != nil {
return existing
}
return data
}
if len(existing) == 0 {
data, err := json.Marshal(incomingMap)
if err != nil {
return existing
}
return data
}
var merged map[string]interface{}
if err := json.Unmarshal(existing, &merged); err != nil || merged == nil {
merged = map[string]interface{}{}
}
merged = mergeStringAnyMaps(merged, incomingMap)
data, err := json.Marshal(merged)
if err != nil {
return existing
}
return data
}
func mergeStringAnyMaps(dst, src map[string]interface{}) map[string]interface{} {
if dst == nil {
dst = map[string]interface{}{}
}
for key, value := range src {
if value == nil {
continue
}
if nestedSrc, ok := value.(map[string]interface{}); ok {
if nestedDst, ok := dst[key].(map[string]interface{}); ok {
dst[key] = mergeStringAnyMaps(nestedDst, nestedSrc)
continue
}
}
dst[key] = value
}
return dst
}
func shouldRetryOAuthQuota(status int, body []byte) bool {
_, retry := classifyOAuthFailure(status, body)
return retry

38
pkg/providers/oauth_test.go
View File

@@ -197,6 +197,44 @@ func TestHTTPProviderOAuthSwitchesAccountOnQuota(t *testing.T) {
}
}
func TestHTTPProviderOpenAICompatStreamMergesLateUsage(t *testing.T) {
t.Parallel()
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/v1/chat/completions" {
http.NotFound(w, r)
return
}
w.Header().Set("Content-Type", "text/event-stream")
_, _ = w.Write([]byte("data: {\"choices\":[{\"index\":0,\"message\":{\"content\":\"hello\"},\"finish_reason\":\"stop\"}]}\n\n"))
_, _ = w.Write([]byte("data: {\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":2,\"total_tokens\":3}}\n\n"))
}))
defer server.Close()
provider := NewHTTPProvider("openai", "token", server.URL+"/v1", "gpt-test", false, "api_key", 5*time.Second, nil)
req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, server.URL+"/v1/chat/completions", nil)
if err != nil {
t.Fatalf("new request failed: %v", err)
}
body, status, _, _, err := provider.doStreamAttempt(req, authAttempt{kind: "api_key", token: "token"}, nil)
if err != nil {
t.Fatalf("stream attempt failed: %v", err)
}
if status != http.StatusOK {
t.Fatalf("unexpected status: %d", status)
}
resp, err := parseOpenAICompatResponse(body)
if err != nil {
t.Fatalf("parse response failed: %v", err)
}
if resp.Content != "hello" {
t.Fatalf("unexpected response content: %q", resp.Content)
}
if resp.Usage == nil || resp.Usage.PromptTokens != 1 || resp.Usage.CompletionTokens != 2 || resp.Usage.TotalTokens != 3 {
t.Fatalf("unexpected usage: %#v", resp.Usage)
}
}
func TestOAuthManagerPreRefreshesExpiringSession(t *testing.T) {
t.Parallel()