Commit 83322120 authored by Mahmoud Aglan's avatar Mahmoud Aglan

fix(chat): make the live feed independent of SSE

The chat showed a spinner for the whole run and only rendered when it finished,
even though the server's stream was correct — verified frame by frame against
production, one per second, each tool call and result as it happened:

  10:01:35  ⚙ $ echo STEP1
  10:01:35    → STEP1
  10:01:36  ⚙ $ echo STEP2

So the transport was fine and the client was not receiving it. Rather than chase
a long-lived stream through a proxy and a mobile network — where buffering, an
idle timeout or a handover all produce exactly this silence — the feed no longer
depends on the stream at all.

Two sources now race: SSE (instant) and a 1.5s poll of the running task
(guaranteed). Whichever is further ahead wins, compared by length because both
render the same append-only text. If the stream never delivers a byte the feed
still updates every 1.5s; if it works, it wins every time and the poll is
invisible. Polling targets GET /api/tasks/:id rather than the whole chat, so it
carries one run's output instead of every message's body on every tick.

Also: the stream now replays with from=0, so opening a chat that is already
mid-run shows what has happened so far instead of only what arrives next.

And "Working…" no longer appears twice — the bubble header says it, so the body
no longer repeats it.
Co-Authored-By: 's avatarClaude Opus 5 (1M context) <noreply@anthropic.com>
parent 84ee14c6
...@@ -198,6 +198,20 @@ class SonApi { ...@@ -198,6 +198,20 @@ class SonApi {
return (j['taskId'] ?? '').toString(); return (j['taskId'] ?? '').toString();
} }
/// One run's current output and status.
///
/// The live feed polls this as its floor. SSE is faster, but it is a
/// long-lived stream over a mobile network and through a proxy — plenty of
/// ways to deliver nothing — and a chat that shows a spinner until the run
/// ends is the failure this exists to prevent.
Future<({String output, RunStatus status})> taskProgress(String taskId) async {
final j = _obj(await _get('/api/tasks/$taskId'));
return (
output: (j['output'] ?? '').toString(),
status: runStatus(j['status']),
);
}
Future<Attachment> uploadAttachment( Future<Attachment> uploadAttachment(
String chatId, { String chatId, {
required String name, required String name,
......
...@@ -30,11 +30,25 @@ class _ChatScreenState extends ConsumerState<ChatScreen> { ...@@ -30,11 +30,25 @@ class _ChatScreenState extends ConsumerState<ChatScreen> {
Object? _error; Object? _error;
bool _sending = false; bool _sending = false;
/// Lines streamed for the run in flight, held separately from history so a /// Output of the run in flight.
/// refresh cannot duplicate what SSE already delivered. ///
final List<String> _live = []; /// Fed from two sources that race deliberately: the SSE stream (instant) and
/// a poll of the running task (guaranteed). Whichever is further ahead wins,
/// so a stream that never delivers — a proxy that buffers, a network that
/// drops it — degrades to a ~1.5s refresh instead of a spinner that sits
/// there until the run ends. Length is the comparison because both sources
/// render the same append-only text.
String _liveText = '';
StreamSubscription<RunEvent>? _sub; StreamSubscription<RunEvent>? _sub;
Timer? _poll; Timer? _poll;
Timer? _retry;
String? _liveTaskId;
void _offer(String text) {
if (text.length <= _liveText.length) return;
setState(() => _liveText = text);
_autoScroll();
}
String? _model; String? _model;
String? _effort; String? _effort;
...@@ -48,8 +62,7 @@ class _ChatScreenState extends ConsumerState<ChatScreen> { ...@@ -48,8 +62,7 @@ class _ChatScreenState extends ConsumerState<ChatScreen> {
@override @override
void dispose() { void dispose() {
_sub?.cancel(); _stopLive();
_poll?.cancel();
_input.dispose(); _input.dispose();
_scroll.dispose(); _scroll.dispose();
super.dispose(); super.dispose();
...@@ -73,7 +86,7 @@ class _ChatScreenState extends ConsumerState<ChatScreen> { ...@@ -73,7 +86,7 @@ class _ChatScreenState extends ConsumerState<ChatScreen> {
// A run that ended while we were away is worth a notification: the whole // A run that ended while we were away is worth a notification: the whole
// workflow is "send it, put the phone away, get told when it lands". // workflow is "send it, put the phone away, get told when it lands".
if (wasRunning && !c.isRunning) { if (wasRunning && !c.isRunning) {
_live.clear(); _liveText = '';
Notifications.runFinished( Notifications.runFinished(
chatTitle: c.title, chatTitle: c.title,
project: c.project, project: c.project,
...@@ -81,11 +94,13 @@ class _ChatScreenState extends ConsumerState<ChatScreen> { ...@@ -81,11 +94,13 @@ class _ChatScreenState extends ConsumerState<ChatScreen> {
); );
} }
if (c.isRunning) { if (c.isRunning) {
// The newest message is the one in flight; both live sources follow it.
final last = c.messages.where((m) => !m.isUser).lastOrNull;
_liveTaskId = last?.taskId;
_follow(); _follow();
_startPolling();
} else { } else {
_sub?.cancel(); _stopLive();
_sub = null;
_poll?.cancel();
} }
} catch (e) { } catch (e) {
if (!mounted) return; if (!mounted) return;
...@@ -98,11 +113,16 @@ class _ChatScreenState extends ConsumerState<ChatScreen> { ...@@ -98,11 +113,16 @@ class _ChatScreenState extends ConsumerState<ChatScreen> {
/// exactly like a hung agent. /// exactly like a hung agent.
void _follow() { void _follow() {
if (_sub != null) return; if (_sub != null) return;
_sub = _api.streamChat(widget.chatId, from: _live.length).listen( // from: 0 replays this run from the beginning, so opening a chat that is
// already mid-run shows what has happened so far rather than only what
// arrives next.
_sub = _api.streamChat(widget.chatId, from: 0).listen(
(ev) { (ev) {
if (!mounted) return; if (!mounted) return;
setState(() => _live.addAll(ev.lines)); if (ev.lines.isNotEmpty) {
_autoScroll(); final joined = ev.lines.join('\n');
_offer(_liveText.isEmpty ? joined : '$_liveText\n$joined');
}
if (ev.done) { if (ev.done) {
_sub?.cancel(); _sub?.cancel();
_sub = null; _sub = null;
...@@ -110,9 +130,11 @@ class _ChatScreenState extends ConsumerState<ChatScreen> { ...@@ -110,9 +130,11 @@ class _ChatScreenState extends ConsumerState<ChatScreen> {
} }
}, },
onError: (_) { onError: (_) {
// Polling carries the feed while the stream is down, so a reconnect is
// an optimisation, not a rescue.
_sub = null; _sub = null;
_poll?.cancel(); _retry?.cancel();
_poll = Timer(const Duration(seconds: 3), () { _retry = Timer(const Duration(seconds: 3), () {
if (mounted && (_chat?.isRunning ?? false)) _follow(); if (mounted && (_chat?.isRunning ?? false)) _follow();
}); });
}, },
...@@ -124,6 +146,34 @@ class _ChatScreenState extends ConsumerState<ChatScreen> { ...@@ -124,6 +146,34 @@ class _ChatScreenState extends ConsumerState<ChatScreen> {
); );
} }
void _startPolling() {
_poll?.cancel();
_poll = Timer.periodic(const Duration(milliseconds: 1500), (_) async {
final id = _liveTaskId;
if (id == null || !mounted) return;
try {
final p = await _api.taskProgress(id);
if (!mounted) return;
_offer(p.output);
if (p.status != RunStatus.running) {
_stopLive();
_load();
}
} catch (_) {
// A dropped poll is not worth surfacing; the next tick retries.
}
});
}
void _stopLive() {
_sub?.cancel();
_sub = null;
_poll?.cancel();
_poll = null;
_retry?.cancel();
_retry = null;
}
void _autoScroll() { void _autoScroll() {
if (!_scroll.hasClients) return; if (!_scroll.hasClients) return;
// Only stick to the bottom if the user is already near it — yanking the // Only stick to the bottom if the user is already near it — yanking the
...@@ -155,7 +205,7 @@ class _ChatScreenState extends ConsumerState<ChatScreen> { ...@@ -155,7 +205,7 @@ class _ChatScreenState extends ConsumerState<ChatScreen> {
); );
_input.clear(); _input.clear();
_pending.clear(); _pending.clear();
_live.clear(); _liveText = '';
HapticFeedback.lightImpact(); HapticFeedback.lightImpact();
await _load(); await _load();
_autoScroll(); _autoScroll();
...@@ -305,10 +355,13 @@ class _ChatScreenState extends ConsumerState<ChatScreen> { ...@@ -305,10 +355,13 @@ class _ChatScreenState extends ConsumerState<ChatScreen> {
// The final assistant turn shows streamed lines while running; history // The final assistant turn shows streamed lines while running; history
// supplies everything else. // supplies everything else.
final isLast = i == msgs.length - 1; final isLast = i == msgs.length - 1;
final live = isLast && !m.isUser && chat.isRunning && _live.isNotEmpty; final isLive = isLast && !m.isUser && chat.isRunning;
// While running, show whichever of the two live sources has more than
// the stored message body — that is the feed.
final body = isLive && _liveText.length > m.content.length ? _liveText : m.content;
return _Bubble( return _Bubble(
message: live ? m.copyWith(content: _live.join('\n')) : m, message: body == m.content ? m : m.copyWith(content: body),
running: isLast && !m.isUser && chat.isRunning, running: isLive,
); );
}, },
); );
......
...@@ -35,8 +35,9 @@ class AgentOutput extends StatelessWidget { ...@@ -35,8 +35,9 @@ class AgentOutput extends StatelessWidget {
if (i > 0) const SizedBox(height: 8), if (i > 0) const SizedBox(height: 8),
blocks[i].build(context), blocks[i].build(context),
], ],
// A live run with nothing rendered yet still needs to look alive. // Nothing here for an empty live run: the bubble header already shows a
if (running && blocks.isEmpty) const _Pulse(label: 'Working…'), // spinner and "Working…", and saying it twice reads as a stuck screen.
if (running && blocks.isEmpty) const _Pulse(label: 'starting…'),
], ],
); );
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment