diff --git a/.forgejo/workflows/ci.yml b/.forgejo/workflows/ci.yml index 16baa49..b29a55e 100644 --- a/.forgejo/workflows/ci.yml +++ b/.forgejo/workflows/ci.yml @@ -1,7 +1,12 @@ # CI runs first; build only proceeds if all checks pass. # -# Push to dev or main: flutter analyze + flutter test -# Tag v* (release): gates + signed APK build + attach to Forgejo Release +# Push to dev: flutter analyze + flutter test +# Tag v* (release): gates + signed APK build + attach to Forgejo Release +# +# main pushes are NOT gated here: a merge to main only happens after +# dev has already passed CI, and the release tag is the sole trigger +# for a signed APK build. Re-running analyze+test on the merge commit +# just burns runner time without changing the outcome. # # To cut a release: # Create a release via the Forgejo UI on main with a v* tag name. @@ -20,7 +25,7 @@ name: CI & Build on: push: - branches: [dev, main] + branches: [dev] tags: ["v*"] # Cancel older runs on the same branch when a newer push lands. Tag runs @@ -63,7 +68,7 @@ jobs: build: name: Build release APK needs: [analyze] - # Only tag pushes produce a signed release build. dev/main pushes + # Only tag pushes produce a signed release build. dev pushes # run the gates above and stop there. if: startsWith(github.ref, 'refs/tags/') runs-on: ci-runner diff --git a/lib/providers/briefing_provider.dart b/lib/providers/briefing_provider.dart index 0e9034e..f5b1560 100644 --- a/lib/providers/briefing_provider.dart +++ b/lib/providers/briefing_provider.dart @@ -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 { 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 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 { 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 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 _consumeStream(Stream 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 { 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 _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 { 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 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; - } - } } diff --git a/lib/providers/chat_provider.dart b/lib/providers/chat_provider.dart index bca71c1..5b882d8 100644 --- a/lib/providers/chat_provider.dart +++ b/lib/providers/chat_provider.dart @@ -118,6 +118,120 @@ class MessagesNotifier extends AsyncNotifier> { } } + /// 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 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 []; + 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 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 sendMessage(String content) async { final convId = _convId; final repo = ref.read(chatRepositoryProvider); diff --git a/lib/screens/briefing/briefing_screen.dart b/lib/screens/briefing/briefing_screen.dart index 7159c3a..df71caf 100644 --- a/lib/screens/briefing/briefing_screen.dart +++ b/lib/screens/briefing/briefing_screen.dart @@ -41,7 +41,14 @@ class _BriefingScreenState extends ConsumerState @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 ref.read(briefingProvider.notifier).silentRefresh(); } + Future _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 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 }, ), ), - ], + ], + ), ), ), diff --git a/lib/screens/chat/chat_screen.dart b/lib/screens/chat/chat_screen.dart index 5cc3811..987155f 100644 --- a/lib/screens/chat/chat_screen.dart +++ b/lib/screens/chat/chat_screen.dart @@ -44,6 +44,16 @@ class _ChatScreenState extends ConsumerState 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