loadConversation method
Implementation
Future<void> loadConversation(String conversationId) async {
try {
resetConversation();
assistantsReplying.value = 0;
isStreaming.value = false;
setDefaultMessageInputFieldHeight();
incomingMessages.clear();
conversationSseSubscription?.cancel();
conversationSseSubscription = null;
messageNotifier = MessageNotifier();
isLoadingConversation.value = true;
loadingMessage.value = LoadingMessage(
message: "",
loadingType: LoadingType.dots,
);
update();
conversation.value = await ConversationService.getConversation(
assistantId,
conversationId,
isMarketplace,
);
if (conversation.value == null) {
isLoadingConversation.value = false;
isConversationHistoryLoaded = false;
update();
return;
}
Get.find<PupauAttachmentsController>().loadAttachments();
// Initialize notifier for possible catch-up/live updates.
messageNotifier.setAssistantId(assistantId);
messageNotifier.setConversationId(conversation.value!.id);
// Try async SSE history + catch-up (feature gated on server).
final String? lastEventId = PupauSharedPreferences.getLastEventId(
conversation.value?.id ?? "",
);
final Stream<SSEModel>? sseStream =
await SSEService.createConversationSseGetStream(
assistantId,
conversation.value!.id,
conversation.value!.token,
lastEventId: lastEventId,
chatController: this,
);
if (sseStream == null) {
// Fallback to previous REST pagination behavior.
// Initialize pagination so that the newest messages are loaded first.
_initializeConversationPagination();
await loadConversationMessages(reset: true);
if (conversation.value != null && messages.length > 1) {
_isFirstMessage = false;
}
isLoadingConversation.value = false;
isConversationHistoryLoaded = true;
update();
PupauEventService.instance.emitPupauEvent(
PupauEvent(
type: UpdateConversationType.conversationChanged,
payload: {
"assistantId": assistantId,
"assistantType": assistant.value?.type ?? AssistantType.assistant,
"conversation": conversation.value!,
},
),
);
setAssistantSettings();
// Guarantee the slack sliver (chatExtraBottomPaddingActive) is gone
// before we measure maxScrollExtent, otherwise scroll targets phantom
// space and visually overshoots past the real last message.
chatExtraBottomPaddingActive.value = false;
latestQueryAssistantClusterHeight.value = 0.0;
update();
_scrollToTrueBottomOnLoadConversation();
return;
}
// SSE flow: first event is always `history`, then catch-up/live events.
isConversationHistoryLoaded = false;
isConversationLastPage = true;
conversationPage = 0;
conversationItemsLoaded = 0;
resetLoadingMessage();
chatExtraBottomPaddingActive.value = false;
latestQueryAssistantClusterHeight.value = 0.0;
update();
setAssistantSettings();
_scrollToTrueBottomOnLoadConversation();
bool historyLoaded = false;
final Completer<void> historyCompleter = Completer<void>();
conversationSseSubscription = sseStream.listen(
(SSEModel sseEvent) async {
_bumpSseIdleTimer();
final String? data = sseEvent.data;
setLastEventId(sseEvent);
if (sseEvent.event == 'history') {
if (historyLoaded) return;
historyLoaded = true;
messages.clear();
resetExtraBottomScrollPadding();
_clearExpandedMessageGroups();
incomingMessages.clear();
try {
final dynamic decoded = jsonDecode(data ?? '[]');
final List<dynamic> items = decoded is List
? decoded
: <dynamic>[];
// Use the exact same message shaping as REST pagination, so
// markdown/thinking/tool-use elements render identically.
final List<PupauMessage> loadedMessages = items
.whereType<Map<String, dynamic>>()
.map(
(Map<String, dynamic> raw) => PupauMessage.fromLoadedChat(
Map<String, dynamic>.from(raw),
),
)
.toList();
// §1.1/§5: backfill the final row's `grounding` onto earlier
// rows in the same `queryGroupId` before splitting into
// user/assistant messages.
backfillGroupGrounding(loadedMessages);
for (final PupauMessage loadedMessage in loadedMessages) {
final PupauMessage userMessage =
MessageService.getUserLoadedMessage(loadedMessage);
final PupauMessage assistantMessage =
MessageService.getAssistantLoadedMessage(loadedMessage);
final String queryGroupId =
InlineThinkingMessageService.queryGroupIdFor(
assistantMessage,
);
if (queryGroupId.isNotEmpty) {
userMessage.groupId = queryGroupId;
assistantMessage.groupId = queryGroupId;
}
if (isFirstMessageInGroup(queryGroupId)) {
messages.insert(0, userMessage);
}
messages.insert(0, assistantMessage);
incomingMessages.add(assistantMessage);
}
_syncAllInlineThinkingMessages();
} catch (_) {
// If parsing fails, keep whatever was loaded by the resetConversation.
}
messages.refresh();
update();
rebuildConversationSkillsFromHistory();
isConversationHistoryLoaded = true;
isConversationLastPage = true;
// Only enter streaming mode immediately when history is not empty.
// If history is empty and no further events arrive, keep non-streaming.
if (messages.isNotEmpty) {
assistantsReplying.value = 1;
isStreaming.value = true;
_bumpSseIdleTimer();
} else {
assistantsReplying.value = 0;
isStreaming.value = false;
setDefaultMessageInputFieldHeight();
}
update();
if (!historyCompleter.isCompleted) {
historyCompleter.complete();
}
return;
}
if (!historyLoaded) return;
if (data == null || data.trim().isEmpty) return;
try {
final Map<String, dynamic> decoded =
jsonDecode(data) as Map<String, dynamic>;
// Async SSE reconnection payloads are wrapped:
// { eventType: "...", payload: {...}, ... }
final String eventType = (decoded['eventType']?.toString() ?? '')
.trim();
// Terminal async events: stop streaming UI.
const List<String> terminalTypes = <String>[
'run_completed',
'run_stopped',
'run_error',
];
if (terminalTypes.contains(eventType)) {
assistantsReplying.value = 0;
isStreaming.value = false;
resetLoadingMessage();
setDefaultMessageInputFieldHeight();
update();
return;
}
// §3.1 catch-up event: can arrive for a turn that finished long
// ago (stale reconnect cursor), so it must NOT be treated as
// "the run is active" like the generic branch below does.
if (eventType == 'grounding_verified') {
applyGroundingVerification(
GroundingVerificationFrame.fromJson(decoded),
);
return;
}
// Any non-terminal event after history means the run is active.
// Mirror the regular sendMessage() behavior (assistantsReplying=1).
if (!isStreaming.value) {
assistantsReplying.value = 1;
isStreaming.value = true;
update();
}
if (eventType == 'message') {
final Map<String, dynamic> payload = decoded['payload'];
manageSSEData(payload, false);
return;
}
} catch (_) {}
},
onError: (e) {
_cancelSseIdleTimer();
if (!historyCompleter.isCompleted) {
historyCompleter.complete();
}
showErrorSnackbar(
"${Strings.apiErrorGeneric.tr} ${Strings.checkConnectionOrRetry.tr}",
);
},
onDone: () {
_cancelSseIdleTimer();
// If the run completed/stopped or the server closes the stream,
// we consider streaming finished.
assistantsReplying.value = 0;
isStreaming.value = false;
resetLoadingMessage();
setDefaultMessageInputFieldHeight();
update();
if (!historyCompleter.isCompleted) {
historyCompleter.complete();
}
},
cancelOnError: false,
);
await historyCompleter.future;
if (conversation.value != null && messages.length > 1) {
_isFirstMessage = false;
}
isLoadingConversation.value = false;
isConversationHistoryLoaded = true;
isConversationLastPage = true;
chatExtraBottomPaddingActive.value = false;
latestQueryAssistantClusterHeight.value = 0.0;
update();
_scrollToTrueBottomOnLoadConversation();
PupauEventService.instance.emitPupauEvent(
PupauEvent(
type: UpdateConversationType.conversationChanged,
payload: {
"assistantId": assistantId,
"assistantType": assistant.value?.type ?? AssistantType.assistant,
"conversation": conversation.value!,
},
),
);
} catch (e) {
String errorMessage = 'Failed to load conversation: ${e.toString()}';
PupauEventService.instance.emitPupauEvent(
PupauEvent(
type: UpdateConversationType.error,
payload: {
"error": errorMessage,
"assistantId": assistantId,
"assistantType": assistant.value?.type ?? AssistantType.assistant,
"conversationId": conversationId,
},
),
);
isLoadingConversation.value = false;
update();
}
}
