Vitto Chat Persistence & Response Recovery

Session: S228 Date: 2026-03-16 Status: Design approved, implementation pending Predecessor: S226 (Sessions Security Unification) — built auth infra that enables this work

Problem Statement

When a user sends a message to Vitto and the frontend disconnects (page refresh, navigation, network hiccup, browser close), Vitto's response stalls permanently. The user sees a frozen "thinking" state or partial response that never completes. The backend continues processing and persists the final result to agent-sessions (vitto-mind DB), but this result never reaches chat_sessions (ai-agents DB) — which is what the frontend loads on page init.

Root cause (verified): The response pipeline is frontend-dependent. The final assistant message only reaches ai-agents DB when the frontend sets isStreaming: false and ChatBackendSyncService fires POST /sessions/:id/messages. If the frontend never reconnects, the response exists only in vitto-mind DB, invisible to the user.

Compounding bug (discovered during analysis): On page reload, reconcileActiveSession() finds the stale isStreaming: true message in IndexedDB, notes it's missing from MongoDB, and pushes the broken partial message to MongoDB — actively corrupting the backend data.

Architecture Context

Current Data Flow (Broken)

User sends message
  POST /engine/message (vitto-api:3003) → { taskId }

  Backend (detached Promise):
    agentLoop.run() → LLM calls + tools
    Events → RxJS Subject<SSEEvent> (in-memory, NOT persisted)
    On completion:
      agentSessions.addMessage() → MongoDB (vitto-mind DB, agent-sessions collection)
      streamSubject.next({ type: 'task.completed', ... })
      setTimeout(30s) → streamSubject.complete() → delete from Map

  Frontend:
    EventSource → /engine/message/:taskId/stream
    Creates message id: "stream-${taskId}", isStreaming: true
    Writes to autoFlashState → IndexedDB (debounced 300ms)
    ChatBackendSyncService FILTERS OUT isStreaming messages (line 156)
    On task.completed: isStreaming → false → syncs to MongoDB (ai-agents DB)

  THE GAP: If frontend disconnects before task.completed is received,
  ai-agents DB never gets the response. vitto-mind DB has it, ai-agents doesn't.
  loadBackendSessions() reads from ai-agents → response is lost to the user.

Key Subsystems

Deployment Model

• backend-vitto-api: Single Railway instance (no horizontal scaling configured)
• Redis available via RedisService (global injectable) and BullMQ TaskQueueService
• No enableShutdownHooks() — deploys hard-kill in-flight tasks
• In-memory Maps (conversationalStreams, taskAbortControllers, taskContexts) are appropriate given single-instance deployment

Multi-Device & Multi-Channel Architecture

Sessions are keyed by userId + spaceId (compound index), NOT by channel or device. A user's web chat and WhatsApp conversations converge into the same chat_sessions document when the phone number is linked to a GoGrab account (via ChannelUserService.resolveUser()).

Sync mechanisms by scope:

Socket.IO broadcast topology (from ChatGateway.emitNewMessages()):

1. chat:{sessionId} room → chat:messages event (full message payload to all session viewers)
2. user:{senderId} room → chat:session-updated event (badge/notification to sender's other devices)
3. For group sessions: user:{participantId} room → chat:messages for each non-sender participant

How the bridge (Change 1) enables multi-device recovery: When bridgeMessageToAiApi() calls POST /sessions/:id/messages on backend-ai-api, appendMessages() unconditionally calls chatGateway.emitNewMessages(). This means:

• Device A sends message, disconnects. Backend completes task, bridge fires.
• Device B (connected to Socket.IO) receives the response in real-time via chat:messages.
• Device C (offline) gets the response from loadBackendSessions() on next page load.
• No additional work needed — the bridge naturally provides multi-device delivery.

Channel-specific outbound paths (independent of web SSE):

Group chat (Whistle): Sessions upgrade from single-user to group via addParticipant(). The participants[] field on ChatSession carries userId, role, historyAccess, and channel per member. emitNewMessages() broadcasts to all participants' user rooms, so group members on any device see new messages. Recovery works identically — the bridge writes to the shared session, all participants receive it.

Industry Validation

Production AI chat systems (ChatGPT, Claude.ai, Gemini) all follow the same pattern:

• Backend persists completed messages
• In-flight streaming tokens are ephemeral — lost on disconnect
• No system uses SSE Last-Event-ID for mid-stream token replay
• Recovery = reload completed conversation from backend

Design: 6 Changes, Ordered by Impact

Change 1: Bridge Completed Responses to ai-agents DB

What: After conversational task completion, call the existing bridgeMessageToAiApi() to push the assistant response to ai-agents DB unconditionally.

Where: vitto-engine.service.ts, post-loop block (~line 1017), after agentSessions.addMessage().

How:

// After existing agentSessions.addMessage() calls
this.backendAiClient.bridgeMessageToAiApi({
  userId,
  spaceId,
  channel: 'web',
  messages: [{ role: 'assistant', content: result.output }],
}).catch(err => this.logger.warn('Bridge to ai-api failed (non-critical)', err.message));

Note: bridgeMessageToAiApi() takes a single params object (not positional args). It generates synthetic message IDs internally (${channel}-${now}-${i}-${random}) and converts timestamps to Unix ms. The method already handles session ID resolution via GET /sessions/by-space?userId=...&spaceId=... with deterministic fallback channel-${userId}-${spaceKey}.

Why unconditional: Checking whether the SSE has subscribers adds complexity for negligible savings. The bridge is fire-and-forget.

Deduplication strategy: The frontend will also sync the same message when it reconnects (normal happy path). appendMessages() on backend-ai-api does NOT deduplicate — it does a raw messages.push(). To prevent double messages:

• The bridge call uses IDs with bridge- prefix (generated by bridgeMessageToAiApi)
• The frontend sync uses IDs with vitto- prefix (set during finalization)
• On the frontend, recoverStaleStreams() (Change 5) should check if the bridge already wrote the message before syncing. When recovery detects status: 'completed', it sets metadata.bridged: true on the recovered message.
• In ChatBackendSyncService.appendMessages(), skip messages with metadata?.bridged === true — the bridge already handled backend persistence.

Impact: Fixes the core bug for the "close browser entirely" case. Even if the frontend never reconnects, loadBackendSessions() will find the response in ai-agents DB.

Effort: ~30 minutes.

Change 2: Subject → ReplaySubject + Extended Window

What: Replace new Subject<SSEEvent>() with new ReplaySubject<SSEEvent>(100, 5 * 60 * 1000) for conversational streams. Extend teardown from 30s to 5 minutes.

Where: vitto-engine.service.ts, startConversationalMessage() (~line 870) and the finally block (~line 1050).

How:

// In startConversationalMessage():
const streamSubject = new ReplaySubject<SSEEvent>(100, 5 * 60 * 1000);

// In finally block:
const STREAM_TTL_MS = 5 * 60 * 1000; // 5 minutes, matches CostGuard max
setTimeout(() => {
  streamSubject.complete();
  this.conversationalStreams.delete(taskId);
}, STREAM_TTL_MS);

Why: A plain Subject loses events when no subscribers are connected. The current 30s window is too short — if a user refreshes and the page takes >30s to load + hydrate + open EventSource, the stream is gone. A ReplaySubject with 100-event buffer and 5-minute window means:

• Reconnecting clients get full event replay (including task.completed)
• Memory impact: ~100KB per active task × concurrent users = negligible
• CostGuard limits tasks to 5 minutes, so the window aligns perfectly

Impact: Fixes "brief disconnect" and "refresh within 5 minutes" cases without any new endpoint.

Effort: ~15 minutes.

Change 3: Fix Reconciliation Corruption

What: Prevent reconcileActiveSession() from pushing stale isStreaming: true messages to MongoDB.

Where: vitto-chat-persistence.service.ts, reconcileActiveSession() (~line 838).

How: Before sending localMessageIds to the reconcile endpoint, filter out messages with isStreaming: true:

const localMessageIds = session.messages
  .filter(m => !m.isStreaming)  // Don't reconcile broken streaming messages
  .map(m => m.id);

Why: Currently, stale streaming messages get pushed to MongoDB via reconciliation, actively corrupting the backend. This is the fastest fix for an ongoing data quality issue.

One-time cleanup: Run against existing corrupted data (only messages older than 1 hour to avoid removing actively-streaming messages during deployment):

const oneHourAgo = Date.now() - (60 * 60 * 1000);
db.chat_sessions.updateMany(
  { 'messages': { $elemMatch: { isStreaming: true, timestamp: { $lt: oneHourAgo } } } },
  { $pull: { messages: { isStreaming: true, timestamp: { $lt: oneHourAgo } } } }
)

Impact: Stops active data corruption. Prerequisite for clean recovery in Changes 4-5.

Effort: ~30 minutes.

Change 4: Task Status Endpoint

What: New endpoint GET /engine/message/:taskId/status on backend-vitto-api that reports whether a conversational task completed and returns the final response.

Where: vitto.controller.ts, new method near the existing POST /engine/message block (~line 3470).

Schema:

// Response
interface TaskStatusResponse {
  status: 'running' | 'completed' | 'failed' | 'unknown';
  response?: string;       // Final assistant content (only if completed)
  completedAt?: number;    // Unix ms
}

New in-memory tracking Map: The status endpoint cannot rely on taskAbortControllers presence because the finally block deletes it immediately on task completion (before the 5-minute ReplaySubject window). Instead, introduce a dedicated completedTaskResults Map:

// New Map alongside conversationalStreams
private readonly completedTaskResults = new Map<string, {
  status: 'completed' | 'failed';
  response?: string;
  completedAt: number;
  userId: string;
}>();

This Map is populated in the post-loop block (on success) and catch block (on failure), before the finally block runs. It is cleaned up by the same setTimeout that cleans conversationalStreams.

Resolution logic (in order):

1. Check completedTaskResults.has(taskId) → return stored status + response (most common recovery path)
2. Check conversationalStreams.has(taskId) AND NOT completedTaskResults.has(taskId) → running (task still in-flight)
3. Neither in memory → query agent-sessions for messages with metadata.taskId === taskId. If found → completed with content. Otherwise → unknown.

Adding taskId to AgentSessionMessage: Extend the addMessage() call to include taskId in a metadata field:

this.agentSessions.addMessage(userId, {
  role: 'assistant',
  content: result.output,
  timestamp: new Date(),
  channel: 'web',
  metadata: { taskId }  // New field
}, spaceId);

This requires adding metadata?: Record<string, unknown> to SessionMessage interface and AgentSessionMessageSchema. Minor schema change. Enables persistent recovery after the 5-minute in-memory window expires.

Auth: Protected by the existing UnifiedAuthGuard. The endpoint verifies that the requesting user matches the task's userId from completedTaskResults (for in-memory hits) or from the agent-sessions query (for DB fallback). Note: taskContexts is deleted in finally and cannot be used for auth.

Impact: Enables frontend recovery (Change 5). Also useful for debugging and monitoring.

Effort: ~2 hours.

Change 5: Frontend Stale Stream Recovery

What: On page load, detect isStreaming: true messages in IndexedDB, query the task status endpoint, and recover or clean up.

Where: vitto-chat-engine.service.ts, new recoverStaleStreams() method called from the init effect() after hydration.

Init ordering (critical): recoverStaleStreams() must run BEFORE reconcileActiveSession(). The init sequence in the auth effect() becomes:

1. loadBackendSessions() (active session + history merge)
2. recoverStaleStreams() ← NEW: recover/cleanup stale messages
3. reconcileActiveSession() (now safe — no stale isStreaming messages remain)

Flow:

On hydration complete + auth resolved (after loadBackendSessions):
  messages = persistence.getMessagesWithStreaming(true)
  await Promise.allSettled(messages.map(staleMsg => recover(staleMsg)))

  For each staleMsg:
    taskId = staleMsg.id.replace('stream-', '')
    response = GET /engine/message/${taskId}/status

    switch (response.status):
      case 'completed':
        persistence.updateMessage(staleMsg.id, {
          id: `vitto-${taskId}`,
          content: response.response,
          isStreaming: false,
          metadata: { recovered: true }  // Flag to suppress emotion inference
        })

      case 'running':
        // ReplaySubject handles replay — just reopen EventSource
        reopenEventSource(taskId)

      case 'failed':
      case 'unknown':
        persistence.updateMessage(staleMsg.id, {
          isStreaming: false,
          content: staleMsg.content + '\n\n[Response interrupted — please try again]',
          metadata: { recovered: true, error: true }
        })

Emotion suppression: When metadata.recovered === true, the chat-shell effect() should skip inferAndCapture(). Add a guard:

// In chat-shell.component.ts emotion effect (~line 309)
if (!msg || msg.isStreaming || msg.metadata?.recovered) return;

Platform considerations:

• Chrome extension: Recovery runs on every popup open (component init). Works correctly.
• Studio dock: Recovery runs after loadProject() resolves. Works correctly.
• Full-page chat: Recovery runs on every page load. Works correctly.
• Multi-tab: BroadcastChannel propagates the recovery update to other tabs.

Impact: Completes the user experience. Users see their response after any kind of disconnection.

Effort: ~3 hours.

Change 6: Emotion Suppression During Recovery

Merged into Change 5 — the metadata.recovered flag and the emotion guard (if (!msg || msg.isStreaming || msg.metadata?.recovered) return;) are part of the same logical unit. Listed separately for clarity but implemented together.

Data Flow After Fix

User sends message
  POST /engine/message (vitto-api:3003) → { taskId }

  Backend (detached Promise):
    agentLoop.run() → LLM calls + tools
    Events → ReplaySubject<SSEEvent>(100, 5min) [CHANGE 2]
    On completion:
      agentSessions.addMessage(metadata: { taskId }) → vitto-mind DB [CHANGE 4]
      bridgeMessageToAiApi() → ai-agents DB [CHANGE 1]
      streamSubject.next({ type: 'task.completed' })
      setTimeout(5min) → complete + delete [CHANGE 2]

  Frontend (normal flow — no disconnect):
    EventSource receives task.completed → isStreaming: false → sync to ai-agents DB
    (Bridge write is duplicate but harmless)

  Frontend (page refresh within 5min):
    Hydrates from IndexedDB → finds isStreaming: true message
    recoverStaleStreams() → GET /engine/message/:taskId/status [CHANGE 5]
    If completed: update message, suppress emotion [CHANGE 6]
    If running: reopen EventSource → ReplaySubject replays all events [CHANGE 2]

  Frontend (page refresh after 5min or browser closed):
    Hydrates from IndexedDB → finds isStreaming: true message
    recoverStaleStreams() → GET /engine/message/:taskId/status → 'completed'
    Response found in agent-sessions (via taskId metadata) [CHANGE 4]
    Update message with recovered content

  Frontend (page refresh, task truly lost — backend restart):
    recoverStaleStreams() → GET /engine/message/:taskId/status → 'unknown'
    Mark message as error with "[Response interrupted]"

  Reconciliation:
    Stale isStreaming messages excluded from reconcile payload [CHANGE 3]
    No broken data pushed to MongoDB

What This Design Does NOT Do (Intentional)

1. No full SSE event persistence — Production AI systems don't do this. In-flight tokens are ephemeral.
2. No Last-Event-ID support — Only works within a page session. Browser EventSource auto-reconnect already handles transient blips.
3. No Redis-based stream buffering — In-memory ReplaySubject is sufficient for single-instance deployment. Redis adds complexity without benefit at current scale.
4. No horizontal scaling support — backend-vitto-api is single-instance. When scaling is needed, migrate conversationalStreams to Redis pub/sub. Not needed now.
5. No graceful shutdown handling — Pre-existing gap. Would need enableShutdownHooks() + drain logic. Out of scope.

Edge Cases

Success Criteria

1. User refreshes page mid-response → response appears when page reloads (within 5min: via ReplaySubject; after 5min: via status endpoint + agent-sessions)
2. User closes browser entirely → response is waiting on next visit (via bridge to ai-agents DB)
3. Network hiccup drops SSE → browser auto-reconnect + ReplaySubject handles replay
4. Backend restart during response → user sees "[Response interrupted]" error, not infinite loading
5. No stale isStreaming: true messages in MongoDB (reconciliation fix)
6. No unexpected avatar animations during recovery (emotion suppression)
7. All platforms (web, extension, studio) recover consistently
8. Multi-device: user sends on Device A, response appears on Device B (via Socket.IO or backend load)
9. Multi-channel: WhatsApp responses continue to bridge to web chat (no regression on existing bridgeMessageToAiApi paths)
10. Group sessions (Whistle): all participants see the recovered response, not just the sender

Files Modified

Supersedes

This spec supersedes the proposed solution in docs/english/sessions/S226/S226-handoff-vitto-chat-persistence.md. The handoff's root cause analysis is accurate and was used as the foundation for this design. The solution direction diverges: this spec opts for a simpler, production-validated approach (backend bridge + ReplaySubject + task status polling) over full event persistence and Last-Event-ID support.