diff --git a/apps/edge/internal/openai/stream.go b/apps/edge/internal/openai/stream.go index 3be0496..a351dda 100644 --- a/apps/edge/internal/openai/stream.go +++ b/apps/edge/internal/openai/stream.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "net/http" + "os" "strings" "time" "unicode" @@ -13,6 +14,11 @@ import ( edgeservice "iop/apps/edge/internal/service" ) +const ( + streamTraceMetadataKey = "iop_trace_stream" + streamTraceEnvKey = "IOP_OPENAI_COMPAT_TRACE_STREAM" +) + func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, req chatCompletionRequest, submitReq edgeservice.SubmitRunRequest, handle edgeservice.RunResult, outputPolicy strictOutputPolicy, validation toolValidationContract) { flusher, ok := w.(http.Flusher) if !ok { @@ -38,6 +44,43 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re created := time.Now().Unix() id := "chatcmpl-" + handle.Dispatch().RunID model := responseModel(req.Model, handle.Dispatch().Target) + traceStream := openAICompatTraceStreamEnabled(submitReq.Metadata) + traceSeq := 0 + writeTracedContentDelta := func(content, source string, filtered bool) { + if content == "" { + return + } + traceSeq++ + if traceStream { + logOpenAICompatStreamOutput(s.logger, handle.Dispatch().RunID, traceSeq, "content", content, + zap.Int("source_len", len(source)), + zap.Bool("filtered", filtered), + zap.Bool("source_has_open_think", hasOpenThinkTag(source)), + zap.Bool("source_has_close_think", hasCloseThinkTag(source)), + zap.String("source_preview", previewString(source, 2000)), + ) + } + writeContentDeltaSSE(w, flusher, id, created, model, content) + } + writeTracedReasoningDelta := func(reasoning string) { + if reasoning == "" { + return + } + traceSeq++ + if traceStream { + logOpenAICompatStreamOutput(s.logger, handle.Dispatch().RunID, traceSeq, "reasoning", reasoning) + } + writeSSE(w, flusher, chatCompletionChunk{ + ID: id, + Object: "chat.completion.chunk", + Created: created, + Model: model, + Choices: []chatCompletionChunkChoice{{ + Index: 0, + Delta: chatDelta{ReasoningContent: reasoning}, + }}, + }) + } writeSSE(w, flusher, chatCompletionChunk{ ID: id, Object: "chat.completion.chunk", @@ -100,16 +143,27 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re if event.GetDelta() == "" { continue } - delta := event.GetDelta() - contentBuilder.WriteString(delta) + sourceDelta := event.GetDelta() + delta := sourceDelta + contentBuilder.WriteString(sourceDelta) if toolTextFilter != nil { delta = toolTextFilter.Append(delta) } if delta == "" { + if traceStream { + s.logger.Info("openai chat completion stream output suppressed", + zap.String("run_id", handle.Dispatch().RunID), + zap.String("type", "content"), + zap.Int("source_len", len(sourceDelta)), + zap.Bool("source_has_open_think", hasOpenThinkTag(sourceDelta)), + zap.Bool("source_has_close_think", hasCloseThinkTag(sourceDelta)), + zap.String("source_preview", previewString(sourceDelta, 2000)), + ) + } continue } emittedContent.WriteString(delta) - writeContentDeltaSSE(w, flusher, id, created, model, delta) + writeTracedContentDelta(delta, sourceDelta, delta != sourceDelta) case "reasoning_delta": if event.GetDelta() == "" { continue @@ -118,16 +172,7 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re if outputPolicy.Strict { continue } - writeSSE(w, flusher, chatCompletionChunk{ - ID: id, - Object: "chat.completion.chunk", - Created: created, - Model: model, - Choices: []chatCompletionChunkChoice{{ - Index: 0, - Delta: chatDelta{ReasoningContent: event.GetDelta()}, - }}, - }) + writeTracedReasoningDelta(event.GetDelta()) case "complete": metadata := event.GetMetadata() finishReason := metadata["finish_reason"] @@ -140,10 +185,18 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re pending := toolTextFilter.Flush() if pending != "" { emittedContent.WriteString(pending) - writeContentDeltaSSE(w, flusher, id, created, model, pending) + writeTracedContentDelta(pending, pending, false) } } nativeToolCalls = normalizeNativeToolCallsForResponse(nativeToolCalls, req.Tools) + if traceStream { + s.logger.Info("openai chat completion stream output chunk", + zap.String("run_id", handle.Dispatch().RunID), + zap.Int("seq", traceSeq+1), + zap.String("type", "tool_calls"), + zap.Int("tool_call_count", len(nativeToolCalls)), + ) + } writeToolCallsDeltaSSE(w, flusher, id, created, model, nativeToolCalls) finishReason = "tool_calls" } else if toolTextFilter != nil && shouldSynthesizeTextToolCalls(handle.Dispatch(), isTextToolFallback(metadata)) { @@ -151,20 +204,37 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re if len(toolCalls) > 0 { if remaining := unstreamedCleanedText(cleaned, emittedContent.String()); remaining != "" { emittedContent.WriteString(remaining) - writeContentDeltaSSE(w, flusher, id, created, model, remaining) + writeTracedContentDelta(remaining, remaining, false) + } + if traceStream { + s.logger.Info("openai chat completion stream output chunk", + zap.String("run_id", handle.Dispatch().RunID), + zap.Int("seq", traceSeq+1), + zap.String("type", "tool_calls"), + zap.Int("tool_call_count", len(toolCalls)), + ) } writeToolCallsDeltaSSE(w, flusher, id, created, model, toolCalls) finishReason = "tool_calls" } else if pending := toolTextFilter.Flush(); pending != "" { emittedContent.WriteString(pending) - writeContentDeltaSSE(w, flusher, id, created, model, pending) + writeTracedContentDelta(pending, pending, false) } } else if toolTextFilter != nil { if pending := toolTextFilter.Flush(); pending != "" { emittedContent.WriteString(pending) - writeContentDeltaSSE(w, flusher, id, created, model, pending) + writeTracedContentDelta(pending, pending, false) } } + if traceStream { + s.logger.Info("openai chat completion stream output done", + zap.String("run_id", handle.Dispatch().RunID), + zap.Int("seq", traceSeq+1), + zap.String("finish_reason", finishReason), + zap.Int("content_len", contentBuilder.Len()), + zap.Int("reasoning_len", reasoningBuilder.Len()), + ) + } writeSSE(w, flusher, chatCompletionChunk{ ID: id, Object: "chat.completion.chunk", @@ -355,7 +425,7 @@ func (s *Server) streamBufferedChatCompletion(w http.ResponseWriter, r *http.Req return } handle.Close() - s.writeBufferedStreamOutput(w, flusher, req, handle.Dispatch(), result, outputPolicy) + s.writeBufferedStreamOutput(w, flusher, req, handle.Dispatch(), result, outputPolicy, openAICompatTraceStreamEnabled(submitReq.Metadata)) return } } @@ -363,7 +433,7 @@ func (s *Server) streamBufferedChatCompletion(w http.ResponseWriter, r *http.Req // writeBufferedStreamOutput emits a validated buffered result as SSE chunks: // the role chunk, an optional content chunk, an optional tool_calls chunk, and // the terminal finish chunk. -func (s *Server) writeBufferedStreamOutput(w http.ResponseWriter, flusher http.Flusher, req chatCompletionRequest, dispatch edgeservice.RunDispatch, result chatCompletionOutput, outputPolicy strictOutputPolicy) { +func (s *Server) writeBufferedStreamOutput(w http.ResponseWriter, flusher http.Flusher, req chatCompletionRequest, dispatch edgeservice.RunDispatch, result chatCompletionOutput, outputPolicy strictOutputPolicy, traceStream bool) { created := time.Now().Unix() id := "chatcmpl-" + dispatch.RunID model := responseModel(req.Model, dispatch.Target) @@ -388,8 +458,28 @@ func (s *Server) writeBufferedStreamOutput(w http.ResponseWriter, flusher http.F Delta: chatDelta{Role: "assistant"}, }}, }) + if traceStream { + logOpenAICompatStreamOutput(s.logger, dispatch.RunID, 1, "content", content) + } writeContentDeltaSSE(w, flusher, id, created, model, content) + if traceStream && len(result.toolCalls) > 0 { + s.logger.Info("openai chat completion stream output chunk", + zap.String("run_id", dispatch.RunID), + zap.Int("seq", 2), + zap.String("type", "tool_calls"), + zap.Int("tool_call_count", len(result.toolCalls)), + ) + } writeToolCallsDeltaSSE(w, flusher, id, created, model, result.toolCalls) + if traceStream { + s.logger.Info("openai chat completion stream output done", + zap.String("run_id", dispatch.RunID), + zap.Int("seq", 3), + zap.String("finish_reason", result.finishReason), + zap.Int("content_len", len(content)), + zap.Int("reasoning_len", result.reasoningLen), + ) + } writeSSE(w, flusher, chatCompletionChunk{ ID: id, Object: "chat.completion.chunk", @@ -423,3 +513,44 @@ func writeSSEErrorWithType(w http.ResponseWriter, flusher http.Flusher, errType, fmt.Fprint(w, "data: [DONE]\n\n") flusher.Flush() } + +func openAICompatTraceStreamEnabled(metadata map[string]string) bool { + if traceBool(os.Getenv(streamTraceEnvKey)) { + return true + } + if metadata == nil { + return false + } + return traceBool(metadata[streamTraceMetadataKey]) +} + +func traceBool(v string) bool { + switch strings.ToLower(strings.TrimSpace(v)) { + case "1", "true", "yes", "on": + return true + default: + return false + } +} + +func logOpenAICompatStreamOutput(logger *zap.Logger, runID string, seq int, typ string, delta string, extra ...zap.Field) { + fields := []zap.Field{ + zap.String("run_id", runID), + zap.Int("seq", seq), + zap.String("type", typ), + zap.Int("delta_len", len(delta)), + zap.Bool("delta_has_open_think", hasOpenThinkTag(delta)), + zap.Bool("delta_has_close_think", hasCloseThinkTag(delta)), + zap.String("delta_preview", previewString(delta, 2000)), + } + fields = append(fields, extra...) + logger.Info("openai chat completion stream output chunk", fields...) +} + +func hasOpenThinkTag(s string) bool { + return strings.Contains(strings.ToLower(s), " 0 { toolCalls.AddDelta(choice.Delta.ToolCalls) } + if traceStream { + a.logger.Info("openai_compat provider stream parsed chunk", + zap.String("run_id", spec.RunID), + zap.Int("seq", traceSeq), + zap.String("finish_reason", finishReason), + zap.Int("tool_call_delta_count", len(choice.Delta.ToolCalls)), + zap.Int("content_len", len(choice.Delta.Content)), + zap.Bool("content_has_open_think", hasOpenThinkTag(choice.Delta.Content)), + zap.Bool("content_has_close_think", hasCloseThinkTag(choice.Delta.Content)), + zap.String("content_preview", tracePreview(choice.Delta.Content, 2000)), + zap.Int("reasoning_len", len(choice.Delta.ReasoningContent)), + zap.Bool("reasoning_has_open_think", hasOpenThinkTag(choice.Delta.ReasoningContent)), + zap.Bool("reasoning_has_close_think", hasCloseThinkTag(choice.Delta.ReasoningContent)), + zap.String("reasoning_preview", tracePreview(choice.Delta.ReasoningContent, 2000)), + ) + } if reasoning := choice.Delta.ReasoningContent; reasoning != "" { if err := sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, @@ -254,6 +299,40 @@ streamResponse: return fmt.Errorf("openai_compat adapter: stream ended without [DONE]") } +func openAICompatTraceStreamEnabled(metadata map[string]string) bool { + if traceBool(os.Getenv(streamTraceEnvKey)) { + return true + } + if metadata == nil { + return false + } + return traceBool(metadata[streamTraceMetadataKey]) +} + +func traceBool(v string) bool { + switch strings.ToLower(strings.TrimSpace(v)) { + case "1", "true", "yes", "on": + return true + default: + return false + } +} + +func tracePreview(s string, max int) string { + if max <= 0 || len(s) <= max { + return s + } + return s[:max] + "...[truncated]" +} + +func hasOpenThinkTag(s string) bool { + return strings.Contains(strings.ToLower(s), "