fix(im): address stream display review feedback

Wait for agent complete without a short timeout, unify Telegram stream
parse mode, remove obsolete FinalAnswer handling, isolate per-section
newline tracking, add lifecycle tests, and update IM streaming docs.

docs/IM集成开发文档.md

@@ -773,8 +773,9 @@ type ReplyMessage struct {
 │ · EventBus 订阅    │
 │ · 300ms 批量刷新   │
 │ · 工具事件展示     │
-│ · SendStreamChunk  │
-│ · EndStream        │
+│ · UpdateStreamContent │
+│ · FinalizeStream      │
+│ · EndStream           │
 └────────────────────┘
             │
             ▼
@@ -833,11 +834,15 @@ type Adapter interface {
 ```go
 type StreamSender interface {
     StartStream(ctx context.Context, incoming *IncomingMessage) (streamID string, err error)
-    SendStreamChunk(ctx context.Context, incoming *IncomingMessage, streamID string, content string) error
+    UpdateStreamContent(ctx context.Context, incoming *IncomingMessage, streamID string, fullContent string) error
+    FinalizeStream(ctx context.Context, incoming *IncomingMessage, streamID string, finalContent string) error
     EndStream(ctx context.Context, incoming *IncomingMessage, streamID string) error
 }
 ```
 
+- `UpdateStreamContent`：流式过程中用**当前可见全文**替换消息（replace 语义，非增量追加）
+- `FinalizeStream`：结束前最后一次替换，通常为折叠思考/工具进度后的最终展示（多为纯答案）
+
 实现此接口后，Service 会自动路由到流式模式。渠道配置 `output_mode: "full"` 可强制关闭。
 
 ### im.FileDownloader — 文件下载 (可选)
@@ -949,14 +954,17 @@ StartStream:
   1. POST /cardkit/v1/cards              → 创建卡片实体 (streaming_mode: true)
   2. POST /im/v1/messages                → 发送卡片消息到聊天
 
-SendStreamChunk:
+UpdateStreamContent:
   3. PUT /cardkit/v1/cards/{id}/elements/{eid}/content  → 更新元素内容 (累积全文)
 
+FinalizeStream:
+  4. PUT /cardkit/v1/cards/{id}/elements/{eid}/content  → 最终可见内容（通常为纯答案）
+
 EndStream:
-  4. PATCH /cardkit/v1/cards/{id}/settings  → 设置 streaming_mode: false
+  5. PATCH /cardkit/v1/cards/{id}/settings  → 设置 streaming_mode: false
 ```
 
-每次 `SendStreamChunk` 发送的是**累积全文**而非增量，由 `feishuStreamState` 跟踪完整内容和严格递增的 `sequence` 序号。
+每次 `UpdateStreamContent` / `FinalizeStream` 发送的是**累积全文**而非增量，由 `feishuStreamState` 跟踪完整内容和严格递增的 `sequence` 序号。
 
 **Think 块处理：** 流式输出中的 `<think>...</think>` 块会被转换为飞书 Markdown 引用块格式：
 
@@ -1017,14 +1025,17 @@ Slack 的流式输出基于消息更新 (chat.update) 实现：
 StartStream:
   1. POST /chat.postMessage              → 发送初始消息，获取 ts (timestamp)
 
-SendStreamChunk:
+UpdateStreamContent:
   2. POST /chat.update                   → 根据 ts 更新消息内容 (累积全文)
 
+FinalizeStream:
+  3. POST /chat.update                   → 最终可见内容替换
+
 EndStream:
-  3. 无需特殊操作
+  4. 无需特殊操作
 ```
 
-每次 `SendStreamChunk` 发送的是**累积全文**而非增量。
+每次 `UpdateStreamContent` / `FinalizeStream` 发送的是**累积全文**而非增量。
 
 #### 源码文件
 
@@ -1079,14 +1090,17 @@ Telegram 的流式输出基于消息编辑 (editMessageText) 实现：
 StartStream:
   1. POST sendMessage               → 发送初始 "正在思考..." 消息，获取 message_id
 
-SendStreamChunk:
-  2. POST editMessageText            → 根据 message_id 更新消息内容（累积全文）
+UpdateStreamContent:
+  2. POST editMessageText            → 根据 message_id 更新消息内容（累积全文，纯文本）
+
+FinalizeStream:
+  3. POST editMessageText            → 最终可见内容替换（与流式阶段相同，纯文本）
 
 EndStream:
-  3. POST editMessageText            → 最终更新，启用 Markdown 解析
+  4. 清理流状态
 ```
 
-每次 `SendStreamChunk` 发送的是**累积全文**而非增量，最小编辑间隔 500ms（避免触发 Telegram 速率限制）。
+每次 `UpdateStreamContent` / `FinalizeStream` 发送的是**累积全文**而非增量，最小编辑间隔 500ms（避免触发 Telegram 速率限制）。中间态与终态均使用纯文本，避免 Markdown 解析失败。
 
 **Think 块处理：** 流式输出中的 `<think>...</think>` 块会被转换为 Telegram 引用块格式：
 
@@ -1156,14 +1170,17 @@ LongConnClient ══Stream══▶ 钉钉 Stream 服务
 StartStream:
   1. POST /v1.0/card/instances/createAndDeliver  → 创建并投递 AI 卡片
 
-SendStreamChunk:
+UpdateStreamContent:
   2. PUT /v1.0/card/streaming                     → 流式更新卡片内容（累积全文）
 
+FinalizeStream:
+  3. PUT /v1.0/card/streaming                     → 最终可见内容替换
+
 EndStream:
-  3. PUT /v1.0/card/streaming (isFinalize=true)   → 标记流式结束
+  4. PUT /v1.0/card/streaming (isFinalize=true)   → 标记流式结束
 ```
 
-每次 `SendStreamChunk` 发送的是**累积全文**（`isFull: true`），最小更新间隔 500ms。
+每次 `UpdateStreamContent` / `FinalizeStream` 发送的是**累积全文**（`isFull: true`），最小更新间隔 500ms。
 
 **无卡片模板时的降级策略：** 未配置 `card_template_id` 时，流式内容在内存中累积，`EndStream` 时一次性通过 `sessionWebhook` 或 OpenAPI 发送完整回复。
 
@@ -1216,11 +1233,14 @@ Mattermost 服务器 ──HTTP POST──▶ /api/v1/im/callback/{channel_id}
 StartStream:
   1. POST /api/v4/posts                    → 创建占位帖（如「正在思考...」），得到 post id
 
-SendStreamChunk:
+UpdateStreamContent:
   2. PUT /api/v4/posts/{post_id}/patch     → Patch message 字段（累积全文）
 
+FinalizeStream:
+  3. PUT /api/v4/posts/{post_id}/patch     → 最终可见内容替换
+
 EndStream:
-  3. 最后一次 Patch，与 SendStreamChunk 相同逻辑
+  4. 清理流状态
 ```
 
 流式刷新间隔由 Service 侧 `streamFlushInterval`（300ms）批量合并，以降低编辑频率、减轻 API 压力。

internal/im/service.go

@@ -196,6 +196,7 @@ func holdbackCutoff(chunk string) int {
 func formatIMOutboundAnswer(ctx context.Context, raw string, tenant *types.Tenant, defaultFileSvc interfaces.FileService) string {
 	return cleanIMContent(ctx, FormatIMDisplayContent(raw, StreamDisplayFinal), tenant, defaultFileSvc)
 }
+
 // cleanIMContent applies all IM-specific content transformations:
 //  1. Collapse <image> XML blocks back to plain markdown
 //  2. Strip <kb/> and <web/> citation tags
@@ -1825,20 +1826,19 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
 	eventBus := event.NewEventBus()
 
 	var (
-		bufMu          sync.Mutex
-		pipelineInner  strings.Builder // quick QA: RAG pipeline steps
-		reasoningInner strings.Builder // quick QA: model reasoning_content
-		agentInner     strings.Builder // agent: retracted text + thoughts + tools
+		bufMu           sync.Mutex
+		reasoningInner  streamSection   // quick QA: model reasoning_content
+		agentInner      streamSection   // agent: retracted text + thoughts
 		agentLiveAnswer strings.Builder // agent: optimistic answer before tool retract
-		answerOuter    strings.Builder // final answer for display persistence
-		answerBuilder  strings.Builder // answer persisted to DB
-		qaErr          error
-		done           = make(chan struct{})
-		completeDone   = make(chan struct{})
-		closeOnce      sync.Once
-		completeOnce   sync.Once
-		agentDone      bool
-		assistantMsg   *types.Message
+		answerOuter     strings.Builder // final answer for display persistence
+		answerBuilder   strings.Builder // answer persisted to DB
+		qaErr           error
+		done            = make(chan struct{})
+		completeDone    = make(chan struct{})
+		closeOnce       sync.Once
+		completeOnce    sync.Once
+		agentDone       bool
+		assistantMsg    *types.Message
 
 		seenToolCalls = make(map[string]bool)
 		agentToolIdx  = make(map[string]int)
@@ -1847,40 +1847,35 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
 		agentToolSteps    []IMToolStep
 		pipelineToolSteps []IMToolStep
 
-		sectionLastNewline = true
-		streamedAny        bool
+		streamedAny bool
 	)
 	closeDone := func() { closeOnce.Do(func() { close(done) }) }
 	closeComplete := func() { completeOnce.Do(func() { close(completeDone) }) }
 
-	sectionWrite := func(b *strings.Builder, s string) {
+	agentWrite := func(s string) {
 		if s == "" {
 			return
 		}
-		b.WriteString(s)
-		sectionLastNewline = s[len(s)-1] == '\n'
+		agentInner.write(s)
 		streamedAny = true
 	}
-
-	sectionEnsureNewlineBefore := func(b *strings.Builder) {
-		if !sectionLastNewline {
-			b.WriteByte('\n')
-			sectionLastNewline = true
+	reasoningWrite := func(s string) {
+		if s == "" {
+			return
 		}
+		reasoningInner.write(s)
+		streamedAny = true
 	}
 
-	agentWrite := func(s string) { sectionWrite(&agentInner, s) }
-	reasoningWrite := func(s string) { sectionWrite(&reasoningInner, s) }
-
 	// retractAgentLiveAnswer moves the optimistic answer into the think block (Web: superseded preamble).
 	retractAgentLiveAnswer := func() {
 		if agentLiveAnswer.Len() == 0 {
 			return
 		}
-		if agentInner.Len() > 0 {
-			sectionEnsureNewlineBefore(&agentInner)
+		if agentInner.text.Len() > 0 {
+			agentInner.ensureNewlineBefore()
 		}
-		sectionWrite(&agentInner, agentLiveAnswer.String())
+		agentInner.write(agentLiveAnswer.String())
 		agentLiveAnswer.Reset()
 	}
 
@@ -1891,10 +1886,9 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
 		}
 		return IMStreamParts{
 			Mode:              mode,
-			PipelineInner:     pipelineInner.String(),
 			PipelineToolSteps: pipelineToolSteps,
-			ReasoningInner:    reasoningInner.String(),
-			AgentInner:        agentInner.String(),
+			ReasoningInner:    reasoningInner.text.String(),
+			AgentInner:        agentInner.text.String(),
 			AgentToolSteps:    agentToolSteps,
 			LiveAnswer:        agentLiveAnswer.String(),
 			Answer:            answerOuter.String(),
@@ -1910,7 +1904,10 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
 
 		bufMu.Lock()
 		if useAgent && !agentDone {
-			sectionWrite(&agentLiveAnswer, data.Content)
+			if data.Content != "" {
+				agentLiveAnswer.WriteString(data.Content)
+				streamedAny = true
+			}
 		} else {
 			answerOuter.WriteString(data.Content)
 			answerBuilder.WriteString(data.Content)
@@ -1960,13 +1957,7 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
 		bufMu.Lock()
 		agentDone = true
 		applyIMCompleteDataToMessage(assistantMsg, data)
-		if data.FinalAnswer != "" {
-			answerBuilder.Reset()
-			answerBuilder.WriteString(data.FinalAnswer)
-			answerOuter.Reset()
-			answerOuter.WriteString(data.FinalAnswer)
-			agentLiveAnswer.Reset()
-		} else if answerBuilder.Len() == 0 && agentLiveAnswer.Len() > 0 {
+		if answerBuilder.Len() == 0 && agentLiveAnswer.Len() > 0 {
 			answerBuilder.WriteString(agentLiveAnswer.String())
 			answerOuter.WriteString(agentLiveAnswer.String())
 		} else if answerBuilder.Len() == 0 && answerOuter.Len() > 0 {
@@ -2120,7 +2111,7 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
 	ticker := time.NewTicker(streamFlushInterval)
 	defer ticker.Stop()
 
-	flush := func(final bool, phase StreamDisplayPhase) {
+	flush := func() {
 		bufMu.Lock()
 		parts := getStreamParts()
 		agentRunning := useAgent && !agentDone
@@ -2131,10 +2122,8 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
 			return
 		}
 
-		if !final {
-			if cut := holdbackCutoff(displaySource); cut < len(displaySource) {
-				displaySource = displaySource[:cut]
-			}
+		if cut := holdbackCutoff(displaySource); cut < len(displaySource) {
+			displaySource = displaySource[:cut]
 		}
 
 		display := cleanIMContent(ctx, displaySource, tenant, s.defaultFileSvc)
@@ -2147,7 +2136,7 @@ loop:
 	for {
 		select {
 		case <-ticker.C:
-			flush(false, StreamDisplayIntermediate)
+			flush()
 		case <-done:
 			break loop
 		case <-qaCtx.Done():
@@ -2158,8 +2147,8 @@ loop:
 	if useAgent {
 		select {
 		case <-completeDone:
-		case <-time.After(500 * time.Millisecond):
-			logger.Warnf(ctx, "[IM] Timed out waiting for agent complete event: session=%s", session.ID)
+		case <-qaCtx.Done():
+			logger.Warnf(ctx, "[IM] QA context ended before agent complete event: session=%s", session.ID)
 		}
 	}
 
@@ -2304,9 +2293,6 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin
 		}
 		answerMu.Lock()
 		applyIMCompleteDataToMessage(assistantMsg, data)
-		if answerBuilder.Len() == 0 && data.FinalAnswer != "" {
-			answerBuilder.WriteString(data.FinalAnswer)
-		}
 		answerMu.Unlock()
 		closeComplete()
 		return nil
@@ -2351,8 +2337,8 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin
 		if useAgent {
 			select {
 			case <-completeDone:
-			case <-time.After(500 * time.Millisecond):
-				logger.Warnf(ctx, "[IM] Timed out waiting for agent complete event: session=%s", session.ID)
+			case <-ctx.Done():
+				logger.Warnf(ctx, "[IM] QA context ended before agent complete event: session=%s", session.ID)
 			}
 		}
 	case <-ctx.Done():
@@ -2772,7 +2758,6 @@ func (s *Service) streamSmartReply(ctx context.Context, chatModel chat.Chat, str
 	// Flush loop with batching (same pattern as handleMessageStream)
 	var (
 		bufMu     sync.Mutex
-		buf       strings.Builder
 		streamRaw strings.Builder
 		done      = make(chan struct{})
 	)
@@ -2782,7 +2767,6 @@ func (s *Service) streamSmartReply(ctx context.Context, chatModel chat.Chat, str
 		for resp := range streamCh {
 			if resp.Content != "" {
 				bufMu.Lock()
-				buf.WriteString(resp.Content)
 				streamRaw.WriteString(resp.Content)
 				bufMu.Unlock()
 			}
@@ -2794,7 +2778,6 @@ func (s *Service) streamSmartReply(ctx context.Context, chatModel chat.Chat, str
 
 	pushStream := func(phase StreamDisplayPhase) {
 		bufMu.Lock()
-		buf.Reset()
 		raw := streamRaw.String()
 		bufMu.Unlock()
 		if raw == "" {

internal/im/stream_display_test.go

@@ -109,19 +109,28 @@ func TestStreamDisplayPipeline_agentScenario_redGreen(t *testing.T) {
 
 func TestStreamDisplayPipeline_quickQA_redGreen(t *testing.T) {
 	during := IMStreamParts{
-		Mode:          IMStreamModeQuickQA,
-		PipelineInner: "⏳ 问题理解\n✅ 问题理解\n⏳ 知识库检索\n",
+		Mode: IMStreamModeQuickQA,
+		PipelineToolSteps: []IMToolStep{
+			{ToolName: "query_understand", Success: true},
+			{ToolName: "knowledge_search", Pending: true, Arguments: map[string]any{"query": "test"}},
+		},
 	}
 	done := IMStreamParts{
-		Mode:          IMStreamModeQuickQA,
-		PipelineInner: "⏳ 问题理解\n✅ 问题理解\n⏳ 知识库检索\n",
-		Answer:        "答案是 B。",
+		Mode: IMStreamModeQuickQA,
+		PipelineToolSteps: []IMToolStep{
+			{ToolName: "query_understand", Success: true},
+			{ToolName: "knowledge_search", Success: true, Arguments: map[string]any{"query": "test"}},
+		},
+		Answer: "答案是 B。",
 	}
 
 	intermediate := FormatIMIntermediateFromParts(during, false)
 	if intermediate == "" {
 		t.Fatal("quick QA should show pipeline progress while streaming")
 	}
+	if !strings.Contains(intermediate, "问题理解") {
+		t.Fatalf("quick QA pipeline should show query_understand step, got: %q", intermediate)
+	}
 	if strings.Contains(intermediate, "思考过程") {
 		t.Fatalf("quick QA pipeline should not use agent think header, got: %q", intermediate)
 	}

internal/im/stream_lifecycle_test.go

@@ -0,0 +1,166 @@
+package im
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+// imStreamDisplayState mirrors the display buffers in handleMessageStream for lifecycle tests.
+type imStreamDisplayState struct {
+	useAgent          bool
+	agentDone         bool
+	agentInner        streamSection
+	agentLiveAnswer   strings.Builder
+	answerOuter       strings.Builder
+	agentToolSteps    []IMToolStep
+	agentToolIdx      map[string]int
+	pipelineToolSteps []IMToolStep
+	pipelineIdx       map[string]int
+}
+
+func newIMStreamDisplayState(useAgent bool) *imStreamDisplayState {
+	return &imStreamDisplayState{
+		useAgent:     useAgent,
+		agentToolIdx: make(map[string]int),
+		pipelineIdx:  make(map[string]int),
+	}
+}
+
+func (s *imStreamDisplayState) retractAgentLiveAnswer() {
+	if s.agentLiveAnswer.Len() == 0 {
+		return
+	}
+	if s.agentInner.text.Len() > 0 {
+		s.agentInner.ensureNewlineBefore()
+	}
+	s.agentInner.write(s.agentLiveAnswer.String())
+	s.agentLiveAnswer.Reset()
+}
+
+func (s *imStreamDisplayState) parts() IMStreamParts {
+	mode := IMStreamModeQuickQA
+	if s.useAgent {
+		mode = IMStreamModeAgent
+	}
+	return IMStreamParts{
+		Mode:              mode,
+		PipelineToolSteps: s.pipelineToolSteps,
+		AgentInner:        s.agentInner.text.String(),
+		AgentToolSteps:    s.agentToolSteps,
+		LiveAnswer:        s.agentLiveAnswer.String(),
+		Answer:            s.answerOuter.String(),
+	}
+}
+
+func (s *imStreamDisplayState) intermediate() string {
+	return FormatIMIntermediateFromParts(s.parts(), s.useAgent && !s.agentDone)
+}
+
+func (s *imStreamDisplayState) final() string {
+	return FormatIMFinalFromParts(s.parts())
+}
+
+// TestIMStreamLifecycle_agentToolRetract simulates handleMessageStream display assembly:
+// live answer → tool retract → tool result → final answer → complete.
+func TestIMStreamLifecycle_agentToolRetract(t *testing.T) {
+	state := newIMStreamDisplayState(true)
+
+	state.agentLiveAnswer.WriteString("好的，让我先搜索知识库。")
+	liveOnly := state.intermediate()
+	if liveOnly != "好的，让我先搜索知识库。" {
+		t.Fatalf("live answer phase = %q", liveOnly)
+	}
+	if strings.Contains(liveOnly, "思考过程") {
+		t.Fatal("think header must not appear before tool retract")
+	}
+
+	state.retractAgentLiveAnswer()
+	upsertIMToolStep(&state.agentToolSteps, state.agentToolIdx, "tool-1", func(step *IMToolStep) {
+		step.ToolName = "grep_chunks"
+		step.Pending = true
+	})
+	duringTools := state.intermediate()
+	if !strings.Contains(duringTools, "思考过程") {
+		t.Fatalf("after retract should show think block, got: %q", duringTools)
+	}
+	if !strings.Contains(duringTools, "好的，让我先搜索知识库") {
+		t.Fatalf("retracted preamble missing, got: %q", duringTools)
+	}
+
+	upsertIMToolStep(&state.agentToolSteps, state.agentToolIdx, "tool-1", func(step *IMToolStep) {
+		step.ToolName = "grep_chunks"
+		step.Pending = false
+		step.Success = true
+	})
+	state.agentLiveAnswer.WriteString("根据检索结果，《文明6》是回合制策略游戏。")
+	withLiveAnswer := state.intermediate()
+	if !strings.Contains(withLiveAnswer, "根据检索结果") {
+		t.Fatalf("live answer after tools missing, got: %q", withLiveAnswer)
+	}
+
+	state.agentDone = true
+	state.answerOuter.WriteString("《文明6》是回合制策略游戏。")
+	final := state.final()
+	if !strings.Contains(final, "《文明6》是回合制策略游戏。") {
+		t.Fatalf("final answer missing, got: %q", final)
+	}
+
+	rec := &recordingStreamSender{}
+	ctx := context.Background()
+	incoming := &IncomingMessage{Platform: PlatformWeCom, UserID: "u1"}
+	streamID, err := rec.StartStream(ctx, incoming)
+	if err != nil {
+		t.Fatalf("StartStream: %v", err)
+	}
+	if err := rec.UpdateStreamContent(ctx, incoming, streamID, duringTools); err != nil {
+		t.Fatalf("UpdateStreamContent: %v", err)
+	}
+	if err := rec.FinalizeStream(ctx, incoming, streamID, final); err != nil {
+		t.Fatalf("FinalizeStream: %v", err)
+	}
+	if err := rec.EndStream(ctx, incoming, streamID); err != nil {
+		t.Fatalf("EndStream: %v", err)
+	}
+
+	chunks, finalized, ended := rec.snapshot()
+	if !ended {
+		t.Fatal("stream should end")
+	}
+	if len(chunks) != 1 {
+		t.Fatalf("expected 1 intermediate update, got %d", len(chunks))
+	}
+	if chunks[0] == finalized {
+		t.Fatal("intermediate and final should differ")
+	}
+}
+
+// TestIMStreamLifecycle_quickQAPipeline verifies quick-QA pipeline steps collapse to answer-only final.
+func TestIMStreamLifecycle_quickQAPipeline(t *testing.T) {
+	state := newIMStreamDisplayState(false)
+
+	upsertIMToolStep(&state.pipelineToolSteps, state.pipelineIdx, "qu-1", func(step *IMToolStep) {
+		step.ToolName = "query_understand"
+		step.Pending = false
+		step.Success = true
+	})
+	upsertIMToolStep(&state.pipelineToolSteps, state.pipelineIdx, "ks-1", func(step *IMToolStep) {
+		step.ToolName = "knowledge_search"
+		step.Pending = true
+		step.Arguments = map[string]any{"query": "文明6"}
+	})
+
+	intermediate := state.intermediate()
+	if intermediate == "" {
+		t.Fatal("pipeline progress should be visible")
+	}
+	if strings.Contains(intermediate, "思考过程") {
+		t.Fatalf("quick QA must not use agent think header, got: %q", intermediate)
+	}
+
+	state.answerOuter.WriteString("答案是 B。")
+	final := state.final()
+	if final != "答案是 B。" {
+		t.Fatalf("final = %q, want answer only", final)
+	}
+}

internal/im/stream_section.go

@@ -0,0 +1,24 @@
+package im
+
+import "strings"
+
+// streamSection tracks newline state for one IM stream text buffer.
+type streamSection struct {
+	lastCharNewline bool
+	text            strings.Builder
+}
+
+func (s *streamSection) write(str string) {
+	if str == "" {
+		return
+	}
+	s.text.WriteString(str)
+	s.lastCharNewline = str[len(str)-1] == '\n'
+}
+
+func (s *streamSection) ensureNewlineBefore() {
+	if !s.lastCharNewline {
+		s.text.WriteByte('\n')
+		s.lastCharNewline = true
+	}
+}

internal/im/telegram/adapter.go

@@ -438,7 +438,7 @@ func (a *Adapter) FinalizeStream(ctx context.Context, incoming *im.IncomingMessa
 	msgID := state.msgID
 	state.mu.Unlock()
 
-	if err := a.editMessage(ctx, chatID, msgID, finalContent, "Markdown"); err != nil {
+	if err := a.editMessage(ctx, chatID, msgID, finalContent, ""); err != nil {
 		logger.Warnf(ctx, "[Telegram] Failed to finalize stream: %v", err)
 	}
 	return nil

internal/im/think.go

@@ -55,7 +55,6 @@ type IMStreamParts struct {
 	Mode IMStreamMode
 
 	// Quick-QA (RagPipelineProgress on Web):
-	PipelineInner     string       // non-tool pipeline narrative (if any)
 	PipelineToolSteps []IMToolStep // query_understand / knowledge_search rows
 	ReasoningInner    string       // model reasoning_content — separate "思考" section
 
@@ -73,8 +72,7 @@ func agentThinkContent(parts IMStreamParts) string {
 }
 
 func quickQAPipelineContent(parts IMStreamParts) string {
-	toolLines := renderIMToolSteps(parts.PipelineToolSteps, FormatIMRagPipelineLine)
-	return mergeIMNarrativeAndTools(parts.PipelineInner, toolLines)
+	return renderIMToolSteps(parts.PipelineToolSteps, FormatIMRagPipelineLine)
 }
 
 // RAGThinkingStyle matches Web RagPipelineProgress thinking row (agent.think).

internal/im/wecom/longconn.go

@@ -380,7 +380,7 @@ func (c *LongConnClient) connectAndRun(ctx context.Context) error {
 		_ = conn.Close()
 		// NOTE: streamBufs is intentionally NOT cleared on reconnect.
 		// Active streams survive reconnections — the WeCom replace-based
-		// protocol means the next SendStreamChunk will resend the full
+		// protocol means the next UpdateStreamContent will resend the full
 		// accumulated content on the new connection. EndStream always
 		// cleans up the buffer, so there is no memory leak.
 	}()