Managing AI Streaming State in Angular 20+: Signals + SignalStore for OpenAI Token Streams, Retries, and Graceful Recovery (IntegrityLens Lessons)

The moment AI streams break demos (and how I fixed it)

Why Angular teams need a streaming state machine for OpenAI

Implementing a resilient SignalStore for AI streams

// ai-stream.store.ts (Angular 20+, @ngrx/signals)
import { signal, computed, inject } from '@angular/core';
import { signalStore, withState, withComputed, withMethods, patchState } from '@ngrx/signals';

export type StreamStatus = 'idle' | 'streaming' | 'complete' | 'error' | 'cancelled' | 'retrying';

interface AiStreamState {
  requestId: string | null;
  prompt: string | null;
  partial: string;
  tokens: number;
  status: StreamStatus;
  error?: string;
  retryCount: number;
  startedAt?: number;
  completedAt?: number;
}

const initialState: AiStreamState = {
  requestId: null,
  prompt: null,
  partial: '',
  tokens: 0,
  status: 'idle',
  retryCount: 0,
};

export const AiStreamStore = signalStore(
  withState(initialState),
  withComputed(({ partial, startedAt, completedAt, tokens }) => ({
    durationMs: computed(() => {
      const start = startedAt();
      const end = completedAt() ?? Date.now();
      return start ? end - start : 0;
    }),
    tokensPerSec: computed(() => {
      const t = tokens();
      const ms = (completedAt() ?? Date.now()) - (startedAt() ?? Date.now());
      return ms > 0 ? +(t / (ms / 1000)).toFixed(2) : 0;
    }),
  })),
  withMethods((store) => {
    let controller: AbortController | null = null;

    const start = async (prompt: string, model = 'gpt-4o-mini') => {
      controller?.abort();
      controller = new AbortController();
      const requestId = crypto.randomUUID();

      patchState(store, {
        status: 'streaming',
        prompt,
        requestId,
        partial: '',
        tokens: 0,
        retryCount: 0,
        error: undefined,
        startedAt: Date.now(),
        completedAt: undefined,
      });

      await streamOnce({ prompt, model, requestId });
    };

    const streamOnce = async ({ prompt, model, requestId }: { prompt: string; model: string; requestId: string; }) => {
      try {
        const res = await fetch('/api/openai/stream', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({ prompt, model, requestId }),
          signal: controller!.signal,
        });
        if (!res.ok || !res.body) throw new Error(`HTTP ${res.status}`);

        const reader = res.body.getReader();
        const decoder = new TextDecoder('utf-8');
        let buffer = '';

        while (true) {
          const { value, done } = await reader.read();
          if (done) break;
          buffer += decoder.decode(value, { stream: true });

          // SSE frames separated by two newlines
          const frames = buffer.split('\n\n');
          buffer = frames.pop() ?? '';
          for (const f of frames) {
            const line = f.trim();
            if (!line.startsWith('data:')) continue;
            const payload = line.replace(/^data:\s*/, '');
            if (payload === '[DONE]') continue;
            try {
              const json = JSON.parse(payload);
              const delta: string | undefined = json.choices?.[0]?.delta?.content;
              if (delta) {
                patchState(store, (s) => ({ partial: s.partial + delta, tokens: s.tokens + 1 }));
              }
            } catch (_) {}
          }
        }

        patchState(store, { status: 'complete', completedAt: Date.now() });
      } catch (e: any) {
        if (controller?.signal.aborted) {
          patchState(store, { status: 'cancelled', completedAt: Date.now() });
          return;
        }
        const retry = store.retryCount() + 1;
        patchState(store, { status: 'retrying', retryCount: retry, error: e?.message ?? 'stream-error' });
        if (retry <= 3) {
          const backoff = Math.min(1000 * 2 ** (retry - 1) + Math.floor(Math.random() * 250), 8000);
          await new Promise((r) => setTimeout(r, backoff));
          await streamOnce({ prompt, model, requestId }); // idempotent by requestId
        } else {
          patchState(store, { status: 'error', completedAt: Date.now() });
        }
      }
    };

    const abort = () => {
      controller?.abort();
    };

    const reset = () => patchState(store, initialState);

    return { start, abort, reset };
  })
);

// firebase/functions/src/streamChat.ts (Node 18+, Firebase Functions v2)
import { onRequest } from 'firebase-functions/v2/https';
import fetch from 'node-fetch';

export const streamChat = onRequest({ cors: true }, async (req, res) => {
  const { prompt, model, requestId } = req.body ?? {};
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('X-Request-Id', requestId ?? '');

  const upstream = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      model: model ?? 'gpt-4o-mini',
      stream: true,
      messages: [{ role: 'user', content: prompt }],
    }),
  });

  if (!upstream.ok || !upstream.body) {
    res.status(upstream.status).end();
    return;
  }

  for await (const chunk of upstream.body as any) {
    res.write(chunk);
  }
  res.end();
});

UI wiring: PrimeNG, Signals, and accessible live updates

// chat.component.ts
import { Component, inject } from '@angular/core';
import { AiStreamStore } from './ai-stream.store';

@Component({
  selector: 'app-chat',
  templateUrl: './chat.component.html',
})
export class ChatComponent {
  ai = inject(AiStreamStore);
  prompt = '';
  send() { if (this.prompt.trim()) this.ai.start(this.prompt.trim()); }
}

<!-- chat.component.html -->
<p-card>
  <div class="input-row">
    <input pInputText [(ngModel)]="prompt" placeholder="Ask..." [disabled]="ai.status()==='streaming'" />
    <button pButton label="Send" (click)="send()" [disabled]="ai.status()==='streaming'"></button>
    <button pButton severity="secondary" label="Cancel" (click)="ai.abort()" [disabled]="ai.status()!=='streaming'"></button>
  </div>

  <p-progressBar *ngIf="ai.status()==='streaming'" [value]="ai.tokensPerSec() * 10"></p-progressBar>

  <pre aria-live="polite" aria-atomic="false">{{ ai.partial() }}</pre>

  <ng-container *ngIf="ai.status()==='error'">
    <p-message severity="error" text="Stream failed. Please retry."></p-message>
    <button pButton label="Retry" (click)="send()"></button>
  </ng-container>
</p-card>

/* chat.component.scss */
pre { white-space: pre-wrap; max-height: 40vh; overflow: auto; }
.input-row { display: flex; gap: .5rem; align-items: center; }

Graceful error recovery: idempotence, dedupe, and telemetry

// telemetry.ts (example)
import { logEvent } from '@angular/fire/analytics';

export function logStreamOutcome(status: string, ctx: { requestId: string; tokens: number; durationMs: number; retries: number; }) {
  logEvent(globalThis as any, 'ai_stream_outcome', ctx);
}

How an Angular consultant approaches Signals for AI streaming

Outcomes and what to instrument next