From e2a358a158806abf8bccf27ea5596bd1c8206e35 Mon Sep 17 00:00:00 2001 From: bvandeusen Date: Mon, 6 Apr 2026 11:46:42 -0400 Subject: [PATCH] feat: show tool status notifications in chat during generation Parse SSE `status` events alongside `chunk` events and display the status text next to the spinner while the assistant is generating. Previously the Android app discarded all non-chunk SSE events. Co-Authored-By: Claude Sonnet 4.6 --- lib/data/api/chat_api.dart | 29 ++++++-- lib/data/repositories/chat_repository.dart | 3 +- lib/providers/briefing_provider.dart | 87 +++++++++++++++++++++- lib/providers/chat_provider.dart | 35 +++++++-- lib/screens/chat/chat_screen.dart | 11 ++- lib/widgets/chat_message_bubble.dart | 39 ++++++++-- 6 files changed, 180 insertions(+), 24 deletions(-) diff --git a/lib/data/api/chat_api.dart b/lib/data/api/chat_api.dart index 4692812..44f90c0 100644 --- a/lib/data/api/chat_api.dart +++ b/lib/data/api/chat_api.dart @@ -6,6 +6,18 @@ import '../models/conversation.dart'; import '../models/message.dart'; import 'api_client.dart'; +sealed class ChatStreamEvent {} + +class ChatTextChunk extends ChatStreamEvent { + final String text; + ChatTextChunk(this.text); +} + +class ChatStatusUpdate extends ChatStreamEvent { + final String status; // empty string = clear status + ChatStatusUpdate(this.status); +} + class ChatApi { final Dio _dio; const ChatApi(this._dio); @@ -71,8 +83,8 @@ class ChatApi { } } - // Step 2: GET the SSE stream and yield text chunks. - Stream streamGeneration(int conversationId) async* { + // Step 2: GET the SSE stream and yield typed events (text chunks + status updates). + Stream streamGeneration(int conversationId) async* { try { final response = await _dio.get( '/api/chat/conversations/$conversationId/generation/stream', @@ -108,14 +120,21 @@ class ChatApi { if (data == '[DONE]') return; if (currentEvent == 'done' || currentEvent == 'error') return; - // Parse as JSON if possible, otherwise yield raw text. if (currentEvent == 'chunk' || currentEvent.isEmpty) { try { final obj = json.decode(data) as Map; final text = obj['text'] as String? ?? ''; - if (text.isNotEmpty) yield text; + if (text.isNotEmpty) yield ChatTextChunk(text); } catch (_) { - if (data.isNotEmpty) yield data; + if (data.isNotEmpty) yield ChatTextChunk(data); + } + } else if (currentEvent == 'status') { + try { + final obj = json.decode(data) as Map; + final status = obj['status'] as String? ?? ''; + yield ChatStatusUpdate(status); + } catch (_) { + // Ignore malformed status events } } } else if (line.isEmpty) { diff --git a/lib/data/repositories/chat_repository.dart b/lib/data/repositories/chat_repository.dart index 8d548c0..26a6d18 100644 --- a/lib/data/repositories/chat_repository.dart +++ b/lib/data/repositories/chat_repository.dart @@ -1,4 +1,5 @@ import '../api/chat_api.dart'; +export '../api/chat_api.dart' show ChatStreamEvent, ChatTextChunk, ChatStatusUpdate; import '../models/conversation.dart'; import '../models/message.dart'; @@ -14,6 +15,6 @@ class ChatRepository { _api.getMessages(conversationId); Future sendMessage(int conversationId, String content) => _api.sendMessage(conversationId, content); - Stream streamGeneration(int conversationId) => + Stream streamGeneration(int conversationId) => _api.streamGeneration(conversationId); } diff --git a/lib/providers/briefing_provider.dart b/lib/providers/briefing_provider.dart index 42d76be..c866509 100644 --- a/lib/providers/briefing_provider.dart +++ b/lib/providers/briefing_provider.dart @@ -1,5 +1,6 @@ import 'package:flutter_riverpod/flutter_riverpod.dart'; +import '../data/api/chat_api.dart'; import '../data/models/briefing_conversation.dart'; import '../data/models/message.dart'; import 'api_client_provider.dart'; @@ -48,6 +49,86 @@ class BriefingNotifier extends AsyncNotifier { await future; } + /// Inject a news article as context and trigger generation. + /// + /// Mirrors sendReply() but calls the /discuss endpoint instead of + /// /messages so the backend injects article content before generating. + Future discussArticle(int convId, int itemId) async { + final conv = state.value; + if (conv == null) return; + final chatApi = ref.read(chatApiProvider); + final briefingApi = ref.read(briefingApiProvider); + + final previous = conv.messages; + final placeholder = Message( + conversationId: convId, + role: MessageRole.assistant, + content: '', + status: 'generating', + ); + state = AsyncData(conv.copyWith(messages: [...previous, placeholder])); + ref.read(isBriefingStreamingProvider.notifier).state = true; + + try { + await briefingApi.discussArticle(convId, itemId); + } catch (e) { + state = AsyncData(conv.copyWith(messages: previous)); + ref.read(isBriefingStreamingProvider.notifier).state = false; + rethrow; + } + + // SSE stream (best-effort) + bool streamedContent = false; + try { + await for (final event in chatApi.streamGeneration(convId)) { + if (event is! ChatTextChunk) continue; + streamedContent = true; + final current = state.value; + if (current == null) break; + final msgs = current.messages; + if (msgs.isEmpty) continue; + final updated = + msgs.last.copyWith(content: msgs.last.content + event.text); + state = AsyncData(current.copyWith( + messages: [...msgs.sublist(0, msgs.length - 1), updated])); + } + } catch (_) { + // Fall through to polling. + } + + // Poll until complete (max 20 attempts, 2s apart) + try { + for (var attempt = 0; attempt < 20; attempt++) { + if (attempt > 0) await Future.delayed(const Duration(seconds: 2)); + final fresh = await briefingApi.getMessages(convId); + final done = fresh.any( + (m) => m.role == MessageRole.assistant && m.status != 'generating', + ); + final hasContent = fresh.any( + (m) => m.role == MessageRole.assistant && m.content.isNotEmpty, + ); + final current = state.value; + if (current != null && (!streamedContent || done || hasContent)) { + state = AsyncData(current.copyWith(messages: fresh)); + } + if (done) break; + } + } catch (_) { + final current = state.value; + if (current != null) { + final msgs = current.messages; + if (msgs.isNotEmpty && msgs.last.status == 'generating') { + state = AsyncData(current.copyWith(messages: [ + ...msgs.sublist(0, msgs.length - 1), + msgs.last.copyWith(status: 'complete'), + ])); + } + } + } finally { + ref.read(isBriefingStreamingProvider.notifier).state = false; + } + } + /// Send a reply to today's briefing conversation. /// /// Mirrors MessagesNotifier.sendMessage(): @@ -87,13 +168,15 @@ class BriefingNotifier extends AsyncNotifier { // SSE stream (best-effort) bool streamedContent = false; try { - await for (final chunk in chatApi.streamGeneration(convId)) { + await for (final event in chatApi.streamGeneration(convId)) { + if (event is! ChatTextChunk) continue; streamedContent = true; final current = state.value; if (current == null) break; final msgs = current.messages; if (msgs.isEmpty) continue; - final updated = msgs.last.copyWith(content: msgs.last.content + chunk); + final updated = + msgs.last.copyWith(content: msgs.last.content + event.text); state = AsyncData(current.copyWith( messages: [...msgs.sublist(0, msgs.length - 1), updated])); } diff --git a/lib/providers/chat_provider.dart b/lib/providers/chat_provider.dart index b52ba69..0a6e61d 100644 --- a/lib/providers/chat_provider.dart +++ b/lib/providers/chat_provider.dart @@ -2,6 +2,7 @@ import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../data/models/conversation.dart'; import '../data/models/message.dart'; +import '../data/repositories/chat_repository.dart'; import 'api_client_provider.dart'; // Separate NotifierProvider.family so UI re-builds immediately when streaming starts/stops. @@ -18,6 +19,20 @@ class _IsStreamingNotifier extends Notifier { bool build() => false; } +// Tracks the current tool status text during generation (empty = no status). +final streamingStatusProvider = + NotifierProvider.family<_StreamingStatusNotifier, String, int>( + (convId) => _StreamingStatusNotifier(convId), +); + +class _StreamingStatusNotifier extends Notifier { + // ignore: avoid_unused_constructor_parameters + _StreamingStatusNotifier(int convId); + + @override + String build() => ''; +} + final conversationsProvider = AsyncNotifierProvider>( ConversationsNotifier.new); @@ -103,12 +118,19 @@ class MessagesNotifier extends AsyncNotifier> { // ── Step 2: Stream the response (best effort — silent on failure). ── bool streamedContent = false; try { - await for (final chunk in repo.streamGeneration(convId)) { - streamedContent = true; - final msgs = state.requireValue; - if (msgs.isEmpty) continue; - final updated = msgs.last.copyWith(content: msgs.last.content + chunk); - state = AsyncData([...msgs.sublist(0, msgs.length - 1), updated]); + await for (final event in repo.streamGeneration(convId)) { + if (event is ChatTextChunk) { + streamedContent = true; + ref.read(streamingStatusProvider(convId).notifier).state = ''; + final msgs = state.requireValue; + if (msgs.isEmpty) continue; + final updated = + msgs.last.copyWith(content: msgs.last.content + event.text); + state = AsyncData([...msgs.sublist(0, msgs.length - 1), updated]); + } else if (event is ChatStatusUpdate) { + ref.read(streamingStatusProvider(convId).notifier).state = + event.status; + } } } catch (_) { // SSE failed — fall through to the polling reload below. @@ -157,6 +179,7 @@ class MessagesNotifier extends AsyncNotifier> { } } finally { ref.read(isStreamingProvider(convId).notifier).state = false; + ref.read(streamingStatusProvider(convId).notifier).state = ''; } } } diff --git a/lib/screens/chat/chat_screen.dart b/lib/screens/chat/chat_screen.dart index e526b31..0dbb479 100644 --- a/lib/screens/chat/chat_screen.dart +++ b/lib/screens/chat/chat_screen.dart @@ -89,6 +89,8 @@ class _ChatScreenState extends ConsumerState { Widget build(BuildContext context) { final messagesAsync = ref.watch(messagesProvider(widget.conversationId)); final isStreaming = ref.watch(isStreamingProvider(widget.conversationId)); + final streamingStatus = + ref.watch(streamingStatusProvider(widget.conversationId)); final voiceState = ref.watch(voiceProvider); // Scroll when messages change. @@ -139,8 +141,13 @@ class _ChatScreenState extends ConsumerState { padding: const EdgeInsets.symmetric( horizontal: 8, vertical: 12), itemCount: messages.length, - itemBuilder: (context, i) => - ChatMessageBubble(message: messages[i]), + itemBuilder: (context, i) => ChatMessageBubble( + message: messages[i], + streamingStatus: (i == messages.length - 1 && + messages[i].status == 'generating') + ? streamingStatus + : '', + ), ); }, ), diff --git a/lib/widgets/chat_message_bubble.dart b/lib/widgets/chat_message_bubble.dart index 19ed6ea..801afc9 100644 --- a/lib/widgets/chat_message_bubble.dart +++ b/lib/widgets/chat_message_bubble.dart @@ -7,7 +7,12 @@ import '../data/models/message.dart'; class ChatMessageBubble extends StatelessWidget { final Message message; - const ChatMessageBubble({super.key, required this.message}); + final String streamingStatus; + const ChatMessageBubble({ + super.key, + required this.message, + this.streamingStatus = '', + }); @override Widget build(BuildContext context) { @@ -53,13 +58,31 @@ class ChatMessageBubble extends StatelessWidget { child: Padding( padding: const EdgeInsets.symmetric(horizontal: 12, vertical: 8), child: isGenerating && message.content.isEmpty - ? SizedBox( - width: 20, - height: 20, - child: CircularProgressIndicator( - strokeWidth: 2, - color: scheme.onSurfaceVariant, - ), + ? Row( + mainAxisSize: MainAxisSize.min, + children: [ + SizedBox( + width: 16, + height: 16, + child: CircularProgressIndicator( + strokeWidth: 2, + color: scheme.onSurfaceVariant, + ), + ), + if (streamingStatus.isNotEmpty) ...[ + const SizedBox(width: 8), + Flexible( + child: Text( + streamingStatus, + style: TextStyle( + fontSize: 12, + color: scheme.onSurfaceVariant, + fontStyle: FontStyle.italic, + ), + ), + ), + ], + ], ) : MarkdownBody( data: message.content.isEmpty ? '…' : message.content,