Managing AI Streaming State in Angular 20+: OpenAI Token Streams, Signals/SignalStore Micro‑Batching, and Graceful Error Recovery (IntegrityLens Lessons)

The stutter isn’t OpenAI—it’s your state

We’ll build a focused SignalStore, wire a streaming reader, micro-batch DOM updates, and handle graceful recovery for 429/5xx/network faults—without losing transcripts.

Why Angular 20 Signals + SignalStore win for AI streams

The key is not RxJS vs Signals—it’s using both where they shine. Signals manage fast UI invalidation; RxJS handles retries and timers. I still use RxJS backoff with jitter for error recovery and keep the steady per-frame UI updates in Signals.

A SignalStore for OpenAI streaming (typed, abortable, resumable)

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

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

interface StreamState {
  requestId: string|null;
  status: StreamStatus;
  partialText: string;        // micro-batched string for UI
  tokens: number;             // count, not the whole array
  usage?: { prompt: number; completion: number };
  error?: { code: number; message: string };
  rateLimited: boolean;
  startedAt?: number;
  ttfuMs?: number;            // time to first update
}

const initialState: StreamState = {
  requestId: null,
  status: 'idle',
  partialText: '',
  tokens: 0,
  rateLimited: false
};

@Injectable({ providedIn: 'root' })
export class AiStreamStore extends signalStore(
  { providedIn: 'root' },
  withState(initialState),
  withMethods((store) => {
    const http = inject(HttpClient);
    const zone = inject(NgZone);

    let abort: AbortController|null = null;
    let queued: string[] = [];
    let scheduled = false;

    const flush = () => {
      if (!queued.length) return;
      const chunk = queued.join('');
      queued = [];
      patchState(store, s => ({
        partialText: s.partialText + chunk,
        tokens: s.tokens + chunk.length
      }));
      scheduled = false;
    };

    const scheduleFlush = () => {
      if (scheduled) return;
      scheduled = true;
      // 1 frame micro-batch; use setTimeout(0) for wider batching if needed
      requestAnimationFrame(() => zone.run(flush));
    };

    const start = async (prompt: string) => {
      // cancel any in-flight
      if (abort) abort.abort();
      abort = new AbortController();

      const requestId = crypto.randomUUID();
      patchState(store, {
        requestId,
        status: 'streaming',
        partialText: '',
        tokens: 0,
        error: undefined,
        rateLimited: false,
        startedAt: performance.now(),
        ttfuMs: undefined
      });

      try {
        // Call your server-side relay (Firebase Function, Node, etc.) that talks to OpenAI with stream: true
        const res = await fetch('/api/openai/stream', {
          method: 'POST',
          body: JSON.stringify({ prompt, requestId }),
          headers: { 'Content-Type': 'application/json' },
          signal: abort.signal
        });

        if (res.status === 429) {
          patchState(store, { status: 'rate_limited', rateLimited: true, error: { code: 429, message: 'Rate limit' } });
          return;
        }
        if (!res.ok || !res.body) throw new Error(`HTTP ${res.status}`);

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

        await zone.runOutsideAngular(async () => {
          while (true) {
            const { value, done } = await reader.read();
            if (done) break;
            const text = decoder.decode(value, { stream: true });
            // Parse SSE lines or raw chunks; here we assume lines starting with 'data:'
            for (const line of text.split('\n')) {
              const trimmed = line.trim();
              if (!trimmed || !trimmed.startsWith('data:')) continue;
              if (trimmed === 'data: [DONE]') continue;
              try {
                const payload = JSON.parse(trimmed.slice(5).trim());
                const token = payload.delta?.content ?? payload.choices?.[0]?.delta?.content ?? '';
                if (token) {
                  queued.push(token);
                  scheduleFlush();
                  if (first) {
                    first = false;
                    zone.run(() => patchState(store, s => ({ ttfuMs: Math.round(performance.now() - (s.startedAt ?? performance.now())) })));
                  }
                }
              } catch {}
            }
          }
        });

        patchState(store, { status: 'complete' });
      } catch (err: any) {
        if (err?.name === 'AbortError') {
          patchState(store, { status: 'cancelled' });
        } else {
          const code = typeof err?.status === 'number' ? err.status : 0;
          patchState(store, { status: code === 429 ? 'rate_limited' : 'error', error: { code, message: String(err?.message ?? err) } });
        }
      }
    };

    const cancel = () => {
      if (abort) abort.abort();
      abort = null;
    };

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

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

Token-by-token updates without jank: micro-batching and templates

<!-- ai-stream.component.html -->
<p-panel header='AI Response'>
  <ng-container *ngIf='store.status() === "streaming"'>
    <p-progressBar mode='indeterminate'></p-progressBar>
  </ng-container>
  <pre class='stream' [textContent]='store.partialText()'></pre>
</p-panel>
<div class='actions'>
  <p-button label='Start' icon='pi pi-play' (onClick)='store.start(prompt.value)'></p-button>
  <p-button label='Cancel' icon='pi pi-stop' severity='warning' (onClick)='store.cancel()' [disabled]='store.status() !== "streaming"'></p-button>
  <input #prompt type='text' pInputText placeholder='Ask something...' />
</div>

/* ai-stream.component.scss */
.stream { white-space: pre-wrap; word-break: break-word; font-family: var(--font-mono, ui-monospace, SFMono-Regular); }
.actions { display: flex; gap: .5rem; align-items: center; }

Graceful error recovery and resume

// retry.policy.ts
export function* backoffDelays(maxRetries = 4, baseMs = 400, capMs = 8000) {
  for (let i = 0; i < maxRetries; i++) {
    const exp = Math.min(capMs, baseMs * Math.pow(2, i));
    const jitter = Math.random() * exp * 0.4; // 40% jitter
    yield Math.round(exp / 2 + jitter);
  }
}

// Example usage (RxJS for timers)
import { timer, lastValueFrom } from 'rxjs';

export async function retryWithPolicy<T>(fn: () => Promise<T>, retries = 3): Promise<T> {
  let lastErr: any;
  for (const delayMs of backoffDelays(retries)) {
    try { return await fn(); } catch (e) { lastErr = e; }
    await lastValueFrom(timer(delayMs));
  }
  throw lastErr;
}

// Example resume flow inside a component
async function resumeStream(store: AiStreamStore, lastPrompt: string) {
  // Wrap your store.start(...) with retry policy on 429/5xx (not within the read loop)
  try {
    await retryWithPolicy(() => store.start(lastPrompt));
  } catch (e) {
    // surface a human-readable message
  }
}

Observability: metrics and CI guardrails that prove streaming quality

// telemetry.types.ts
export type StreamEvent =
  | { type: 'stream_start'; requestId: string; ttfuTargetMs: number }
  | { type: 'stream_update'; requestId: string; tokens: number }
  | { type: 'stream_complete'; requestId: string; tokens: number; durationMs: number }
  | { type: 'stream_error'; requestId: string; code: number; message: string };

export interface TelemetryPort { emit: (e: StreamEvent) => void }

# .github/workflows/ci.yml (excerpt)
- name: Unit
  run: npm run test -- --watch=false --browsers=ChromeHeadless
- name: E2E Stream
  run: npx cypress run --spec 'cypress/e2e/ai-stream.cy.ts'
- name: Lighthouse Budget
  run: npx @lhci/cli autorun --upload.target=temporary-public-storage

When to hire an Angular developer for AI streaming integrations

If you need a remote Angular developer or Angular consultant, I can review your build, wire SignalStore patterns, and land CI guardrails so you ship confidently in Q1.

Practical takeaways

• Model streaming as a lifecycle in a SignalStore, not ad-hoc flags in components.
• Parse outside Angular; flush inside on rAF to avoid jank.
• Micro-batch tokens—aim for ~16ms frames. It feels realtime and keeps INP happy.
• Design explicit error states with Resume. Users forgive errors; they don’t forgive silent stalls.
• Instrument TTFU, p95 completion, tokens/sec, and retries. Show the numbers in interviews and reviews.