Fix duplicated terminal replay on reconnect

apps/mobile_app/lib/features/terminal/terminal_page.dart

@@ -105,8 +105,10 @@ class _TerminalPageState extends ConsumerState<TerminalPage>
   String? _pendingHistorySeed;
   bool _receivedSocketFrame = false;
   bool _historySeeded = false;
+  bool _awaitingAttachReplayFrame = true;
   bool _isKeyTrayVisible = false;
   bool _shouldReconnectOnResume = false;
+  TerminalConnectionState? _lastConnectionState;
 
   @override
   void initState() {
@@ -318,6 +320,15 @@ class _TerminalPageState extends ConsumerState<TerminalPage>
   }
 
   void _handleTerminalFrame(String frame) {
+    if (_awaitingAttachReplayFrame) {
+      _awaitingAttachReplayFrame = false;
+      if (_shouldSuppressAttachReplay(frame)) {
+        _receivedSocketFrame = true;
+        _cancelHistorySeedTimer();
+        return;
+      }
+    }
+
     _receivedSocketFrame = true;
     _cancelHistorySeedTimer();
     terminal.write(frame);
@@ -334,6 +345,18 @@ class _TerminalPageState extends ConsumerState<TerminalPage>
   }
 
   void _handlePageStateChanged() {
+    final connectionState = _connectionState;
+    if (_lastConnectionState != connectionState) {
+      if (connectionState == TerminalConnectionState.connecting ||
+          connectionState == TerminalConnectionState.reconnecting) {
+        _awaitingAttachReplayFrame = true;
+        if (connectionState == TerminalConnectionState.reconnecting) {
+          _resetTerminalForReplay();
+        }
+      }
+      _lastConnectionState = connectionState;
+    }
+
     _scheduleHistorySeedIfNeeded();
   }
 
@@ -382,6 +405,44 @@ class _TerminalPageState extends ConsumerState<TerminalPage>
   bool get _terminalHasVisibleContent =>
       terminal.buffer.getText().trim().isNotEmpty;
 
+  void _resetTerminalForReplay() {
+    if (!_terminalHasVisibleContent) {
+      return;
+    }
+
+    terminal.buffer.clear();
+    terminal.buffer.setCursor(0, 0);
+    terminal.notifyListeners();
+  }
+
+  bool _shouldSuppressAttachReplay(String frame) {
+    final normalizedFrame = _normalizeTerminalText(frame);
+    if (normalizedFrame.isEmpty) {
+      return false;
+    }
+
+    if (!_historySeeded && !_terminalHasVisibleContent) {
+      return false;
+    }
+
+    final normalizedTerminalText = _normalizeTerminalText(
+      terminal.buffer.getText(),
+    );
+    if (normalizedTerminalText.isNotEmpty &&
+        normalizedTerminalText.endsWith(normalizedFrame)) {
+      return true;
+    }
+
+    final pendingHistorySeed = _pendingHistorySeed;
+    return _historySeeded &&
+        pendingHistorySeed != null &&
+        _normalizeTerminalText(pendingHistorySeed) == normalizedFrame;
+  }
+
+  static String _normalizeTerminalText(String text) {
+    return text.replaceAll('\r\n', '\n').replaceAll('\r', '\n');
+  }
+
   Future<void> _showDiagnostics() {
     return showModalBottomSheet<void>(
       context: context,

apps/mobile_app/test/widget_test.dart

@@ -749,6 +749,37 @@ void main() {
     },
   );
 
+  testWidgets(
+    'terminal ignores attach replay when seeded history already restored the same output',
+    (tester) async {
+      final transportFactory = _QueuedTerminalSocketTransportFactory(
+        connectionStartupFrames: [
+          const [
+            _StartupFrame('{"type":"attached","sessionId":"session-1"}'),
+            _StartupFrame('one\r\ntwo', delay: Duration(milliseconds: 200)),
+          ],
+        ],
+      );
+
+      await _pumpTerminalPage(
+        tester,
+        session: _session('session-1', 'codex-main'),
+        apiClient: _FakeAgentApiClient(),
+        socketFactory: TerminalSocketSessionFactory(
+          transportFactory: transportFactory.create,
+        ),
+      );
+      await tester.pump(const Duration(milliseconds: 300));
+      await tester.pumpAndSettle();
+
+      final terminal = tester
+          .widget<TerminalView>(find.byType(TerminalView))
+          .terminal;
+
+      expect(_countOccurrences(terminal.buffer.getText(), 'one\ntwo'), 1);
+    },
+  );
+
   testWidgets(
     'terminal attach replay keeps the cursor on the last restored line',
     (tester) async {
@@ -810,6 +841,48 @@ void main() {
     },
   );
 
+  testWidgets(
+    'terminal reconnect replay does not append duplicate visible output',
+    (tester) async {
+      final transportFactory = _QueuedTerminalSocketTransportFactory(
+        connectionStartupFrames: [
+          const [
+            _StartupFrame('{"type":"attached","sessionId":"session-1"}'),
+            _StartupFrame('one\r\ntwo'),
+          ],
+          const [
+            _StartupFrame('{"type":"attached","sessionId":"session-1"}'),
+            _StartupFrame('one\r\ntwo'),
+          ],
+        ],
+      );
+
+      await _pumpTerminalPage(
+        tester,
+        session: _session('session-1', 'codex-main'),
+        socketFactory: TerminalSocketSessionFactory(
+          transportFactory: transportFactory.create,
+        ),
+      );
+
+      var terminal = tester
+          .widget<TerminalView>(find.byType(TerminalView))
+          .terminal;
+      expect(_countOccurrences(terminal.buffer.getText(), 'one\ntwo'), 1);
+
+      await transportFactory.createdTransports.first.close();
+      await tester.pump();
+      await tester.pump(const Duration(seconds: 2));
+      await tester.pumpAndSettle();
+
+      terminal = tester
+          .widget<TerminalView>(find.byType(TerminalView))
+          .terminal;
+      expect(transportFactory.createCount, 2);
+      expect(_countOccurrences(terminal.buffer.getText(), 'one\ntwo'), 1);
+    },
+  );
+
   testWidgets(
     're-entering an existing session restores the terminal cursor to the last line',
     (tester) async {
@@ -1136,6 +1209,24 @@ Session _session(String sessionId, String name) {
   );
 }
 
+int _countOccurrences(String source, String pattern) {
+  if (pattern.isEmpty) {
+    return 0;
+  }
+
+  var count = 0;
+  var start = 0;
+  while (true) {
+    final index = source.indexOf(pattern, start);
+    if (index < 0) {
+      return count;
+    }
+
+    count += 1;
+    start = index + pattern.length;
+  }
+}
+
 class _FailingSessionRepository extends SessionRepository {
   _FailingSessionRepository() : super(_FakeAgentApiClient());
 
@@ -1204,24 +1295,27 @@ class _FakeTerminalSocketTransport implements TerminalSocketTransport {
   _FakeTerminalSocketTransport({
     this.autoAttach = false,
     this.startupFrames = const <String>[],
+    this.scheduledStartupFrames = const <_StartupFrame>[],
   }) {
-    if (autoAttach && startupFrames.isEmpty) {
+    final framesToEmit = scheduledStartupFrames.isNotEmpty
+        ? scheduledStartupFrames
+        : startupFrames.map(_StartupFrame.new).toList(growable: false);
+
+    if (autoAttach && framesToEmit.isEmpty) {
       Future<void>.microtask(() {
         emit('{"type":"attached","sessionId":"session-1"}');
       });
-    } else if (startupFrames.isNotEmpty) {
+    } else if (framesToEmit.isNotEmpty) {
-      Future<void>.microtask(() {
+      _scheduleFrames(framesToEmit);
-        for (final frame in startupFrames) {
-          emit(frame);
-        }
-      });
     }
   }
 
   final bool autoAttach;
   final List<String> startupFrames;
+  final List<_StartupFrame> scheduledStartupFrames;
   final _incoming = StreamController<dynamic>.broadcast();
   final sentMessages = <String>[];
+  final List<Timer> _frameTimers = <Timer>[];
 
   @override
   Stream<dynamic> get stream => _incoming.stream;
@@ -1233,30 +1327,69 @@ class _FakeTerminalSocketTransport implements TerminalSocketTransport {
 
   @override
   Future<void> close() async {
+    for (final timer in _frameTimers) {
+      timer.cancel();
+    }
+    _frameTimers.clear();
     await _incoming.close();
   }
 
   void emit(String message) {
     _incoming.add(message);
   }
+
+  void _scheduleFrames(List<_StartupFrame> frames) {
+    var cumulativeDelay = Duration.zero;
+    for (final frame in frames) {
+      cumulativeDelay += frame.delay;
+      if (cumulativeDelay == Duration.zero) {
+        Future<void>.microtask(() {
+          if (!_incoming.isClosed) {
+            emit(frame.message);
+          }
+        });
+        continue;
+      }
+
+      final timer = Timer(cumulativeDelay, () {
+        if (!_incoming.isClosed) {
+          emit(frame.message);
+        }
+      });
+      _frameTimers.add(timer);
+    }
+  }
 }
 
 class _QueuedTerminalSocketTransportFactory {
   _QueuedTerminalSocketTransportFactory({
     this.startupFrames = const <String>[],
+    this.connectionStartupFrames = const <List<_StartupFrame>>[],
   });
 
   final List<String> startupFrames;
+  final List<List<_StartupFrame>> connectionStartupFrames;
   final createdTransports = <_FakeTerminalSocketTransport>[];
   int createCount = 0;
 
   TerminalSocketTransport create(Uri _) {
+    final scheduledFrames = createCount < connectionStartupFrames.length
+        ? connectionStartupFrames[createCount]
+        : const <_StartupFrame>[];
     final transport = _FakeTerminalSocketTransport(
-      autoAttach: startupFrames.isEmpty,
+      autoAttach: startupFrames.isEmpty && scheduledFrames.isEmpty,
       startupFrames: startupFrames,
+      scheduledStartupFrames: scheduledFrames,
     );
     createdTransports.add(transport);
     createCount += 1;
     return transport;
   }
 }
+
+class _StartupFrame {
+  const _StartupFrame(this.message, {this.delay = Duration.zero});
+
+  final String message;
+  final Duration delay;
+}