bvandeusen/FabledApp
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 32 Wiki Activity

feat: show tool status notifications in chat during generation

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
bvandeusen
2026-04-06 11:46:42 -04:00
parent 4919f7a185
commit e2a358a158
6 changed files with 180 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/data/api/chat_api.dart
+24 -5
View File
@@ -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<String> streamGeneration(int conversationId) async* {
// Step 2: GET the SSE stream and yield typed events (text chunks + status updates).
Stream<ChatStreamEvent> 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<String, dynamic>;
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<String, dynamic>;
final status = obj['status'] as String? ?? '';
yield ChatStatusUpdate(status);
} catch (_) {
// Ignore malformed status events
}
}
} else if (line.isEmpty) {
lib/data/repositories/chat_repository.dart
+2 -1
View File
@@ -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<void> sendMessage(int conversationId, String content) =>
_api.sendMessage(conversationId, content);
Stream<String> streamGeneration(int conversationId) =>
Stream<ChatStreamEvent> streamGeneration(int conversationId) =>
_api.streamGeneration(conversationId);
}
lib/providers/briefing_provider.dart
+85 -2
View File
@@ -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<BriefingConversation> {
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<void> 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<BriefingConversation> {
// 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]));
}
lib/providers/chat_provider.dart
+29 -6
View File
@@ -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> {
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<String> {
// ignore: avoid_unused_constructor_parameters
_StreamingStatusNotifier(int convId);
@override
String build() => '';
}
final conversationsProvider =
AsyncNotifierProvider<ConversationsNotifier, List<Conversation>>(
ConversationsNotifier.new);
@@ -103,12 +118,19 @@ class MessagesNotifier extends AsyncNotifier<List<Message>> {
// ── 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<List<Message>> {
}
} finally {
ref.read(isStreamingProvider(convId).notifier).state = false;
ref.read(streamingStatusProvider(convId).notifier).state = '';
}
}
}
lib/screens/chat/chat_screen.dart
+9 -2
View File
@@ -89,6 +89,8 @@ class _ChatScreenState extends ConsumerState<ChatScreen> {
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<ChatScreen> {
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
: '',
),
);
},
),
lib/widgets/chat_message_bubble.dart
+31 -8
View File
@@ -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,