diff --git a/docs/IM集成开发文档.md b/docs/IM集成开发文档.md
index 1632e8ca..117ae1c1 100644
--- a/docs/IM集成开发文档.md
+++ b/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 块处理：** 流式输出中的 `...` 块会被转换为飞书 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 块处理：** 流式输出中的 `...` 块会被转换为 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 压力。
diff --git a/internal/im/service.go b/internal/im/service.go
index 59989cd0..a1d81fce 100644
--- a/internal/im/service.go
+++ b/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 XML blocks back to plain markdown
// 2. Strip and 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 == "" {
diff --git a/internal/im/stream_display_test.go b/internal/im/stream_display_test.go
index c6cd6e72..b1ec92ec 100644
--- a/internal/im/stream_display_test.go
+++ b/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)
}
diff --git a/internal/im/stream_lifecycle_test.go b/internal/im/stream_lifecycle_test.go
new file mode 100644
index 00000000..d92b4465
--- /dev/null
+++ b/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)
+ }
+}
diff --git a/internal/im/stream_section.go b/internal/im/stream_section.go
new file mode 100644
index 00000000..5ad0166e
--- /dev/null
+++ b/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
+ }
+}
diff --git a/internal/im/telegram/adapter.go b/internal/im/telegram/adapter.go
index 29ddb85f..4e32618e 100644
--- a/internal/im/telegram/adapter.go
+++ b/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
diff --git a/internal/im/think.go b/internal/im/think.go
index 6989f100..30f9cd5e 100644
--- a/internal/im/think.go
+++ b/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).
diff --git a/internal/im/wecom/longconn.go b/internal/im/wecom/longconn.go
index 7afa539e..addd6b41 100644
--- a/internal/im/wecom/longconn.go
+++ b/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.
}()