blame | history | raw | HEAD

PAILot Message System Rewrite Spec

Problem

The current message handling has accumulated race conditions from incremental fixes: - switchSession is async — messages arriving during the async gap get overwritten by loadAll - iOS kills the MQTT socket in background but the client reports "connected" - File corruption from concurrent read/write on the same session file - Notification tap triggers switchSession which reloads from disk, losing in-memory messages - addMessage and switchSession compete on the same state and disk files

Architecture: Single Message Bus

1. One MQTT topic for everything

2. MessageStore redesign

Replace per-session debounced saves with a single append-only log:

``` ~/.../messages/log.jsonl — append-only, one JSON line per message ~/.../messages/index.json — { sessionId: [lineNumbers] } for fast lookup ```

Operations: - append(message) — append one line to log.jsonl (sync, atomic, no race) - loadSession(sessionId) — read index, seek to lines, return messages - compact() — rewrite log removing old messages (run on app start, not during use)

Benefits: - No per-session files — no file-level races - Append-only — no read-modify-write cycle - No debounce needed — each append is a single writeAsStringSync with mode: FileMode.append

3. Connection state machine

``` States: disconnected → connecting → connected → suspended → reconnecting → connected ↑ | └──────────────────────────┘

Transitions: - App launch: disconnected → connecting → connected - App background: connected → suspended (keep client, mark state) - App resume: suspended → reconnecting → connected (always force-reconnect) - Connection lost: connected → reconnecting → connected (autoReconnect) - User disconnect: any → disconnected ```

Key rule: In suspended state, do NOT process buffered MQTT messages. Process them only after reconnect + catch_up completes. This prevents the race where a buffered message is added to state before loadAll overwrites it.

4. Message routing (synchronous, no async gaps)

```dart void _onMessage(Map<String, dynamic> json) { final type = json['type'] as String?; final sessionId = json['sessionId'] as String?; final currentId = _currentSessionId;

if (type == 'text' || type == 'voice' || type == 'image') {
    // Append to log immediately (sync)
    MessageStore.append(Message.fromMqtt(json));

    // Display only if for current session
    if (sessionId == currentId) {
        _messages.add(Message.fromMqtt(json));
        notifyListeners(); // or setState
    } else {
        _incrementUnread(sessionId);
    }
}

} ```

No async. No switchSession during message handling. No race.

5. Session switching

```dart void switchSession(String sessionId) { _currentSessionId = sessionId; _messages = MessageStore.loadSession(sessionId); // sync read from index notifyListeners(); } ```

Synchronous. No async gap. No race with incoming messages.

6. Resume flow

```dart void onResume() { state = suspended; // Kill old client (disable autoReconnect first) _client?.autoReconnect = false; _client?.disconnect(); _client = null;

// Fast reconnect to last host
await _fastReconnect(connectedHost);

// Now process: sync → sessions → catch_up
// catch_up messages go through same _onMessage path (append + display if current)
state = connected;

} ```

7. Notification tap

```dart void onNotificationTap(String sessionId) { if (sessionId != _currentSessionId) { switchSession(sessionId); // sync, no async } // Message is already in the log from MQTT delivery or catch_up // switchSession loads it } ```

Migration path

  1. Create MessageStoreV2 with append-only log
  2. Create ConnectionStateMachine with explicit states
  3. Rewrite _handleIncomingMessage to use sync append
  4. Rewrite switchSession to be sync
  5. Remove debounced saves, per-session file locks, merge protection
  6. Test each step before moving to next

What stays the same