feat(boost): add streaming chat endpoint and normalized event pipeline (#3302) (#3555)

* feat(boost): add streaming chat endpoint and normalized event pipeline (#3302)

Implement the SSE streaming endpoint, synchronous chat route, and
supporting infrastructure that connects AI providers to the frontend
via NormalizedStreamEvent.

New files:
- chat/routes.ts: POST /chat (sync) and POST /chat/stream (SSE) endpoints
  with permission gating (boost.chat.create + boost.admin fallback),
  input validation, and error-resilient streaming
- chat/ConversationAgentCache.ts: session-scoped cacheService mapping
  conversation IDs to provider IDs (task 1.8, 24h TTL)
- chat/RateLimiter.ts: per-window rate limiter backed by cacheService
  (task 1.9, configurable max/window)
- chat/index.ts: barrel exports

Modified files:
- plugin.ts: wire chat routes, ConversationAgentCache, RateLimiter into
  plugin init; add /chat auth policy
- index.ts: export new public API surface

All caches use Backstage cacheService per Decision 3 (design.md).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: address review feedback on streaming chat pipeline

- Track provider-sent done event to prevent duplicate done in SSE stream
- Replace double-cast (as unknown as string) with JSON.stringify/parse in RateLimiter
- Tighten SSE test assertion from toBeGreaterThanOrEqual(1) to toBe(1)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: gabemontero <gmontero@redhat.com>

* feat(#3302): stub full NormalizedStreamEvent union and add missing test coverage

- Add 8 stub event interfaces to NormalizedStreamEvent union per
  normalized-streaming spec: reasoning, rag_result, handoff, approval,
  form, auth, artifact, citation
- Add test for provider-without-done dedup path in streaming route
- Add clarifying comments for sync endpoint limitation and anonymous
  rate-limit bucket

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: gabemontero <gmontero@redhat.com>

* fix(#3302): handle auto-deserializing cache backends in RateLimiter

RateLimiter.consume() now handles both string and pre-parsed object
returns from cacheService.get(). Some cache backends (e.g., Keyv with
JSON serializer) auto-deserialize stored JSON into objects, which would
bypass the typeof raw === 'string' check and reset the rate-limit
window on every request.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: gabemontero <gmontero@redhat.com>

---------

Signed-off-by: gabemontero <gmontero@redhat.com>
Co-authored-by: fullsend-code <278716306+fullsend-ai-coder[bot]@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: gabemontero <gmontero@redhat.com>

Author: 3 people

workspaces/boost/plugins/boost-backend/report.api.md

@@ -180,6 +180,16 @@ export type BoostConfigKey = keyof typeof boostConfigFields;
 const boostPlugin: BackendFeature;
 export default boostPlugin;
 
+// @public
+export interface ChatRoutesOptions {
+  conversationAgentCache: ConversationAgentCache;
+  httpAuth: HttpAuthService;
+  logger: LoggerService;
+  permissions: PermissionsService;
+  providerManager: ProviderManager;
+  rateLimiter: RateLimiter;
+}
+
 // @public
 export interface ConfigFieldMeta<T extends z.ZodTypeAny = z.ZodTypeAny> {
   configScope: ConfigScope;
@@ -191,12 +201,29 @@ export interface ConfigFieldMeta<T extends z.ZodTypeAny = z.ZodTypeAny> {
 // @public
 export type ConfigScope = 'yaml-only' | 'db-overridable' | 'db-only';
 
+// @public
+export class ConversationAgentCache {
+  constructor(options: ConversationAgentCacheOptions);
+  delete(conversationId: string): Promise<void>;
+  get(conversationId: string): Promise<string | undefined>;
+  set(conversationId: string, providerId: string): Promise<void>;
+}
+
+// @public
+export interface ConversationAgentCacheOptions {
+  cache: CacheService;
+  logger: LoggerService;
+}
+
 // @public
 export function createAgentResourceLoader(): ResourceLoader;
 
 // @public
 export function createAgentRoutes(options: AgentRoutesOptions): Router;
 
+// @public
+export function createChatRoutes(options: ChatRoutesOptions): Router;
+
 // @public
 export function createKagentiAdminRoutes(
   options: KagentiAdminRoutesOptions,
@@ -288,6 +315,24 @@ export class ProviderManager {
   switchProvider(providerId: string): void;
 }
 
+// @public
+export class RateLimiter {
+  constructor(options: RateLimiterOptions);
+  consume(identity: string): Promise<{
+    allowed: boolean;
+    remaining: number;
+    retryAfterMs?: number;
+  }>;
+}
+
+// @public
+export interface RateLimiterOptions {
+  cache: CacheService;
+  logger: LoggerService;
+  maxRequests?: number;
+  windowMs?: number;
+}
+
 // @public
 export type ResourceLoader = (req: Request_2) => Promise<
   | {

workspaces/boost/plugins/boost-backend/src/chat/ConversationAgentCache.test.ts

@@ -0,0 +1,83 @@
+/*
+ * Copyright Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type {
+  CacheService,
+  LoggerService,
+} from '@backstage/backend-plugin-api';
+import { ConversationAgentCache } from './ConversationAgentCache';
+
+function createMockLogger(): LoggerService {
+  return {
+    info: jest.fn(),
+    warn: jest.fn(),
+    error: jest.fn(),
+    debug: jest.fn(),
+    child: jest.fn().mockReturnThis(),
+  };
+}
+
+function createMockCache(): CacheService {
+  const store = new Map<string, unknown>();
+  const cache: CacheService = {
+    get: jest.fn(async (key: string) => store.get(key)) as CacheService['get'],
+    set: jest.fn(async (key: string, value: unknown) => {
+      store.set(key, value);
+    }),
+    delete: jest.fn(async (key: string) => {
+      store.delete(key);
+    }),
+    withOptions: jest.fn().mockReturnThis(),
+  };
+  return cache;
+}
+
+describe('ConversationAgentCache', () => {
+  let cache: CacheService;
+  let conversationCache: ConversationAgentCache;
+
+  beforeEach(() => {
+    cache = createMockCache();
+    conversationCache = new ConversationAgentCache({
+      cache,
+      logger: createMockLogger(),
+    });
+  });
+
+  it('stores and retrieves a conversation-agent mapping', async () => {
+    await conversationCache.set('conv-1', 'llamastack');
+    const result = await conversationCache.get('conv-1');
+    expect(result).toBe('llamastack');
+  });
+
+  it('returns undefined for unknown conversation', async () => {
+    const result = await conversationCache.get('unknown');
+    expect(result).toBeUndefined();
+  });
+
+  it('deletes a conversation-agent mapping', async () => {
+    await conversationCache.set('conv-1', 'llamastack');
+    await conversationCache.delete('conv-1');
+    const result = await conversationCache.get('conv-1');
+    expect(result).toBeUndefined();
+  });
+
+  it('uses cacheService withOptions for namespace isolation', () => {
+    expect(cache.withOptions).toHaveBeenCalledWith({
+      defaultTtl: 24 * 60 * 60 * 1000,
+    });
+  });
+});

workspaces/boost/plugins/boost-backend/src/chat/ConversationAgentCache.ts

@@ -0,0 +1,93 @@
+/*
+ * Copyright Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type {
+  CacheService,
+  LoggerService,
+} from '@backstage/backend-plugin-api';
+
+/**
+ * Options for creating a ConversationAgentCache.
+ *
+ * @public
+ */
+export interface ConversationAgentCacheOptions {
+  /** The Backstage cache service. */
+  cache: CacheService;
+  /** The Backstage logger service. */
+  logger: LoggerService;
+}
+
+/**
+ * Session-scoped cache that maps conversation IDs to the agent (provider)
+ * ID that is handling the conversation. Uses Backstage cacheService with
+ * namespace isolation per Decision 3 (design.md).
+ *
+ * @public
+ */
+export class ConversationAgentCache {
+  private readonly cache: CacheService;
+  private readonly logger: LoggerService;
+
+  /** Cache TTL for conversation-agent mappings: 24 hours. */
+  private static readonly TTL_MS = 24 * 60 * 60 * 1000;
+
+  constructor(options: ConversationAgentCacheOptions) {
+    this.cache = options.cache.withOptions({
+      defaultTtl: ConversationAgentCache.TTL_MS,
+    });
+    this.logger = options.logger;
+  }
+
+  /**
+   * Associates a conversation with a provider (agent) ID.
+   *
+   * @param conversationId - The conversation identifier.
+   * @param providerId - The provider identifier handling this conversation.
+   */
+  async set(conversationId: string, providerId: string): Promise<void> {
+    const key = `conversation-agent:${conversationId}`;
+    await this.cache.set(key, providerId);
+    this.logger.debug(
+      `Mapped conversation ${conversationId} to provider ${providerId}`,
+    );
+  }
+
+  /**
+   * Retrieves the provider ID for a conversation.
+   *
+   * @param conversationId - The conversation identifier.
+   * @returns The provider ID or undefined if not cached.
+   */
+  async get(conversationId: string): Promise<string | undefined> {
+    const key = `conversation-agent:${conversationId}`;
+    const value = await this.cache.get(key);
+    return typeof value === 'string' ? value : undefined;
+  }
+
+  /**
+   * Removes the conversation-agent mapping.
+   *
+   * @param conversationId - The conversation identifier.
+   */
+  async delete(conversationId: string): Promise<void> {
+    const key = `conversation-agent:${conversationId}`;
+    await this.cache.delete(key);
+    this.logger.debug(
+      `Removed conversation-agent mapping for ${conversationId}`,
+    );
+  }
+}

workspaces/boost/plugins/boost-backend/src/chat/RateLimiter.test.ts

@@ -0,0 +1,128 @@
+/*
+ * Copyright Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type {
+  CacheService,
+  LoggerService,
+} from '@backstage/backend-plugin-api';
+import { RateLimiter } from './RateLimiter';
+
+function createMockLogger(): LoggerService {
+  return {
+    info: jest.fn(),
+    warn: jest.fn(),
+    error: jest.fn(),
+    debug: jest.fn(),
+    child: jest.fn().mockReturnThis(),
+  };
+}
+
+function createMockCache(): CacheService {
+  const store = new Map<string, unknown>();
+  const cache: CacheService = {
+    get: jest.fn(async (key: string) => store.get(key)) as CacheService['get'],
+    set: jest.fn(async (key: string, value: unknown) => {
+      store.set(key, value);
+    }),
+    delete: jest.fn(async (key: string) => {
+      store.delete(key);
+    }),
+    withOptions: jest.fn().mockReturnThis(),
+  };
+  return cache;
+}
+
+describe('RateLimiter', () => {
+  let cache: CacheService;
+  let limiter: RateLimiter;
+
+  beforeEach(() => {
+    cache = createMockCache();
+    limiter = new RateLimiter({
+      cache,
+      logger: createMockLogger(),
+      maxRequests: 3,
+      windowMs: 60_000,
+    });
+  });
+
+  it('allows requests within the limit', async () => {
+    const r1 = await limiter.consume('user-1');
+    expect(r1.allowed).toBe(true);
+    expect(r1.remaining).toBe(2);
+
+    const r2 = await limiter.consume('user-1');
+    expect(r2.allowed).toBe(true);
+    expect(r2.remaining).toBe(1);
+  });
+
+  it('denies requests exceeding the limit', async () => {
+    await limiter.consume('user-1');
+    await limiter.consume('user-1');
+    await limiter.consume('user-1');
+
+    const r4 = await limiter.consume('user-1');
+    expect(r4.allowed).toBe(false);
+    expect(r4.remaining).toBe(0);
+    expect(r4.retryAfterMs).toBeGreaterThan(0);
+  });
+
+  it('tracks separate limits per identity', async () => {
+    await limiter.consume('user-1');
+    await limiter.consume('user-1');
+    await limiter.consume('user-1');
+
+    // user-2 should still be allowed
+    const r = await limiter.consume('user-2');
+    expect(r.allowed).toBe(true);
+    expect(r.remaining).toBe(2);
+  });
+
+  it('handles cache backends that auto-deserialize JSON into objects', async () => {
+    const objectCache = createMockCache();
+    const origSet = objectCache.set as jest.Mock;
+    const origGet = objectCache.get as jest.Mock;
+
+    // Intercept set to store parsed objects instead of raw strings
+    origSet.mockImplementation(async (key: string, value: unknown) => {
+      const parsed = typeof value === 'string' ? JSON.parse(value) : value;
+      (origGet as jest.Mock).mockImplementation(async (k: string) =>
+        k === key ? parsed : undefined,
+      );
+    });
+
+    const objLimiter = new RateLimiter({
+      cache: objectCache,
+      logger: createMockLogger(),
+      maxRequests: 3,
+      windowMs: 60_000,
+    });
+
+    const r1 = await objLimiter.consume('user-obj');
+    expect(r1.allowed).toBe(true);
+    expect(r1.remaining).toBe(2);
+
+    const r2 = await objLimiter.consume('user-obj');
+    expect(r2.allowed).toBe(true);
+    expect(r2.remaining).toBe(1);
+  });
+
+  it('uses cacheService withOptions for namespace isolation', () => {
+    expect(cache.withOptions).toHaveBeenCalledWith({
+      defaultTtl: 60_000,
+    });
+  });
+});