fix(webui): OpenAI 兼容面 —— 补 /v1/models + 真流式（原先是假流式）

/v1/* 是**给外部程序用的**（IDE、脚本、agent 框架），不是给人看的聊天页。它的行为必须真符合 OpenAI 协议，否则调用方直接坏掉。下面两条都在 **生产实测**中确认过，不是推理。 ## ① GET /v1/models → 404 几乎每个 OpenAI 客户端（curl 脚本、LangChain、OpenAI SDK、IDE 插件）启动时都会先列模型来探测服务可用性。404 让它们直接判定「服务不可用」，连试都不试 —— 这是集成方最容易踩空、也最难自查的缺口（表现为「连不上」，而实际端点是通的）。新增 handleOpenAIModels。返回什么模型**不重要**，结构合法才重要：本端点不做模型选择（model 只是回显），所以只暴露 HomeAgent 自身。不谎报 GPT 之类名字 —— 那会让用户以为能选模型，实际不能。 ## ② stream=true 是假流式实测：首字节 7.79s，随后**整段**内容在一个 chunk 里到达。根因：两条路径都走 InjectTextSyncNoMemory —— **同步等完整回复**才返回，之后才把已拼好的全文切成 3 个 chunk 吐出去。客户端的「生成中」/取消/ 超时/进度条全部失效；300s 超时表现为「卡 5 分钟然后一次性出现」。重写为真流式：先订阅 EventContentDelta / EventReasoningDelta **再**启动注入（顺序反了会漏开头几个分片），边收边转成 chunk，最后用同步调用拿到的完整回复补 usage、发 finish、[DONE]。沿用 handleSSE 的成熟结构（批量 16ms 合并、独立 writer goroutine、done channel 而非 close）。顺带处理内核的 reset 事件：流式失败回退非流式时内核会发 content="" + reset=true（见 internal/agent/core/process.go）。忽略它会让客户端看到半截内容后又接上完整内容（重复且自相矛盾），故识别并丢弃累积。 ## 判据（4 条，变异验证） - TestOpenAIModelsEndpoint / RequiresAuth - TestOpenAIStreamIsActuallyStreaming - TestOpenAINonStreamUnchanged（别把非流式改坏） ★ **判据本身踩了两个坑，都已修正并记在测试注释里**： 1. `httptest.ResponseRecorder` 把整个响应**缓冲在内存里**，请求结束才交付 —— 它**根本观察不到流式**。用它写的流式判据必然是假的。故改用 `httptest.NewServer` + `bufio.Reader` 逐帧读。 2. 「要求首帧早于末帧」**抓不住**假流式：假流式确实是分多次 write 的，帧间间隔是微秒级 > 0，任何 `> 0` 判据都绿（已实测）。真正能区分的是：**首帧是否早于「内核产出最终答案」那一刻**。于是假内核被构造成：发 3 个增量后**扣住**最终响应，直到消费者表现出「已在读帧」才放行 —— 真流式首帧 0.35s，假流式首帧 3.30s。 3. 鉴权判据一度写错：未登录时门户 302 到 /login，若跟随重定向就会拿到登录页的 200，把「被重定向」误判成「鉴权通过」。改用不跟随重定向的客户端。变异验证：去掉 /v1/models 路由 → 判红（还原 404）；不转发增量 → 判红（首帧 3.30s）。全量：35 包全绿。
2026-10-03 15:53:56 +00:00 · 2026-09-26 09:48:56 +08:00
parent b4c6b17721
commit f146781b4b
3 changed files with 509 additions and 65 deletions
--- a/internal/plugins/webui/handler_openai.go
+++ b/internal/plugins/webui/handler_openai.go
@ -6,8 +6,11 @@ import (
 	"time"

 	"encoding/json"
-	agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
 	"net/http"
+	"sync"
+
+	agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
+	"gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
 )

 // OpenAI 兼容面：/v1/chat/completions（含流式）。
@ -49,6 +52,20 @@ func (h *Handler) handleOpenAICompletions(w http.ResponseWriter, r *http.Request
 	ctx, cancel := context.WithTimeout(r.Context(), 300*time.Second)
 	defer cancel()

+	// ★ 流式与非流式是**两条不同的实现路径**，不能共用「先同步等完再分帧」。
+	//
+	// 原实现两条路都走 InjectTextSyncNoMemory（同步等完整回复），流式只是
+	// 把已拼好的全文切成 3 个 chunk 吐出去 —— 实测首字节 7.79s、随后
+	// 整段到达，客户端的「生成中」/取消/进度条全部失效。
+	//
+	// 真流式必须**订阅增量事件边收边转**：先写 SSE 头（把响应状态锁定为
+	// 200，之后再出错也无法改状态码），随后把 EventContentDelta 逐块转成
+	// chunk，最后用同步调用拿到的完整回复收尾（补 usage、发 finish）。
+	if req.Stream {
+		h.streamOpenAI(ctx, w, req.Model, lastMsg.Content)
+		return
+	}
+
 	respCh := make(chan *agentIO.OutputEvent, 1)
 	go func() {
 		// NoMemory：本端点（OpenAI 兼容 /v1/chat/completions）的调用方是
@ -75,11 +92,6 @@ func (h *Handler) handleOpenAICompletions(w http.ResponseWriter, r *http.Request
 	reasoningContent, _ := response.Payload["reasoning_content"].(string)
 	usage, _ := response.Payload["usage"].(map[string]interface{})

-	if req.Stream {
-		h.writeOpenAIStream(w, req.Model, content, reasoningContent, usage)
-		return
-	}
-
 	resp := map[string]interface{}{
 		"id":      fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano()),
 		"object":  "chat.completion",
@ -107,88 +119,221 @@ func (h *Handler) handleOpenAICompletions(w http.ResponseWriter, r *http.Request
 	json.NewEncoder(w).Encode(resp)
 }

-func (h *Handler) writeOpenAIStream(w http.ResponseWriter, model, content, reasoningContent string, usage map[string]interface{}) {
+// streamOpenAI 是 /v1/chat/completions 的真流式实现。
+//
+// 结构与 handleSSE 一致：订阅增量 → 收集 → 写 writer goroutine → 收尾。
+// 关键差异：SSE 端点是在「最终回复已就绪」后才开始收尾，本端点必须**边收边发**。
+func (h *Handler) streamOpenAI(ctx context.Context, w http.ResponseWriter, model, userText string) {
 	flusher, ok := w.(http.Flusher)
 	if !ok {
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "streaming not supported"})
 		return
 	}

+	// 先锁状态码：SSE 一旦写出第一个字节，之后上游再出错也只能以
+	// 「在流里报错」的方式告知（data: {"error":...}），无法改成 4xx/5xx。
+	// 这是所有 SSE 实现的固有限制，写明以免后人误以为能改。
 	w.Header().Set("Content-Type", "text/event-stream")
 	w.Header().Set("Cache-Control", "no-cache")
 	w.Header().Set("Connection", "keep-alive")
+	w.Header().Set("X-Accel-Buffering", "no") // 关掉 nginx/frp 的缓冲，否则流式白做
 	w.WriteHeader(http.StatusOK)
 	flusher.Flush()

-	// 如果有 reasoning_content，先发送一个 reasoning chunk
-	if reasoningContent != "" {
-		reasoningChunk := map[string]interface{}{
-			"id":      fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano()),
-			"object":  "chat.completion.chunk",
-			"created": time.Now().Unix(),
-			"model":   model,
-			"choices": []map[string]interface{}{
-				{
-					"index": 0,
-					"delta": map[string]interface{}{
-						"content":           "",
-						"reasoning_content": reasoningContent,
-					},
-					"finish_reason": nil,
-				},
-			},
+	completionID := fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano())
+	created := time.Now().Unix()
+
+	// delta 收集通道。缓冲 2048 与 handleSSE 同理：LLM token 级高频小包，
+	// 缓冲过小会溢出丢帧。
+	deltaCh := make(chan deltaItem, 2048)
+	done := make(chan struct{})
+
+	// 写 goroutine：批量合并（16ms 窗口）后 flush，降 syscall 次数。
+	// 不 close(deltaCh) —— 订阅回调可能在 handler 返回后仍被总线异步触发，
+	// close 后再发会 panic（生产日志里出现过单日数千次）。
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		pending := make([]string, 0, 64)
+		flushPending := func() {
+			if len(pending) == 0 {
+				return
+			}
+			for _, line := range pending {
+				fmt.Fprintf(w, "%s\n", line)
+			}
+			flusher.Flush()
+			pending = pending[:0]
 		}
-		data, _ := json.Marshal(reasoningChunk)
-		fmt.Fprintf(w, "data: %s\n\n", data)
+		flushTicker := time.NewTicker(16 * time.Millisecond)
+		defer flushTicker.Stop()
+		for {
+			select {
+			case d := <-deltaCh:
+				if d.reset {
+					pending = pending[:0]
+				}
+				if d.content != "" {
+					pending = append(pending, openAIChunkLine(completionID, created, model,
+						map[string]interface{}{"content": d.content}, nil, nil))
+				}
+				if d.reasoning != "" {
+					pending = append(pending, openAIChunkLine(completionID, created, model,
+						map[string]interface{}{"content": "", "reasoning_content": d.reasoning}, nil, nil))
+				}
+				if len(pending) >= 128 {
+					flushPending()
+				}
+			case <-flushTicker.C:
+				flushPending()
+			case <-done:
+				flushPending()
+				return
+			}
+		}
+	}()
+
+	send := func(d deltaItem) {
+		select {
+		case deltaCh <- d:
+		case <-done:
+		}
+	}
+
+	// 订阅增量事件。必须在启动注入**之前**订阅，否则会漏掉开头几个分片。
+	var unsubs []func()
+	deltaSub := func(typ sdk.EventType, pick func(payload map[string]interface{}) (string, bool)) {
+		unsub := h.sdk.Subscribe(typ, func(evt *sdk.Event) {
+			text, ok := pick(evt.Payload)
+			if !ok {
+				return
+			}
+			send(deltaItem{content: text, reset: isResetMarked(evt.Payload)})
+		})
+		unsubs = append(unsubs, unsub)
+	}
+	deltaSub(sdk.EventContentDelta, func(p map[string]interface{}) (string, bool) {
+		v, _ := p["content"].(string)
+		return v, v != ""
+	})
+	deltaSub(sdk.EventReasoningDelta, func(p map[string]interface{}) (string, bool) {
+		v, _ := p["reasoning_content"].(string)
+		return v, v != ""
+	})
+	defer func() {
+		close(done)
+		wg.Wait()
+		for _, u := range unsubs {
+			u()
+		}
+	}()
+
+	// 同步拿最终回复：用于补 usage、发 finish chunk。
+	// 此时增量早已转发完毕，所以这次等待不会拖慢首字节。
+	respCh := make(chan *agentIO.OutputEvent, 1)
+	go func() {
+		respCh <- h.sdk.InjectTextSyncNoMemory("http", "http", userText)
+	}()
+	var response *agentIO.OutputEvent
+	select {
+	case response = <-respCh:
+	case <-ctx.Done():
+		fmt.Fprintf(w, "data: %s\n\n", `{"error":{"message":"agent timeout (300s)"}}`)
+		fmt.Fprintf(w, "data: [DONE]\n\n")
 		flusher.Flush()
+		return
 	}

-	// content chunk
-	contentChunk := map[string]interface{}{
-		"id":      fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano()),
-		"object":  "chat.completion.chunk",
-		"created": time.Now().Unix(),
-		"model":   model,
-		"choices": []map[string]interface{}{
-			{
-				"index": 0,
-				"delta": map[string]interface{}{
-					"content": content,
-				},
-				"finish_reason": nil,
-			},
-		},
+	var usage map[string]interface{}
+	if response != nil {
+		usage, _ = response.Payload["usage"].(map[string]interface{})
 	}
-	data, _ := json.Marshal(contentChunk)
-	fmt.Fprintf(w, "data: %s\n\n", data)
-	flusher.Flush()
-
-	// finish chunk
-	finishChunk := map[string]interface{}{
-		"id":      fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano()),
-		"object":  "chat.completion.chunk",
-		"created": time.Now().Unix(),
-		"model":   model,
-		"choices": []map[string]interface{}{
-			{
-				"index":         0,
-				"delta":         map[string]interface{}{},
-				"finish_reason": "stop",
-			},
-		},
-	}
-	if usage != nil {
-		finishChunk["usage"] = usage
-	}
-	data, _ = json.Marshal(finishChunk)
-	fmt.Fprintf(w, "data: %s\n\n", data)
-	flusher.Flush()

+	// finish chunk + [DONE]
+	fmt.Fprintf(w, "%s\n", openAIChunkLine(completionID, created, model,
+		map[string]interface{}{}, "stop", usage))
 	fmt.Fprintf(w, "data: [DONE]\n\n")
 	flusher.Flush()
 }

+// openAIChunkLine 把一个 chunk 序列化成一行 SSE。
+func openAIChunkLine(id string, created int64, model string, delta map[string]interface{}, finishReason interface{}, usage map[string]interface{}) string {
+	choice := map[string]interface{}{
+		"index": 0,
+		"delta": delta,
+	}
+	if finishReason != nil {
+		choice["finish_reason"] = finishReason
+	} else {
+		choice["finish_reason"] = nil
+	}
+	chunk := map[string]interface{}{
+		"id":      id,
+		"object":  "chat.completion.chunk",
+		"created": created,
+		"model":   model,
+		"choices": []map[string]interface{}{choice},
+	}
+	if usage != nil {
+		chunk["usage"] = usage
+	}
+	data, _ := json.Marshal(chunk)
+	return "data: " + string(data)
+}
+
+type deltaItem struct {
+	content   string
+	reasoning string
+	// reset 为真表示上游要求清空已显示内容（流式作废回退），前端需要丢弃累积。
+	reset bool
+}
+
+// isResetMarked 判定增量事件是否要求重置。
+//
+// 内核在「流式失败、回退到非流式」时会发一个 content="" + reset=true 的
+// 事件（见 internal/agent/core/process.go）。若忽略它，客户端会看到
+// 半截内容后又接上完整内容（重复且自相矛盾）。
+func isResetMarked(payload map[string]interface{}) bool {
+	reset, _ := payload["reset"].(bool)
+	return reset
+}
+
 type openAIMessage struct {
 	Role    string `json:"role"`
 	Content string `json:"content"`
 }
+
+// handleOpenAIModels 实现 GET /v1/models。
+//
+// 为什么本端点「返回什么模型」并不重要，重要的是**结构合法**：
+// OpenAI 客户端把它当作能力探测（probe）。请求方要的不是模型清单本身，
+// 而是一个「这个端点讲 OpenAI 协议」的确认。
+//
+// 所以这里返回 HomeAgent 自身作为唯一条目：调用方无论填哪个 id 都会
+// 被路由到同一个 agent（本端点的 model 参数确实只是透传）。
+// 不谎报 GPT 之类的名字 —— 那会让用户以为能选模型，而实际不能。
+func (h *Handler) handleOpenAIModels(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+	id := openAIModelID()
+	writeJSON(w, http.StatusOK, map[string]interface{}{
+		"object": "list",
+		"data": []map[string]interface{}{
+			{
+				"id":       id,
+				"object":   "model",
+				"created":  time.Now().Unix(),
+				"owned_by": "homeagent",
+			},
+		},
+	})
+}
+
+// openAIModelID 返回对外暴露的模型标识。
+//
+// 与 /v1/chat/completions 接受任意 model 值保持一致：本端点不做模型选择，
+// 请求里的 model 只是回显。给一个稳定的名字，便于客户端写死在配置里。
+func openAIModelID() string { return "homeagent" }