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

feat(briefing): stall watchdog + pull-to-refresh; chat attaches to in-flight streams

Browse Source 
Briefing chat now uses the same StreamIterator + 45s per-event timeout
as the main chat provider, so a dropped SSE socket no longer leaves
isBriefingStreaming stuck true. Adds refreshMessages() that unfreezes
state when the server-side message is complete, wired to a new
pull-to-refresh gesture and the app lifecycle-resume hook.

Chat provider gains attachToGeneration() — safe to call unconditionally
on screen init, so landing on a conversation that's already mid-stream
(e.g. the /news discuss button, which now creates a conv and auto-kicks
generation server-side) picks up live tokens instead of freezing on the
placeholder.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
bvandeusen
2026-04-14 07:55:47 -04:00
parent 1c4e3c018b
commit fa84e40efc
4 changed files with 266 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/providers/briefing_provider.dart
+113 -102
View File
@@ -1,3 +1,5 @@
import 'dart:async';
import 'package:flutter_riverpod/flutter_riverpod.dart';
import '../data/api/chat_api.dart';
@@ -49,6 +51,40 @@ class BriefingNotifier extends AsyncNotifier<BriefingConversation> {
await future;
}
/// Re-fetch the current briefing conversation and unfreeze a stuck
/// streaming state if the server-side message is already complete.
///
/// Same role as MessagesNotifier.refresh() in chat_provider: when an SSE
/// socket dies silently (mobile network handoff, app backgrounded mid-stream,
/// reverse proxy dropping idle sockets) the send loop never observes close
/// and [isBriefingStreamingProvider] stays stuck true. This is the manual
/// recovery path hit by pull-to-refresh, the AppBar refresh button, and the
/// lifecycle-resume hook.
Future<void> refreshMessages() async {
final current = state.value;
if (current == null) {
ref.invalidateSelf();
return;
}
try {
final fresh = await ref.read(briefingApiProvider).getToday();
state = AsyncData(fresh);
final messages = fresh.messages;
Message? lastAssistant;
for (var i = messages.length - 1; i >= 0; i--) {
if (messages[i].role == MessageRole.assistant) {
lastAssistant = messages[i];
break;
}
}
if (lastAssistant != null && lastAssistant.status != 'generating') {
ref.read(isBriefingStreamingProvider.notifier).state = false;
}
} catch (_) {
// Network hiccup — keep existing state; user can retry.
}
}
/// Inject a news article as context and trigger generation.
///
/// Mirrors sendReply() but calls the /discuss endpoint instead of
@@ -77,10 +113,71 @@ class BriefingNotifier extends AsyncNotifier<BriefingConversation> {
rethrow;
}
// SSE stream (best-effort)
bool streamedContent = false;
final streamedContent = await _consumeStream(chatApi.streamGeneration(convId));
await _pollUntilComplete(convId, streamedContent);
}
/// Send a reply to today's briefing conversation.
///
/// Mirrors MessagesNotifier.sendMessage() in chat_provider with the same
/// stall-watchdog pattern:
/// 1. Optimistic UI update
/// 2. POST message to chat endpoint
/// 3. SSE stream with per-event timeout (stall watchdog)
/// 4. Poll until complete
Future<void> sendReply(String content) async {
final conv = state.value;
if (conv == null) return;
final convId = conv.id;
final chatApi = ref.read(chatApiProvider);
final previous = conv.messages;
final userMsg = Message(
conversationId: convId,
role: MessageRole.user,
content: content,
);
final placeholder = Message(
conversationId: convId,
role: MessageRole.assistant,
content: '',
status: 'generating',
);
state = AsyncData(conv.copyWith(messages: [...previous, userMsg, placeholder]));
ref.read(isBriefingStreamingProvider.notifier).state = true;
try {
await for (final event in chatApi.streamGeneration(convId)) {
await chatApi.sendMessage(convId, content);
} catch (e) {
state = AsyncData(conv.copyWith(messages: previous));
ref.read(isBriefingStreamingProvider.notifier).state = false;
rethrow;
}
final streamedContent = await _consumeStream(chatApi.streamGeneration(convId));
await _pollUntilComplete(convId, streamedContent);
}
/// Consume an SSE stream into the current briefing conversation state.
///
/// Uses a StreamIterator with a per-event timeout as a stall watchdog —
/// same rationale as MessagesNotifier.sendMessage() in chat_provider.dart.
/// Mobile networks occasionally drop SSE sockets silently: the TCP
/// connection is half-closed, Dio never sees the close, and `await for`
/// hangs forever with [isBriefingStreamingProvider] stuck true. If no event
/// arrives within the watchdog window we bail out and let the polling pass
/// below reconcile state from the server.
///
/// Returns whether any text content was actually streamed (the polling
/// pass uses this to decide whether it's safe to overwrite with a possibly
/// empty server-side row).
Future<bool> _consumeStream(Stream<ChatStreamEvent> stream) async {
const stallTimeout = Duration(seconds: 45);
bool streamedContent = false;
final iter = StreamIterator(stream);
try {
while (await iter.moveNext().timeout(stallTimeout)) {
final event = iter.current;
final current = state.value;
if (current == null) break;
final msgs = current.messages;
@@ -100,11 +197,22 @@ class BriefingNotifier extends AsyncNotifier<BriefingConversation> {
messages: [...msgs.sublist(0, msgs.length - 1), updated]));
}
}
} on TimeoutException {
// Stall watchdog — no SSE event for stallTimeout. Fall through to
// polling so the UI eventually unfreezes even if the socket is dead.
} catch (_) {
// Fall through to polling.
// SSE failed — fall through to polling.
} finally {
await iter.cancel();
}
return streamedContent;
}
// Poll until complete (max 20 attempts, 2s apart)
/// Poll /api/briefing messages until the last assistant row is complete,
/// same reconcile pattern as MessagesNotifier.sendMessage(). Always clears
/// [isBriefingStreamingProvider] at the end so the input can't stay locked.
Future<void> _pollUntilComplete(int convId, bool streamedContent) async {
final briefingApi = ref.read(briefingApiProvider);
try {
for (var attempt = 0; attempt < 20; attempt++) {
if (attempt > 0) await Future.delayed(const Duration(seconds: 2));
@@ -136,101 +244,4 @@ class BriefingNotifier extends AsyncNotifier<BriefingConversation> {
ref.read(isBriefingStreamingProvider.notifier).state = false;
}
}
/// Send a reply to today's briefing conversation.
///
/// Mirrors MessagesNotifier.sendMessage():
/// 1. Optimistic UI update
/// 2. POST message to chat endpoint
/// 3. SSE stream (best-effort)
/// 4. Poll until complete
Future<void> sendReply(String content) async {
final conv = state.value;
if (conv == null) return;
final convId = conv.id;
final chatApi = ref.read(chatApiProvider);
final previous = conv.messages;
final userMsg = Message(
conversationId: convId,
role: MessageRole.user,
content: content,
);
final placeholder = Message(
conversationId: convId,
role: MessageRole.assistant,
content: '',
status: 'generating',
);
state = AsyncData(conv.copyWith(messages: [...previous, userMsg, placeholder]));
ref.read(isBriefingStreamingProvider.notifier).state = true;
try {
await chatApi.sendMessage(convId, content);
} 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)) {
final current = state.value;
if (current == null) break;
final msgs = current.messages;
if (msgs.isEmpty) continue;
if (event is ChatTextChunk) {
streamedContent = true;
final updated =
msgs.last.copyWith(content: msgs.last.content + event.text);
state = AsyncData(current.copyWith(
messages: [...msgs.sublist(0, msgs.length - 1), updated]));
} else if (event is ChatToolCall) {
final last = msgs.last;
if (last.role != MessageRole.assistant) continue;
final nextCalls = [...?last.toolCalls, event.toolCall];
final updated = last.copyWith(toolCalls: nextCalls);
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 ref.read(briefingApiProvider).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 (_) {
// Clear the generating placeholder so UI doesn't spin forever.
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;
}
}
}
lib/providers/chat_provider.dart
+114
View File
@@ -118,6 +118,120 @@ class MessagesNotifier extends AsyncNotifier<List<Message>> {
}
}
/// Attach to an already-running generation for this conversation.
///
/// Used when the chat screen lands on a conversation that was started by
/// something other than a direct user message — e.g. the /news discuss
/// button, which creates a conversation on the backend and auto-kicks a
/// generation before navigating. Without this the stream runs to
/// completion invisibly and the screen only shows the final persisted
/// message after a manual refresh.
///
/// Safe to call unconditionally on screen init: no-ops when there is no
/// generating assistant message. Mirrors the web chat store's
/// reconnectIfGenerating() helper.
Future<void> attachToGeneration() async {
final convId = _convId;
final repo = ref.read(chatRepositoryProvider);
// Make sure we're looking at fresh server state before deciding whether
// to attach. The provider's build() fetches once; if the conversation
// was seeded via a POST that happened between build() and this call the
// generating placeholder won't be in our in-memory list yet.
try {
final (_, fresh) = await repo.getMessages(convId);
state = AsyncData(fresh);
} catch (_) {
// If we can't load messages we can't attach either — bail cleanly.
return;
}
if (ref.read(isStreamingProvider(convId))) return;
final msgs = state.value ?? const <Message>[];
final hasGeneratingAssistant = msgs.any(
(m) => m.role == MessageRole.assistant && m.status == 'generating',
);
if (!hasGeneratingAssistant) return;
ref.read(isStreamingProvider(convId).notifier).state = true;
const stallTimeout = Duration(seconds: 45);
bool streamedContent = false;
final iter = StreamIterator(repo.streamGeneration(convId));
try {
while (await iter.moveNext().timeout(stallTimeout)) {
final event = iter.current;
if (event is ChatTextChunk) {
streamedContent = true;
ref.read(streamingStatusProvider(convId).notifier).state = '';
final cur = state.requireValue;
if (cur.isEmpty) continue;
// Route text chunks into the generating assistant message. The
// last message is usually the placeholder, but tool-call fan-in
// means we can't rely on that universally.
final idx = _findGeneratingAssistantIndex(cur);
if (idx < 0) continue;
final updated = cur[idx].copyWith(content: cur[idx].content + event.text);
state = AsyncData([
...cur.sublist(0, idx),
updated,
...cur.sublist(idx + 1),
]);
} else if (event is ChatStatusUpdate) {
ref.read(streamingStatusProvider(convId).notifier).state = event.status;
} else if (event is ChatToolCall) {
final cur = state.requireValue;
if (cur.isEmpty) continue;
final idx = _findGeneratingAssistantIndex(cur);
if (idx < 0) continue;
final nextCalls = [...?cur[idx].toolCalls, event.toolCall];
final updated = cur[idx].copyWith(toolCalls: nextCalls);
state = AsyncData([
...cur.sublist(0, idx),
updated,
...cur.sublist(idx + 1),
]);
}
}
} on TimeoutException {
// Stall — fall through to polling.
} catch (_) {
// Stream failed — fall through to polling.
} finally {
await iter.cancel();
}
try {
for (var attempt = 0; attempt < 20; attempt++) {
if (attempt > 0) await Future.delayed(const Duration(seconds: 2));
final (_, fresh) = await repo.getMessages(convId);
final done = fresh.any(
(m) => m.role == MessageRole.assistant && m.status != 'generating',
);
final polledHasContent = fresh.any(
(m) => m.role == MessageRole.assistant && m.content.isNotEmpty,
);
if (!streamedContent || done || polledHasContent) {
state = AsyncData(fresh);
}
if (done) break;
}
} catch (_) {
// Give up silently — user can pull-to-refresh.
} finally {
ref.read(isStreamingProvider(convId).notifier).state = false;
ref.read(streamingStatusProvider(convId).notifier).state = '';
}
}
int _findGeneratingAssistantIndex(List<Message> msgs) {
for (var i = msgs.length - 1; i >= 0; i--) {
final m = msgs[i];
if (m.role == MessageRole.assistant && m.status == 'generating') return i;
}
return -1;
}
Future<void> sendMessage(String content) async {
final convId = _convId;
final repo = ref.read(chatRepositoryProvider);
lib/screens/briefing/briefing_screen.dart
+29 -4
View File
@@ -41,7 +41,14 @@ class _BriefingScreenState extends ConsumerState<BriefingScreen>
@override
void didChangeAppLifecycleState(AppLifecycleState state) {
final wasBackground = !_appInForeground;
_appInForeground = state == AppLifecycleState.resumed;
// On resume, force a message refresh. This also unfreezes a stuck
// streaming state if the SSE socket died while the app was backgrounded
// — silentRefresh won't do that because it guards on isStreaming.
if (_appInForeground && wasBackground && mounted) {
ref.read(briefingProvider.notifier).refreshMessages();
}
}
void _pollSilently() {
@@ -51,6 +58,18 @@ class _BriefingScreenState extends ConsumerState<BriefingScreen>
ref.read(briefingProvider.notifier).silentRefresh();
}
Future<void> _pullToRefresh() async {
try {
await ref.read(briefingProvider.notifier).refreshMessages();
} catch (_) {
if (mounted) {
ScaffoldMessenger.of(context).showSnackBar(
const SnackBar(content: Text('Could not refresh.')),
);
}
}
}
@override
void dispose() {
_pollTimer?.cancel();
@@ -250,9 +269,14 @@ class _BriefingScreenState extends ConsumerState<BriefingScreen>
return Column(
children: [
Expanded(
child: CustomScrollView(
controller: _scrollController,
slivers: [
child: RefreshIndicator(
onRefresh: _pullToRefresh,
child: CustomScrollView(
controller: _scrollController,
// AlwaysScrollable so pull-to-refresh fires even when
// the briefing is empty or shorter than the viewport.
physics: const AlwaysScrollableScrollPhysics(),
slivers: [
if (conv.messages.isEmpty)
SliverFillRemaining(
child: Center(
@@ -293,7 +317,8 @@ class _BriefingScreenState extends ConsumerState<BriefingScreen>
},
),
),
],
],
),
),
),
lib/screens/chat/chat_screen.dart
+10
View File
@@ -44,6 +44,16 @@ class _ChatScreenState extends ConsumerState<ChatScreen>
void initState() {
super.initState();
WidgetsBinding.instance.addObserver(this);
// If we land on a conversation whose last assistant message is already
// mid-stream (e.g. the /news discuss button creates a conv and
// auto-kicks generation), attach to the running stream so the user sees
// live tokens instead of a frozen placeholder.
WidgetsBinding.instance.addPostFrameCallback((_) {
if (!mounted) return;
ref
.read(messagesProvider(widget.conversationId).notifier)
.attachToGeneration();
});
}
@override