feat(openai): stream 개선 및 테스트 추가

apps/edge/internal/openai/server_test.go

@@ -1,6 +1,7 @@
 package openai
 
 import (
+	"bufio"
 	"context"
 	"encoding/json"
 	"net/http"
@@ -1092,6 +1093,61 @@ func TestChatCompletionsStreamSynthesizesToolCallsFromClineTextBlock(t *testing.
 	}
 }
 
+func TestChatCompletionsStreamWithToolsFlushesTextBeforeComplete(t *testing.T) {
+	fake := &fakeRunService{events: make(chan *iop.RunEvent)}
+	srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil)
+	httpSrv := httptest.NewServer(srv.routes())
+	defer httpSrv.Close()
+
+	resp, err := http.Post(httpSrv.URL+"/v1/chat/completions", "application/json", strings.NewReader(`{
+		"model":"stream-model",
+		"stream":true,
+		"messages":[{"role":"user","content":"explain sudo"}],
+		"tools":[{"type":"function","function":{"name":"run_commands"}}],
+		"tool_choice":"auto"
+	}`))
+	if err != nil {
+		t.Fatalf("post: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status: got %d", resp.StatusCode)
+	}
+
+	lines := make(chan string, 16)
+	scanDone := make(chan struct{})
+	go func() {
+		defer close(scanDone)
+		scanner := bufio.NewScanner(resp.Body)
+		for scanner.Scan() {
+			lines <- scanner.Text()
+		}
+	}()
+
+	fake.events <- &iop.RunEvent{Type: "delta", Delta: "streaming before complete"}
+
+	deadline := time.After(2 * time.Second)
+	for {
+		select {
+		case line := <-lines:
+			if strings.Contains(line, `"finish_reason"`) {
+				t.Fatalf("saw finish before streamed content: %s", line)
+			}
+			if strings.Contains(line, `"content":"streaming before complete"`) {
+				fake.events <- &iop.RunEvent{Type: "complete"}
+				select {
+				case <-scanDone:
+				case <-time.After(2 * time.Second):
+					t.Fatal("stream did not close after complete")
+				}
+				return
+			}
+		case <-deadline:
+			t.Fatal("tools request did not stream content before complete")
+		}
+	}
+}
+
 func TestChatCompletionsStreamSynthesizesToolCallsFromClineTemplateCall(t *testing.T) {
 	fake := &fakeRunService{events: make(chan *iop.RunEvent, 3)}
 	fake.events <- &iop.RunEvent{Type: "delta", Delta: "먼저 현재 git 상태를 확인하겠습니다.\n\n"}

apps/edge/internal/openai/stream.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"strings"
 	"time"
+	"unicode"
 
 	"go.uber.org/zap"
 
@@ -36,13 +37,18 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re
 		}},
 	})
 
-	if len(req.Tools) > 0 || (outputPolicy.Strict && outputPolicy.StreamBuffer) {
+	if outputPolicy.Strict && outputPolicy.StreamBuffer {
 		s.streamBufferedChatCompletion(w, r, handle, id, created, model, flusher, outputPolicy, req.Tools)
 		return
 	}
 
 	var contentBuilder strings.Builder
 	var reasoningBuilder strings.Builder
+	var emittedContent strings.Builder
+	var toolTextFilter *streamToolTextFilter
+	if len(req.Tools) > 0 {
+		toolTextFilter = &streamToolTextFilter{}
+	}
 	defer func() {
 		s.logger.Info("openai chat completion stream closed",
 			zap.String("run_id", handle.Dispatch().RunID),
@@ -86,17 +92,16 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re
 				if event.GetDelta() == "" {
 					continue
 				}
-				contentBuilder.WriteString(event.GetDelta())
-				writeSSE(w, flusher, chatCompletionChunk{
-					ID:      id,
-					Object:  "chat.completion.chunk",
-					Created: created,
-					Model:   model,
-					Choices: []chatCompletionChunkChoice{{
-						Index: 0,
-						Delta: chatDelta{Content: event.GetDelta()},
-					}},
-				})
+				delta := event.GetDelta()
+				contentBuilder.WriteString(delta)
+				if toolTextFilter != nil {
+					delta = toolTextFilter.Append(delta)
+				}
+				if delta == "" {
+					continue
+				}
+				emittedContent.WriteString(delta)
+				writeContentDeltaSSE(w, flusher, id, created, model, delta)
 			case "reasoning_delta":
 				if event.GetDelta() == "" {
 					continue
@@ -116,10 +121,42 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re
 					}},
 				})
 			case "complete":
-				finishReason := event.GetMetadata()["finish_reason"]
+				metadata := event.GetMetadata()
+				finishReason := metadata["finish_reason"]
 				if finishReason == "" {
 					finishReason = "stop"
 				}
+				nativeToolCalls := toolCallsFromRunMetadata(metadata)
+				if len(nativeToolCalls) > 0 {
+					if toolTextFilter != nil {
+						pending := toolTextFilter.Flush()
+						if pending != "" {
+							emittedContent.WriteString(pending)
+							writeContentDeltaSSE(w, flusher, id, created, model, pending)
+						}
+					}
+					nativeToolCalls = normalizeNativeToolCallsForResponse(nativeToolCalls, req.Tools)
+					writeToolCallsDeltaSSE(w, flusher, id, created, model, nativeToolCalls)
+					finishReason = "tool_calls"
+				} else if toolTextFilter != nil && shouldSynthesizeTextToolCalls(handle.Dispatch(), isTextToolFallback(metadata)) {
+					cleaned, toolCalls := synthesizeToolCallsFromText(contentBuilder.String(), req.Tools, handle.Dispatch().RunID)
+					if len(toolCalls) > 0 {
+						if remaining := unstreamedCleanedText(cleaned, emittedContent.String()); remaining != "" {
+							emittedContent.WriteString(remaining)
+							writeContentDeltaSSE(w, flusher, id, created, model, remaining)
+						}
+						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)
+					}
+				} else if toolTextFilter != nil {
+					if pending := toolTextFilter.Flush(); pending != "" {
+						emittedContent.WriteString(pending)
+						writeContentDeltaSSE(w, flusher, id, created, model, pending)
+					}
+				}
 				writeSSE(w, flusher, chatCompletionChunk{
 					ID:      id,
 					Object:  "chat.completion.chunk",
@@ -152,6 +189,116 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re
 	}
 }
 
+type streamToolTextFilter struct {
+	pending string
+}
+
+func (f *streamToolTextFilter) Append(delta string) string {
+	f.pending += delta
+	if idx := firstStreamToolTextCandidateIndex(f.pending); idx >= 0 {
+		out := strings.TrimRightFunc(f.pending[:idx], unicode.IsSpace)
+		f.pending = f.pending[idx:]
+		return out
+	}
+	flushLen := streamToolTextSafeFlushLen(f.pending)
+	out := strings.TrimRightFunc(f.pending[:flushLen], unicode.IsSpace)
+	f.pending = f.pending[len(out):]
+	return out
+}
+
+func (f *streamToolTextFilter) Flush() string {
+	out := f.pending
+	f.pending = ""
+	return out
+}
+
+func firstStreamToolTextCandidateIndex(s string) int {
+	xml := strings.Index(strings.ToLower(s), "<tool_call")
+	mustache := strings.Index(s, textMustacheToolCallOpenHint)
+	switch {
+	case xml < 0:
+		return mustache
+	case mustache < 0:
+		return xml
+	case xml < mustache:
+		return xml
+	default:
+		return mustache
+	}
+}
+
+func streamToolTextSafeFlushLen(s string) int {
+	if s == "" {
+		return 0
+	}
+	const xmlHint = "<tool_call"
+	lower := strings.ToLower(s)
+	start := len(s) - len(xmlHint) + 1
+	if start < 0 {
+		start = 0
+	}
+	for i := start; i < len(s); i++ {
+		if strings.HasPrefix(xmlHint, lower[i:]) ||
+			strings.HasPrefix(textMustacheToolCallOpenHint, s[i:]) {
+			return i
+		}
+	}
+	return len(s)
+}
+
+func unstreamedCleanedText(cleaned, emitted string) string {
+	if cleaned == "" {
+		return ""
+	}
+	if emitted == "" {
+		return cleaned
+	}
+	if strings.HasPrefix(cleaned, emitted) {
+		return cleaned[len(emitted):]
+	}
+	cleaned = strings.TrimSpace(cleaned)
+	emitted = strings.TrimSpace(emitted)
+	if cleaned == "" || cleaned == emitted {
+		return ""
+	}
+	if emitted != "" && strings.HasPrefix(cleaned, emitted) {
+		return strings.TrimSpace(strings.TrimPrefix(cleaned, emitted))
+	}
+	return ""
+}
+
+func writeContentDeltaSSE(w http.ResponseWriter, flusher http.Flusher, id string, created int64, model string, content string) {
+	if content == "" {
+		return
+	}
+	writeSSE(w, flusher, chatCompletionChunk{
+		ID:      id,
+		Object:  "chat.completion.chunk",
+		Created: created,
+		Model:   model,
+		Choices: []chatCompletionChunkChoice{{
+			Index: 0,
+			Delta: chatDelta{Content: content},
+		}},
+	})
+}
+
+func writeToolCallsDeltaSSE(w http.ResponseWriter, flusher http.Flusher, id string, created int64, model string, toolCalls []any) {
+	if len(toolCalls) == 0 {
+		return
+	}
+	writeSSE(w, flusher, chatCompletionChunk{
+		ID:      id,
+		Object:  "chat.completion.chunk",
+		Created: created,
+		Model:   model,
+		Choices: []chatCompletionChunkChoice{{
+			Index: 0,
+			Delta: chatDelta{ToolCalls: toolCallsForStreamDelta(toolCalls)},
+		}},
+	})
+}
+
 func (s *Server) streamBufferedChatCompletion(w http.ResponseWriter, r *http.Request, handle edgeservice.RunResult, id string, created int64, model string, flusher http.Flusher, outputPolicy strictOutputPolicy, tools []any) {
 	text, reasoning, finishReason, nativeToolCalls, _, textToolFallback, err := collectRunResult(r.Context(), handle.Stream(), handle.WaitTimeout())
 	if err != nil {