오프라인 메시지 큐(N4-11) 구현: since_id 캐치업 + WS 재연결 backoff

서버가 이미 모든 메시지를 DB에 durable하게 저장하므로 별도 큐를 새로
만들지 않고, 연결이 끊겼던 클라이언트가 그 DB에서 놓친 부분만 다시
받아오는 방식으로 풀었다.

- core-backend: GET /conversations/:id/messages?since_id=<id> 추가
  (파라미터 없으면 기존 전체 히스토리 그대로, 비정상 값은 400, 최댓값
  초과는 빈 배열). TestListMessagesSinceID로 확인.
- mobile/ws_client.dart: 소켓 onDone/onError에서 그냥 멈추던 것을
  backoff(1s→2s→4s→8s→...→30s 캡, 성공 시 리셋) 재연결로 교체하고,
  재연결 성공 신호를 reconnects 스트림으로 노출.
- mobile/chat_screen.dart: WidgetsBindingObserver를 추가해 소켓
  재연결 신호 + didChangeAppLifecycleState(resumed) 양쪽에서 since_id
  캐치업을 호출, 기존 메시지와 id 기준으로 중복 없이 병합.
- mobile/services/message_sync.dart: backoff 계산과 중복 없는 병합을
  순수 함수로 뽑아 message_sync_test.dart에서 결정적으로 검증.

실제 소켓 재연결 타이밍이나 앱 백그라운드/포그라운드 전환에서 OS가
소켓을 어떻게 처리하는지는 단위 테스트로 증명할 수 없어 실기기
Android QA가 남아 있음 — roadmap.md/deploy-checklist.md N4-11에
부분 검증으로 기록.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014YSB5PqF38raTxP5ABgr9m

core-backend/a1_a2_routes.go

@@ -241,8 +241,20 @@ func registerA1A2Routes(r *gin.Engine, db *gorm.DB) {
 			c.JSON(http.StatusForbidden, gin.H{"detail": "not a participant of this conversation"})
 			return
 		}
+		query := db.Where("conversation_id = ?", convID)
+		// 오프라인 큐 캐치업(roadmap.md "멀티 디바이스 동기화" / deploy-checklist N4-11):
+		// 재연결·앱 복귀 시 클라이언트가 이미 가진 마지막 메시지 id 이후만
+		// 다시 받아 gap을 메운다. 파라미터가 없으면 기존 동작(전체 히스토리) 그대로.
+		if q := c.Query("since_id"); q != "" {
+			sinceID, err := strconv.ParseUint(q, 10, 64)
+			if err != nil {
+				c.JSON(http.StatusBadRequest, gin.H{"detail": "since_id must be a positive integer"})
+				return
+			}
+			query = query.Where("id > ?", sinceID)
+		}
 		var messages []Message
-		db.Where("conversation_id = ?", convID).Order("id asc").Find(&messages)
+		query.Order("id asc").Find(&messages)
 		out := make([]gin.H, 0, len(messages))
 		for _, m := range messages {
 			out = append(out, gin.H{

core-backend/a1_a2_test.go

@@ -160,6 +160,90 @@ func TestConversationContactMessageHistoryAndEscalationLogs(t *testing.T) {
 	}
 }
 
+// 오프라인 큐 캐치업(roadmap.md "멀티 디바이스 동기화" / deploy-checklist N4-11):
+// 재연결·앱 복귀 시 since_id로 놓친 메시지만 다시 받아올 수 있어야 한다.
+func TestListMessagesSinceID(t *testing.T) {
+	server, _ := setupTestServer(t)
+	ownerID, ownerToken := mustSignup(t, server.URL, "캐치업주인")
+	peerID, _ := mustSignup(t, server.URL, "캐치업상대")
+
+	convResp := postJSONAuth(t, server.URL+"/conversations", ownerToken, createConversationRequest{
+		UserIDs: []uint{ownerID, peerID},
+	})
+	if convResp.StatusCode != http.StatusOK {
+		t.Fatalf("create conversation: %d", convResp.StatusCode)
+	}
+	var conv map[string]interface{}
+	json.NewDecoder(convResp.Body).Decode(&conv)
+	convID := uint(conv["id"].(float64))
+	convPath := server.URL + "/conversations/" + strconv.FormatUint(uint64(convID), 10) + "/messages"
+
+	var sentIDs []uint
+	for _, text := range []string{"하나", "둘", "셋"} {
+		resp := postJSON(t, convPath, sendMessageRequest{SenderID: ownerID, Text: text})
+		if resp.StatusCode != http.StatusOK {
+			t.Fatalf("send %q: %d", text, resp.StatusCode)
+		}
+		var sent map[string]interface{}
+		json.NewDecoder(resp.Body).Decode(&sent)
+		sentIDs = append(sentIDs, uint(sent["id"].(float64)))
+	}
+
+	getMessages := func(query string) *http.Response {
+		req, _ := http.NewRequest(http.MethodGet, convPath+query, nil)
+		req.Header.Set("Authorization", "Bearer "+ownerToken)
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return resp
+	}
+
+	// No since_id -- existing behavior unchanged: full history.
+	full := getMessages("")
+	if full.StatusCode != http.StatusOK {
+		t.Fatalf("full history: %d", full.StatusCode)
+	}
+	var fullOut map[string]interface{}
+	json.NewDecoder(full.Body).Decode(&fullOut)
+	if len(fullOut["messages"].([]interface{})) != 3 {
+		t.Fatalf("expected 3 messages without since_id, got %v", fullOut)
+	}
+
+	// since_id=<first message> -- only the later two.
+	since := getMessages("?since_id=" + strconv.FormatUint(uint64(sentIDs[0]), 10))
+	if since.StatusCode != http.StatusOK {
+		t.Fatalf("since_id history: %d", since.StatusCode)
+	}
+	var sinceOut map[string]interface{}
+	json.NewDecoder(since.Body).Decode(&sinceOut)
+	sinceMsgs := sinceOut["messages"].([]interface{})
+	if len(sinceMsgs) != 2 {
+		t.Fatalf("expected 2 messages after since_id, got %v", sinceOut)
+	}
+	firstReturned := sinceMsgs[0].(map[string]interface{})
+	if uint(firstReturned["id"].(float64)) != sentIDs[1] {
+		t.Fatalf("expected first returned message to be %d, got %v", sentIDs[1], firstReturned["id"])
+	}
+
+	// since_id beyond the highest message id -- empty array, not an error.
+	beyond := getMessages("?since_id=" + strconv.FormatUint(uint64(sentIDs[2])+1000, 10))
+	if beyond.StatusCode != http.StatusOK {
+		t.Fatalf("beyond history: %d", beyond.StatusCode)
+	}
+	var beyondOut map[string]interface{}
+	json.NewDecoder(beyond.Body).Decode(&beyondOut)
+	if len(beyondOut["messages"].([]interface{})) != 0 {
+		t.Fatalf("expected 0 messages beyond highest id, got %v", beyondOut)
+	}
+
+	// Non-numeric since_id -- 400.
+	bad := getMessages("?since_id=abc")
+	if bad.StatusCode != http.StatusBadRequest {
+		t.Fatalf("expected 400 for non-numeric since_id, got %d", bad.StatusCode)
+	}
+}
+
 func TestContactScopedWhitelistMatch(t *testing.T) {
 	server, db := setupTestServer(t)
 	ownerID, ownerToken := mustSignup(t, server.URL, "화이트주인")

docs/deploy-checklist.md

@@ -233,7 +233,7 @@ API 계층은 프로덕션에서 검증됨 (`CORE_API_BASE=https://msn.iykyka.co
 
 | ID | 작업 | Status | 비고 |
 |----|------|--------|------|
-| **N4-11** | 오프라인 메시지 큐 | todo | 멀티디바이스 고도화 |
+| **N4-11** | 오프라인 메시지 큐 | **done (부분 검증)** (2026-08-03) | 서버는 이미 모든 메시지를 DB에 durable하게 저장 — 별도 큐를 새로 만들지 않고 `GET /conversations/:id/messages?since_id=`(신규, 파라미터 없으면 기존 전체 히스토리 그대로) + 클라이언트 WS 재연결(backoff 1s→2s→...→30s 캡, 성공 시 리셋, `mobile/lib/services/ws_client.dart`) + 재연결/앱 포그라운드 복귀 시 since_id 캐치업(`mobile/lib/screens/chat_screen.dart`)으로 gap을 메움. `core-backend/a1_a2_test.go`(`TestListMessagesSinceID`)와 `mobile/test/message_sync_test.dart`(backoff 계산 + 중복 없는 병합을 순수 함수로 뽑아 검증)로 확인한 부분과, **실제 소켓 재연결 타이밍·앱 백그라운드/포그라운드 전환에서 OS가 소켓을 어떻게 처리하는지는 단위 테스트로 증명 불가 — 실기기 Android QA 남음**(N4-C6b와 같은 프레이밍) |
 | **N4-12** | 자연스러움 피드백 UI | todo | vision 지표 |
 | **N4-13** | `prototype.md` `SHARE_URL` | todo | Master 기입 |
 | **N4-14** | 내부 release APK | todo | `docs/android-release.md` |

docs/roadmap.md

@@ -35,7 +35,10 @@
 **2.1 코어 백엔드** (Go, PoC 결과 무관 — 지금 착수 가능)
 - [x] 계정/인증 (초대 코드 기반 가입) — `core-backend/` (Go, Gin), 중복 코드 409 실제 테스트로 확인함
 - [x] 메시지 릴레이 서버 (송수신) — `core-backend/` WebSocket + REST, 실제 테스트로 브로드캐스트 확인함.
-  멀티 디바이스 동기화(같은 유저 여러 기기)는 아직 — 지금은 대화방 단위 인메모리 커넥션 매니저뿐
+  멀티 디바이스 동기화(같은 유저 여러 기기)는 아직 — 지금은 대화방 단위 인메모리 커넥션 매니저뿐.
+  오프라인 큐(deploy-checklist N4-11)는 DB가 이미 모든 메시지를 durable하게 들고 있으므로
+  서버에 별도 큐를 새로 만들지 않고, `GET /conversations/:id/messages?since_id=`로 재연결 시
+  gap을 메우는 방식으로 구현함(Go 테스트로 확인) — 자세한 내용은 §4 N4-11 항목 참고
 - [x] DB 스키마: users, invite_codes, contacts, conversations, messages, twin_settings,
   escalation_logs, whitelist_rules — `core-backend/models.go` (GORM), `backend/app/models.py`
   (Python 프로토타입)와 동일 스키마(+ invite_codes는 여기서 새로 추가)
@@ -342,7 +345,16 @@ Master 합의 착수 순서: **A → B → C → D(맨 마지막)**. E는 Phase
   `FCM_SERVER_KEY`와 실제 FCM registration token이 있으면 전송, 없으면 soft-skip.
   Flutter는 install-id 플레이스홀더 등록(Firebase Messaging 앱 키는 배포 환경에서 교체)
 - [~] 멀티 디바이스 동기화 — 세션 목록 + 세션 종료(`DELETE /users/:id/sessions/:id`).
-  메시지 히스토리 서버 동기화는 이미 REST/WS; 오프라인 큐는 후속
+  메시지 히스토리 서버 동기화는 이미 REST/WS. 오프라인 메시지 큐(deploy-checklist N4-11)는
+  구현 완료(부분 검증) — 서버는 이미 모든 메시지를 DB에 durable하게 저장하므로 별도
+  큐를 새로 만들지 않고, `GET /conversations/:id/messages?since_id=`(신규, 없으면 기존
+  전체 히스토리 동작 그대로) + 클라이언트 WS 재연결(backoff 1s→2s→4s→8s→...→30s 캡, 성공
+  시 리셋) + 재연결/앱 포그라운드 복귀 시 그 since_id 캐치업으로 gap을 메움.
+  `core-backend/a1_a2_test.go`(`TestListMessagesSinceID`)와
+  `mobile/test/message_sync_test.dart`(backoff 계산 + 중복 없는 병합 순수 함수)로
+  결정적으로 검증한 부분과, 실제 소켓 재연결 타이밍·앱 백그라운드/포그라운드 전환에서
+  OS가 소켓을 실제로 어떻게 처리하는지는 단위 테스트로 증명할 수 없어 실기기 Android
+  QA가 남아 있음(N4-C6b 스누즈 알림과 같은 "부분 검증" 프레이밍)
 - [x] drift + SQLCipher 로컬 저장 — `mobile/lib/db/` (말투 샘플·KV). 키는
   `flutter_secure_storage`. Linux CI는 SQLCipher SO 없으면 메모리 폴백
 - [x] 말투 이력 기기 내 저장 + 서버 최소 전송 — drift 암호화 저장, draft에 샘플만 전달

mobile/lib/screens/chat_screen.dart

@@ -5,6 +5,7 @@ import 'package:provider/provider.dart';
 
 import '../models/models.dart';
 import '../services/api_client.dart';
+import '../services/message_sync.dart';
 import '../services/snooze_service.dart';
 import '../services/ws_client.dart';
 import '../state/session_state.dart';
@@ -32,13 +33,19 @@ class ChatScreen extends StatefulWidget {
   State<ChatScreen> createState() => _ChatScreenState();
 }
 
-class _ChatScreenState extends State<ChatScreen> {
+class _ChatScreenState extends State<ChatScreen> with WidgetsBindingObserver {
   final _input = TextEditingController();
   final _draftEdit = TextEditingController();
   final _scroll = ScrollController();
   final _messages = <ChatMessage>[];
   ConversationSocket? _socket;
   StreamSubscription? _sub;
+  // 오프라인 큐 캐치업(roadmap.md "멀티 디바이스 동기화" / deploy-checklist N4-11):
+  // 소켓 재연결 신호를 듣기 위한 별도 구독. 실제 소켓 끊김/재연결 타이밍은
+  // 단위 테스트로 검증할 수 없어 — 신호가 왔을 때 REST since_id 캐치업을
+  // 호출한다는 것만 여기서 보장하고, 나머지(진짜 네트워크 단절, 백그라운드
+  // 전환에서 OS가 소켓을 실제로 끊는지)는 실기기 QA 대상이다.
+  StreamSubscription? _reconnectSub;
   String? _banner;
   DraftResult? _pendingDraft;
   bool _busy = false;
@@ -52,6 +59,7 @@ class _ChatScreenState extends State<ChatScreen> {
   @override
   void initState() {
     super.initState();
+    WidgetsBinding.instance.addObserver(this);
     _api = context.read<SessionState>().api;
     _floodBlocked = widget.twinDisabledByFlood;
     if (_floodBlocked) {
@@ -59,10 +67,45 @@ class _ChatScreenState extends State<ChatScreen> {
     }
     _socket = ConversationSocket(widget.conversationId)..connect();
     _sub = _socket!.events.listen(_onEvent);
+    // 소켓이 끊겼다가 다시 붙으면(백그라운드, 네트워크 hiccup 등) 그 사이에
+    // 놓친 메시지를 REST since_id로 다시 받아온다 — ws_client.dart 참고.
+    _reconnectSub = _socket!.reconnects.listen((_) => _catchUp());
     _loadHistory();
     _loadSnooze();
   }
 
+  @override
+  void didChangeAppLifecycleState(AppLifecycleState state) {
+    // 앱이 백그라운드에 있는 동안 소켓이 조용히 죽어 있었을 수 있다 — 포그라운드로
+    // 돌아올 때마다 캐치업을 한 번 더 시도한다(소켓 자체의 재연결 신호와는 별개의
+    // 안전망). 실기기에서 OS가 실제로 어떻게 동작하는지는 QA 대상.
+    if (state == AppLifecycleState.resumed) {
+      _catchUp();
+    }
+  }
+
+  /// 오프라인 큐 캐치업 본체: 지금까지 로드된 메시지 중 가장 큰 id 이후를
+  /// REST로 다시 받아, 소켓/REST가 겹쳐 온 것이 있어도 중복 없이 합친다.
+  Future<void> _catchUp() async {
+    if (!mounted || _loadingHistory) return;
+    final sinceId = highestMessageId(_messages);
+    try {
+      final fresh = await _api.listMessages(widget.conversationId, sinceId: sinceId);
+      if (!mounted || fresh.isEmpty) return;
+      final merged = mergeNewMessages(_messages, fresh);
+      setState(() {
+        _messages
+          ..clear()
+          ..addAll(merged);
+      });
+      _scrollToEnd();
+      _markLatestRead();
+    } on ApiException {
+      // Best-effort — 실시간 소켓이 살아 있으면 다음 이벤트로 계속 채워지고,
+      // 다음 재연결/앱 복귀 때 다시 시도되므로 여기서 에러를 굳이 보여주지 않는다.
+    }
+  }
+
   Future<void> _loadSnooze() async {
     final controller = context.read<SessionState>().snoozeController;
     if (controller == null) return;
@@ -223,7 +266,9 @@ class _ChatScreenState extends State<ChatScreen> {
     // Fire-and-forget: dispose는 동기라 기다릴 수 없고, 실패해도 안 읽음
     // 배지가 조금 늦게 갱신되는 정도라 굳이 에러를 보여줄 필요 없음.
     _markLatestRead();
+    WidgetsBinding.instance.removeObserver(this);
     _sub?.cancel();
+    _reconnectSub?.cancel();
     _socket?.dispose();
     _input.dispose();
     _draftEdit.dispose();

mobile/lib/services/api_client.dart

@@ -101,8 +101,15 @@ class ApiClient {
     );
   }
 
-  Future<List<ChatMessage>> listMessages(int conversationId) async {
+  /// [sinceId]가 있으면 그 id보다 큰 메시지만 받아온다 — 오프라인 큐
-    final obj = await _getObject('/conversations/$conversationId/messages');
+  /// 캐치업(roadmap.md "멀티 디바이스 동기화" / deploy-checklist N4-11)에서
+  /// 재연결·앱 복귀 시 이미 로드된 마지막 메시지 이후만 다시 받아 gap을 메우는 용도.
+  /// 생략하면 기존 동작(전체 히스토리) 그대로.
+  Future<List<ChatMessage>> listMessages(int conversationId, {int? sinceId}) async {
+    final path = sinceId != null
+        ? '/conversations/$conversationId/messages?since_id=$sinceId'
+        : '/conversations/$conversationId/messages';
+    final obj = await _getObject(path);
     final list = (obj['messages'] as List<dynamic>? ?? const []);
     return list.map((e) => ChatMessage.fromJson(e as Map<String, dynamic>)).toList();
   }

mobile/lib/services/message_sync.dart

@@ -0,0 +1,55 @@
+/// 오프라인 큐 캐치업(roadmap.md "멀티 디바이스 동기화" / deploy-checklist N4-11)
+/// 순수 로직. `DateTime.now()`/타이머에 의존하지 않고 값을 인자로 받는 함수들이라
+/// 테스트에서 결정적으로 검증할 수 있다.
+///
+/// 배경: 서버(core-backend)는 모든 메시지를 DB에 그대로 durable하게 저장한다 —
+/// 클라이언트가 끊겨 있는 동안 온 메시지도 서버에는 이미 있다. 그래서 여기서
+/// "큐"를 새로 만드는 게 아니라, 소켓이 재연결되거나 앱이 포그라운드로 돌아올 때
+/// 마지막으로 본 메시지 id 이후를 `GET .../messages?since_id=` 로 다시 받아
+/// 그 gap을 메우는 전략을 쓴다. 이 파일은 그 재연결 backoff 계산과, REST로
+/// 받아온 새 메시지를 이미 있는 목록에 중복 없이 합치는 부분만 담는다.
+library;
+
+import '../models/models.dart';
+
+/// 재연결 시도 backoff의 초기 지연.
+const Duration kReconnectInitialDelay = Duration(seconds: 1);
+
+/// 재연결 시도 backoff의 상한 — 이보다 더 늘어나지 않는다.
+const Duration kReconnectMaxDelay = Duration(seconds: 30);
+
+/// 몇 번째 재연결 시도([attempt], 0-based: 첫 시도가 0)인지에 따른 다음 지연을
+/// 계산한다. 1s → 2s → 4s → 8s → 16s → 30s(상한)로 두 배씩 늘고, 상한을
+/// 넘지 않는다. 성공적으로 재연결되면 호출부에서 attempt를 0으로 되돌린다.
+/// [attempt]가 아무리 커져도(재연결이 오래 실패해도) 오버플로 없이 상한에서
+/// 멈추도록 시프트 전에 attempt 자체를 안전한 범위로 클램프한다.
+Duration nextReconnectDelay(int attempt) {
+  if (attempt <= 0) return kReconnectInitialDelay;
+  const maxUsefulAttempt = 10; // 1s * 2^10 = 1024s, 상한(30s)을 한참 넘어 충분.
+  final clamped = attempt > maxUsefulAttempt ? maxUsefulAttempt : attempt;
+  final ms = kReconnectInitialDelay.inMilliseconds * (1 << clamped);
+  final capped = ms > kReconnectMaxDelay.inMilliseconds ? kReconnectMaxDelay.inMilliseconds : ms;
+  return Duration(milliseconds: capped);
+}
+
+/// 이미 로드되어 있는 메시지 목록의 최댓값 id (없으면 null) — 캐치업 요청에
+/// `since_id`로 넘길 값을 계산할 때 쓴다.
+int? highestMessageId(List<ChatMessage> messages) {
+  if (messages.isEmpty) return null;
+  return messages.map((m) => m.id).reduce((a, b) => a > b ? a : b);
+}
+
+/// [existing] 뒤에 [fresh]에서 아직 없는(id 기준) 메시지만 id 순서로 이어붙인다.
+/// 소켓으로 실시간 수신한 메시지와 REST 캐치업이 같은 메시지를 중복으로
+/// 가져오는 경우(재연결 race)를 여기서 걸러낸다. [existing]의 기존 순서는
+/// 그대로 보존한다.
+List<ChatMessage> mergeNewMessages(
+  List<ChatMessage> existing,
+  List<ChatMessage> fresh,
+) {
+  final knownIds = existing.map((m) => m.id).toSet();
+  final toAdd = fresh.where((m) => !knownIds.contains(m.id)).toList()
+    ..sort((a, b) => a.id.compareTo(b.id));
+  if (toAdd.isEmpty) return List.of(existing);
+  return [...existing, ...toAdd];
+}

mobile/lib/services/ws_client.dart

@@ -5,31 +5,78 @@ import 'package:web_socket_channel/web_socket_channel.dart';
 
 import '../config.dart';
 import '../models/models.dart';
+import 'message_sync.dart';
 
 /// Conversation-scoped WebSocket relay (`GET /ws/conversations/:id`).
+///
+/// 오프라인 큐 캐치업(roadmap.md "멀티 디바이스 동기화" / deploy-checklist N4-11):
+/// 소켓이 끊기면(백그라운드, 네트워크 hiccup 등) `onDone`/`onError`에서 그냥
+/// 멈추는 게 아니라 [nextReconnectDelay]로 계산한 backoff 후 재연결을 시도한다.
+/// 재연결에 성공하면 [reconnects] 스트림으로 신호를 보내 — 화면(chat_screen.dart)이
+/// 그 신호를 받아 REST `since_id` 캐치업을 트리거할 수 있게 한다. 실제 소켓
+/// 타이밍(진짜 네트워크 단절 등)은 단위 테스트로 결정적으로 검증할 수 없어
+/// 실기기 QA가 필요하다 — [nextReconnectDelay] 자체의 계산 로직만 순수
+/// 함수로 분리해 테스트했다(mobile/test/message_sync_test.dart).
 class ConversationSocket {
   ConversationSocket(this.conversationId);
 
   final int conversationId;
   WebSocketChannel? _channel;
+  StreamSubscription? _channelSub;
   final _controller = StreamController<Map<String, dynamic>>.broadcast();
+  final _reconnectController = StreamController<void>.broadcast();
+  Timer? _reconnectTimer;
+  int _reconnectAttempt = 0;
+  bool _disposed = false;
+  bool _everConnected = false;
 
   Stream<Map<String, dynamic>> get events => _controller.stream;
 
+  /// 재연결에 성공할 때마다 한 번씩 이벤트가 발생한다 (최초 연결은 포함하지
+  /// 않음 — chat_screen.dart는 이미 initState에서 `_loadHistory()`로 전체
+  /// 히스토리를 로드하므로, 이 신호는 "끊겼다가 다시 붙었을 때"만 의미가 있다).
+  Stream<void> get reconnects => _reconnectController.stream;
+
   void connect() {
+    final wasReconnect = _everConnected;
+    // 이전 연결이 onError로 끊겼을 때는 cancelOnError:false라 구독이 아직
+    // 살아 있을 수 있다 — 새 채널을 만들기 전에 정리해 leak을 막는다.
+    _channelSub?.cancel();
     final uri = Uri.parse('${AppConfig.wsBase()}/ws/conversations/$conversationId');
     _channel = WebSocketChannel.connect(uri);
-    _channel!.stream.listen(
+    _everConnected = true;
+    _channelSub = _channel!.stream.listen(
       (raw) {
+        // 데이터가 왔다는 건 연결이 살아 있다는 뜻 — backoff를 리셋한다.
+        _reconnectAttempt = 0;
         if (raw is! String) return;
         final decoded = jsonDecode(raw);
         if (decoded is Map<String, dynamic>) {
           _controller.add(decoded);
         }
       },
-      onError: _controller.addError,
+      onError: (Object e, StackTrace st) {
-      onDone: () {},
+        _controller.addError(e, st);
+        _scheduleReconnect();
+      },
+      onDone: _scheduleReconnect,
+      cancelOnError: false,
     );
+    if (wasReconnect) {
+      _reconnectAttempt = 0;
+      if (!_reconnectController.isClosed) _reconnectController.add(null);
+    }
+  }
+
+  void _scheduleReconnect() {
+    if (_disposed) return;
+    _reconnectTimer?.cancel();
+    final delay = nextReconnectDelay(_reconnectAttempt);
+    _reconnectAttempt++;
+    _reconnectTimer = Timer(delay, () {
+      if (_disposed) return;
+      connect();
+    });
   }
 
   ChatMessage? parseMessageEvent(Map<String, dynamic> event) {
@@ -47,7 +94,11 @@ class ConversationSocket {
   }
 
   Future<void> dispose() async {
+    _disposed = true;
+    _reconnectTimer?.cancel();
+    await _channelSub?.cancel();
     await _channel?.sink.close();
     await _controller.close();
+    await _reconnectController.close();
   }
 }

mobile/test/message_sync_test.dart

@@ -0,0 +1,85 @@
+import 'package:flutter_test/flutter_test.dart';
+import 'package:ykavu_mobile/models/models.dart';
+import 'package:ykavu_mobile/services/message_sync.dart';
+
+ChatMessage _msg(int id, {bool retracted = false}) => ChatMessage(
+      id: id,
+      conversationId: 1,
+      senderId: 1,
+      senderMode: SenderMode.human,
+      text: 'msg-$id',
+      retracted: retracted,
+      createdAt: DateTime(2026, 8, 3),
+    );
+
+void main() {
+  group('nextReconnectDelay', () {
+    test('first attempt (0) is the initial 1s delay', () {
+      expect(nextReconnectDelay(0), const Duration(seconds: 1));
+    });
+
+    test('doubles each attempt: 1s, 2s, 4s, 8s, 16s', () {
+      expect(nextReconnectDelay(1), const Duration(seconds: 2));
+      expect(nextReconnectDelay(2), const Duration(seconds: 4));
+      expect(nextReconnectDelay(3), const Duration(seconds: 8));
+      expect(nextReconnectDelay(4), const Duration(seconds: 16));
+    });
+
+    test('caps at 30s once doubling would exceed it', () {
+      expect(nextReconnectDelay(5), const Duration(seconds: 30));
+      expect(nextReconnectDelay(6), const Duration(seconds: 30));
+    });
+
+    test('stays capped even for a very large attempt count (no overflow)', () {
+      expect(nextReconnectDelay(1000000), const Duration(seconds: 30));
+    });
+
+    test('negative attempt is treated like the first attempt', () {
+      expect(nextReconnectDelay(-1), const Duration(seconds: 1));
+    });
+  });
+
+  group('highestMessageId', () {
+    test('empty list -> null', () {
+      expect(highestMessageId(const []), isNull);
+    });
+
+    test('returns the max id regardless of list order', () {
+      expect(highestMessageId([_msg(3), _msg(1), _msg(5), _msg(2)]), 5);
+    });
+  });
+
+  group('mergeNewMessages', () {
+    test('appends fresh messages not already present, sorted by id', () {
+      final existing = [_msg(1), _msg(2)];
+      final fresh = [_msg(4), _msg(3)];
+      final merged = mergeNewMessages(existing, fresh);
+      expect(merged.map((m) => m.id).toList(), [1, 2, 3, 4]);
+    });
+
+    test('drops fresh messages whose id is already loaded (WS/REST race)', () {
+      final existing = [_msg(1), _msg(2), _msg(3)];
+      final fresh = [_msg(2), _msg(3), _msg(4)];
+      final merged = mergeNewMessages(existing, fresh);
+      expect(merged.map((m) => m.id).toList(), [1, 2, 3, 4]);
+    });
+
+    test('fresh entirely a subset of existing -> unchanged, same order', () {
+      final existing = [_msg(1), _msg(2)];
+      final merged = mergeNewMessages(existing, [_msg(1), _msg(2)]);
+      expect(merged.map((m) => m.id).toList(), [1, 2]);
+    });
+
+    test('existing empty -> result is just fresh, sorted', () {
+      final merged = mergeNewMessages(const [], [_msg(2), _msg(1)]);
+      expect(merged.map((m) => m.id).toList(), [1, 2]);
+    });
+
+    test('does not mutate the existing list instance', () {
+      final existing = [_msg(1)];
+      final merged = mergeNewMessages(existing, [_msg(2)]);
+      expect(existing.length, 1);
+      expect(merged.length, 2);
+    });
+  });
+}