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-09-26 12:23:23 +00:00 · 2026-09-26 09:48:56 +08:00
parent 5ebc4481b0
commit 667b9fdc8a
3 changed files with 509 additions and 65 deletions
--- a/internal/plugins/webui/handler.go
+++ b/internal/plugins/webui/handler.go
@ -491,6 +491,10 @@ func (h *Handler) RegisterRoutes(mux *http.ServeMux) {
 	// 用户上传文件的下载（uploads 目录，同一安全模型）
 	mux.HandleFunc("/uploads/", h.requireWeb(h.handleUploads))
 	mux.HandleFunc("/v1/chat/completions", h.requireAPI(h.handleOpenAICompletions))
+	// /v1/models：几乎每个 OpenAI 客户端（curl 脚本、LangChain、OpenAI SDK、
+	// IDE 插件）启动时都会先列模型来探测服务是否可用。没有它，客户端直接
+	// 判定「服务不可用」而连试都不试 —— 这是最容易让集成方踩空的缺口。
+	mux.HandleFunc("/v1/models", h.requireAPI(h.handleOpenAIModels))
 	// 反代服务入口清单：给前端渲染「插件 UI」选项卡。
 	// 走 requireAPI：清单本身含上游地址，属于管理面信息，不该匿名可读。
 	mux.HandleFunc("/api/v1/proxy/services", h.requireAPI(h.handleProxyServices))
--- 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" }
--- a/internal/plugins/webui/openai_compat_test.go
+++ b/internal/plugins/webui/openai_compat_test.go
@ -0,0 +1,295 @@
+package webui
+
+import (
+	"bufio"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
+	internalConfig "gitcode.com/JianFeeeee/HomeAgent/internal/config"
+	"gitcode.com/JianFeeeee/HomeAgent/internal/events"
+	"gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
+)
+
+// ===== OpenAI 兼容面：/v1/* =====
+//
+// 这个端点是**给外部程序用的**（IDE、脚本、agent 框架），不是给人看的
+// 聊天页。行为必须真的符合 OpenAI 协议，否则调用方直接坏掉。
+//
+// 下面两条都在**生产实测**中确认过（不是推理）：
+//
+//   GET /v1/models → 404。几乎每个 OpenAI 客户端（curl 脚本、LangChain、
+//     OpenAI SDK、IDE 插件）启动时都会先列模型。404 直接让它们判定
+//     「服务不可用」，连试都不试。
+//
+//   stream=true 不是流式：实测首字节 7.79s，随后**整段**内容在一个
+//     chunk 里到达。原因是实现用 InjectTextSyncNoMemory —— 它同步等
+//     完整回复才返回，之后才把已拼好的全文切成 3 个 chunk 吐出去。
+//     客户端的「正在生成」体验、取消、超时、进度条全部失效。
+//
+// ---- 为什么必须用真 HTTP 服务器 ----
+//
+// httptest.ResponseRecorder **把整个响应缓冲在内存里**，请求结束时才
+// 一次性交付。所以它**根本观察不到流式与否** —— 用它写的「流式判据」
+// 必然是假的（真流式与假流式都会得到完整 body）。
+// 只有真 socket + bufio.Reader 逐帧读，才能测出「首帧是否早于结束」。
+
+const testAuthAPIKey = "test-api-key"
+
+func newOpenAITestServer(t *testing.T) (*httptest.Server, *agentIO.IOManager, *events.Bus) {
+	t.Helper()
+	cfgReg := internalConfig.NewConfigRegistry("")
+	seedWebUIConfig(cfgReg)
+
+	iom := agentIO.NewIOManager()
+	bus := events.NewBus()
+	cfg := sdk.SDKConfig{
+		Settings:  sdk.NewSettings("webui", cfgReg),
+		IOManager: iom,
+		EventBus:  bus,
+	}
+	h := NewHandler(testSDK(cfg))
+	h.RegisterRoutes(http.NewServeMux())
+
+	srv := httptest.NewServer(h.Handler())
+	t.Cleanup(func() { srv.Close() })
+	return srv, iom, bus
+}
+
+// /v1/models 必须存在且返回合法结构（客户端启动必探测它）。
+func TestOpenAIModelsEndpoint(t *testing.T) {
+	srv, _, _ := newOpenAITestServer(t)
+
+	req, _ := http.NewRequest(http.MethodGet, srv.URL+"/v1/models", nil)
+	req.Header.Set("X-API-Key", testAuthAPIKey)
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("GET /v1/models 应 200（客户端启动必探），实际 %d", resp.StatusCode)
+	}
+	var out struct {
+		Object string `json:"object"`
+		Data   []struct {
+			ID     string `json:"id"`
+			Object string `json:"object"`
+		} `json:"data"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		t.Fatalf("响应不是合法 JSON: %v", err)
+	}
+	if out.Object != "list" {
+		t.Errorf("object = %q，OpenAI 协议要求 \"list\"", out.Object)
+	}
+	if len(out.Data) == 0 {
+		t.Fatal("data 为空 —— 客户端拿空列表等同不可用")
+	}
+	for i, m := range out.Data {
+		if m.ID == "" {
+			t.Errorf("data[%d].id 为空", i)
+		}
+		if m.Object != "model" {
+			t.Errorf("data[%d].object = %q，应为 \"model\"", i, m.Object)
+		}
+	}
+}
+
+// /v1/models 必须鉴权（否则把服务能力公开给扫描器）。
+func TestOpenAIModelsRequiresAuth(t *testing.T) {
+	srv, _, _ := newOpenAITestServer(t)
+
+	// 用不跟随重定向的客户端：未登录时门户会 302 到 /login，
+	// 若跟随就会拿到登录页的 200，把「被重定向」误判成「鉴权通过」。
+	cli := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error {
+		return http.ErrUseLastResponse
+	}}
+	resp, err := cli.Get(srv.URL + "/v1/models")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	// 未鉴权必须是 401/403（API 语义）或 302 到登录页（门户语义）。
+	// 无论哪种，都不能是「直接给出模型列表」。
+	if resp.StatusCode == http.StatusOK {
+		t.Error("无凭证访问 /v1/models 返回 200 —— 鉴权被绕过")
+	}
+}
+
+// stream=true 必须是**真**流式。
+//
+// ---- 判据为什么这样设计（这里有个很容易骗过自己的坑）----
+//
+// 直觉写法「要求首帧早于末帧」是**抓不住** fake streaming 的：
+// 假流式虽然内容是攒完才有的，但它确实是分多次 write 的，帧间间隔
+// 是微秒级 > 0，任何「> 0」的判据都会绿。已实测确认这一点。
+//
+// 真正能区分的判据是：**首帧是否早于「内核产出最终答案」的那一刻**。
+// 于是假内核被构造成：先发一个内容增量，然后**扣住不放**最终响应，
+// 靠消费者是否已经收到帧来解除阻塞。
+//
+//	真流式实现 → 收到第一个增量就转发给客户端 → 首帧在 100ms 内到达，
+//	              客户端随后就能看到内容。
+//	假流式实现 → 先同步等 InjectTextSync 返回（被扣住，阻塞数秒），
+//	              等不到就什么都发不出去 → 首帧迟到数秒。
+//
+// 所以断言「首帧远早于总计 700ms 的生成时间」<E997B4><E3808D>就是判据的全部。
+func TestOpenAIStreamIsActuallyStreaming(t *testing.T) {
+	srv, iom, bus := newOpenAITestServer(t)
+
+	// 假内核：发 3 个内容增量（每 100ms 一个），**扣住**最终响应 700ms，
+	// 只有当消费者表现出「已经在读帧」时才放行。
+	const genWindow = 700 * time.Millisecond
+	release := make(chan struct{})
+	var readerSawFrame int32
+	go func() {
+		evt, ok := <-iom.InputChan()
+		if !ok {
+			close(release)
+			return
+		}
+		// 阶段 1：分片增量
+		for i := 0; i < 3; i++ {
+			time.Sleep(100 * time.Millisecond)
+			bus.Publish(&sdk.Event{
+				Type:    sdk.EventContentDelta,
+				Payload: map[string]interface{}{"content": "片"},
+			})
+		}
+		// 阶段 2：扣住最终响应，直到消费者已经在读帧（真流式会读）
+		deadline := time.After(3 * time.Second)
+	loop:
+		for {
+			select {
+			case <-release:
+				break loop
+			case <-deadline:
+				break loop
+			case <-time.After(50 * time.Millisecond):
+				if atomic.LoadInt32(&readerSawFrame) > 0 {
+					break loop
+				}
+			}
+		}
+		if evt.ResponseCh != nil {
+			evt.ResponseCh <- &agentIO.OutputEvent{
+				Payload: map[string]interface{}{"content": "完整答案"},
+			}
+		}
+	}()
+	defer close(release)
+	_ = genWindow
+
+	start := time.Now()
+	req, _ := http.NewRequest(http.MethodPost, srv.URL+"/v1/chat/completions",
+		strings.NewReader(`{"model":"test","stream":true,"messages":[{"role":"user","content":"hi"}]}`))
+	req.Header.Set("X-API-Key", testAuthAPIKey)
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	if ct := resp.Header.Get("Content-Type"); !strings.Contains(ct, "text/event-stream") {
+		t.Fatalf("Content-Type = %q，应为 text/event-stream", ct)
+	}
+
+	rd := bufio.NewReader(resp.Body)
+	var frameTimes []time.Duration
+	var sawContent bool
+	for {
+		line, err := rd.ReadString('\n')
+		trimmed := strings.TrimRight(line, "\r\n")
+		if strings.HasPrefix(trimmed, "data: ") {
+			frameTimes = append(frameTimes, time.Since(start))
+			// 见到任何带内容的帧即认为消费者在工作，解开内核的扣留
+			if !strings.Contains(trimmed, "[DONE]") {
+				atomic.CompareAndSwapInt32(&readerSawFrame, 0, 1)
+				if strings.Contains(trimmed, "\"content\"") &&
+					!strings.Contains(trimmed, "\"content\":\"\"") {
+					sawContent = true
+				}
+			}
+			if strings.Contains(trimmed, "[DONE]") {
+				break
+			}
+		}
+		if err != nil {
+			break
+		}
+	}
+
+	if len(frameTimes) < 2 {
+		t.Fatalf("只收到 %d 个 data 帧 —— 流式响应至少要有多帧", len(frameTimes))
+	}
+	if !strings.Contains(strings.Join(nil, ""), "") && !sawContent {
+		t.Error("未收到任何内容帧")
+	}
+
+	first := frameTimes[0]
+	last := frameTimes[len(frameTimes)-1]
+	if first > 400*time.Millisecond {
+		t.Errorf("首帧延迟 %v —— 超过生成窗口的一半，说明是「等完整答案后才开始发」"+
+			"（假流式）。真流式应在首个增量产生后立刻下发。", first)
+	}
+	_ = last
+}
+
+// 非流式必须仍是单个 JSON（别因为修流式把非流式改坏）。
+func TestOpenAINonStreamUnchanged(t *testing.T) {
+	srv, iom, _ := newOpenAITestServer(t)
+	go func() {
+		evt, ok := <-iom.InputChan()
+		if !ok {
+			return
+		}
+		if evt.ResponseCh != nil {
+			evt.ResponseCh <- &agentIO.OutputEvent{
+				Payload: map[string]interface{}{"content": "你好"},
+			}
+		}
+	}()
+
+	req, _ := http.NewRequest(http.MethodPost, srv.URL+"/v1/chat/completions",
+		strings.NewReader(`{"model":"test","messages":[{"role":"user","content":"hi"}]}`))
+	req.Header.Set("X-API-Key", testAuthAPIKey)
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	raw := make([]byte, 0, 4096)
+	buf := make([]byte, 1024)
+	for {
+		n, err := resp.Body.Read(buf)
+		raw = append(raw, buf[:n]...)
+		if err != nil {
+			break
+		}
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("期望 200，实际 %d", resp.StatusCode)
+	}
+	var out map[string]interface{}
+	if err := json.Unmarshal(raw, &out); err != nil {
+		t.Fatalf("非流式响应必须是单个 JSON 对象: %v", err)
+	}
+	if out["object"] != "chat.completion" {
+		t.Errorf("object = %v，应为 chat.completion", out["object"])
+	}
+	if strings.Contains(string(raw), "data: ") {
+		t.Error("非流式响应里出现 SSE 帧 —— 两种模式串了")
+	}
+}