Skip to content

feat: implement message continuation and error handling improvements #65

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chenyme merged 2 commits into from
May 27, 2026
Merged

feat: implement message continuation and error handling improvements #65

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
db9f187
feat: implement message continuation and error handling improvements
chenyme May 27, 2026
cdeeb39
Merge branch 'dev' into error_message
chenyme May 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion backend/internal/application/conversation/service_branch.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,7 +155,14 @@ func normalizeDefaultBranchContext(
}

func isContextMessage(item *model.Message) bool {
return item != nil && strings.EqualFold(strings.TrimSpace(item.Status), "success")
if item == nil {
return false
}
status := strings.TrimSpace(item.Status)
if strings.EqualFold(status, "success") {
return true
}
return item.Role == "assistant" && strings.EqualFold(status, "interrupted")
}

func selectLatestDefaultParentCandidate(messages []model.Message) *model.Message {
Expand Down
26 changes: 25 additions & 1 deletion backend/internal/application/conversation/service_helpers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,17 +286,41 @@ func upstreamErrorSummary(err *llm.UpstreamError) string {
return ""
}
lines := make([]string, 0, 3)
if isSuccessfulUpstreamStatus(err.StatusCode) {
lines = append(lines, fmt.Sprintf("模型响应格式不兼容（HTTP %d）", err.StatusCode))
lines = append(lines, "错误：上游返回成功状态码，但响应格式与当前协议不兼容")
return strings.Join(lines, "\n")
}
if err.StatusCode > 0 {
lines = append(lines, fmt.Sprintf("模型请求失败（HTTP %d）", err.StatusCode))
} else {
lines = append(lines, "模型请求失败")
}
if message := strings.TrimSpace(err.Message); message != "" {
if message := normalizeUpstreamErrorMessage(err.Message); message != "" {
lines = append(lines, "错误："+message)
}
return strings.Join(lines, "\n")
}

func isSuccessfulUpstreamStatus(statusCode int) bool {
return statusCode >= 200 && statusCode < 300
}

func normalizeUpstreamErrorMessage(message string) string {
value := strings.TrimSpace(message)
if value == "" || looksLikeRawSSEBody(value) {
return ""
}
return value
}

func looksLikeRawSSEBody(value string) bool {
normalized := strings.TrimSpace(value)
return strings.HasPrefix(normalized, "data:") ||
strings.Contains(normalized, "\ndata:") ||
strings.Contains(normalized, " data:")
}

func wrapUpstreamRequestError(cause error) error {
if cause == nil {
return ErrUpstreamRequestFailed
Expand Down
279 changes: 254 additions & 25 deletions backend/internal/application/conversation/service_message_completion.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,15 @@ package conversation

import (
"context"
"errors"
"strings"
"time"

"github.com/DEEIX-AI/DEEIX-Chat/backend/internal/application/channel"
model "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/conversation"
"github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/llm"
"github.com/DEEIX-AI/DEEIX-Chat/backend/internal/pkg/traceid"
"go.uber.org/zap"
)

type persistMessageGenerationInput struct {
Expand All @@ -25,6 +30,42 @@ type persistMessageGenerationInput struct {
ToolCallRows []model.ToolCall
}

type persistInterruptedMessageGenerationInput struct {
SendInput SendMessageInput
UserMessage *model.Message
AssistantMessage *model.Message
AssistantText string
EstimatedInputTokens int64
Usage llm.Usage
AssistantLatency int64
Error error
ToolCallRows []model.ToolCall
TraceRecorder *messageTraceRecorder
Route *channel.ResolvedRoute
EffectiveOptions map[string]interface{}
ServerSideToolUsage map[string]int64
StartedAt time.Time
}

type interruptedMessageGenerationMetrics struct {
InputTokens int64
OutputTokens int64
LatencyMS int64
ErrorCode string
ErrorMessage string
CacheReadTokens int64
CacheWriteTokens int64
ReasoningTokens int64
}

type persistMessageToolCallsInput struct {
SendInput SendMessageInput
UserMessageID uint
AssistantMessageID uint
RunID string
Rows []model.ToolCall
}

func (s *Service) persistSuccessfulMessageGeneration(ctx context.Context, input persistMessageGenerationInput) error {
input.UserMessage.InputTokens = input.InputTokens
input.UserMessage.CacheReadTokens = input.CacheReadTokens
Expand Down Expand Up @@ -56,31 +97,14 @@ func (s *Service) persistSuccessfulMessageGeneration(ctx context.Context, input
input.AssistantMessage.LatencyMS = input.AssistantLatency
input.AssistantMessage.Status = "success"

if len(input.ToolCallRows) > 0 {
for i := range input.ToolCallRows {
if input.ToolCallRows[i].ConversationID == 0 {
input.ToolCallRows[i].ConversationID = input.SendInput.ConversationID
}
if input.ToolCallRows[i].UserID == 0 {
input.ToolCallRows[i].UserID = input.SendInput.UserID
}
if input.ToolCallRows[i].MessageID == 0 {
input.ToolCallRows[i].MessageID = input.AssistantMessage.ID
}
if strings.TrimSpace(input.ToolCallRows[i].RunID) == "" {
input.ToolCallRows[i].RunID = input.AssistantMessage.RunID
}
}
if err := s.repo.CreateConversationToolCalls(ctx, input.ToolCallRows); err != nil {
return err
}
s.persistToolContextArtifacts(ctx, toolContextArtifactInput{
ConversationID: input.SendInput.ConversationID,
UserID: input.SendInput.UserID,
MessageID: input.UserMessage.ID,
RunID: input.AssistantMessage.RunID,
Rows: input.ToolCallRows,
})
if err := s.persistMessageToolCalls(ctx, persistMessageToolCallsInput{
SendInput: input.SendInput,
UserMessageID: input.UserMessage.ID,
AssistantMessageID: input.AssistantMessage.ID,
RunID: input.AssistantMessage.RunID,
Rows: input.ToolCallRows,
}); err != nil {
return err
}

s.updateStatefulResponseAsync(input.SendInput.ConversationID, input.ResponseID, input.StatefulPromptFingerprint)
Expand All @@ -90,6 +114,211 @@ func (s *Service) persistSuccessfulMessageGeneration(ctx context.Context, input
return nil
}

// persistInterruptedMessageGeneration 在模型调用已经产生可见内容或工具轨迹后失败时，保留本轮 assistant 消息。
// 显式取消由取消流程单独处理，避免把用户主动停止误标为异常中断。
func (s *Service) persistInterruptedMessageGeneration(ctx context.Context, input persistInterruptedMessageGenerationInput) *SendMessageResult {
if !shouldPersistInterruptedMessageGeneration(input) {
return nil
}
persistCtx := ctx
var cancel context.CancelFunc
if persistCtx == nil || persistCtx.Err() != nil {
persistCtx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
}

metrics := resolveInterruptedMessageGenerationMetrics(input)

if err := s.repo.UpdateMessageUsage(
persistCtx,
input.UserMessage.ID,
metrics.InputTokens,
0,
metrics.CacheReadTokens,
metrics.CacheWriteTokens,
0,
); err != nil {
s.logger.Warn("persist_interrupted_user_usage_failed",
zap.String("trace_id", traceid.FromContext(ctx)),
zap.Uint("message_id", input.UserMessage.ID),
zap.Error(err),
)
}
if err := s.repo.UpdateMessageState(persistCtx, input.UserMessage.ID, "success", "", ""); err != nil {
s.logger.Warn("persist_interrupted_user_state_failed",
zap.String("trace_id", traceid.FromContext(ctx)),
zap.Uint("message_id", input.UserMessage.ID),
zap.Error(err),
)
}
if err := s.repo.UpdateAssistantMessageCompletion(
persistCtx,
input.AssistantMessage.ID,
input.AssistantText,
metrics.OutputTokens,
metrics.ReasoningTokens,
metrics.LatencyMS,
"interrupted",
metrics.ErrorCode,
metrics.ErrorMessage,
); err != nil {
s.logger.Error("persist_interrupted_assistant_completion_failed",
zap.String("trace_id", traceid.FromContext(ctx)),
zap.Uint("message_id", input.AssistantMessage.ID),
zap.Error(err),
)
return nil
}
applyInterruptedMessageGenerationState(input, metrics)

if err := s.persistMessageToolCalls(persistCtx, persistMessageToolCallsInput{
SendInput: input.SendInput,
UserMessageID: input.UserMessage.ID,
AssistantMessageID: input.AssistantMessage.ID,
RunID: input.AssistantMessage.RunID,
Rows: input.ToolCallRows,
}); err != nil {
s.logger.Warn("persist_interrupted_tool_calls_failed",
zap.String("trace_id", traceid.FromContext(ctx)),
zap.Uint("message_id", input.AssistantMessage.ID),
zap.Error(err),
)
}
if input.TraceRecorder != nil {
input.TraceRecorder.fail(input.Error)
input.TraceRecorder.attachToMessage(input.AssistantMessage)
}

return buildInterruptedSendMessageResult(input, metrics)
}

// shouldPersistInterruptedMessageGeneration 只在已有可展示内容或可追踪工具结果时保留中断消息。
func shouldPersistInterruptedMessageGeneration(input persistInterruptedMessageGenerationInput) bool {
if input.Error == nil || input.UserMessage == nil || input.AssistantMessage == nil {
return false
}
if errors.Is(input.Error, ErrMessageGenerationCanceled) {
return false
}
hasRetainedToolTrace := len(input.ToolCallRows) > 0 || len(input.ServerSideToolUsage) > 0
return strings.TrimSpace(input.AssistantText) != "" || hasRetainedToolTrace
}

// resolveInterruptedMessageGenerationMetrics 统一处理中断消息的真实 usage 与估算兜底。
func resolveInterruptedMessageGenerationMetrics(input persistInterruptedMessageGenerationInput) interruptedMessageGenerationMetrics {
inputTokens := input.Usage.InputTokens
if inputTokens <= 0 {
inputTokens = input.EstimatedInputTokens
}
outputTokens := input.Usage.OutputTokens
if outputTokens <= 0 && strings.TrimSpace(input.AssistantText) != "" {
outputTokens = estimateTokens(input.AssistantText)
}
latencyMS := input.AssistantLatency
if latencyMS < 0 {
latencyMS = time.Since(input.StartedAt).Milliseconds()
}
if latencyMS < 0 {
latencyMS = 0
}
return interruptedMessageGenerationMetrics{
InputTokens: inputTokens,
OutputTokens: outputTokens,
LatencyMS: latencyMS,
ErrorCode: classifyRunErrorCode(input.Error),
ErrorMessage: truncateError(messageErrorSummary(input.Error), 255),
CacheReadTokens: input.Usage.CacheReadTokens,
CacheWriteTokens: input.Usage.CacheWriteTokens,
ReasoningTokens: input.Usage.ReasoningTokens,
}
}

// applyInterruptedMessageGenerationState 同步内存消息对象，保证后续响应、run 记录和持久化状态一致。
func applyInterruptedMessageGenerationState(input persistInterruptedMessageGenerationInput, metrics interruptedMessageGenerationMetrics) {
input.UserMessage.Status = "success"
input.UserMessage.ErrorCode = ""
input.UserMessage.ErrorMessage = ""
input.UserMessage.InputTokens = metrics.InputTokens
input.UserMessage.CacheReadTokens = metrics.CacheReadTokens
input.UserMessage.CacheWriteTokens = metrics.CacheWriteTokens
input.UserMessage.TokenUsage = metrics.InputTokens + metrics.CacheReadTokens + metrics.CacheWriteTokens

input.AssistantMessage.Content = input.AssistantText
input.AssistantMessage.TokenUsage = metrics.OutputTokens + metrics.ReasoningTokens
input.AssistantMessage.OutputTokens = metrics.OutputTokens
input.AssistantMessage.ReasoningTokens = metrics.ReasoningTokens
input.AssistantMessage.LatencyMS = metrics.LatencyMS
input.AssistantMessage.Status = "interrupted"
input.AssistantMessage.ErrorCode = metrics.ErrorCode
input.AssistantMessage.ErrorMessage = metrics.ErrorMessage
}

// buildInterruptedSendMessageResult 构造中断回复响应，供 handler 继续走计费和前端展示链路。
func buildInterruptedSendMessageResult(input persistInterruptedMessageGenerationInput, metrics interruptedMessageGenerationMetrics) *SendMessageResult {
result := &SendMessageResult{
UserMessage: *input.UserMessage,
AssistantMessage: *input.AssistantMessage,
EffectiveOptions: input.EffectiveOptions,
UsageSpeed: input.Usage.Speed,
UsageServiceTier: input.Usage.ServiceTier,
CacheWrite5mTokens: input.Usage.CacheWrite5mTokens,
CacheWrite1hTokens: input.Usage.CacheWrite1hTokens,
ServerSideToolUsage: input.ServerSideToolUsage,
LatencyMS: metrics.LatencyMS,
}
if input.Route != nil {
result.UpstreamID = input.Route.UpstreamID
result.UpstreamName = input.Route.UpstreamName
result.PlatformModelName = input.Route.PlatformModelName
result.RoutedBindingCode = input.Route.BindingCode
result.UpstreamModelName = input.Route.UpstreamModel
result.UpstreamProtocol = input.Route.Protocol
}
return result
}

// persistMessageToolCalls 持久化工具调用并写入上下文 artifact，成功和中断路径共用同一套归属规则。
func (s *Service) persistMessageToolCalls(ctx context.Context, input persistMessageToolCallsInput) error {
rows := normalizeMessageToolCallRows(input)
if len(rows) == 0 {
return nil
}
if err := s.repo.CreateConversationToolCalls(ctx, rows); err != nil {
return err
}
s.persistToolContextArtifacts(ctx, toolContextArtifactInput{
ConversationID: input.SendInput.ConversationID,
UserID: input.SendInput.UserID,
MessageID: input.UserMessageID,
RunID: input.RunID,
Rows: rows,
})
return nil
}

// normalizeMessageToolCallRows 补齐工具调用归属字段，避免不同路径写入的 trace 缺少 message/run 关联。
func normalizeMessageToolCallRows(input persistMessageToolCallsInput) []model.ToolCall {
if len(input.Rows) == 0 {
return nil
}
rows := append([]model.ToolCall(nil), input.Rows...)
for i := range rows {
if rows[i].ConversationID == 0 {
rows[i].ConversationID = input.SendInput.ConversationID
}
if rows[i].UserID == 0 {
rows[i].UserID = input.SendInput.UserID
}
if rows[i].MessageID == 0 {
rows[i].MessageID = input.AssistantMessageID
}
if strings.TrimSpace(rows[i].RunID) == "" {
rows[i].RunID = input.RunID
}
}
return rows
}

func (s *Service) updateStatefulResponseAsync(conversationID uint, responseID string, promptFingerprint string) {
respID := strings.TrimSpace(responseID)
if respID == "" {
Expand Down
Loading
Loading