refactor(gateway+model): 拆 gateway/server.go 719→204 + 重构 scanSSE/processFrom 到 ≤80行

- server.go 204 + 新 routes.go(193,路由注册/Handler/withMiddleware)/deps.go(108)/contracts.go(231,17依赖接口)/broadcast.go(78,BroadcastPersisted/CapacityAwaken/Mood); 导出面不变; 74条路由pattern+RBAC门逐条比对一致 - model/openai_stream.go: scanSSE 138→25(parse/apply/finalize 对齐 anthropic_stream); openaiStreamState结构体 - channelbridge/inbound.go: processFrom 112→46(blacklistDrop/inboundTargetFor/runInboundTurn/deliverReply) - 纯行为等价; build/vet/test绿(56包)+win/darwin交叉编译
2026-09-30 11:58:49 +08:00
1 parent 4c3c98d5d4
commit 4ea68d17f3
@@ -27,6 +27,22 @@ func (a *openAIToolAcc) idOrName() string {
 	return a.name
 }

+// openAIStreamState is the running aggregation of one streamed
+// /chat/completions response: the assembled Response, the bracketing flags of the
+// currently open segment, and the per-index tool-call accumulators. It is the
+// OpenAI counterpart of anthropic's streamBlock bookkeeping, so both adapters
+// share one parse → apply → finalize shape. emit is the nil-tolerant event sink
+// (a nil onEvent simply drops every event).
+type openAIStreamState struct {
+	agg             Response
+	text            strings.Builder
+	textStarted     bool
+	thinkingStarted bool
+	toolCalls       map[int]*openAIToolAcc
+	toolOrder       []int
+	onEvent         func(StreamEvent)
+}
+
 // scanSSE incrementally decodes stream chunks, accumulates the full body text
 // with a strings.Builder (never reading the whole body at once) and captures
 // usage from the final chunk. It splits each delta into its segment and brackets
@@ -37,19 +53,7 @@ func (a *openAIToolAcc) idOrName() string {
 func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) {
 	scanner := bufio.NewScanner(r)
 	scanner.Buffer(make([]byte, 0, sseBufInit), maxSSELineBytes)
-	agg := Response{}
-	var text strings.Builder
-	textStarted := false
-	thinkingStarted := false
-	toolCalls := map[int]*openAIToolAcc{}
-	var toolOrder []int
-
-	endThinking := func() {
-		if onEvent != nil && thinkingStarted {
-			thinkingStarted = false
-			onEvent(StreamEvent{Kind: StreamKindThinkingEnd})
-		}
-	}
+	st := &openAIStreamState{toolCalls: map[int]*openAIToolAcc{}, onEvent: onEvent}

 	for scanner.Scan() {
 		line := strings.TrimSpace(scanner.Text())
@@ -64,99 +68,143 @@ func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) {
 		if err := json.Unmarshal([]byte(data), &chunk); err != nil {
 			return Response{}, fmt.Errorf("model: %w: decode stream chunk: %v", ErrUpstream, err)
 		}
-		if u := chunk.Usage; u != nil {
-			agg.PromptTokens = u.PromptTokens
-			agg.CompletionTokens = u.CompletionTokens
-			agg.TotalTokens = u.TotalTokens
-		}
-		if len(chunk.Choices) > 0 && chunk.Choices[0].FinishReason != "" {
-			agg.StopReason = chunk.Choices[0].FinishReason
-		}
-		if len(chunk.Choices) == 0 || chunk.Choices[0].Delta == nil {
-			continue
-		}
-		delta := chunk.Choices[0].Delta
-
-		// Reasoning / thinking segment (OpenAI o-series, DeepSeek, DashScope).
-		if delta.ReasoningContent != "" {
-			if onEvent != nil && !thinkingStarted {
-				thinkingStarted = true
-				onEvent(StreamEvent{Kind: StreamKindThinkingStart})
-			}
-			if onEvent != nil {
-				onEvent(StreamEvent{Kind: StreamKindThinkingDelta, Delta: delta.ReasoningContent})
-			}
-			continue
-		}
-
-		// Streaming tool-call fragments.
-		if len(delta.ToolCalls) > 0 {
-			endThinking()
-			for _, tc := range delta.ToolCalls {
-				acc := toolCalls[tc.Index]
-				if acc == nil {
-					acc = &openAIToolAcc{}
-					toolCalls[tc.Index] = acc
-					toolOrder = append(toolOrder, tc.Index)
-				}
-				if acc.id == "" {
-					acc.id = tc.ID
-				}
-				if acc.name == "" {
-					acc.name = tc.Function.Name
-				}
-				if onEvent != nil && !acc.started && acc.idOrName() != "" {
-					acc.started = true
-					onEvent(StreamEvent{Kind: StreamKindToolCallStart, Index: tc.Index, ID: acc.idOrName()})
-				}
-				if tc.Function.Arguments != "" {
-					acc.args.WriteString(tc.Function.Arguments)
-					if onEvent != nil {
-						onEvent(StreamEvent{Kind: StreamKindToolCallDelta, Index: tc.Index, ID: acc.idOrName(), Delta: tc.Function.Arguments})
-					}
-				}
-			}
-			continue
-		}
-
-		// Body-text segment.
-		if delta.Content == "" {
-			continue
-		}
-		endThinking()
-		text.WriteString(delta.Content)
-		if onEvent != nil {
-			if !textStarted {
-				textStarted = true
-				onEvent(StreamEvent{Kind: StreamKindTextStart})
-			}
-			onEvent(StreamEvent{Kind: StreamKindTextDelta, Delta: delta.Content})
-		}
+		st.applyChunk(chunk)
 	}
 	if err := scanner.Err(); err != nil {
 		return Response{}, fmt.Errorf("model: %w: read stream: %v", ErrUpstream, err)
 	}
+	return finalizeOpenAIStream(st), nil
+}

-	if onEvent != nil {
-		if thinkingStarted {
-			onEvent(StreamEvent{Kind: StreamKindThinkingEnd})
-		}
-		if textStarted {
-			onEvent(StreamEvent{Kind: StreamKindTextEnd})
-		}
-		// OpenAI has no per-block stop event, so each tool_call block is closed
-		// when the stream terminates.
-		for _, idx := range toolOrder {
-			acc := toolCalls[idx]
-			if acc == nil || !acc.started {
-				continue
-			}
-			onEvent(StreamEvent{Kind: StreamKindToolCallEnd, Index: idx, ID: acc.idOrName()})
-		}
+// emit forwards one structured event, tolerating a nil sink.
+func (st *openAIStreamState) emit(ev StreamEvent) {
+	if st.onEvent != nil {
+		st.onEvent(ev)
 	}
+}

-	for _, idx := range toolOrder {
-		acc := toolCalls[idx]
+// applyChunk folds a single decoded stream chunk into the running aggregation:
+// usage and stop reason are recorded even on a choice-less terminal chunk, and
+// the segment delta (if any) is routed to its handler.
+func (st *openAIStreamState) applyChunk(chunk openAIChatResponse) {
+	if u := chunk.Usage; u != nil {
+		st.agg.PromptTokens = u.PromptTokens
+		st.agg.CompletionTokens = u.CompletionTokens
+		st.agg.TotalTokens = u.TotalTokens
+	}
+	if len(chunk.Choices) == 0 {
+		return
+	}
+	if chunk.Choices[0].FinishReason != "" {
+		st.agg.StopReason = chunk.Choices[0].FinishReason
+	}
+	if chunk.Choices[0].Delta == nil {
+		return
+	}
+	st.applyDelta(chunk.Choices[0].Delta)
+}
+
+// applyDelta routes one delta fragment to its segment handler. The precedence is
+// the provider's: a reasoning fragment wins over a tool-call fragment, which wins
+// over body text, so a delta never contributes to two segments at once.
+func (st *openAIStreamState) applyDelta(delta *openAIDelta) {
+	switch {
+	case delta.ReasoningContent != "":
+		st.applyThinking(delta.ReasoningContent)
+	case len(delta.ToolCalls) > 0:
+		st.endThinking()
+		for _, tc := range delta.ToolCalls {
+			st.applyToolCall(tc)
+		}
+	case delta.Content != "":
+		st.endThinking()
+		st.applyText(delta.Content)
+	}
+}
+
+// applyThinking forwards one reasoning fragment, opening the thinking segment on
+// the first fragment only.
+func (st *openAIStreamState) applyThinking(fragment string) {
+	if !st.thinkingStarted {
+		st.thinkingStarted = true
+		st.emit(StreamEvent{Kind: StreamKindThinkingStart})
+	}
+	st.emit(StreamEvent{Kind: StreamKindThinkingDelta, Delta: fragment})
+}
+
+// endThinking closes the open thinking segment (no-op when none is open), so a
+// following tool-call or text segment starts a fresh one.
+func (st *openAIStreamState) endThinking() {
+	if !st.thinkingStarted {
+		return
+	}
+	st.thinkingStarted = false
+	st.emit(StreamEvent{Kind: StreamKindThinkingEnd})
+}
+
+// applyToolCall folds one streaming tool-call fragment into its per-index
+// accumulator, opening the block's start event once its id/name is known and
+// forwarding the argument fragments as they arrive.
+func (st *openAIStreamState) applyToolCall(tc openAIToolCallDelta) {
+	acc := st.toolCalls[tc.Index]
+	if acc == nil {
+		acc = &openAIToolAcc{}
+		st.toolCalls[tc.Index] = acc
+		st.toolOrder = append(st.toolOrder, tc.Index)
+	}
+	if acc.id == "" {
+		acc.id = tc.ID
+	}
+	if acc.name == "" {
+		acc.name = tc.Function.Name
+	}
+	if !acc.started && acc.idOrName() != "" {
+		acc.started = true
+		st.emit(StreamEvent{Kind: StreamKindToolCallStart, Index: tc.Index, ID: acc.idOrName()})
+	}
+	if tc.Function.Arguments == "" {
+		return
+	}
+	acc.args.WriteString(tc.Function.Arguments)
+	st.emit(StreamEvent{Kind: StreamKindToolCallDelta, Index: tc.Index, ID: acc.idOrName(), Delta: tc.Function.Arguments})
+}
+
+// applyText appends one body-text fragment, opening the text segment on the first
+// fragment only.
+func (st *openAIStreamState) applyText(fragment string) {
+	st.text.WriteString(fragment)
+	if !st.textStarted {
+		st.textStarted = true
+		st.emit(StreamEvent{Kind: StreamKindTextStart})
+	}
+	st.emit(StreamEvent{Kind: StreamKindTextDelta, Delta: fragment})
+}
+
+// closeSegments emits the end event of every segment still open at stream
+// termination. OpenAI has no per-block stop event, so each started tool_call
+// block is closed here.
+func (st *openAIStreamState) closeSegments() {
+	if st.thinkingStarted {
+		st.emit(StreamEvent{Kind: StreamKindThinkingEnd})
+	}
+	if st.textStarted {
+		st.emit(StreamEvent{Kind: StreamKindTextEnd})
+	}
+	for _, idx := range st.toolOrder {
+		acc := st.toolCalls[idx]
+		if acc == nil || !acc.started {
+			continue
+		}
+		st.emit(StreamEvent{Kind: StreamKindToolCallEnd, Index: idx, ID: acc.idOrName()})
+	}
+}
+
+// assembleToolCalls decodes each accumulated tool-call argument buffer into its
+// structured Args map, in provider order. Unparseable fragments degrade to an
+// empty map (the call is still surfaced).
+func (st *openAIStreamState) assembleToolCalls() {
+	for _, idx := range st.toolOrder {
+		acc := st.toolCalls[idx]
 		if acc == nil {
 			continue
 		}
@@ -166,9 +214,17 @@ func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) {
 				args = map[string]any{}
 			}
 		}
-		agg.ToolCalls = append(agg.ToolCalls, ToolCall{Name: acc.name, ID: acc.id, Args: args})
+		st.agg.ToolCalls = append(st.agg.ToolCalls, ToolCall{Name: acc.name, ID: acc.id, Args: args})
 	}
-	agg.Content = text.String()
-	agg.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: agg.Content}}}
-	return agg, nil
+}
+
+// finalizeOpenAIStream closes the open segments and assembles the aggregated
+// Response: the concatenated body text (reasoning never leaks into it) plus the
+// re-assembled tool calls.
+func finalizeOpenAIStream(st *openAIStreamState) Response {
+	st.closeSegments()
+	st.assembleToolCalls()
+	st.agg.Content = st.text.String()
+	st.agg.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: st.agg.Content}}}
+	return st.agg
 }