Automated cherry pick of #25145: feat(aiproxy): support streaming responses for visual messages (#25146)

* feat(aiproxy): support streaming responses for visual messages

* refactor(aiproxy): share API log helpers and support input/output token usage

Extract common chatlog record lifecycle across proxy handlers, and fall back
to Responses/Anthropic usage field aliases when filling token counts.

pkg/aiproxy/chatlog/chatlog.go

@@ -191,6 +191,8 @@ func FillUsageFromJSON(rec *Record, data []byte) bool {
 			PromptTokens     int `json:"prompt_tokens"`
 			CompletionTokens int `json:"completion_tokens"`
 			TotalTokens      int `json:"total_tokens"`
+			InputTokens      int `json:"input_tokens"`
+			OutputTokens     int `json:"output_tokens"`
 		} `json:"usage"`
 	}
 	if rec == nil || json.Unmarshal(data, &wrap) != nil || wrap.Usage == nil {
@@ -199,9 +201,27 @@ func FillUsageFromJSON(rec *Record, data []byte) bool {
 		}
 		return false
 	}
-	rec.PromptTokens = wrap.Usage.PromptTokens
-	rec.CompletionTokens = wrap.Usage.CompletionTokens
-	rec.TotalTokens = wrap.Usage.TotalTokens
+	u := wrap.Usage
+	prompt := u.PromptTokens
+	completion := u.CompletionTokens
+	total := u.TotalTokens
+	// Prefer OpenAI fields; fall back to Responses/Anthropic aliases.
+	if prompt == 0 {
+		prompt = u.InputTokens
+	}
+	if completion == 0 {
+		completion = u.OutputTokens
+	}
+	if total == 0 {
+		total = prompt + completion
+	}
+	if prompt == 0 && completion == 0 && total == 0 {
+		rec.UsageMissing = true
+		return false
+	}
+	rec.PromptTokens = prompt
+	rec.CompletionTokens = completion
+	rec.TotalTokens = total
 	rec.UsageMissing = false
 	return true
 }

pkg/aiproxy/chatlog/chatlog_usage_test.go

@@ -0,0 +1,47 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package chatlog
+
+import "testing"
+
+func TestFillUsageFromJSONOpenAI(t *testing.T) {
+	rec := &Record{}
+	ok := FillUsageFromJSON(rec, []byte(`{"usage":{"prompt_tokens":11,"completion_tokens":22,"total_tokens":33}}`))
+	if !ok || rec.UsageMissing {
+		t.Fatalf("expected openai usage ok, got ok=%v missing=%v", ok, rec.UsageMissing)
+	}
+	if rec.PromptTokens != 11 || rec.CompletionTokens != 22 || rec.TotalTokens != 33 {
+		t.Fatalf("unexpected tokens: %+v", rec)
+	}
+}
+
+func TestFillUsageFromJSONAnthropicAliases(t *testing.T) {
+	rec := &Record{}
+	ok := FillUsageFromJSON(rec, []byte(`{"usage":{"input_tokens":7,"output_tokens":9}}`))
+	if !ok || rec.UsageMissing {
+		t.Fatalf("expected anthropic usage ok, got ok=%v missing=%v", ok, rec.UsageMissing)
+	}
+	if rec.PromptTokens != 7 || rec.CompletionTokens != 9 || rec.TotalTokens != 16 {
+		t.Fatalf("unexpected tokens: prompt=%d completion=%d total=%d", rec.PromptTokens, rec.CompletionTokens, rec.TotalTokens)
+	}
+}
+
+func TestFillUsageFromJSONMissing(t *testing.T) {
+	rec := &Record{}
+	ok := FillUsageFromJSON(rec, []byte(`{"id":"resp"}`))
+	if ok || !rec.UsageMissing {
+		t.Fatalf("expected missing usage, got ok=%v missing=%v", ok, rec.UsageMissing)
+	}
+}

pkg/aiproxy/extensions/visual/messages.go

@@ -71,24 +71,25 @@ func anthropicContentHasImage(raw json.RawMessage) bool {
 	return false
 }
 
-// ShouldHandleMessages reports whether the Messages visual path should run (non-stream only).
+// ShouldHandleMessages reports whether the Messages visual path should run.
+// Streaming requests with images use non-streaming orchestration and synthetic SSE.
 func ShouldHandleMessages(dict *jsonutils.JSONDict, up *models.ChatUpstream, isStream bool) bool {
-	if isStream || dict == nil || up == nil || !Enabled(up) {
+	_ = isStream
+	if dict == nil || up == nil || !Enabled(up) {
 		return false
 	}
 	return AnthropicMessagesHasImage(dict)
 }
 
-// ShouldRejectMessagesStreaming reports stream+visual+image (unsupported).
+// ShouldRejectMessagesStreaming is deprecated: stream+visual+image now uses synthetic SSE.
+// Kept for compatibility; always returns false.
 func ShouldRejectMessagesStreaming(dict *jsonutils.JSONDict, up *models.ChatUpstream, isStream bool) bool {
-	if !isStream || dict == nil || up == nil || !Enabled(up) {
-		return false
-	}
-	return AnthropicMessagesHasImage(dict)
+	_, _, _ = dict, up, isStream
+	return false
 }
 
-// HandleMessagesCreate runs visual orchestration for a non-streaming Anthropic Messages request.
-func HandleMessagesCreate(
+// HandleMessagesCreateChat runs visual orchestration and returns the upstream chat completion body.
+func HandleMessagesCreateChat(
 	ctx context.Context,
 	dict *jsonutils.JSONDict,
 	textUp *models.ChatUpstream,
@@ -111,11 +112,21 @@ func HandleMessagesCreate(
 		return nil, err
 	}
 	chatClone := cloneDict(chatBody)
+	forceNonStreamChatBody(chatClone)
 	textUpChat := *textUp
 	textUpChat.BaseURL = ChatBaseURL(textUp.BaseURL)
 	visUpChat := *visUp
 	visUpChat.BaseURL = ChatBaseURL(visUp.BaseURL)
-	respBody, err := RunChatOrchestrator(ctx, &textUpChat, &visUpChat, chatClone, runtime)
+	return RunChatOrchestrator(ctx, &textUpChat, &visUpChat, chatClone, runtime)
+}
+
+// HandleMessagesCreate runs visual orchestration for a non-streaming Anthropic Messages request.
+func HandleMessagesCreate(
+	ctx context.Context,
+	dict *jsonutils.JSONDict,
+	textUp *models.ChatUpstream,
+) ([]byte, error) {
+	respBody, err := HandleMessagesCreateChat(ctx, dict, textUp)
 	if err != nil {
 		return nil, err
 	}

pkg/aiproxy/extensions/visual/visual_test.go

@@ -128,6 +128,47 @@ func TestAnthropicMessagesHasImage(t *testing.T) {
 	}
 }
 
+func TestShouldHandleMessagesIgnoresStream(t *testing.T) {
+	cfg := &api.SAiModelConfig{
+		Extensions: &api.SAiModelExtensions{
+			Visual: &api.SAiModelVisualConfig{Enabled: true},
+		},
+	}
+	up := &models.ChatUpstream{
+		ModelConfig:      cfg,
+		VisualProviderId: "prov-1",
+		VisualModelKey:   "vision-model",
+	}
+	body, _ := jsonutils.Parse([]byte(`{"messages":[{"role":"user","content":[{"type":"image","source":{"type":"url","url":"https://x/a.png"}}]}]}`))
+	if !ShouldHandleMessages(body.(*jsonutils.JSONDict), up, false) {
+		t.Fatal("expected visual handle non-stream with image")
+	}
+	if !ShouldHandleMessages(body.(*jsonutils.JSONDict), up, true) {
+		t.Fatal("stream with image should use visual orchestrator")
+	}
+	bodyText, _ := jsonutils.Parse([]byte(`{"messages":[{"role":"user","content":"hello"}]}`))
+	if ShouldHandleMessages(bodyText.(*jsonutils.JSONDict), up, true) {
+		t.Fatal("stream without image should not use visual orchestrator")
+	}
+}
+
+func TestShouldRejectMessagesStreamingAlwaysFalse(t *testing.T) {
+	cfg := &api.SAiModelConfig{
+		Extensions: &api.SAiModelExtensions{
+			Visual: &api.SAiModelVisualConfig{Enabled: true},
+		},
+	}
+	up := &models.ChatUpstream{
+		ModelConfig:      cfg,
+		VisualProviderId: "prov-1",
+		VisualModelKey:   "vision-model",
+	}
+	body, _ := jsonutils.Parse([]byte(`{"messages":[{"role":"user","content":[{"type":"image","source":{"type":"url","url":"https://x/a.png"}}]}]}`))
+	if ShouldRejectMessagesStreaming(body.(*jsonutils.JSONDict), up, true) {
+		t.Fatal("stream+visual+image must not reject after synthetic SSE support")
+	}
+}
+
 func TestForceNonStreamChatBody(t *testing.T) {
 	body := jsonutils.NewDict()
 	body.Set("stream", jsonutils.JSONTrue)

pkg/aiproxy/handlers/chat_completions.go

@@ -25,7 +25,6 @@ import (
 	"time"
 
 	"yunion.io/x/jsonutils"
-	"yunion.io/x/pkg/appctx"
 
 	"yunion.io/x/onecloud/pkg/aiproxy/chatlog"
 	"yunion.io/x/onecloud/pkg/aiproxy/models"
@@ -106,30 +105,11 @@ func streamChunksWithCancel(ch <-chan upstream.StreamChunk, cancel context.Cance
 // Upstream is resolved: ai_virtual_key -> project ai_routing -> ai_routing_model -> ai_key (by catalog model_key).
 func chatCompletionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) {
 	start := time.Now()
-	rec := &chatlog.Record{
-		RequestID: appctx.AppContextRequestId(ctx),
-		Timestamp: start,
-		Path:      r.URL.Path,
-		Client:    r.RemoteAddr,
-	}
-	defer func() {
-		if rec.StatusCode == 0 {
-			rec.StatusCode = http.StatusInternalServerError
-		}
-		rec.LatencyMs = time.Since(start).Milliseconds()
-		chatlog.Write(rec)
-	}()
-	fail := func(status int, code string, err error) {
-		rec.StatusCode = status
-		rec.Success = false
-		rec.ErrorCode = code
-		if err != nil {
-			rec.ErrorMessage = err.Error()
-		}
-	}
+	rec := newAPILogRecord(ctx, r, start)
+	defer finishAPILogRecord(rec, start)
 
 	if r.Method != http.MethodPost {
-		fail(http.StatusBadRequest, "invalid_method", nil)
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_method", nil)
 		httperrors.InvalidInputError(ctx, w, "only POST is supported")
 		return
 	}
@@ -137,29 +117,24 @@ func chatCompletionsHandler(ctx context.Context, w http.ResponseWriter, r *http.
 	defer r.Body.Close()
 	raw, err := io.ReadAll(r.Body)
 	if err != nil {
-		fail(http.StatusBadRequest, "read_body", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "read_body", err)
 		httperrors.InvalidInputError(ctx, w, "read body: %v", err)
 		return
 	}
 
 	body, err := jsonutils.Parse(raw)
 	if err != nil {
-		fail(http.StatusBadRequest, "invalid_json", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_json", err)
 		httperrors.InvalidInputError(ctx, w, "invalid JSON body: %v", err)
 		return
 	}
 	dict, ok := body.(*jsonutils.JSONDict)
 	if !ok {
-		fail(http.StatusBadRequest, "invalid_body", nil)
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_body", nil)
 		httperrors.InvalidInputError(ctx, w, "body must be a JSON object")
 		return
 	}
-	rec.ModelRequested, _ = dict.GetString("model")
-	rec.Stream, _ = dict.Bool("stream")
-	rec.ToolCallEnabled = dict.Contains("tools") || dict.Contains("tool_choice")
-	if metadata, err := dict.Get("metadata"); err == nil {
-		rec.Metadata = json.RawMessage([]byte(metadata.String()))
-	}
+	fillAPILogFromBody(rec, dict)
 
 	dbg := NewProxyDebugSession(ctx, "openai-chat")
 	dbg.ClientRequest(r, dict, nil, rec.Stream)
@@ -169,32 +144,16 @@ func chatCompletionsHandler(ctx context.Context, w http.ResponseWriter, r *http.
 	up, err := models.ResolveChatUpstream(ctx, userCred, vk, dict)
 	if err != nil {
 		dbg.Error("resolve upstream: %v", err)
-		fail(http.StatusInternalServerError, "resolve_upstream", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "resolve_upstream", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
 	dbg.RoutingResolved(dict, up)
-	rec.VirtualKey = up.VirtualKeyId
-	rec.ProjectID = up.ProjectId
-	rec.DomainID = up.DomainId
-	rec.AiKey = up.AiKeyId
-	rec.ModelFinal = up.UpstreamModel
-	rec.Provider = up.ProviderKey
-	if up.RoutingLog != nil {
-		rec.RoutingEnabled = up.RoutingLog.Enabled
-		rec.RoutingCandidates = up.RoutingLog.Candidates
-		rec.RoutingSelectedModel = up.RoutingLog.SelectedModel
-		rec.RoutingMethod = up.RoutingLog.Method
-		rec.RoutingScores = up.RoutingLog.Scores
-		rec.RoutingConfidence = up.RoutingLog.Confidence
-		rec.RoutingReason = up.RoutingLog.Reason
-		rec.RoutingLatencyMs = up.RoutingLog.LatencyMs
-		rec.RoutingError = up.RoutingLog.Error
-	}
+	fillAPILogFromUpstream(rec, up)
 
 	if err := models.TakeVirtualKeyRequestsPerMinute(up.VirtualKeyId, up.RequestsPerMinute); err != nil {
 		dbg.Error("rate limit: %v", err)
-		fail(http.StatusInternalServerError, "rate_limit", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "rate_limit", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
@@ -206,7 +165,7 @@ func chatCompletionsHandler(ctx context.Context, w http.ResponseWriter, r *http.
 	}
 	if err := models.EnforceVirtualKeyMaxTokens(dict, vkLim); err != nil {
 		dbg.Error("max tokens: %v", err)
-		fail(http.StatusInternalServerError, "max_tokens", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "max_tokens", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
@@ -216,7 +175,7 @@ func chatCompletionsHandler(ctx context.Context, w http.ResponseWriter, r *http.
 	chatCtx := providers.ChatContextFromUpstream(up.ProviderKey, up.BaseURL, up.APIKey, up.UpstreamModel, up.APIMode)
 	if _, err := prov.BuildUpstreamRequest(chatCtx, dict, isStream); err != nil {
 		dbg.Error("provider request: %v", err)
-		fail(http.StatusBadRequest, "provider_request", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "provider_request", err)
 		httperrors.InvalidInputError(ctx, w, "provider request: %v", err)
 		return
 	}
@@ -241,18 +200,16 @@ func chatCompletionsHandler(ctx context.Context, w http.ResponseWriter, r *http.
 			return
 		}
 		dbg.UpstreamResponse(resp.Body)
-		body := resp.Body
-		if norm, nerr := prov.NormalizeResponse(body); nerr == nil && len(norm) > 0 {
-			body = norm
+		out := resp.Body
+		if norm, nerr := prov.NormalizeResponse(out); nerr == nil && len(norm) > 0 {
+			out = norm
 		}
-		dbg.ClientResponse(body)
-		chatlog.FillUsageFromJSON(rec, body)
-		chatlog.FillToolCallsFromJSON(rec, body)
-		rec.Success = true
-		rec.StatusCode = http.StatusOK
+		dbg.ClientResponse(out)
+		chatlog.FillToolCallsFromJSON(rec, out)
+		markAPILogSuccess(rec, out)
 		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
-		_, _ = w.Write(body)
+		_, _ = w.Write(out)
 		return
 	}
 
@@ -282,8 +239,8 @@ func chatCompletionsHandler(ctx context.Context, w http.ResponseWriter, r *http.
 		if len(chunk.Data) == 0 {
 			continue
 		}
-		if bytes.Contains(chunk.Data, []byte(`"usage"`)) && chatlog.FillUsageFromJSON(rec, chunk.Data) {
-			sawUsage = true
+		if bytes.Contains(chunk.Data, []byte(`"usage"`)) {
+			noteStreamUsage(rec, chunk.Data, &sawUsage)
 		}
 		chatlog.FillToolCallsFromJSON(rec, chunk.Data)
 		if isUpstreamErrorChunk(chunk.Data) {
@@ -305,12 +262,8 @@ func chatCompletionsHandler(ctx context.Context, w http.ResponseWriter, r *http.
 	}
 	_, _ = fmt.Fprintf(w, "data: [DONE]\n\n")
 	flushIf(w)
-	rec.StatusCode = http.StatusOK
-	if !sawUsage {
-		rec.UsageMissing = true
-	}
+	finishAPILogStream(rec, streamOK, sawUsage)
 	if streamOK {
-		rec.Success = true
 		models.RecordAiKeySuccess(up.AiKeyId)
 	}
 }

pkg/aiproxy/handlers/chatlog_record.go

@@ -0,0 +1,132 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package handlers
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"time"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/pkg/appctx"
+
+	"yunion.io/x/onecloud/pkg/aiproxy/chatlog"
+	"yunion.io/x/onecloud/pkg/aiproxy/models"
+)
+
+func newAPILogRecord(ctx context.Context, r *http.Request, start time.Time) *chatlog.Record {
+	return &chatlog.Record{
+		RequestID: appctx.AppContextRequestId(ctx),
+		Timestamp: start,
+		Path:      r.URL.Path,
+		Client:    r.RemoteAddr,
+	}
+}
+
+func finishAPILogRecord(rec *chatlog.Record, start time.Time) {
+	if rec == nil {
+		return
+	}
+	if rec.StatusCode == 0 {
+		rec.StatusCode = http.StatusInternalServerError
+	}
+	rec.LatencyMs = time.Since(start).Milliseconds()
+	chatlog.Write(rec)
+}
+
+func failAPILogRecord(rec *chatlog.Record, status int, code string, err error) {
+	if rec == nil {
+		return
+	}
+	rec.StatusCode = status
+	rec.Success = false
+	rec.ErrorCode = code
+	if err != nil {
+		rec.ErrorMessage = err.Error()
+	}
+}
+
+func fillAPILogFromBody(rec *chatlog.Record, dict *jsonutils.JSONDict) {
+	if rec == nil || dict == nil {
+		return
+	}
+	rec.ModelRequested, _ = dict.GetString("model")
+	rec.Stream, _ = dict.Bool("stream")
+	rec.ToolCallEnabled = dict.Contains("tools") || dict.Contains("tool_choice")
+	if metadata, err := dict.Get("metadata"); err == nil {
+		rec.Metadata = json.RawMessage([]byte(metadata.String()))
+	}
+}
+
+func fillAPILogFromUpstream(rec *chatlog.Record, up *models.ChatUpstream) {
+	if rec == nil || up == nil {
+		return
+	}
+	rec.VirtualKey = up.VirtualKeyId
+	rec.ProjectID = up.ProjectId
+	rec.DomainID = up.DomainId
+	rec.AiKey = up.AiKeyId
+	rec.ModelFinal = up.UpstreamModel
+	rec.Provider = up.ProviderKey
+	if up.RoutingLog != nil {
+		rec.RoutingEnabled = up.RoutingLog.Enabled
+		rec.RoutingCandidates = up.RoutingLog.Candidates
+		rec.RoutingSelectedModel = up.RoutingLog.SelectedModel
+		rec.RoutingMethod = up.RoutingLog.Method
+		rec.RoutingScores = up.RoutingLog.Scores
+		rec.RoutingConfidence = up.RoutingLog.Confidence
+		rec.RoutingReason = up.RoutingLog.Reason
+		rec.RoutingLatencyMs = up.RoutingLog.LatencyMs
+		rec.RoutingError = up.RoutingLog.Error
+	}
+}
+
+func markAPILogSuccess(rec *chatlog.Record, body []byte) {
+	if rec == nil {
+		return
+	}
+	if len(body) > 0 {
+		chatlog.FillUsageFromJSON(rec, body)
+	} else {
+		rec.UsageMissing = true
+	}
+	rec.Success = true
+	rec.StatusCode = http.StatusOK
+}
+
+func finishAPILogStream(rec *chatlog.Record, streamOK, sawUsage bool) {
+	if rec == nil {
+		return
+	}
+	rec.StatusCode = http.StatusOK
+	if sawUsage {
+		rec.UsageMissing = false
+	} else {
+		rec.UsageMissing = true
+	}
+	if streamOK {
+		rec.Success = true
+	}
+}
+
+func noteStreamUsage(rec *chatlog.Record, data []byte, sawUsage *bool) {
+	if rec == nil || len(data) == 0 || sawUsage == nil {
+		return
+	}
+	if chatlog.FillUsageFromJSON(rec, data) {
+		*sawUsage = true
+	}
+}

pkg/aiproxy/handlers/completions.go

@@ -15,6 +15,7 @@
 package handlers
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"io"
@@ -33,7 +34,12 @@ import (
 
 // completionsHandler implements OpenAI-compatible POST /openai/v1/completions.
 func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	start := time.Now()
+	rec := newAPILogRecord(ctx, r, start)
+	defer finishAPILogRecord(rec, start)
+
 	if r.Method != http.MethodPost {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_method", nil)
 		httperrors.InvalidInputError(ctx, w, "only POST is supported")
 		return
 	}
@@ -41,37 +47,43 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	defer r.Body.Close()
 	raw, err := io.ReadAll(r.Body)
 	if err != nil {
+		failAPILogRecord(rec, http.StatusBadRequest, "read_body", err)
 		httperrors.InvalidInputError(ctx, w, "read body: %v", err)
 		return
 	}
 
 	body, err := jsonutils.Parse(raw)
 	if err != nil {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_json", err)
 		httperrors.InvalidInputError(ctx, w, "invalid JSON body: %v", err)
 		return
 	}
 	dict, ok := body.(*jsonutils.JSONDict)
 	if !ok {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_body", nil)
 		httperrors.InvalidInputError(ctx, w, "body must be a JSON object")
 		return
 	}
+	fillAPILogFromBody(rec, dict)
 
 	dbg := NewProxyDebugSession(ctx, "openai-completions")
-	isStream, _ := dict.Bool("stream")
-	dbg.ClientRequest(r, dict, nil, isStream)
+	dbg.ClientRequest(r, dict, nil, rec.Stream)
 
 	vk := extractVirtualKey(r)
 	userCred := auth.AdminCredential()
 	up, err := models.ResolveChatUpstream(ctx, userCred, vk, dict)
 	if err != nil {
 		dbg.Error("resolve upstream: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "resolve_upstream", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
 	dbg.RoutingResolved(dict, up)
+	fillAPILogFromUpstream(rec, up)
 
 	if err := models.TakeVirtualKeyRequestsPerMinute(up.VirtualKeyId, up.RequestsPerMinute); err != nil {
 		dbg.Error("rate limit: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "rate_limit", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
@@ -83,6 +95,7 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	}
 	if err := models.EnforceVirtualKeyMaxTokens(dict, vkLim); err != nil {
 		dbg.Error("max tokens: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "max_tokens", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
@@ -90,10 +103,12 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	compProv, err := providers.GetCompletions(up.ProviderKey)
 	if err != nil {
 		dbg.Error("completions provider: %v", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "provider_request", err)
 		httperrors.InvalidInputError(ctx, w, "%v", err)
 		return
 	}
 
+	isStream := rec.Stream
 	if _, err := compProv.BuildCompletionsRequest(&providers.ChatContext{
 		ProviderKey:   up.ProviderKey,
 		BaseURL:       up.BaseURL,
@@ -101,6 +116,7 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 		UpstreamModel: up.UpstreamModel,
 	}, dict, isStream); err != nil {
 		dbg.Error("provider request: %v", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "provider_request", err)
 		httperrors.InvalidInputError(ctx, w, "provider request: %v", err)
 		return
 	}
@@ -114,6 +130,7 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 		resp, uerr := completionsWithKeyFailover(ctx, up, dict, isStream, timeout, dbg)
 		if uerr != nil {
 			dbg.Error("upstream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+			recordUpstreamError(rec, uerr)
 			writeUpstreamError(ctx, w, uerr)
 			return
 		}
@@ -123,6 +140,7 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 			out = norm
 		}
 		dbg.ClientResponse(out)
+		markAPILogSuccess(rec, out)
 		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
 		_, _ = w.Write(out)
@@ -132,6 +150,7 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	ch, uerr := completionsStreamWithKeyFailover(ctx, up, dict, isStream, compProv, timeout, dbg)
 	if uerr != nil {
 		dbg.Error("upstream stream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+		recordUpstreamError(rec, uerr)
 		writeUpstreamError(ctx, w, uerr)
 		return
 	}
@@ -143,6 +162,7 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	flushIf(w)
 
 	streamOK := true
+	sawUsage := false
 	clientSeq := 0
 	for chunk := range ch {
 		if chunk.Done {
@@ -151,9 +171,14 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 		if len(chunk.Data) == 0 {
 			continue
 		}
+		if bytes.Contains(chunk.Data, []byte(`"usage"`)) {
+			noteStreamUsage(rec, chunk.Data, &sawUsage)
+		}
 		if isUpstreamErrorChunk(chunk.Data) {
 			dbg.Error("upstream stream error chunk: %s", truncateLogBytes(chunk.Data, proxyDebugLogMax))
 			streamOK = false
+			rec.Success = false
+			rec.ErrorCode, rec.ErrorMessage = parseUpstreamErrorInfo(chunk.Data)
 			models.RecordAiKeyFailure(up.AiKeyId, parseUpstreamErrorStatus(chunk.Data))
 			clientSeq++
 			dbg.ClientStreamData(clientSeq, chunk.Data)
@@ -168,6 +193,7 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	}
 	_, _ = fmt.Fprintf(w, "data: [DONE]\n\n")
 	flushIf(w)
+	finishAPILogStream(rec, streamOK, sawUsage)
 	if streamOK {
 		models.RecordAiKeySuccess(up.AiKeyId)
 	}

pkg/aiproxy/handlers/embeddings.go

@@ -31,7 +31,12 @@ import (
 
 // embeddingsHandler implements OpenAI-compatible POST /openai/v1/embeddings.
 func embeddingsHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	start := time.Now()
+	rec := newAPILogRecord(ctx, r, start)
+	defer finishAPILogRecord(rec, start)
+
 	if r.Method != http.MethodPost {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_method", nil)
 		httperrors.InvalidInputError(ctx, w, "only POST is supported")
 		return
 	}
@@ -39,20 +44,25 @@ func embeddingsHandler(ctx context.Context, w http.ResponseWriter, r *http.Reque
 	defer r.Body.Close()
 	raw, err := io.ReadAll(r.Body)
 	if err != nil {
+		failAPILogRecord(rec, http.StatusBadRequest, "read_body", err)
 		httperrors.InvalidInputError(ctx, w, "read body: %v", err)
 		return
 	}
 
 	body, err := jsonutils.Parse(raw)
 	if err != nil {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_json", err)
 		httperrors.InvalidInputError(ctx, w, "invalid JSON body: %v", err)
 		return
 	}
 	dict, ok := body.(*jsonutils.JSONDict)
 	if !ok {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_body", nil)
 		httperrors.InvalidInputError(ctx, w, "body must be a JSON object")
 		return
 	}
+	fillAPILogFromBody(rec, dict)
+	rec.Stream = false
 
 	dbg := NewProxyDebugSession(ctx, "openai-embeddings")
 	dbg.ClientRequest(r, dict, nil, false)
@@ -62,13 +72,16 @@ func embeddingsHandler(ctx context.Context, w http.ResponseWriter, r *http.Reque
 	up, err := models.ResolveChatUpstream(ctx, userCred, vk, dict)
 	if err != nil {
 		dbg.Error("resolve upstream: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "resolve_upstream", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
 	dbg.RoutingResolved(dict, up)
+	fillAPILogFromUpstream(rec, up)
 
 	if err := models.TakeVirtualKeyRequestsPerMinute(up.VirtualKeyId, up.RequestsPerMinute); err != nil {
 		dbg.Error("rate limit: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "rate_limit", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
@@ -81,6 +94,7 @@ func embeddingsHandler(ctx context.Context, w http.ResponseWriter, r *http.Reque
 		UpstreamModel: up.UpstreamModel,
 	}, dict); err != nil {
 		dbg.Error("provider request: %v", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "provider_request", err)
 		httperrors.InvalidInputError(ctx, w, "provider request: %v", err)
 		return
 	}
@@ -88,6 +102,7 @@ func embeddingsHandler(ctx context.Context, w http.ResponseWriter, r *http.Reque
 	resp, uerr := embeddingsWithKeyFailover(ctx, up, dict, 60*time.Second, dbg)
 	if uerr != nil {
 		dbg.Error("upstream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+		recordUpstreamError(rec, uerr)
 		writeUpstreamError(ctx, w, uerr)
 		return
 	}
@@ -97,6 +112,7 @@ func embeddingsHandler(ctx context.Context, w http.ResponseWriter, r *http.Reque
 		out = norm
 	}
 	dbg.ClientResponse(out)
+	markAPILogSuccess(rec, out)
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusOK)
 	_, _ = w.Write(out)

pkg/aiproxy/handlers/images_generations.go

@@ -31,7 +31,12 @@ import (
 
 // imagesGenerationsHandler implements OpenAI-compatible POST /openai/v1/images/generations.
 func imagesGenerationsHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	start := time.Now()
+	rec := newAPILogRecord(ctx, r, start)
+	defer finishAPILogRecord(rec, start)
+
 	if r.Method != http.MethodPost {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_method", nil)
 		httperrors.InvalidInputError(ctx, w, "only POST is supported")
 		return
 	}
@@ -39,20 +44,25 @@ func imagesGenerationsHandler(ctx context.Context, w http.ResponseWriter, r *htt
 	defer r.Body.Close()
 	raw, err := io.ReadAll(r.Body)
 	if err != nil {
+		failAPILogRecord(rec, http.StatusBadRequest, "read_body", err)
 		httperrors.InvalidInputError(ctx, w, "read body: %v", err)
 		return
 	}
 
 	body, err := jsonutils.Parse(raw)
 	if err != nil {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_json", err)
 		httperrors.InvalidInputError(ctx, w, "invalid JSON body: %v", err)
 		return
 	}
 	dict, ok := body.(*jsonutils.JSONDict)
 	if !ok {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_body", nil)
 		httperrors.InvalidInputError(ctx, w, "body must be a JSON object")
 		return
 	}
+	fillAPILogFromBody(rec, dict)
+	rec.Stream = false
 
 	dbg := NewProxyDebugSession(ctx, "openai-images")
 	dbg.ClientRequest(r, dict, nil, false)
@@ -62,13 +72,16 @@ func imagesGenerationsHandler(ctx context.Context, w http.ResponseWriter, r *htt
 	up, err := models.ResolveChatUpstream(ctx, userCred, vk, dict)
 	if err != nil {
 		dbg.Error("resolve upstream: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "resolve_upstream", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
 	dbg.RoutingResolved(dict, up)
+	fillAPILogFromUpstream(rec, up)
 
 	if err := models.TakeVirtualKeyRequestsPerMinute(up.VirtualKeyId, up.RequestsPerMinute); err != nil {
 		dbg.Error("rate limit: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "rate_limit", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
@@ -81,6 +94,7 @@ func imagesGenerationsHandler(ctx context.Context, w http.ResponseWriter, r *htt
 		UpstreamModel: up.UpstreamModel,
 	}, dict); err != nil {
 		dbg.Error("provider request: %v", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "provider_request", err)
 		httperrors.InvalidInputError(ctx, w, "provider request: %v", err)
 		return
 	}
@@ -88,6 +102,7 @@ func imagesGenerationsHandler(ctx context.Context, w http.ResponseWriter, r *htt
 	resp, uerr := imagesGenerationsWithKeyFailover(ctx, up, dict, 180*time.Second, dbg)
 	if uerr != nil {
 		dbg.Error("upstream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+		recordUpstreamError(rec, uerr)
 		writeUpstreamError(ctx, w, uerr)
 		return
 	}
@@ -97,6 +112,7 @@ func imagesGenerationsHandler(ctx context.Context, w http.ResponseWriter, r *htt
 		out = norm
 	}
 	dbg.ClientResponse(out)
+	markAPILogSuccess(rec, out)
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusOK)
 	_, _ = w.Write(out)

pkg/aiproxy/handlers/messages.go

@@ -15,6 +15,7 @@
 package handlers
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -24,6 +25,7 @@ import (
 
 	"yunion.io/x/jsonutils"
 
+	"yunion.io/x/onecloud/pkg/aiproxy/chatlog"
 	"yunion.io/x/onecloud/pkg/aiproxy/extensions/visual"
 	"yunion.io/x/onecloud/pkg/aiproxy/models"
 	"yunion.io/x/onecloud/pkg/aiproxy/providerapi"
@@ -38,7 +40,12 @@ import (
 
 // messagesHandler implements Anthropic-compatible POST /ai/anthropic/v1/messages.
 func messagesHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	start := time.Now()
+	rec := newAPILogRecord(ctx, r, start)
+	defer finishAPILogRecord(rec, start)
+
 	if r.Method != http.MethodPost {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_method", nil)
 		writeAnthropicError(ctx, w, http.StatusBadRequest, "invalid_request_error", "only POST is supported")
 		return
 	}
@@ -46,37 +53,43 @@ func messagesHandler(ctx context.Context, w http.ResponseWriter, r *http.Request
 	defer r.Body.Close()
 	raw, err := io.ReadAll(r.Body)
 	if err != nil {
+		failAPILogRecord(rec, http.StatusBadRequest, "read_body", err)
 		writeAnthropicError(ctx, w, http.StatusBadRequest, "invalid_request_error", "read body: %v", err)
 		return
 	}
 
 	body, err := jsonutils.Parse(raw)
 	if err != nil {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_json", err)
 		writeAnthropicError(ctx, w, http.StatusBadRequest, "invalid_request_error", "invalid JSON body: %v", err)
 		return
 	}
 	dict, ok := body.(*jsonutils.JSONDict)
 	if !ok {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_body", nil)
 		writeAnthropicError(ctx, w, http.StatusBadRequest, "invalid_request_error", "body must be a JSON object")
 		return
 	}
+	fillAPILogFromBody(rec, dict)
 
 	dbg := NewProxyDebugSession(ctx, "anthropic-messages")
-	isStream, _ := dict.Bool("stream")
-	dbg.ClientRequest(r, dict, nil, isStream)
+	dbg.ClientRequest(r, dict, nil, rec.Stream)
 
 	vk := extractVirtualKey(r)
 	userCred := auth.AdminCredential()
 	up, err := models.ResolveChatUpstream(ctx, userCred, vk, dict)
 	if err != nil {
 		dbg.Error("resolve upstream: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "resolve_upstream", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
 	dbg.RoutingResolved(dict, up)
+	fillAPILogFromUpstream(rec, up)
 
 	if err := models.TakeVirtualKeyRequestsPerMinute(up.VirtualKeyId, up.RequestsPerMinute); err != nil {
 		dbg.Error("rate limit: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "rate_limit", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
@@ -88,35 +101,61 @@ func messagesHandler(ctx context.Context, w http.ResponseWriter, r *http.Request
 	}
 	if err := models.EnforceVirtualKeyMaxTokens(dict, vkLim); err != nil {
 		dbg.Error("max tokens: %v", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "max_tokens", err)
 		writeAnthropicError(ctx, w, http.StatusBadRequest, "invalid_request_error", "%v", err)
 		return
 	}
 
-	if visual.ShouldRejectMessagesStreaming(dict, up, isStream) {
-		dbg.Error("%v", visual.ErrVisualStreamingUnsupported)
-		writeAnthropicError(ctx, w, http.StatusBadRequest, "invalid_request_error", "%v", visual.ErrVisualStreamingUnsupported)
-		return
-	}
+	isStream := rec.Stream
 	if visual.ShouldHandleMessages(dict, up, isStream) {
-		bodyOut, err := visual.HandleMessagesCreate(ctx, dict, up)
+		if !isStream {
+			bodyOut, err := visual.HandleMessagesCreate(ctx, dict, up)
+			if err != nil {
+				dbg.Error("visual orchestration: %v", err)
+				failAPILogRecord(rec, http.StatusBadGateway, "visual_orchestration", err)
+				writeAnthropicError(ctx, w, http.StatusBadGateway, "api_error", "visual orchestration: %v", err)
+				return
+			}
+			dbg.ClientResponse(bodyOut)
+			markAPILogSuccess(rec, bodyOut)
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write(bodyOut)
+			if up.AiKeyId != "" {
+				models.RecordAiKeySuccess(up.AiKeyId)
+			}
+			return
+		}
+
+		chatBody, err := visual.HandleMessagesCreateChat(ctx, dict, up)
 		if err != nil {
 			dbg.Error("visual orchestration: %v", err)
+			failAPILogRecord(rec, http.StatusBadGateway, "visual_orchestration", err)
 			writeAnthropicError(ctx, w, http.StatusBadGateway, "api_error", "visual orchestration: %v", err)
 			return
 		}
-		dbg.ClientResponse(bodyOut)
-		w.Header().Set("Content-Type", "application/json")
-		w.WriteHeader(http.StatusOK)
-		_, _ = w.Write(bodyOut)
-		if up.AiKeyId != "" {
-			models.RecordAiKeySuccess(up.AiKeyId)
+		payloads, err := openai.NonStreamChatCompletionToStreamPayloads(chatBody)
+		if err != nil {
+			dbg.Error("visual synthetic stream: %v", err)
+			failAPILogRecord(rec, http.StatusBadGateway, "visual_synthetic_stream", err)
+			writeAnthropicError(ctx, w, http.StatusBadGateway, "api_error", "visual synthetic stream: %v", err)
+			return
 		}
+		adapter, err := messages.GetAdapter(up.ProviderKey, up.APIMode)
+		if err != nil {
+			dbg.Error("adapter: %v", err)
+			failAPILogRecord(rec, http.StatusBadRequest, "adapter", err)
+			writeAnthropicError(ctx, w, http.StatusBadRequest, "invalid_request_error", "%v", err)
+			return
+		}
+		writeAnthropicSyntheticStream(ctx, w, adapter, payloads, up.UpstreamModel, up.AiKeyId, dbg, rec)
 		return
 	}
 
 	adapter, err := messages.GetAdapter(up.ProviderKey, up.APIMode)
 	if err != nil {
 		dbg.Error("adapter: %v", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "adapter", err)
 		writeAnthropicError(ctx, w, http.StatusBadRequest, "invalid_request_error", "%v", err)
 		return
 	}
@@ -124,6 +163,7 @@ func messagesHandler(ctx context.Context, w http.ResponseWriter, r *http.Request
 	chatCtx := providers.ChatContextFromUpstream(up.ProviderKey, up.BaseURL, up.APIKey, up.UpstreamModel, up.APIMode)
 	if _, err := adapter.BuildUpstreamRequest(chatCtx, dict, isStream); err != nil {
 		dbg.Error("provider request: %v", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "provider_request", err)
 		writeAnthropicError(ctx, w, http.StatusBadRequest, "invalid_request_error", "provider request: %v", err)
 		return
 	}
@@ -146,6 +186,7 @@ func messagesHandler(ctx context.Context, w http.ResponseWriter, r *http.Request
 		resp, uerr := upstreamWithKeyFailover(ctx, up, timeout, build)
 		if uerr != nil {
 			dbg.Error("upstream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+			recordUpstreamError(rec, uerr)
 			writeMessagesUpstreamError(ctx, w, adapter, uerr)
 			return
 		}
@@ -155,6 +196,7 @@ func messagesHandler(ctx context.Context, w http.ResponseWriter, r *http.Request
 			bodyOut = norm
 		}
 		dbg.ClientResponse(bodyOut)
+		markAPILogSuccess(rec, bodyOut)
 		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
 		_, _ = w.Write(bodyOut)
@@ -165,10 +207,11 @@ func messagesHandler(ctx context.Context, w http.ResponseWriter, r *http.Request
 		ch, uerr := upstreamRawStreamWithKeyFailover(ctx, up, timeout, build, upstream.ChatCompletionStreamRaw)
 		if uerr != nil {
 			dbg.Error("upstream stream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+			recordUpstreamError(rec, uerr)
 			writeMessagesUpstreamError(ctx, w, adapter, uerr)
 			return
 		}
-		writeAnthropicPassthroughStream(ctx, w, ch, up.AiKeyId, dbg)
+		writeAnthropicPassthroughStream(ctx, w, ch, up.AiKeyId, dbg, rec)
 		return
 	}
 
@@ -178,10 +221,11 @@ func messagesHandler(ctx context.Context, w http.ResponseWriter, r *http.Request
 	})
 	if uerr != nil {
 		dbg.Error("upstream stream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+		recordUpstreamError(rec, uerr)
 		writeMessagesUpstreamError(ctx, w, adapter, uerr)
 		return
 	}
-	writeAnthropicTranslatedStream(ctx, w, ch, adapter, up.UpstreamModel, up.AiKeyId, dbg)
+	writeAnthropicTranslatedStream(ctx, w, ch, adapter, up.UpstreamModel, up.AiKeyId, dbg, rec)
 }
 
 func buildMessagesUpstream(
@@ -300,7 +344,7 @@ func writeMessagesUpstreamError(ctx context.Context, w http.ResponseWriter, adap
 	_, _ = w.Write(openai.NewAnthropicErrorBody("api_error", msg))
 }
 
-func writeAnthropicPassthroughStream(ctx context.Context, w http.ResponseWriter, ch <-chan upstream.RawSSEEvent, aiKeyId string, dbg *ProxyDebugSession) {
+func writeAnthropicPassthroughStream(ctx context.Context, w http.ResponseWriter, ch <-chan upstream.RawSSEEvent, aiKeyId string, dbg *ProxyDebugSession, rec *chatlog.Record) {
 	w.Header().Set("Content-Type", "text/event-stream")
 	w.Header().Set("Cache-Control", "no-cache")
 	w.Header().Set("Connection", "keep-alive")
@@ -308,10 +352,14 @@ func writeAnthropicPassthroughStream(ctx context.Context, w http.ResponseWriter,
 	flushIf(w)
 
 	streamOK := true
+	sawUsage := false
 	seq := 0
 	for evt := range ch {
 		seq++
 		dbg.ClientStreamSSE(seq, evt.Event, evt.Data)
+		if len(evt.Data) > 0 && bytes.Contains(evt.Data, []byte(`"usage"`)) {
+			noteStreamUsage(rec, evt.Data, &sawUsage)
+		}
 		if evt.Event != "" {
 			_, _ = fmt.Fprintf(w, "event: %s\n", evt.Event)
 		}
@@ -322,6 +370,62 @@ func writeAnthropicPassthroughStream(ctx context.Context, w http.ResponseWriter,
 		}
 		flushIf(w)
 	}
+	finishAPILogStream(rec, streamOK, sawUsage)
+	if streamOK && aiKeyId != "" {
+		models.RecordAiKeySuccess(aiKeyId)
+	}
+}
+
+func writeAnthropicSyntheticStream(
+	ctx context.Context,
+	w http.ResponseWriter,
+	adapter providerapi.MessagesAdapter,
+	payloads [][]byte,
+	requestModel string,
+	aiKeyId string,
+	dbg *ProxyDebugSession,
+	rec *chatlog.Record,
+) {
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.WriteHeader(http.StatusOK)
+	flushIf(w)
+
+	state := adapter.NewStreamState(requestModel)
+	streamOK := true
+	sawUsage := false
+	outSeq := 0
+	for _, payload := range payloads {
+		if len(payload) > 0 && bytes.Contains(payload, []byte(`"usage"`)) {
+			noteStreamUsage(rec, payload, &sawUsage)
+		}
+		events, err := adapter.ConvertStreamPayload(state, payload, false)
+		if err != nil {
+			dbg.Error("visual synthetic stream convert error: %v", err)
+			streamOK = false
+			failAPILogRecord(rec, http.StatusOK, "stream_convert", err)
+			break
+		}
+		outSeq++
+		dbg.ClientStreamEvents(outSeq, events)
+		noteAnthropicSSEUsage(rec, events, &sawUsage)
+		writeAnthropicSSEEvents(w, events)
+	}
+	if streamOK {
+		events, err := adapter.ConvertStreamPayload(state, nil, true)
+		if err != nil {
+			dbg.Error("visual synthetic stream end error: %v", err)
+			streamOK = false
+			failAPILogRecord(rec, http.StatusOK, "stream_convert", err)
+		} else {
+			outSeq++
+			dbg.ClientStreamEvents(outSeq, events)
+			noteAnthropicSSEUsage(rec, events, &sawUsage)
+			writeAnthropicSSEEvents(w, events)
+		}
+	}
+	finishAPILogStream(rec, streamOK, sawUsage)
 	if streamOK && aiKeyId != "" {
 		models.RecordAiKeySuccess(aiKeyId)
 	}
@@ -335,6 +439,7 @@ func writeAnthropicTranslatedStream(
 	requestModel string,
 	aiKeyId string,
 	dbg *ProxyDebugSession,
+	rec *chatlog.Record,
 ) {
 	w.Header().Set("Content-Type", "text/event-stream")
 	w.Header().Set("Cache-Control", "no-cache")
@@ -344,6 +449,7 @@ func writeAnthropicTranslatedStream(
 
 	state := adapter.NewStreamState(requestModel)
 	streamOK := true
+	sawUsage := false
 	outSeq := 0
 	for chunk := range ch {
 		if chunk.Done {
@@ -351,19 +457,26 @@ func writeAnthropicTranslatedStream(
 			if err != nil {
 				dbg.Error("stream convert end error: %v", err)
 				streamOK = false
+				failAPILogRecord(rec, http.StatusOK, "stream_convert", err)
 				break
 			}
 			outSeq++
 			dbg.ClientStreamEvents(outSeq, events)
+			noteAnthropicSSEUsage(rec, events, &sawUsage)
 			writeAnthropicSSEEvents(w, events)
 			break
 		}
 		if len(chunk.Data) == 0 {
 			continue
 		}
+		if bytes.Contains(chunk.Data, []byte(`"usage"`)) {
+			noteStreamUsage(rec, chunk.Data, &sawUsage)
+		}
 		if isAnthropicUpstreamErrorChunk(chunk.Data) {
 			dbg.Error("upstream stream error chunk: %s", truncateLogBytes(chunk.Data, proxyDebugLogMax))
 			streamOK = false
+			rec.Success = false
+			rec.ErrorCode, rec.ErrorMessage = parseUpstreamErrorInfo(chunk.Data)
 			if aiKeyId != "" {
 				models.RecordAiKeyFailure(aiKeyId, parseUpstreamErrorStatus(chunk.Data))
 			}
@@ -375,12 +488,15 @@ func writeAnthropicTranslatedStream(
 		if err != nil {
 			dbg.Error("stream convert error: %v upstream_chunk=%s", err, truncateLogBytes(chunk.Data, proxyDebugLogMax))
 			streamOK = false
+			failAPILogRecord(rec, http.StatusOK, "stream_convert", err)
 			break
 		}
 		outSeq++
 		dbg.ClientStreamEvents(outSeq, events)
+		noteAnthropicSSEUsage(rec, events, &sawUsage)
 		writeAnthropicSSEEvents(w, events)
 	}
+	finishAPILogStream(rec, streamOK, sawUsage)
 	if streamOK && aiKeyId != "" {
 		models.RecordAiKeySuccess(aiKeyId)
 	}
@@ -400,6 +516,14 @@ func writeAnthropicSSEEvents(w http.ResponseWriter, events []providerapi.Anthrop
 	}
 }
 
+func noteAnthropicSSEUsage(rec *chatlog.Record, events []providerapi.AnthropicStreamChunk, sawUsage *bool) {
+	for _, evt := range events {
+		if len(evt.Data) > 0 && bytes.Contains(evt.Data, []byte(`"usage"`)) {
+			noteStreamUsage(rec, evt.Data, sawUsage)
+		}
+	}
+}
+
 func isAnthropicUpstreamErrorChunk(data []byte) bool {
 	var wrap struct {
 		Error interface{} `json:"error"`

pkg/aiproxy/handlers/responses.go

@@ -15,6 +15,7 @@
 package handlers
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -26,6 +27,7 @@ import (
 
 	"yunion.io/x/jsonutils"
 
+	"yunion.io/x/onecloud/pkg/aiproxy/chatlog"
 	"yunion.io/x/onecloud/pkg/aiproxy/extensions/visual"
 	"yunion.io/x/onecloud/pkg/aiproxy/models"
 	"yunion.io/x/onecloud/pkg/aiproxy/providerapi"
@@ -41,6 +43,10 @@ import (
 
 func responsesHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
+		start := time.Now()
+		rec := newAPILogRecord(ctx, r, start)
+		defer finishAPILogRecord(rec, start)
+		failAPILogRecord(rec, http.StatusMethodNotAllowed, "invalid_method", nil)
 		writeResponsesError(ctx, w, http.StatusMethodNotAllowed, "invalid_request_error", "only POST is supported")
 		return
 	}
@@ -60,25 +66,33 @@ func responsesDeleteHandler(ctx context.Context, w http.ResponseWriter, r *http.
 }
 
 func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	start := time.Now()
+	rec := newAPILogRecord(ctx, r, start)
+	defer finishAPILogRecord(rec, start)
+
 	defer r.Body.Close()
 	raw, err := io.ReadAll(r.Body)
 	if err != nil {
+		failAPILogRecord(rec, http.StatusBadRequest, "read_body", err)
 		writeResponsesError(ctx, w, http.StatusBadRequest, "invalid_request_error", "read body: %v", err)
 		return
 	}
 	body, err := jsonutils.Parse(raw)
 	if err != nil {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_json", err)
 		writeResponsesError(ctx, w, http.StatusBadRequest, "invalid_request_error", "invalid JSON body: %v", err)
 		return
 	}
 	dict, ok := body.(*jsonutils.JSONDict)
 	if !ok {
+		failAPILogRecord(rec, http.StatusBadRequest, "invalid_body", nil)
 		writeResponsesError(ctx, w, http.StatusBadRequest, "invalid_request_error", "body must be a JSON object")
 		return
 	}
+	fillAPILogFromBody(rec, dict)
 
 	dbg := NewProxyDebugSession(ctx, "openai-responses")
-	isStream, _ := dict.Bool("stream")
+	isStream := rec.Stream
 	dbg.ClientRequest(r, dict, nil, isStream)
 
 	vk := extractVirtualKey(r)
@@ -86,12 +100,15 @@ func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.R
 	up, err := models.ResolveChatUpstream(ctx, userCred, vk, dict)
 	if err != nil {
 		dbg.Error("resolve upstream: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "resolve_upstream", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
 	dbg.RoutingResolved(dict, up)
+	fillAPILogFromUpstream(rec, up)
 	if err := models.TakeVirtualKeyRequestsPerMinute(up.VirtualKeyId, up.RequestsPerMinute); err != nil {
 		dbg.Error("rate limit: %v", err)
+		failAPILogRecord(rec, http.StatusInternalServerError, "rate_limit", err)
 		httperrors.GeneralServerError(ctx, w, err)
 		return
 	}
@@ -101,6 +118,7 @@ func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.R
 	}
 	if err := models.EnsureResponsesMaxOutputTokens(dict, vkLim); err != nil {
 		dbg.Error("max tokens: %v", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "max_tokens", err)
 		writeResponsesError(ctx, w, http.StatusBadRequest, "invalid_request_error", "%v", err)
 		return
 	}
@@ -110,10 +128,12 @@ func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.R
 			bodyOut, _, err := visual.HandleResponsesCreate(ctx, dict, up)
 			if err != nil {
 				dbg.Error("visual orchestration: %v", err)
+				failAPILogRecord(rec, http.StatusBadGateway, "visual_orchestration", err)
 				writeResponsesError(ctx, w, http.StatusBadGateway, "api_error", "visual orchestration: %v", err)
 				return
 			}
 			dbg.ClientResponse(bodyOut)
+			markAPILogSuccess(rec, bodyOut)
 			w.Header().Set("Content-Type", "application/json")
 			w.WriteHeader(http.StatusOK)
 			_, _ = w.Write(bodyOut)
@@ -126,28 +146,32 @@ func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.R
 		chatBody, _, err := visual.HandleResponsesCreateChat(ctx, dict, up)
 		if err != nil {
 			dbg.Error("visual orchestration: %v", err)
+			failAPILogRecord(rec, http.StatusBadGateway, "visual_orchestration", err)
 			writeResponsesError(ctx, w, http.StatusBadGateway, "api_error", "visual orchestration: %v", err)
 			return
 		}
 		payloads, err := openai.NonStreamChatCompletionToStreamPayloads(chatBody)
 		if err != nil {
 			dbg.Error("visual synthetic stream: %v", err)
+			failAPILogRecord(rec, http.StatusBadGateway, "visual_synthetic_stream", err)
 			writeResponsesError(ctx, w, http.StatusBadGateway, "api_error", "visual synthetic stream: %v", err)
 			return
 		}
 		adapter, err := responses.GetAdapter(up.ProviderKey, up.APIMode)
 		if err != nil {
 			dbg.Error("adapter: %v", err)
+			failAPILogRecord(rec, http.StatusBadRequest, "adapter", err)
 			writeResponsesError(ctx, w, http.StatusBadRequest, "invalid_request_error", "%v", err)
 			return
 		}
-		writeResponsesSyntheticStream(ctx, w, adapter, up.UpstreamModel, dict, payloads, up.AiKeyId, dbg)
+		writeResponsesSyntheticStream(ctx, w, adapter, up.UpstreamModel, dict, payloads, up.AiKeyId, dbg, rec)
 		return
 	}
 
 	adapter, err := responses.GetAdapter(up.ProviderKey, up.APIMode)
 	if err != nil {
 		dbg.Error("adapter: %v", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "adapter", err)
 		writeResponsesError(ctx, w, http.StatusBadRequest, "invalid_request_error", "%v", err)
 		return
 	}
@@ -155,6 +179,7 @@ func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.R
 	chatCtx := providers.ChatContextFromUpstream(up.ProviderKey, up.BaseURL, up.APIKey, up.UpstreamModel, up.APIMode)
 	if _, err := adapter.BuildUpstreamRequest(chatCtx, dict, isStream); err != nil {
 		dbg.Error("provider request: %v", err)
+		failAPILogRecord(rec, http.StatusBadRequest, "provider_request", err)
 		writeResponsesError(ctx, w, http.StatusBadRequest, "invalid_request_error", "provider request: %v", err)
 		return
 	}
@@ -176,6 +201,7 @@ func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.R
 		resp, uerr := upstreamWithKeyFailover(ctx, up, timeout, build)
 		if uerr != nil {
 			dbg.Error("upstream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+			recordUpstreamError(rec, uerr)
 			writeResponsesUpstreamError(ctx, w, uerr)
 			return
 		}
@@ -199,6 +225,7 @@ func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.R
 			}
 		}
 		dbg.ClientResponse(bodyOut)
+		markAPILogSuccess(rec, bodyOut)
 		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
 		_, _ = w.Write(bodyOut)
@@ -209,10 +236,11 @@ func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.R
 		ch, uerr := upstreamRawStreamWithKeyFailover(ctx, up, timeout, build, upstream.ChatCompletionStreamRaw)
 		if uerr != nil {
 			dbg.Error("upstream stream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+			recordUpstreamError(rec, uerr)
 			writeResponsesUpstreamError(ctx, w, uerr)
 			return
 		}
-		writeResponsesPassthroughStream(ctx, w, ch, up.AiKeyId, dbg)
+		writeResponsesPassthroughStream(ctx, w, ch, up.AiKeyId, dbg, rec)
 		return
 	}
 
@@ -220,10 +248,11 @@ func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.R
 		ch, uerr := upstreamRawStreamWithKeyFailover(ctx, up, timeout, build, upstream.ChatCompletionStreamRaw)
 		if uerr != nil {
 			dbg.Error("upstream stream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+			recordUpstreamError(rec, uerr)
 			writeResponsesUpstreamError(ctx, w, uerr)
 			return
 		}
-		writeResponsesAnthropicTranslatedStream(ctx, w, ch, adapter, up.UpstreamModel, dict, up.AiKeyId, dbg)
+		writeResponsesAnthropicTranslatedStream(ctx, w, ch, adapter, up.UpstreamModel, dict, up.AiKeyId, dbg, rec)
 		return
 	}
 
@@ -233,10 +262,11 @@ func handleResponsesCreate(ctx context.Context, w http.ResponseWriter, r *http.R
 	})
 	if uerr != nil {
 		dbg.Error("upstream stream error status=%d body=%s", uerr.StatusCode, truncateLogBytes(uerr.Body, proxyDebugLogMax))
+		recordUpstreamError(rec, uerr)
 		writeResponsesUpstreamError(ctx, w, uerr)
 		return
 	}
-	writeResponsesTranslatedStream(ctx, w, ch, adapter, up.UpstreamModel, dict, up.AiKeyId, dbg)
+	writeResponsesTranslatedStream(ctx, w, ch, adapter, up.UpstreamModel, dict, up.AiKeyId, dbg, rec)
 }
 
 func handleResponsesSubResource(ctx context.Context, w http.ResponseWriter, r *http.Request, method, subAction string, params *appsrv.SAppParams) {
@@ -425,17 +455,21 @@ func writeResponsesUpstreamError(ctx context.Context, w http.ResponseWriter, uer
 	writeUpstreamError(ctx, w, uerr)
 }
 
-func writeResponsesPassthroughStream(ctx context.Context, w http.ResponseWriter, ch <-chan upstream.RawSSEEvent, aiKeyId string, dbg *ProxyDebugSession) {
+func writeResponsesPassthroughStream(ctx context.Context, w http.ResponseWriter, ch <-chan upstream.RawSSEEvent, aiKeyId string, dbg *ProxyDebugSession, rec *chatlog.Record) {
 	w.Header().Set("Content-Type", "text/event-stream")
 	w.Header().Set("Cache-Control", "no-cache")
 	w.Header().Set("Connection", "keep-alive")
 	w.WriteHeader(http.StatusOK)
 	flushIf(w)
 	streamOK := true
+	sawUsage := false
 	seq := 0
 	for evt := range ch {
 		seq++
 		dbg.ClientStreamSSE(seq, evt.Event, evt.Data)
+		if len(evt.Data) > 0 && bytes.Contains(evt.Data, []byte(`"usage"`)) {
+			noteStreamUsage(rec, evt.Data, &sawUsage)
+		}
 		if evt.Event != "" {
 			_, _ = fmt.Fprintf(w, "event: %s\n", evt.Event)
 		}
@@ -446,6 +480,7 @@ func writeResponsesPassthroughStream(ctx context.Context, w http.ResponseWriter,
 		}
 		flushIf(w)
 	}
+	finishAPILogStream(rec, streamOK, sawUsage)
 	if streamOK && aiKeyId != "" {
 		models.RecordAiKeySuccess(aiKeyId)
 	}
@@ -460,6 +495,7 @@ func writeResponsesSyntheticStream(
 	payloads [][]byte,
 	aiKeyId string,
 	dbg *ProxyDebugSession,
+	rec *chatlog.Record,
 ) {
 	w.Header().Set("Content-Type", "text/event-stream")
 	w.Header().Set("Cache-Control", "no-cache")
@@ -469,15 +505,21 @@ func writeResponsesSyntheticStream(
 
 	state := adapter.NewStreamState(requestModel, requestBody)
 	streamOK := true
+	sawUsage := false
 	outSeq := 0
 	for _, payload := range payloads {
+		if len(payload) > 0 && bytes.Contains(payload, []byte(`"usage"`)) {
+			noteStreamUsage(rec, payload, &sawUsage)
+		}
 		events, err := adapter.ConvertStreamPayload(state, payload, false)
 		if err != nil {
 			dbg.Error("visual synthetic stream convert error: %v", err)
 			streamOK = false
+			failAPILogRecord(rec, http.StatusOK, "stream_convert", err)
 			break
 		}
 		outSeq++
+		noteResponsesSSEUsage(rec, events, &sawUsage)
 		writeResponsesSSEEvents(w, events, dbg, outSeq)
 	}
 	if streamOK {
@@ -485,11 +527,14 @@ func writeResponsesSyntheticStream(
 		if err != nil {
 			dbg.Error("visual synthetic stream end error: %v", err)
 			streamOK = false
+			failAPILogRecord(rec, http.StatusOK, "stream_convert", err)
 		} else {
 			outSeq++
+			noteResponsesSSEUsage(rec, events, &sawUsage)
 			writeResponsesSSEEvents(w, events, dbg, outSeq)
 		}
 	}
+	finishAPILogStream(rec, streamOK, sawUsage)
 	if streamOK && aiKeyId != "" {
 		models.RecordAiKeySuccess(aiKeyId)
 	}
@@ -504,6 +549,7 @@ func writeResponsesTranslatedStream(
 	requestBody *jsonutils.JSONDict,
 	aiKeyId string,
 	dbg *ProxyDebugSession,
+	rec *chatlog.Record,
 ) {
 	w.Header().Set("Content-Type", "text/event-stream")
 	w.Header().Set("Cache-Control", "no-cache")
@@ -513,6 +559,7 @@ func writeResponsesTranslatedStream(
 
 	state := adapter.NewStreamState(requestModel, requestBody)
 	streamOK := true
+	sawUsage := false
 	outSeq := 0
 	for chunk := range ch {
 		if chunk.Done {
@@ -520,18 +567,25 @@ func writeResponsesTranslatedStream(
 			if err != nil {
 				dbg.Error("stream convert end error: %v", err)
 				streamOK = false
+				failAPILogRecord(rec, http.StatusOK, "stream_convert", err)
 				break
 			}
 			outSeq++
+			noteResponsesSSEUsage(rec, events, &sawUsage)
 			writeResponsesSSEEvents(w, events, dbg, outSeq)
 			break
 		}
 		if len(chunk.Data) == 0 {
 			continue
 		}
+		if bytes.Contains(chunk.Data, []byte(`"usage"`)) {
+			noteStreamUsage(rec, chunk.Data, &sawUsage)
+		}
 		if isResponsesUpstreamErrorChunk(chunk.Data) {
 			dbg.Error("upstream stream error chunk: %s", truncateLogBytes(chunk.Data, proxyDebugLogMax))
 			streamOK = false
+			rec.Success = false
+			rec.ErrorCode, rec.ErrorMessage = parseUpstreamErrorInfo(chunk.Data)
 			if aiKeyId != "" {
 				models.RecordAiKeyFailure(aiKeyId, parseUpstreamErrorStatus(chunk.Data))
 			}
@@ -543,11 +597,14 @@ func writeResponsesTranslatedStream(
 		if err != nil {
 			dbg.Error("stream convert error: %v", err)
 			streamOK = false
+			failAPILogRecord(rec, http.StatusOK, "stream_convert", err)
 			break
 		}
 		outSeq++
+		noteResponsesSSEUsage(rec, events, &sawUsage)
 		writeResponsesSSEEvents(w, events, dbg, outSeq)
 	}
+	finishAPILogStream(rec, streamOK, sawUsage)
 	if streamOK && aiKeyId != "" {
 		models.RecordAiKeySuccess(aiKeyId)
 	}
@@ -562,6 +619,7 @@ func writeResponsesAnthropicTranslatedStream(
 	requestBody *jsonutils.JSONDict,
 	aiKeyId string,
 	dbg *ProxyDebugSession,
+	rec *chatlog.Record,
 ) {
 	w.Header().Set("Content-Type", "text/event-stream")
 	w.Header().Set("Cache-Control", "no-cache")
@@ -571,23 +629,31 @@ func writeResponsesAnthropicTranslatedStream(
 
 	state := adapter.NewStreamState(requestModel, requestBody)
 	streamOK := true
+	sawUsage := false
 	outSeq := 0
 	for evt := range ch {
 		outSeq++
 		dbg.UpstreamStreamChunk(outSeq, evt.Data)
+		if len(evt.Data) > 0 && bytes.Contains(evt.Data, []byte(`"usage"`)) {
+			noteStreamUsage(rec, evt.Data, &sawUsage)
+		}
 		events, err := responses.ConvertAnthropicStreamEvent(state, evt.Event, evt.Data, false)
 		if err != nil {
 			dbg.Error("stream convert error: %v", err)
 			streamOK = false
+			failAPILogRecord(rec, http.StatusOK, "stream_convert", err)
 			break
 		}
+		noteResponsesSSEUsage(rec, events, &sawUsage)
 		writeResponsesSSEEvents(w, events, dbg, outSeq)
 	}
 	events, err := responses.ConvertAnthropicStreamEvent(state, "", nil, true)
 	if err == nil {
 		outSeq++
+		noteResponsesSSEUsage(rec, events, &sawUsage)
 		writeResponsesSSEEvents(w, events, dbg, outSeq)
 	}
+	finishAPILogStream(rec, streamOK, sawUsage)
 	if streamOK && aiKeyId != "" {
 		models.RecordAiKeySuccess(aiKeyId)
 	}
@@ -613,6 +679,14 @@ func writeResponsesSSEEvents(w http.ResponseWriter, events []providerapi.Respons
 	}
 }
 
+func noteResponsesSSEUsage(rec *chatlog.Record, events []providerapi.ResponsesStreamChunk, sawUsage *bool) {
+	for _, evt := range events {
+		if len(evt.Data) > 0 && bytes.Contains(evt.Data, []byte(`"usage"`)) {
+			noteStreamUsage(rec, evt.Data, sawUsage)
+		}
+	}
+}
+
 func isResponsesUpstreamErrorChunk(data []byte) bool {
 	var wrap struct {
 		Error interface{} `json:"error"`