feat: add OpenAI compat stream trace logging

- Add stream trace logging to edge server streamChatCompletion
- Add stream trace logging to node adapter openai_compat provider
- Enable trace via IOP_OPENAI_COMPAT_TRACE_STREAM env var or iop_trace_stream metadata
- Track content, reasoning, tool_calls chunks with think tag detection

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), "<think")
+}
+
+func hasCloseThinkTag(s string) bool {
+	return strings.Contains(strings.ToLower(s), "</think")
+}

apps/node/internal/adapters/openai_compat/openai_compat.go

@@ -14,6 +14,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
 	"time"
 
@@ -28,6 +29,8 @@ const Name = "openai_compat"
 const (
 	runtimeMetadataOpenAIToolCalls        = "openai_tool_calls"
 	runtimeMetadataOpenAITextToolFallback = "openai_text_tool_fallback"
+	streamTraceMetadataKey                = "iop_trace_stream"
+	streamTraceEnvKey                     = "IOP_OPENAI_COMPAT_TRACE_STREAM"
 )
 
 type Adapter struct {
@@ -195,6 +198,8 @@ streamResponse:
 	finishReason := ""
 	var usage *runtime.UsageStats
 	var toolCalls openAIToolCallAccumulator
+	traceStream := openAICompatTraceStreamEnabled(spec.Metadata)
+	traceSeq := 0
 	for scanner.Scan() {
 		line := scanner.Text()
 		if !strings.HasPrefix(line, "data: ") {
@@ -202,10 +207,34 @@ streamResponse:
 		}
 		payload := strings.TrimPrefix(line, "data: ")
 		if payload == "[DONE]" {
+			if traceStream {
+				a.logger.Info("openai_compat provider stream done",
+					zap.String("run_id", spec.RunID),
+					zap.Int("seq", traceSeq+1),
+				)
+			}
 			return sink.Emit(ctx, completeEvent(spec.RunID, finishReason, usage, outputTokens, toolCalls.ToolCalls(), textToolFallback))
 		}
+		traceSeq++
+		if traceStream {
+			a.logger.Info("openai_compat provider stream raw chunk",
+				zap.String("run_id", spec.RunID),
+				zap.Int("seq", traceSeq),
+				zap.Int("raw_len", len(payload)),
+				zap.Bool("raw_has_open_think", hasOpenThinkTag(payload)),
+				zap.Bool("raw_has_close_think", hasCloseThinkTag(payload)),
+				zap.String("raw_preview", tracePreview(payload, 4000)),
+			)
+		}
 		var chunk chatChunk
 		if err := json.Unmarshal([]byte(payload), &chunk); err != nil {
+			if traceStream {
+				a.logger.Warn("openai_compat provider stream raw chunk ignored",
+					zap.String("run_id", spec.RunID),
+					zap.Int("seq", traceSeq),
+					zap.Error(err),
+				)
+			}
 			continue
 		}
 		if chunk.Usage != nil {
@@ -224,6 +253,22 @@ streamResponse:
 		if len(choice.Delta.ToolCalls) > 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), "<think")
+}
+
+func hasCloseThinkTag(s string) bool {
+	return strings.Contains(strings.ToLower(s), "</think")
+}
+
 // ProbeProvider checks endpoint availability and target presence using the
 // OpenAI-compatible /v1/models endpoint, returning only the public baseline
 // status values.