refactor: 重构聊天模块支持 SSE 断线重连及用户上下文隔离

2026-03-30 09:06:10 +08:00
parent 1aac62f39e
commit 4285b4ec80
28 changed files with 1624 additions and 658 deletions
@@ -317,13 +317,20 @@ class HistoryMessage {
    seq: _asInt(json['seq']),
    role: _asString(json['role']),
    content: _asString(json['content']),
-    timestamp:
-        DateTime.tryParse(_asString(json['timestamp'])) ?? DateTime.now(),
+    timestamp: _parseTimestamp(_asString(json['timestamp'])),
    attachments: _parseHistoryAttachments(json['attachments']),
    uiSchema: _asMap(json['ui_schema']),
  );
 }

+DateTime _parseTimestamp(String value) {
+  final parsed = DateTime.tryParse(value);
+  if (parsed == null) {
+    return DateTime.now();
+  }
+  return parsed.isUtc ? parsed.toLocal() : parsed;
+}
+
 class HistoryAttachment {
  const HistoryAttachment({required this.url, required this.mimeType});

@@ -54,6 +54,8 @@ class _RunInputPayload {
 }

 class AgUiService {
+  static const int _maxSseResumeAttempts = 2;
+
  final ChatApi _chatApi;
  final ChatHistoryRepository? _historyRepository;
  EventCallback onEvent;
@@ -63,6 +65,7 @@ class AgUiService {
  Completer<void>? _activeSseDoneCompleter;

  String? _threadId;
+  String? _userId;
  String? _activeThreadIdForRun;
  String? _activeRunId;
  bool _hasMoreHistory = false;
@@ -98,7 +101,7 @@ class AgUiService {
    _activeThreadIdForRun = threadId;
    _activeRunId = runId;
    try {
-      await _streamEventsFromApi(
+      await _streamEventsWithResume(
        threadId,
        expectedRunId: runId,
        streamToken: streamToken,
@@ -114,12 +117,16 @@ class AgUiService {
    );
  }

-  Future<HistorySnapshot> loadHistory({DateTime? beforeDate}) async {
+  Future<HistorySnapshot> loadHistory({
+    DateTime? beforeDate,
+    bool forceRefresh = false,
+  }) async {
    final repository = _historyRepository;
    final snapshot = repository != null
        ? await repository.loadHistory(
            threadId: _threadId,
            beforeDate: beforeDate,
+            forceRefresh: forceRefresh,
          )
        : await _loadHistoryFromApi(beforeDate: beforeDate);
    if (snapshot.threadId != null && snapshot.threadId!.isNotEmpty) {
@@ -129,6 +136,21 @@ class AgUiService {
    return snapshot;
  }

+  Future<void> setUserContext(String? userId) async {
+    final normalizedUserId = userId?.trim();
+    if (_userId == normalizedUserId) {
+      return;
+    }
+    _userId = normalizedUserId;
+    _threadId = null;
+    _activeThreadIdForRun = null;
+    _activeRunId = null;
+    _hasMoreHistory = false;
+    _lastEventIdByThread.clear();
+    _activeStreamToken += 1;
+    await _cancelActiveSseSubscription();
+  }
+
  Future<HistorySnapshot> _loadHistoryFromApi({DateTime? beforeDate}) async {
    final payload = await _chatApi.fetchHistory(
      threadId: _threadId,
@@ -186,6 +208,7 @@ class AgUiService {
  }) async {
    final sseLines = await _chatApi.streamRunEvents(
      threadId,
+      runId: expectedRunId,
      lastEventId: _lastEventIdByThread[threadId],
    );

@@ -332,6 +355,42 @@ class AgUiService {
    }
  }

+  Future<void> _streamEventsWithResume(
+    String threadId, {
+    required String expectedRunId,
+    required int streamToken,
+  }) async {
+    var attempt = 0;
+    while (true) {
+      try {
+        await _streamEventsFromApi(
+          threadId,
+          expectedRunId: expectedRunId,
+          streamToken: streamToken,
+        );
+        return;
+      } catch (error) {
+        final canResume =
+            _isPrematureSseClose(error) &&
+            attempt < _maxSseResumeAttempts &&
+            streamToken == _activeStreamToken &&
+            _activeThreadIdForRun == threadId &&
+            _activeRunId == expectedRunId;
+        if (!canResume) {
+          rethrow;
+        }
+        attempt += 1;
+        await Future<void>.delayed(Duration(milliseconds: 250 * attempt));
+      }
+    }
+  }
+
+  bool _isPrematureSseClose(Object error) {
+    return error.toString().toLowerCase().contains(
+      'sse closed before terminal event',
+    );
+  }
+
  Future<_RunInputPayload> _buildRunInput({
    required String content,
    List<AttachmentUploadInput>? attachments,
@@ -3,7 +3,8 @@ import '../l10n/l10n.dart';
 enum AgentStage { routing, execution, memory }

 AgentStage? stageFromStepName(String value) {
-  switch (value) {
+  final normalized = value.trim().toLowerCase();
+  switch (normalized) {
    case 'router':
      return AgentStage.routing;
    case 'worker':
@@ -5,6 +5,7 @@ abstract class ChatApi {

  Future<Stream<String>> streamRunEvents(
    String threadId, {
+    required String runId,
    String? lastEventId,
  });

@@ -20,6 +20,7 @@ class TextMessageItem extends ChatListItem {
  @override
  final MessageSender sender;
  final bool isStreaming;
+  final bool isLocalEcho;
  final List<Map<String, dynamic>> attachments;

  TextMessageItem({
@@ -28,6 +29,7 @@ class TextMessageItem extends ChatListItem {
    required this.timestamp,
    required this.sender,
    this.isStreaming = false,
+    this.isLocalEcho = false,
    this.attachments = const [],
  });

@@ -40,6 +42,7 @@ class TextMessageItem extends ChatListItem {
    DateTime? timestamp,
    MessageSender? sender,
    bool? isStreaming,
+    bool? isLocalEcho,
    List<Map<String, dynamic>>? attachments,
  }) => TextMessageItem(
    id: id ?? this.id,
@@ -47,6 +50,7 @@ class TextMessageItem extends ChatListItem {
    timestamp: timestamp ?? this.timestamp,
    sender: sender ?? this.sender,
    isStreaming: isStreaming ?? this.isStreaming,
+    isLocalEcho: isLocalEcho ?? this.isLocalEcho,
    attachments: attachments ?? this.attachments,
  );
 }
@@ -0,0 +1,169 @@
+import 'package:social_app/core/chat/chat_list_item.dart';
+
+class ChatTimelineReconciler {
+  const ChatTimelineReconciler._();
+
+  static List<ChatListItem> merge({
+    required List<ChatListItem> localItems,
+    required List<ChatListItem> remoteItems,
+  }) {
+    final merged = List<ChatListItem>.from(localItems);
+
+    for (final remote in remoteItems) {
+      final sameIdIndex = merged.indexWhere((item) => item.id == remote.id);
+      if (sameIdIndex >= 0) {
+        merged[sameIdIndex] = remote;
+        continue;
+      }
+
+      _reconcileOptimisticUserEcho(merged, remote);
+      merged.add(remote);
+    }
+
+    final byId = <String, ChatListItem>{};
+    for (final item in merged) {
+      byId[item.id] = item;
+    }
+    final collapsed = byId.values.toList();
+    collapsed.sort(_compareItems);
+    return collapsed;
+  }
+
+  static void _reconcileOptimisticUserEcho(
+    List<ChatListItem> merged,
+    ChatListItem remote,
+  ) {
+    if (remote is! TextMessageItem || remote.sender != MessageSender.user) {
+      return;
+    }
+
+    var optimisticIndex = -1;
+    Duration? nearestGap;
+    for (var i = 0; i < merged.length; i++) {
+      final item = merged[i];
+      if (item is! TextMessageItem) {
+        continue;
+      }
+      if (!item.isLocalEcho || item.sender != MessageSender.user) {
+        continue;
+      }
+      if (!isLikelySameUserMessage(item, remote)) {
+        continue;
+      }
+      final gap = item.timestamp.difference(remote.timestamp).abs();
+      if (nearestGap == null || gap < nearestGap) {
+        nearestGap = gap;
+        optimisticIndex = i;
+      }
+    }
+
+    if (optimisticIndex >= 0) {
+      merged.removeAt(optimisticIndex);
+    }
+  }
+
+  static bool isLikelySameUserMessage(
+    TextMessageItem local,
+    TextMessageItem remote,
+  ) {
+    if (_normalizeText(local.content) != _normalizeText(remote.content)) {
+      return false;
+    }
+
+    final diff = local.timestamp.difference(remote.timestamp).abs();
+    if (diff > const Duration(minutes: 1)) {
+      return false;
+    }
+
+    return _attachmentsLikelySame(
+      local.attachments,
+      remote.attachments,
+      timeGap: diff,
+    );
+  }
+
+  static String _normalizeText(String text) {
+    return text.trim().replaceAll(RegExp(r'\s+'), ' ');
+  }
+
+  static bool _attachmentsLikelySame(
+    List<Map<String, dynamic>> local,
+    List<Map<String, dynamic>> remote, {
+    required Duration timeGap,
+  }) {
+    if (local.length != remote.length) {
+      return false;
+    }
+    if (_attachmentMimeSignature(local) != _attachmentMimeSignature(remote)) {
+      return false;
+    }
+
+    final localIdentity = _attachmentIdentitySignature(local);
+    final remoteIdentity = _attachmentIdentitySignature(remote);
+    if (localIdentity.isNotEmpty && remoteIdentity.isNotEmpty) {
+      return localIdentity == remoteIdentity;
+    }
+
+    return timeGap <= const Duration(seconds: 20);
+  }
+
+  static String _attachmentMimeSignature(
+    List<Map<String, dynamic>> attachments,
+  ) {
+    if (attachments.isEmpty) {
+      return '';
+    }
+    final parts = attachments.map((entry) {
+      final mimeType = entry['mimeType'] as String? ?? '';
+      return mimeType;
+    }).toList()..sort();
+    return parts.join('||');
+  }
+
+  static String _attachmentIdentitySignature(
+    List<Map<String, dynamic>> attachments,
+  ) {
+    final parts =
+        attachments
+            .map(_attachmentIdentity)
+            .where((value) => value.isNotEmpty)
+            .toList()
+          ..sort();
+    if (parts.isEmpty) {
+      return '';
+    }
+    return parts.join('||');
+  }
+
+  static String _attachmentIdentity(Map<String, dynamic> entry) {
+    final url = entry['url'] as String?;
+    if (url != null && url.isNotEmpty) {
+      final uri = Uri.tryParse(url);
+      if (uri != null && uri.pathSegments.isNotEmpty) {
+        return uri.pathSegments.last.toLowerCase();
+      }
+      return url.toLowerCase();
+    }
+
+    final path = entry['path'] as String?;
+    if (path != null && path.isNotEmpty) {
+      final normalized = path.replaceAll('\\', '/');
+      final slashIndex = normalized.lastIndexOf('/');
+      return slashIndex >= 0
+          ? normalized.substring(slashIndex + 1).toLowerCase()
+          : normalized.toLowerCase();
+    }
+    return '';
+  }
+
+  static int _compareItems(ChatListItem a, ChatListItem b) {
+    final tsCompare = a.timestamp.compareTo(b.timestamp);
+    if (tsCompare != 0) {
+      return tsCompare;
+    }
+    if (a.sender != b.sender) {
+      return a.sender == MessageSender.user ? -1 : 1;
+    }
+    return a.id.compareTo(b.id);
+  }
+}