2026 03 16 Vitto Chat Persistence Recovery

Documentation for Plans

Vitto Chat Persistence & Response Recovery — Implementation Plan

For agentic workers: REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Ensure Vitto's responses survive any kind of frontend disconnection — page refresh, browser close, device switch, or network hiccup.

Architecture: 6 targeted changes across backend-vitto-api (3 changes) and frontend vitto-chat (3 changes). The backend bridge writes completed responses to ai-agents DB unconditionally. ReplaySubject buffers events for reconnecting clients. A new status endpoint enables frontend recovery of stale streams. No new databases, no new collections — leverages existing bridgeMessageToAiApi() and agent-sessions.

Tech Stack: NestJS (backend-vitto-api), Angular 20+ with signals (frontend), RxJS ReplaySubject, Mongoose/MongoDB, Socket.IO

Spec: docs/superpowers/specs/2026-03-16-vitto-chat-persistence-recovery-design.md

Git constraint (S228): Do NOT commit during this session — another session is running concurrently on the same branch.

Chunk 1: Backend Changes (Tasks 1-4)

Task 1: Bridge Completed Responses to ai-agents DB

Files:

This is the highest-impact change. After the conversational loop completes, call bridgeMessageToAiApi() to push the response to ai-agents DB. This ensures the response is visible to loadBackendSessions() even if the frontend never reconnects.

Check that VittoEngineService already has BackendAiClientService injected. The actual property name is aiClient (verified at line 73: private readonly aiClient: BackendAiClientService).

Run: grep -n 'aiClient\|BackendAiClientService' APIs/backend-vitto-api/src/app/engine/vitto-engine.service.ts

If NOT injected, add it to the constructor. The service is exported by ConnectivityModule which is imported in AppModule.

In vitto-engine.service.ts, after line 1017 (the assistant addMessage call), add:

    // S228: Bridge response to ai-agents DB for multi-device recovery.
    // Even if the frontend never reconnects, loadBackendSessions() will find this.
    this.aiClient.bridgeMessageToAiApi({
      userId,
      spaceId,
      channel: 'web',
      messages: [{ role: 'assistant', content: result.output }],
      title: `Vitto conversation`,
    }).catch(err => this.logger.debug(`Bridge to ai-api failed (non-critical): ${(err as Error).message}`));

This goes between line 1017 and line 1019 (the comment // GAP 6: Emit task.completed).

Run: npx nx build backend-vitto-api --skip-nx-cache 2>&1 | tail -20

Expected: BUILD SUCCESS (or only pre-existing warnings).

Task 2: Subject to ReplaySubject + Extended Teardown Window

Files:

Replace the plain Subject with ReplaySubject so late-connecting clients get buffered event replay.

In vitto-engine.service.ts, line 12, change:

import { Subject, Subscription } from 'rxjs';

to:

import { Subject, ReplaySubject, Subscription } from 'rxjs';

Line 60, change:

private readonly conversationalStreams = new Map<string, Subject<SSEEvent>>();

to:

private readonly conversationalStreams = new Map<string, ReplaySubject<SSEEvent>>();

Line 872, change:

    const streamSubject = new Subject<SSEEvent>();

to:

    // S228: ReplaySubject buffers events for late subscribers (reconnecting clients).
    // 100 events max, 5-minute window (matches CostGuard max task duration).
    const streamSubject = new ReplaySubject<SSEEvent>(100, 5 * 60 * 1000);

Lines 1048-1052, change:

    // Keep the stream alive for 30s so late SSE subscribers can still read the final event
    setTimeout(() => {
      streamSubject.complete();
      this.conversationalStreams.delete(taskId);
    }, 30000);

to:

    // S228: Keep stream alive for 5min so reconnecting clients get full event replay.
    // Matches CostGuard max task duration. ReplaySubject buffers up to 100 events.
    const STREAM_TTL_MS = 5 * 60 * 1000;
    setTimeout(() => {
      streamSubject.complete();
      this.conversationalStreams.delete(taskId);
      this.completedTaskResults.delete(taskId);
    }, STREAM_TTL_MS);

Note: this.completedTaskResults is added in Task 3.

Run: npx nx build backend-vitto-api --skip-nx-cache 2>&1 | tail -20

Expected: May fail until Task 3 adds completedTaskResults. If so, temporarily comment out that line and verify the ReplaySubject changes compile. Uncomment after Task 3.

Task 3: Add completedTaskResults Map + Task Status Endpoint

Files:

In vitto-engine.service.ts, after line 65 (after taskContexts declaration), add:

  /** S228: Track completed conversational task results for status queries.
   *  Populated on task success/failure, cleaned up alongside conversationalStreams. */
  private readonly completedTaskResults = new Map<string, {
    status: 'completed' | 'failed';
    response?: string;
    error?: string;
    completedAt: number;
    userId: string;
  }>();

In vitto-engine.service.ts, after line 1017 (after assistant addMessage) and BEFORE the bridge call added in Task 1, add:

    // S228: Record terminal state for status endpoint
    this.completedTaskResults.set(taskId, {
      status: 'completed',
      response: result.output,
      completedAt: Date.now(),
      userId,
    });

In the catch block (after line 1042, after the task.failed event emission), add:

    // S228: Record terminal state for status endpoint
    this.completedTaskResults.set(taskId, {
      status: 'failed',
      error: errMessage,
      completedAt: Date.now(),
      userId,
    });

Add this public method after the existing getConversationalStream method (after line 1137):

  /** S228: Query the status of a conversational task for frontend recovery. */
  async getConversationalTaskStatus(taskId: string, userId: string): Promise<{
    status: 'running' | 'completed' | 'failed' | 'unknown';
    response?: string;
    error?: string;
    completedAt?: number;
    userId?: string;
  }> {
    // 1. Check completed results map (most common recovery path)
    const completed = this.completedTaskResults.get(taskId);
    if (completed) {
      return {
        status: completed.status,
        response: completed.response,
        error: completed.error,
        completedAt: completed.completedAt,
        userId: completed.userId,
      };
    }

    // 2. Check if stream still exists (task is running)
    if (this.conversationalStreams.has(taskId)) {
      return { status: 'running' };
    }

    // 3. Fallback: query agent-sessions for this taskId in message metadata.
    //    We can't use getSession(userId) because we don't know spaceId.
    //    Instead, query the model directly for any active session belonging
    //    to this user that has a message with the matching taskId.
    try {
      const session = await this.agentSessions.findSessionByTaskId(userId, taskId);
      if (session) {
        const match = [...session.messages].reverse().find(
          m => m.role === 'assistant' && (m.metadata as Record<string, unknown>)?.taskId === taskId,
        );
        if (match) {
          return { status: 'completed', response: match.content, completedAt: match.timestamp?.getTime?.(), userId };
        }
      }
    } catch { /* non-critical */ }

    return { status: 'unknown' };
  }

Note: Check if AgentSessionService has a getSession(userId) method. If it uses a different method name (e.g., getOrCreateSession), use that. The key is finding the session by userId and scanning its messages for matching taskId in metadata. See Task 4 if the method is missing.

In agent-session.service.ts, line 18-25, change:

export interface SessionMessage {
  role: 'user' | 'assistant' | 'system' | 'tool';
  content: string;
  toolName?: string;
  toolCallId?: string;
  timestamp: Date;
  channel?: string;
}

to:

export interface SessionMessage {
  role: 'user' | 'assistant' | 'system' | 'tool';
  content: string;
  toolName?: string;
  toolCallId?: string;
  timestamp: Date;
  channel?: string;
  /** S228: Optional metadata for task tracking (taskId, etc.) */
  metadata?: Record<string, unknown>;
}

In agent-session.schema.ts, after line 29 (after channel prop), add:

  @Prop({ type: Object })
  metadata?: Record<string, unknown>;

In vitto-engine.service.ts, line 1017, change:

    this.agentSessions.addMessage(userId, { role: 'assistant', content: result.output, timestamp: new Date(), channel: 'web' }, spaceId).catch(() => { /* non-critical */ });

to:

    this.agentSessions.addMessage(userId, { role: 'assistant', content: result.output, timestamp: new Date(), channel: 'web', metadata: { taskId } }, spaceId).catch(() => { /* non-critical */ });

In vitto.controller.ts, BEFORE the @Sse('message/:taskId/stream') endpoint (before line 3555), add:

  /** S228: Task status endpoint for frontend recovery after disconnection. */
  @Public()
  @Get('message/:taskId/status')
  async getConversationalTaskStatus(
    @Param('taskId') taskId: string,
    @CurrentUser() user: AuthenticatedUser,
  ) {
    const userId = user?.userId || 'anonymous';
    const result = await this.engine.getConversationalTaskStatus(taskId, userId);

    // Auth check: for completed/failed, verify the requesting user owns the task.
    // The taskId format is conv-{timestamp}-{random} — cryptographically random but
    // we still enforce ownership for defense-in-depth.
    if ((result.status === 'completed' || result.status === 'failed') && result.userId && result.userId !== userId) {
      return { status: 'unknown' as const }; // Hide other users' task results
    }

    // Strip internal userId from response
    const { userId: _uid, ...publicResult } = result as any;
    return publicResult;
  }

Important route ordering: This @Get('message/:taskId/status') must be declared BEFORE @Sse('message/:taskId/stream') in the controller, because NestJS matches routes top-to-bottom and @Sse could swallow the /status path. Verify the route order after adding.

Run: npx nx build backend-vitto-api --skip-nx-cache 2>&1 | tail -20

Expected: BUILD SUCCESS.

Task 4: Add findSessionByTaskId method to AgentSessionService

Files:

The status endpoint needs to find a session by taskId (stored in message metadata). Since sessions are keyed by userId:spaceId and we don't know the spaceId at query time, we need a query that searches across all of a user's sessions.

  /** S228: Find a user's session containing a message with the given taskId in metadata. */
  async findSessionByTaskId(userId: string, taskId: string): Promise<AgentSessionDoc | null> {
    return this.sessionModel.findOne({
      sessionKey: { $regex: `^${userId}` },
      status: 'active',
      'messages.metadata.taskId': taskId,
    }).exec();
  }

Note: This uses a regex prefix match on sessionKey to match both userId (no space) and userId:spaceId (with space) formats. The messages.metadata.taskId query works because Mongoose stores the nested object directly.

Run: npx nx build backend-vitto-api --skip-nx-cache 2>&1 | tail -20

Chunk 2: Frontend Changes (Tasks 5-7)

Task 5: Fix Reconciliation Corruption

Files:

In vitto-chat-persistence.service.ts, line 845, change:

    const localMessageIds = localSession.messages.map(m => m.id);

to:

    // S228: Exclude stale isStreaming messages from reconciliation.
    // Without this filter, broken partial messages get pushed to MongoDB.
    const localMessageIds = localSession.messages
      .filter(m => !m.isStreaming)
      .map(m => m.id);

In the same method, lines 871-879, the missingMessages are sent to the backend. Add an isStreaming filter there too for defense-in-depth. Change:

        const missingMessages = localSession.messages.filter(
          m => result.missingOnBackend.includes(m.id)
        );

to:

        const missingMessages = localSession.messages.filter(
          m => result.missingOnBackend.includes(m.id) && !m.isStreaming  // S228: Never sync stale streaming messages
        );

Run: npx nx build vitto-chat --skip-nx-cache 2>&1 | tail -20

Expected: BUILD SUCCESS.

Task 6: Add Skip-Sync for Bridged Messages

Files:

In chat-backend-sync.service.ts, line 156, change:

    const finalized = messages.filter(m => !m.isStreaming);

to:

    // S228: Skip streaming messages AND messages already bridged by the backend
    const finalized = messages.filter(m => !m.isStreaming && !(m as any).metadata?.bridged);

Check if SyncMessage has a metadata field. If not, the (m as any) cast handles it. If you want type safety, add metadata?: Record<string, unknown> to the SyncMessage interface in chat-sync.models.ts.

Run: grep -n 'interface SyncMessage' libs/libs-ai/chat-sync/src/lib/models/chat-sync.models.ts

Run: npx nx build chat-sync --skip-nx-cache 2>&1 | tail -20

Task 7: Frontend Stale Stream Recovery + Emotion Suppression

Files:

This is the largest task. It adds recovery logic that runs on page load.

In vitto-chat-persistence.service.ts, add a public method (near the other message accessor methods):

  /** S228: Find all messages with isStreaming: true in the active session. */
  getStaleStreamingMessages(): PersistedChatMessage[] {
    const sessionId = this._activeSessionId();
    if (!sessionId) return [];
    const session = this.state().sessions[sessionId];
    if (!session) return [];
    return session.messages.filter(m => m.isStreaming === true);
  }

  /** S228: Signal that stale stream recovery is complete. Unblocks reconciliation. */
  private _recoveryComplete$ = new Subject<void>();
  setRecoveryComplete(): void { this._recoveryComplete$.next(); this._recoveryComplete$.complete(); }
  private _recoveryDone = false;

Note: Import Subject from rxjs if not already imported in this file.

In reconcileActiveSession(), after line 840 (if (!localSession || localSession.messageCount === 0) return;), add:

    // S228: Wait for stale stream recovery before reconciling (RxJS, no setInterval).
    if (!this._recoveryDone) {
      this._recoveryComplete$.pipe(
        take(1),
        timeout(10000),  // Safety: unblock after 10s regardless
        catchError(() => of(undefined)),
      ).subscribe(() => {
        this._recoveryDone = true;
        this.reconcileActiveSession();
      });
      return;
    }

Note: Add take, timeout, catchError, of to RxJS imports if not already present.

In vitto-chat-persistence.service.ts, find the PersistedChatMessage interface's metadata field (around line 70-81). Add recovery-specific fields:

  metadata?: {
    // ... existing fields (model, provider, tokens, cost, etc.) ...
    /** S228: Recovery metadata */
    recovered?: boolean;
    bridged?: boolean;
    error?: boolean;
  };

This prevents TypeScript compilation errors when recovery code writes metadata: { recovered: true }.

In vitto-chat-engine.service.ts, add near checkAndReconnectActiveGeneration (around line 1305):

  /**
   * S228: Detect and recover stale isStreaming messages after page load.
   * Called after hydration + auth, BEFORE reconciliation completes.
   */
  private async recoverStaleStreams(): Promise<void> {
    if (!this.vittoApiUrl) {
      this.persistence.setRecoveryComplete();
      return;
    }

    const staleMessages = this.persistence.getStaleStreamingMessages();
    if (staleMessages.length === 0) {
      this.persistence.setRecoveryComplete();
      return;
    }

    this.logger.info(`[VittoChatEngine] Recovering ${staleMessages.length} stale streaming message(s)`);

    const recoveries = staleMessages.map(async (msg) => {
      const taskId = msg.id?.replace('stream-', '');
      if (!taskId) {
        this.persistence.updateMessage(msg.id, {
          isStreaming: false,
          content: (msg.content || '') + '\n\n*[Response interrupted]*',
          metadata: { ...msg.metadata, recovered: true, error: true },
        });
        return;
      }

      try {
        const response = await fetch(`${this.vittoApiUrl}/engine/message/${taskId}/status`);
        if (!response.ok) throw new Error(`Status ${response.status}`);

        const status = await response.json() as {
          status: 'running' | 'completed' | 'failed' | 'unknown';
          response?: string;
          error?: string;
        };

        switch (status.status) {
          case 'completed':
            this.logger.info(`[VittoChatEngine] Recovered completed task ${taskId}`);
            this.persistence.updateMessage(msg.id, {
              id: `vitto-${taskId}`,
              content: status.response || msg.content || '',
              isStreaming: false,
              metadata: { ...msg.metadata, recovered: true, bridged: true },
            });
            break;

          case 'running':
            this.logger.info(`[VittoChatEngine] Reconnecting to running task ${taskId}`);
            this.reconnectToStream(taskId, msg.id);
            break;

          case 'failed':
            this.persistence.updateMessage(msg.id, {
              isStreaming: false,
              content: (msg.content || '') + '\n\n*[Response failed]*',
              metadata: { ...msg.metadata, recovered: true, error: true },
            });
            break;

          default:
            this.persistence.updateMessage(msg.id, {
              isStreaming: false,
              content: (msg.content || '') + '\n\n*[Response interrupted]*',
              metadata: { ...msg.metadata, recovered: true, error: true },
            });
            break;
        }
      } catch (err) {
        this.logger.warn(`[VittoChatEngine] Recovery query failed for ${taskId}: ${(err as Error).message}`);
        // Network error — leave as-is, will retry on next page load
      }
    });

    await Promise.allSettled(recoveries);
    this.persistence.setRecoveryComplete();
  }

Add below recoverStaleStreams:

  /** S228: Reopen EventSource to a running task (ReplaySubject replays buffered events). */
  private reconnectToStream(taskId: string, streamingMsgId: string): void {
    const streamUrl = `${this.vittoApiUrl}/engine/message/${taskId}/stream`;
    const eventSource = new EventSource(streamUrl);

    eventSource.onmessage = (event: MessageEvent) => {
      try {
        const sseEvent = JSON.parse(event.data as string) as { type: string; data?: Record<string, unknown> };

        if (sseEvent.type === 'agent.text' || sseEvent.type === 'content_delta' || sseEvent.type === 'text_delta') {
          const delta = (sseEvent.data?.['content'] as string) || (sseEvent.data?.['text'] as string) || '';
          if (delta) {
            // Accumulate content (delta is partial text, not replacement)
            const currentMsgs = this.persistence.messages();
            const existing = currentMsgs.find(m => m.id === streamingMsgId);
            const updatedContent = (existing?.content || '') + delta;
            this.persistence.updateMessage(streamingMsgId, { content: updatedContent });
          }
        }

        if (sseEvent.type === 'task.completed') {
          const response = sseEvent.data?.['response'] as string || '';
          this.persistence.updateMessage(streamingMsgId, {
            id: `vitto-${taskId}`,
            content: response,
            isStreaming: false,
            metadata: { recovered: true, bridged: true },
          });
          this.stopLoadingWithTimeout();
          eventSource.close();
        }

        if (sseEvent.type === 'task.failed') {
          const msgs = this.persistence.messages();
          const current = msgs.find(m => m.id === streamingMsgId);
          this.persistence.updateMessage(streamingMsgId, {
            isStreaming: false,
            content: (current?.content || '') + '\n\n*[Response failed]*',
            metadata: { recovered: true, error: true },
          });
          this.stopLoadingWithTimeout();
          eventSource.close();
        }
      } catch { /* skip parse errors */ }
    };

    eventSource.onerror = () => {
      eventSource.close();
    };

    this.startLoadingWithTimeout();
  }

In the constructor effect, add the recovery call INSIDE the if (!this.backendLoaded) block, between line 214 (this.checkAndReconnectActiveGeneration(spaceId || '');) and the closing brace of the block at line 215. It must be BEFORE the return; at line 216:

            // S228: Recover stale streaming messages from previous sessions
            this.recoverStaleStreams().catch(err =>
              this.logger.warn('[VittoChatEngine] Stale stream recovery failed', err),
            );

In chat-shell.component.ts, line 308, change:

      if (!msg || msg.isStreaming) {

to:

      // S228: Skip emotion inference for recovered messages (avoids unexpected avatar animation)
      if (!msg || msg.isStreaming || (msg.metadata as any)?.recovered) {

Run:

npx nx build vitto-chat --skip-nx-cache 2>&1 | tail -20
npx nx build chat-sync --skip-nx-cache 2>&1 | tail -20

Expected: All BUILD SUCCESS.

Chunk 3: Cleanup + Integration Verification (Tasks 8-9)

Task 8: One-Time MongoDB Cleanup

Run after deployment to clean existing corrupted data (stale isStreaming: true messages in ai-agents DB):

// Run in mongosh connected to ai-agents DB
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 } } } }
)

The 1-hour filter ensures we don't remove actively-streaming messages during a live deployment.

db.chat_sessions.countDocuments({ 'messages.isStreaming': true })

Expected: 0 (or only messages created within the last hour).

Task 9: Manual Integration Verification

Files: None (testing only)

Start backend-api (3000), backend-ai-api (3001), and backend-vitto-api (3003). Wait for all to report listening.

Start the space app on port 6200. Wait for it to be available.

  1. Open browser to localhost:6200, navigate to Vitto chat
  2. Send a message, verify response appears normally
  3. Check DevTools Network tab: confirm both SSE stream AND bridge POST occurred
  1. Send a message to Vitto
  2. While processing, refresh the page (F5)
  3. On reload, check console for [VittoChatEngine] Recovering X stale streaming message(s)
  4. Verify one of: response appears (completed), loading resumes (running), or error shown (unknown)
  1. Send a message, close the browser tab entirely
  2. Wait 10 seconds, open a new tab to localhost:6200
  3. Navigate to Vitto chat, verify the response is present (loaded from ai-agents DB)
curl http://localhost:3003/vitto/engine/message/conv-XXXXX/status

Expected: { "status": "completed", "response": "...", "completedAt": ... } or { "status": "unknown" }.

Check ai-agents DB for isStreaming: true corruption. Expected: 0 new stale messages.

  1. Open two browser tabs to Vitto chat
  2. Send a message in Tab A
  3. Verify response appears in Tab B (via Socket.IO from bridge write)

Summary

Task Change Files Modified Effort
1 Bridge to ai-agents DB vitto-engine.service.ts ~30 min
2 ReplaySubject + extended window vitto-engine.service.ts ~15 min
3 completedTaskResults Map + status endpoint vitto-engine.service.ts, vitto.controller.ts, agent-session.service.ts, agent-session.schema.ts ~2 hr
4 findSessionByTaskId helper agent-session.service.ts ~15 min
5 Reconciliation corruption fix vitto-chat-persistence.service.ts ~30 min
6 Skip-sync for bridged messages chat-backend-sync.service.ts ~30 min
7 Frontend recovery + emotion suppression vitto-chat-engine.service.ts, vitto-chat-persistence.service.ts, chat-shell.component.ts ~3 hr
8 MongoDB cleanup script None (manual) ~15 min
9 Manual integration testing None ~1 hr

Total estimated: ~8.5 hours

Dependencies:

Deviation from spec: The reconnectToStream method lives in vitto-chat-engine.service.ts (not vitto-chat-streaming.service.ts as listed in the spec's Files Modified table). This is intentional — the engine service already manages EventSource lifecycle and the streaming service would require additional cross-service wiring.