fix(dashboard): resolver bugs críticos de tiempo real en HITL — race conditions, In-Band Auth y multi-stream buffer

- AppShell: corregir condición de carrera REST/WS que perdía tokens de agent_stream_chunk
- init_state atómico + eliminación de doble fuente REST/WS para actualización en tiempo real
- conversation_ended e idempotencia de eventos en máquina de estados por conversación
- Seguridad: migrar JWT de query param a In-Band Auth (primer mensaje {action:auth}) con timeout 5s y cierre 1008
- Multi-stream buffer: reemplazar buffer plano por TTL LRU (200 entradas, 60s TTL) para evitar pisado de tokens entre agentes
- agent_stream_completed ya no borra buffer incondicionalmente — delega purge a la política LRU
- Timer: corregir display de 00:00 en estado PENDING con visualización inmediata + cleanup en stop()
- Tests: 8 tests multi-stream, tests In-Band Auth, tests idempotencia y máquina de estados, tests Timer
- Resultado: 86/86 tests pasan | TypeScript 0 errores
This commit is contained in:
2026-07-29 04:32:27 -05:00
parent 8f044567c0
commit 83e3ec2cff
37 changed files with 6979 additions and 1129 deletions
+23 -7
View File
@@ -1,19 +1,35 @@
import { BrowserRouter, Routes, Route, Navigate } from 'react-router-dom';
import { BrowserRouter, Routes, Route, Navigate, Outlet } from 'react-router-dom';
import { AppShell } from './components/layout/AppShell';
import { LoginPage } from './components/auth/LoginPage';
import { ProtectedRoute } from './components/auth/ProtectedRoute';
import CasesPage from './pages/CasesPage';
import MonitorPage from './pages/MonitorPage';
function ProtectedLayout() {
return (
<ProtectedRoute>
<AppShell>
<Outlet />
</AppShell>
</ProtectedRoute>
);
}
export default function App() {
return (
<BrowserRouter>
<AppShell>
<Routes>
<Route path="/" element={<Navigate to="/cases" replace />} />
<Routes>
{/* Login — standalone, sin header ni sidebar */}
<Route path="/login" element={<LoginPage />} />
{/* Rutas protegidas — envueltas en AppShell + auth guard */}
<Route element={<ProtectedLayout />}>
<Route path="/cases" element={<CasesPage />} />
<Route path="/monitor" element={<MonitorPage />} />
<Route path="*" element={<Navigate to="/cases" replace />} />
</Routes>
</AppShell>
</Route>
<Route path="*" element={<Navigate to="/cases" replace />} />
</Routes>
</BrowserRouter>
);
}
+277
View File
@@ -0,0 +1,277 @@
import { useState, useCallback, useEffect } from 'react';
import { useNavigate, useSearchParams } from 'react-router-dom';
import { Loader2, AlertCircle, LogIn, ExternalLink } from 'lucide-react';
import { auth, type Session } from '@/services/auth';
const EXTENSION_STORE_URL = 'https://chromewebstore.google.com/detail/kcfpmlgjjldalkcajjjdfmpjcccbnkeo';
type LoginState =
| 'detecting_redirect'
| 'extension_missing'
| 'idle'
| 'opening_popup'
| 'exchanging_token'
| 'success'
| 'error';
/**
* Check whether the Linguo browser extension is installed by
* looking for the <linguo-component id="linguo-component"> element
* that it injects into every page at document_end.
*/
function detectExtension(): boolean {
return document.getElementById('linguo-component') !== null;
}
export function LoginPage() {
const navigate = useNavigate();
const [searchParams] = useSearchParams();
const [loginState, setLoginState] = useState<LoginState>('detecting_redirect');
const [errorMessage, setErrorMessage] = useState<string>('');
// hasExtension tracked via loginState
// ── Check for pre-existing token (redirect or extension) ──
useEffect(() => {
// A) Redirect flow: Okan passed token via ?token=
const urlToken = searchParams.get('token');
if (urlToken) {
handleExchange(urlToken);
return;
}
// B) Extension already injected tokenOkan into localStorage
const extToken = auth.readExtensionToken();
if (extToken) {
handleExchange(extToken);
return;
}
// C) Detect extension
const installed = detectExtension();
// eslint-disable-next-line no-unused-expressions
if (!installed) {
setLoginState('extension_missing');
} else {
setLoginState('idle');
}
// eslint-disable-next-line react-hooks/exhaustive-deps
}, []);
// ── Exchange token → store session → redirect ───────────────
const handleExchange = useCallback(
async (rawToken: string) => {
setErrorMessage('');
setLoginState('exchanging_token');
let session: Session;
try {
session = await auth.exchangeToken(rawToken);
} catch (exchangeError) {
const msg =
exchangeError instanceof Error
? exchangeError.message
: 'Error de red al verificar credenciales';
// Clear stale/expired token so next attempt opens fresh Okan popup
auth.clearExtensionToken();
setErrorMessage(msg);
setLoginState('error');
return;
}
auth.storeSession(session);
auth.clearExtensionToken();
setLoginState('success');
setTimeout(() => {
navigate('/cases', { replace: true });
}, 500);
},
[navigate],
);
// ── Open popup, wait for extension to inject token ──────────
const handleAutoLogin = useCallback(async () => {
setErrorMessage('');
setLoginState('opening_popup');
try {
const rawToken = await auth.captureOkanToken();
await handleExchange(rawToken);
} catch (err) {
const msg = err instanceof Error ? err.message : 'Error desconocido';
if (msg === 'popup_blocked') {
setErrorMessage(
'No se pudo abrir la ventana de inicio de sesión. ' +
'Permite ventanas emergentes (pop-ups) para este sitio e intenta de nuevo.',
);
} else if (msg === 'cancelado') {
setErrorMessage('Inicio de sesión cancelado. Intenta de nuevo.');
} else if (msg === 'timeout') {
setErrorMessage(
'Tiempo de espera agotado. Asegúrate de iniciar sesión en la ventana de Okan.',
);
} else {
setErrorMessage(msg);
}
setLoginState('error');
}
}, [handleExchange]);
// ── Re-check extension and retry ────────────────────────────
const handleRetry = useCallback(() => {
const installed = detectExtension();
// eslint-disable-next-line no-unused-expressions
if (!installed) {
setLoginState('extension_missing');
} else {
handleAutoLogin();
}
}, [handleAutoLogin]);
// ── Render ──────────────────────────────────────────────────
return (
<div className="flex items-center justify-center h-full bg-bg-base">
<div className="w-full max-w-[420px] mx-4">
<div className="bg-surface border border-border rounded-xl shadow-lg p-8">
{/* Logo */}
<div className="text-center mb-8">
<span className="text-[32px] leading-none" role="img" aria-label="Claro">🔴</span>
<h1 className="text-[22px] font-extrabold bg-gradient-to-r from-accent-orange to-accent-yellow bg-clip-text text-transparent mt-1">
Claro Cases
</h1>
<p className="text-[12px] text-text-muted mt-1">
Inicia sesión para gestionar casos
</p>
</div>
{/* ── Detecting ── */}
{loginState === 'detecting_redirect' && (
<div className="flex flex-col items-center gap-4 py-4">
<Loader2 size={32} className="text-accent-orange animate-spin" />
<p className="text-[14px] text-text-primary font-medium">Verificando sesión...</p>
</div>
)}
{/* ── Extension missing ── */}
{loginState === 'extension_missing' && (
<div className="flex flex-col gap-4">
<div className="flex items-start gap-2 w-full p-3 rounded-lg bg-accent-yellow/10 border border-accent-yellow/20">
<AlertCircle size={16} className="text-accent-yellow shrink-0 mt-0.5" />
<div>
<p className="text-[12px] text-text-secondary leading-relaxed">
No se detectó la extensión de <strong>Linguo</strong> en tu navegador.
Es necesaria para capturar tus credenciales de Okan de forma segura.
</p>
</div>
</div>
<a
href={EXTENSION_STORE_URL}
target="_blank"
rel="noopener noreferrer"
className="w-full flex items-center justify-center gap-2 px-4 py-3
bg-accent-orange text-white font-semibold text-[14px]
rounded-lg hover:bg-[#e04600] active:bg-[#c93d00]
transition-colors shadow-sm no-underline"
>
<ExternalLink size={16} />
Instalar extensión de Linguo
</a>
<button
type="button"
onClick={handleRetry}
className="w-full px-4 py-2.5 text-[12px] font-medium text-text-secondary
border border-border rounded-lg hover:bg-bg-hover transition-colors"
>
Ya instalé la extensión verificar de nuevo
</button>
</div>
)}
{/* ── Idle ── */}
{loginState === 'idle' && (
<div className="flex flex-col items-center gap-4">
<button
type="button"
onClick={handleAutoLogin}
className="w-full flex items-center justify-center gap-2 px-4 py-3
bg-accent-orange text-white font-semibold text-[14px]
rounded-lg hover:bg-[#e04600] active:bg-[#c93d00]
transition-colors shadow-sm"
>
<LogIn size={18} />
Iniciar sesión con Okan
</button>
<p className="text-[10px] text-text-muted text-center leading-relaxed">
Se abrirá una ventana de <strong>Okan Tools</strong> para autenticarte.
Tus credenciales se capturarán automáticamente.
</p>
</div>
)}
{/* ── Opening popup ── */}
{loginState === 'opening_popup' && (
<div className="flex flex-col items-center gap-4 py-4">
<Loader2 size={32} className="text-accent-orange animate-spin" />
<p className="text-[14px] text-text-primary font-medium">
Abriendo ventana de Okan...
</p>
<p className="text-[11px] text-text-muted text-center">
Inicia sesión en la ventana de Okan. Tus credenciales se capturarán automáticamente.
</p>
</div>
)}
{/* ── Exchanging token ── */}
{loginState === 'exchanging_token' && (
<div className="flex flex-col items-center gap-4 py-4">
<Loader2 size={32} className="text-accent-orange animate-spin" />
<p className="text-[14px] text-text-primary font-medium">
Verificando credenciales...
</p>
<p className="text-[11px] text-text-muted text-center">
Intercambiando token de acceso de forma segura.
</p>
</div>
)}
{/* ── Success ── */}
{loginState === 'success' && (
<div className="flex flex-col items-center gap-4 py-4">
<div className="w-[48px] h-[48px] rounded-full bg-accent-green/10 flex items-center justify-center">
<span className="text-accent-green text-[24px]"></span>
</div>
<p className="text-[14px] text-text-primary font-medium">¡Autenticado!</p>
<p className="text-[11px] text-text-muted">Redirigiendo al panel...</p>
</div>
)}
{/* ── Error ── */}
{loginState === 'error' && (
<div className="flex flex-col items-center gap-4">
<div className="flex items-start gap-2 w-full p-3 rounded-lg bg-accent-red/5 border border-accent-red/15">
<AlertCircle size={16} className="text-accent-red shrink-0 mt-0.5" />
<p className="text-[12px] text-accent-red leading-relaxed">{errorMessage}</p>
</div>
<button
type="button"
onClick={handleRetry}
className="w-full flex items-center justify-center gap-2 px-4 py-3
bg-accent-orange text-white font-semibold text-[14px]
rounded-lg hover:bg-[#e04600] active:bg-[#c93d00]
transition-colors shadow-sm"
>
<LogIn size={18} />
Intentar de nuevo
</button>
</div>
)}
</div>
</div>
</div>
);
}
+44
View File
@@ -0,0 +1,44 @@
import { type ReactNode } from 'react';
import { Navigate } from 'react-router-dom';
import { useAuth } from '@/hooks/useAuth';
import { Loader2 } from 'lucide-react';
// ─────────────────────────────────────────────────────────────
// Props
// ─────────────────────────────────────────────────────────────
interface ProtectedRouteProps {
children: ReactNode;
}
// ─────────────────────────────────────────────────────────────
// Component
// ─────────────────────────────────────────────────────────────
/**
* Route guard that wraps protected pages.
*
* - While auth status is 'loading', shows a centered spinner.
* - If not authenticated (anonymous or expired), redirects to /login.
* - If authenticated, renders the children.
*/
export function ProtectedRoute({ children }: ProtectedRouteProps) {
const { isAuthenticated, isLoading } = useAuth();
if (isLoading) {
return (
<div className="flex items-center justify-center h-screen bg-bg-base">
<div className="flex flex-col items-center gap-3">
<Loader2 size={32} className="text-accent-orange animate-spin" />
<span className="text-[13px] text-text-muted">Cargando...</span>
</div>
</div>
);
}
if (!isAuthenticated) {
return <Navigate to="/login" replace />;
}
return <>{children}</>;
}
+9 -2
View File
@@ -37,12 +37,20 @@ function formatDate(iso: string): string {
export default function CaseDetail({ case: caseData }: CaseDetailProps) {
const resolveCase = useAppStore((s) => s.resolveCase);
const startCase = useAppStore((s) => s.startCase);
const [stepsOpen, setStepsOpen] = useState(false);
const timerRef = useRef<TimerHandle>(null);
// Look up the CaseTypeDefinition for this case's toolName
const caseType = caseTypeByToolName[caseData.tipoSolicitud] ?? null;
// Start case (PENDING → IN_PROGRESS) and timer when viewing a PENDING case
useEffect(() => {
if (caseData.status === CaseStatus.PENDING) {
startCase(caseData.id);
}
}, [caseData.id, caseData.status, startCase]);
// Start timer when case is IN_PROGRESS and detail is mounted
useEffect(() => {
if (caseData.status === CaseStatus.IN_PROGRESS && timerRef.current) {
@@ -53,9 +61,8 @@ export default function CaseDetail({ case: caseData }: CaseDetailProps) {
const handleFormSubmit = useCallback(
async (formData: Record<string, unknown>) => {
try {
const actionName = caseType?.toolName ?? 'resolver';
await resolveCase(caseData.id, {
action: actionName,
action: 'approved',
payload: formData,
});
+289 -54
View File
@@ -1,11 +1,15 @@
import { useEffect, useCallback, useRef, type ReactNode } from 'react';
import { useEffect, useCallback, useRef, useState, type ReactNode } from 'react';
import { useLocation, useNavigate } from 'react-router-dom';
import { auth } from '@/services/auth';
import { wsClient } from '@/services/wsClient';
import { useAppStore } from '@/store/useAppStore';
import { streamBuffer } from '@/services/streamBuffer';
import { api } from '@/services/api';
import { useAppStore, type ConversationState } from '@/store/useAppStore';
import { useNotification } from '@/hooks/useNotification';
import { useSound } from '@/hooks/useSound';
import { useTitleFlash } from '@/hooks/useTitleFlash';
import type { WSEnvelope } from '@/types/wsProtocol';
import { CaseStatus, type CaseRequest } from '@/types';
import Header from '@/components/layout/Header';
import Sidebar from '@/components/layout/Sidebar';
@@ -24,15 +28,23 @@ interface AppShellProps {
export function AppShell({ children }: AppShellProps) {
const isDarkMode = useAppStore((s) => s.isDarkMode);
const fetchCases = useAppStore((s) => s.fetchCases);
const fetchConversations = useAppStore((s) => s.fetchConversations);
const setWsStatus = useAppStore((s) => s.setWsStatus);
const upsertCase = useAppStore((s) => s.upsertCase);
const upsertConversation = useAppStore((s) => s.upsertConversation);
const addMessage = useAppStore((s) => s.addMessage);
const appendToken = useAppStore((s) => s.appendToken);
const completeStream = useAppStore((s) => s.completeStream);
const setResolvedCaseAlert = useAppStore((s) => s.setResolvedCaseAlert);
const setConversations = useAppStore((s) => s.setConversations);
const setCases = useAppStore((s) => s.setCases);
const setInitStateReceived = useAppStore((s) => s.setInitStateReceived);
const addProcessedEventId = useAppStore((s) => s.addProcessedEventId);
const setConversationState = useAppStore((s) => s.setConversationState);
const setConversationEndedBanner = useAppStore((s) => s.setConversationEndedBanner);
const initStateReceived = useAppStore((s) => s.initStateReceived);
const location = useLocation();
const navigate = useNavigate();
const [authReady, setAuthReady] = useState(false);
// ── Hooks for preserved features (Paso 9) ──────────────────
const { notify } = useNotification();
@@ -42,39 +54,81 @@ export function AppShell({ children }: AppShellProps) {
// ── Incoming WebSocket message handler ─────────────────────
const handleIncomingMessage = useCallback(
(envelope: WSEnvelope) => {
const { type, payload } = envelope;
const { type, payload, eventId } = envelope;
// ══════════════════════════════════════════════════════════
// Paso 5: Idempotencia — descartar eventos duplicados
// ══════════════════════════════════════════════════════════
if (eventId) {
if (!addProcessedEventId(eventId)) {
if (import.meta.env.DEV) {
console.debug('[WS] Duplicate eventId ignored:', eventId);
}
return;
}
}
switch (type) {
// ── Full state sync on (re)connect ──────────────────
// ── Full state sync on (re)connect (Paso 2) ─────────
case 'init_state': {
const conversations = payload.conversations;
if (Array.isArray(conversations)) {
for (const conv of conversations) {
upsertConversation(conv as any);
}
}
const activeCases = payload.activeCases;
if (Array.isArray(activeCases)) {
for (const c of activeCases) {
upsertCase(c as any);
}
if (Array.isArray(conversations)) {
setConversations(conversations as any[]);
}
if (Array.isArray(activeCases)) {
setCases(activeCases as any[]);
}
// Mark init_state as received
setInitStateReceived(true);
break;
}
// ── New conversation started ────────────────────────
case 'conversation_started': {
const conv = payload.conversation;
if (conv) {
upsertConversation(conv as any);
const startedConvId = payload.conversationId as string;
if (startedConvId && typeof startedConvId === 'string' && startedConvId.length > 0) {
// Check for duplicate before upserting
const existing = useAppStore.getState().conversations.find((c) => c.id === startedConvId);
if (!existing) {
upsertConversation({
id: startedConvId,
clientId: '',
agentId: (payload.agentId as string) ?? '',
status: 'active',
createdAt: new Date().toISOString(),
});
}
}
break;
}
// ── Conversation ended ──────────────────────────────
// ── Conversation ended (Paso 3) ─────────────────────
case 'conversation_ended': {
// The store could mark the conversation as ended;
// currently handled on next init_state sync.
const endedConvId = payload.conversationId as string | undefined;
if (!endedConvId || typeof endedConvId !== 'string') break;
const convs = useAppStore.getState().conversations;
const idx = convs.findIndex((c) => c.id === endedConvId);
if (idx >= 0) {
const updated = [...convs];
updated[idx] = { ...updated[idx], status: 'ended' as const };
useAppStore.setState({ conversations: updated });
// Si está seleccionada, mostrar banner "Conversación finalizada"
const selConvId = useAppStore.getState().selectedConversationId;
if (selConvId === endedConvId) {
useAppStore.setState({ conversationEndedBanner: endedConvId });
}
// Update state machine
setConversationState(endedConvId, 'completed');
}
// Limpiar buffer para esta conversación (Paso 4)
streamBuffer.clear(endedConvId);
break;
}
@@ -88,7 +142,35 @@ export function AppShell({ children }: AppShellProps) {
break;
}
// ── Agent streaming: chunk ──────────────────────────
// ── Agent streaming: started ───────────────────────
case 'agent_stream_started': {
const streamConvId = payload.conversationId as string | undefined;
const streamMsgId = payload.messageId as string | undefined;
if (streamConvId && streamMsgId) {
// Update state machine: if hydrating, transition to streaming
const currentState = useAppStore.getState().conversationStates[streamConvId];
if (currentState === 'hydrating') {
setConversationState(streamConvId, 'streaming');
}
// Only create placeholder if conversation is selected
const selConv = useAppStore.getState().selectedConversation;
if (selConv && selConv.id === streamConvId) {
useAppStore.getState().addMessage(streamConvId, {
id: streamMsgId,
conversationId: streamConvId,
role: 'agent' as any,
content: '',
timestamp: new Date().toISOString(),
isStreaming: true,
});
}
}
break;
}
// ── Agent streaming: chunk (Paso 1) ─────────────────
case 'agent_stream_chunk': {
const chunkConvId = payload.conversationId as string | undefined;
const msgId = payload.messageId as string | undefined;
@@ -96,19 +178,38 @@ export function AppShell({ children }: AppShellProps) {
const index = payload.index as number | undefined;
if (chunkConvId && msgId && token !== undefined && index !== undefined) {
appendToken(chunkConvId, msgId, token, index);
// Paso 1: Use selectedConversation (not selectedConversationId) for routing
const selConv = useAppStore.getState().selectedConversation;
if (selConv && selConv.id === chunkConvId) {
// Conversación cargada → stream directo al store
appendToken(chunkConvId, msgId, token, index);
} else {
// Conversación NO cargada (o null) → buffer externo
streamBuffer.addToken(chunkConvId, msgId, token, index);
}
}
break;
}
// ── Agent streaming: complete ───────────────────────
// ── Agent streaming: complete (Fix #1: gatear buffer clear al merge) ──
case 'agent_stream_completed': {
const compConvId = payload.conversationId as string | undefined;
const compMsgId = payload.messageId as string | undefined;
const fullContent = payload.fullContent as string | undefined;
if (compConvId && compMsgId && fullContent !== undefined) {
completeStream(compConvId, compMsgId, fullContent);
const sel = useAppStore.getState().selectedConversation;
if (sel && sel.id === compConvId) {
// Conversación cargada → merge exitoso al store
completeStream(compConvId, compMsgId, fullContent);
setConversationState(compConvId, 'completed');
// Limpiar solo ESTE stream del buffer (no toda la conversación)
streamBuffer.clearMessage(compConvId, compMsgId);
}
// Si sel es null (loadingConversation), NO limpiar —
// handleConversationClick mergeará el buffer más tarde.
// El TTL del buffer (60s) garantiza limpieza eventual si nunca se abre.
}
break;
}
@@ -124,20 +225,27 @@ export function AppShell({ children }: AppShellProps) {
// HITL Request — trigger all preserved features
// ═══════════════════════════════════════════════════
case 'hitl_request': {
const caseData = payload.case as
| Record<string, unknown>
| undefined;
const caseTitle = (payload.title as string) || 'Nuevo caso';
const caseDescription = 'Se requiere intervención humana';
const caseData = {
id: payload.id as number,
title: caseTitle,
description: caseDescription,
status: (payload.status as string) || 'PENDING',
tipoSolicitud: (payload.tipoSolicitud as string) || '',
uiPattern: (payload.uiPattern as string) || 'SIMPLE_CONFIRMATION',
applicative: '',
payload: {},
handlingTime: 0,
createdAt: new Date().toISOString(),
externalId: (payload.correlationId as string) || undefined,
};
const caseTitle: string =
(caseData?.title as string) ?? 'Nuevo caso HITL';
const caseDescription: string =
(caseData?.description as string) ??
'Se requiere intervención humana';
upsertCase(caseData as any);
// 1) Desktop notification — click handler navigates to /cases
notify(caseTitle, caseDescription, () => {
const caseId = (caseData?.id ?? payload.conversationId) as string | number;
useAppStore.setState({ selectedCaseId: caseId });
useAppStore.setState({ selectedCaseId: payload.id as string | number });
navigate('/cases');
});
@@ -146,18 +254,103 @@ export function AppShell({ children }: AppShellProps) {
// 3) Flash the tab title if the tab is hidden
triggerNotification();
// 4) Insert the new case into the store
if (caseData) {
upsertCase(caseData as any);
}
break;
}
// ── Case resolved (broadcast) ───────────────────────
case 'hitl_resolved': {
// The store could update the case status here;
// the authoritative update comes via REST polling as well.
const resolvedCaseId = payload.caseId as number | string;
const existingCase = useAppStore.getState().cases.find(
(c) => c.id === resolvedCaseId,
);
if (existingCase) {
// Actualizar estado a RESOLVED
upsertCase({
...existingCase,
status: CaseStatus.RESOLVED,
} as CaseRequest);
}
// Si el caso está seleccionado, mostrar alerta temporal
const selectedId = useAppStore.getState().selectedCaseId;
if (selectedId === resolvedCaseId) {
setResolvedCaseAlert({
caseId: resolvedCaseId,
caseTitle: existingCase?.title || 'Caso',
});
}
break;
}
// ── Internal note re-diffused by server ──────────────
case 'internal_note': {
const noteConvId = payload.conversationId as string | undefined;
const noteMessage = payload.message as
| Record<string, unknown>
| undefined;
if (noteConvId && noteMessage) {
const selectedConvId = useAppStore.getState().selectedConversationId;
if (selectedConvId === noteConvId) {
const sel = useAppStore.getState().selectedConversation;
if (sel && sel.id === noteConvId) {
useAppStore.setState({
selectedConversation: {
...sel,
messages: [
...sel.messages,
{
id: noteMessage.id as string,
conversationId: noteConvId,
role: 'internal' as any,
content: noteMessage.content as string,
timestamp:
(noteMessage.timestamp as string) ||
new Date().toISOString(),
},
],
},
});
}
}
}
break;
}
// ── Conversation assigned to advisor ────────────────
case 'conversation_assigned': {
const assignedConvId = payload.conversationId as string;
if (assignedConvId && typeof assignedConvId === 'string' && assignedConvId.length > 0) {
// Check if conversation already exists in store (avoid duplicates)
const existing = useAppStore.getState().conversations.find((c) => c.id === assignedConvId);
if (existing) break; // Already in list, skip
// Fetch full conversation data via REST y upsert en store
api
.getConversation(assignedConvId)
.then((conv) => {
upsertConversation({
id: conv.id,
clientId: conv.clientId || '',
agentId: conv.agentId || '',
status: (conv.status as 'active' | 'paused' | 'ended') || 'active',
createdAt: conv.createdAt || new Date().toISOString(),
});
})
.catch((err) => {
console.error(
'[WS] Failed to fetch assigned conversation:',
err,
);
});
}
break;
}
// ── Heartbeat — connection health ───────────────────
case 'heartbeat': {
// Connection health — no UI action needed
break;
}
@@ -187,6 +380,12 @@ export function AppShell({ children }: AppShellProps) {
addMessage,
appendToken,
completeStream,
setResolvedCaseAlert,
setConversations,
setCases,
setInitStateReceived,
addProcessedEventId,
setConversationState,
navigate,
],
);
@@ -206,33 +405,69 @@ export function AppShell({ children }: AppShellProps) {
}
}, [isDarkMode]);
// ── Initialize WebSocket connection and data fetching ─────
// ── Auth check + redirect ─────────────────────────────────
// On mount, verify authentication. If not authenticated and
// not already on /login, redirect to /login.
useEffect(() => {
// Set up WebSocket status sync
const isLoginPage = location.pathname === '/login';
if (!isLoginPage && !auth.isAuthenticated()) {
navigate('/login', { replace: true });
}
}, [location.pathname, navigate]);
// ── Initialize WebSocket with In-Band Auth (Paso 0) ──────
useEffect(() => {
// Only initialize WS if authenticated
if (!auth.isAuthenticated()) return;
// Paso 0: First set up status callback
wsClient.onStatusChange = (status) => {
setWsStatus(status);
// On disconnect, reset init_state flag and clear buffers
if (status === 'disconnected' || status === 'reconnecting') {
setInitStateReceived(false);
streamBuffer.clearAll();
// Clear processed events on reconnect (Paso 5 — invalidate by epoch)
useAppStore.getState().clearProcessedEventIds();
// Reset conversation ended banner
useAppStore.setState({ conversationEndedBanner: null });
}
};
// Connect WebSocket
wsClient.connect();
// Set up incoming message handler (delegates through ref)
wsClient.onMessage = (envelope) => {
handleIncomingMessageRef.current(envelope);
// Paso 5: Set up authenticated callback — register business handlers only after auth
wsClient.onAuthenticated = () => {
setAuthReady(true);
// Register business event handler only after authentication
wsClient.onMessage = (envelope) => {
handleIncomingMessageRef.current(envelope);
};
};
// Initial data fetch based on route
if (location.pathname.startsWith('/cases')) {
fetchCases();
} else if (location.pathname.startsWith('/monitor')) {
fetchConversations();
// Connect WebSocket (guard against double-connect in StrictMode dev)
if (wsClient.getStatus() === 'disconnected') {
wsClient.connect();
}
// Paso 2: Fallback — if init_state doesn't arrive within 5s, show "Conectando..." UI
// (handled via initStateReceived flag in the store; MonitorPage checks this)
const initTimeout = setTimeout(() => {
if (!useAppStore.getState().initStateReceived) {
console.warn('[AppShell] init_state not received within 5s — showing connecting UI');
// The store flag remains false; MonitorPage reads it to show "Conectando..."
}
}, 5_000);
// Cleanup on unmount
return () => {
clearTimeout(initTimeout);
wsClient.onStatusChange = null;
wsClient.onAuthenticated = null;
wsClient.onMessage = null;
wsClient.disconnect();
setInitStateReceived(false);
streamBuffer.clearAll();
useAppStore.getState().clearProcessedEventIds();
setAuthReady(false);
};
// NOTE: intentionally running only on mount; route changes handled by pages
// eslint-disable-next-line react-hooks/exhaustive-deps
+33 -2
View File
@@ -1,5 +1,7 @@
import { Sun, Moon } from 'lucide-react';
import { Sun, Moon, LogOut } from 'lucide-react';
import { useAppStore } from '@/store/useAppStore';
import { auth } from '@/services/auth';
import { useNavigate } from 'react-router-dom';
// ─────────────────────────────────────────────────────────────
// Helpers
@@ -28,9 +30,18 @@ export default function Header() {
const isDarkMode = useAppStore((s) => s.isDarkMode);
const toggleDarkMode = useAppStore((s) => s.toggleDarkMode);
const wsStatus = useAppStore((s) => s.wsStatus);
const navigate = useNavigate();
const session = auth.getSession();
const fullName = session?.fullName ?? null;
const { dot: dotColor, label: wsLabel } = wsStatusConfig(wsStatus);
function handleLogout(): void {
auth.logout();
navigate('/login', { replace: true });
}
return (
<header className="flex items-center justify-between h-[50px] px-4 border-b border-border bg-surface shadow-sm shrink-0">
{/* ── Left: Logo ──────────────────────────────────────── */}
@@ -49,7 +60,7 @@ export default function Header() {
</span>
</div>
{/* ── Right: WS indicator + Theme toggle ──────────────── */}
{/* ── Right: WS indicator + Theme toggle + User info ─── */}
<div className="flex items-center gap-3">
{/* WebSocket status */}
<div className="flex items-center gap-1.5 text-[11px] text-text-muted">
@@ -61,6 +72,26 @@ export default function Header() {
<span>{wsLabel}</span>
</div>
{/* User full name */}
{fullName && (
<span className="text-[12px] text-text-primary font-medium max-w-[160px] truncate">
{fullName}
</span>
)}
{/* Logout button */}
<button
type="button"
onClick={handleLogout}
className="flex items-center justify-center w-[28px] h-[28px] rounded-md
text-text-muted hover:text-accent-red hover:bg-accent-red/5
transition-colors"
aria-label="Cerrar sesión"
title="Cerrar sesión"
>
<LogOut size={15} />
</button>
{/* Dark mode toggle */}
<button
type="button"
+8 -16
View File
@@ -1,12 +1,12 @@
import { Loader2, User } from 'lucide-react';
import type { Conversation } from '@/types';
import type { ConversationSummary } from '@/types';
// ─────────────────────────────────────────────────────────────
// Props
// ─────────────────────────────────────────────────────────────
interface ConversationCardProps {
conversation: Conversation;
conversation: ConversationSummary;
isActive: boolean;
onClick: () => void;
}
@@ -15,27 +15,19 @@ interface ConversationCardProps {
// Helpers
// ─────────────────────────────────────────────────────────────
function getLastMessage(conversation: Conversation): string {
if (conversation.messages.length === 0) return 'Sin mensajes';
const last = conversation.messages[conversation.messages.length - 1];
const truncated =
last.content.length > 80
? last.content.slice(0, 80) + '...'
: last.content;
return truncated;
function getClientLabel(conversation: ConversationSummary): string {
return conversation.clientId || 'Sin identificar';
}
function isLastMessageStreaming(conversation: Conversation): boolean {
if (conversation.messages.length === 0) return false;
const last = conversation.messages[conversation.messages.length - 1];
return last.isStreaming === true;
function isLastMessageStreaming(_conversation: ConversationSummary): boolean {
return false;
}
// ─────────────────────────────────────────────────────────────
// Status label helper
// ─────────────────────────────────────────────────────────────
function statusLabel(status: Conversation['status']): string {
function statusLabel(status: ConversationSummary['status']): string {
switch (status) {
case 'active':
return 'Activa';
@@ -57,7 +49,7 @@ export default function ConversationCard({
isActive,
onClick,
}: ConversationCardProps) {
const lastMsg = getLastMessage(conversation);
const lastMsg = getClientLabel(conversation);
const streaming = isLastMessageStreaming(conversation);
return (
+378
View File
@@ -0,0 +1,378 @@
// @vitest-environment jsdom
import { describe, it, expect, vi, beforeAll, beforeEach, afterEach } from 'vitest';
import { render, screen, act, cleanup } from '@testing-library/react';
import Timer, { TimerHandle, clearTimerStorage } from './Timer';
// ─────────────────────────────────────────────────────────────
// localStorage mock (jsdom no está instalado en el proyecto)
// ─────────────────────────────────────────────────────────────
function createLocalStorageMock(): Storage {
let store: Record<string, string> = {};
return {
getItem: (key: string) => store[key] ?? null,
setItem: (key: string, value: string) => { store[key] = value; },
removeItem: (key: string) => { delete store[key]; },
clear: () => { store = {}; },
get length() { return Object.keys(store).length; },
key: (index: number) => Object.keys(store)[index] ?? null,
};
}
beforeAll(() => {
vi.stubGlobal('localStorage', createLocalStorageMock());
});
// ─────────────────────────────────────────────────────────────
// Helpers
// ─────────────────────────────────────────────────────────────
function createRef(): { current: TimerHandle | null } {
return { current: null };
}
function renderTimer(caseId: string | number = 'case-123') {
const ref = createRef();
const result = render(<Timer ref={ref} caseId={caseId} />);
return { ref, ...result };
}
const CASE_ID = 'test-case-1';
const STORAGE_KEY = `timer_case_${CASE_ID}`;
describe('Timer — Cronómetro (Fase 9, CA-12..CA-15)', () => {
beforeEach(() => {
vi.useFakeTimers();
localStorage.clear();
});
afterEach(() => {
vi.useRealTimers();
cleanup();
});
// ── CA-12: Display inmediato en start() ───────────────────
describe('CA-12: Display inmediato al llamar start()', () => {
it('debe mostrar accumulatedRef inmediatamente sin esperar el primer tick', () => {
const { ref } = renderTimer(CASE_ID);
// Simular que ya hay tiempo acumulado en localStorage
localStorage.setItem(
STORAGE_KEY,
JSON.stringify({ startTimestamp: 0, accumulated: 42 }),
);
// Re-render para que el useEffect cargue el stored value
cleanup();
const ref2 = createRef();
render(<Timer ref={ref2} caseId={CASE_ID} />);
// start() debe setear display inmediatamente
act(() => {
ref2.current!.start();
});
// displaySeconds se actualizó sincrónicamente → muestra 00:42
expect(screen.getByText('00:42')).toBeDefined();
});
it('debe mostrar 00:00 inmediatamente si no hay tiempo acumulado', () => {
const { ref } = renderTimer(CASE_ID);
act(() => {
ref.current!.start();
});
// Sin accumulated, arranca en 00:00
const display = screen.getByText('00:00');
expect(display).toBeDefined();
});
it('debe actualizar el display después del primer tick del intervalo', () => {
const { ref } = renderTimer(CASE_ID);
act(() => {
ref.current!.start();
});
// Avanzar 1s → el display debe pasar de 00:00 a 00:01
act(() => {
vi.advanceTimersByTime(1000);
});
expect(screen.getByText('00:01')).toBeDefined();
});
});
// ── CA-13: Tick cada segundo ──────────────────────────────
describe('CA-13: Tick cada segundo', () => {
it('debe avanzar el display cada segundo (00:00 → 00:01 → 00:02)', () => {
const { ref } = renderTimer(CASE_ID);
act(() => {
ref.current!.start();
});
expect(screen.getByText('00:00')).toBeDefined();
act(() => {
vi.advanceTimersByTime(1000);
});
expect(screen.getByText('00:01')).toBeDefined();
act(() => {
vi.advanceTimersByTime(1000);
});
expect(screen.getByText('00:02')).toBeDefined();
act(() => {
vi.advanceTimersByTime(3000);
});
expect(screen.getByText('00:05')).toBeDefined();
});
it('debe acumular sobre tiempo previo almacenado', () => {
// Simular 30s acumulados antes de start()
localStorage.setItem(
STORAGE_KEY,
JSON.stringify({ startTimestamp: 0, accumulated: 30 }),
);
const { ref } = renderTimer(CASE_ID);
// accumulated se carga desde localStorage
act(() => {
ref.current!.start();
});
// Display muestra 00:30 inmediatamente (CA-12)
expect(screen.getByText('00:30')).toBeDefined();
// Avanza 1s → 00:31
act(() => {
vi.advanceTimersByTime(1000);
});
expect(screen.getByText('00:31')).toBeDefined();
});
});
// ── CA-14: Persistencia en stop (desmontaje) ─────────────
describe('CA-14: Persistencia en localStorage al desmontar', () => {
it('debe persistir accumulated via stop() al desmontar el componente', () => {
const { ref, unmount } = renderTimer(CASE_ID);
act(() => {
ref.current!.start();
});
// Dejar correr 3s
act(() => {
vi.advanceTimersByTime(3000);
});
// Desmontar (el cleanup llama a stop())
unmount();
// Verificar que localStorage tiene el tiempo acumulado
const stored = JSON.parse(localStorage.getItem(STORAGE_KEY)!);
expect(stored).not.toBeNull();
expect(stored.accumulated).toBe(3);
expect(stored.startTimestamp).toBe(0); // stop setea startTimestamp a 0
});
it('debe persistir con la key correcta timer_case_{caseId}', () => {
const customCaseId = 'custom-999';
const customKey = `timer_case_${customCaseId}`;
const { ref, unmount } = renderTimer(customCaseId);
act(() => {
ref.current!.start();
});
act(() => {
vi.advanceTimersByTime(5000);
});
unmount();
const stored = JSON.parse(localStorage.getItem(customKey)!);
expect(stored).not.toBeNull();
expect(stored.accumulated).toBe(5);
});
it('debe persistir solo el caso correcto al cambiar de caso', () => {
const ref1 = createRef();
const { unmount: unmount1 } = render(
<Timer ref={ref1} caseId="case-A" />,
);
act(() => {
ref1.current!.start();
});
act(() => {
vi.advanceTimersByTime(10_000);
});
// Desmontar caso A
unmount1();
// Montar caso B
const ref2 = createRef();
render(<Timer ref={ref2} caseId="case-B" />);
act(() => {
ref2.current!.start();
});
act(() => {
vi.advanceTimersByTime(5000);
});
act(() => {
ref2.current!.stop();
});
// Caso A debe tener 10s
const storedA = JSON.parse(localStorage.getItem('timer_case_case-A')!);
expect(storedA.accumulated).toBe(10);
// Caso B debe tener 5s
const storedB = JSON.parse(localStorage.getItem('timer_case_case-B')!);
expect(storedB.accumulated).toBe(5);
});
});
// ── CA-15: StrictMode — cleanup resetea isRunningRef ─────
describe('CA-15: StrictMode — cleanup resetea isRunningRef', () => {
it('debe permitir start() después de unmount/remount simulado', () => {
// Simular ciclo de StrictMode: render → cleanup → render
const ref = createRef();
const { unmount } = render(<Timer ref={ref} caseId={CASE_ID} />);
// Primer start
act(() => {
ref.current!.start();
});
expect(screen.getByText('00:00')).toBeDefined();
avanza: act(() => {
vi.advanceTimersByTime(2000);
});
expect(screen.getByText('00:02')).toBeDefined();
// Simular unmount de StrictMode (cleanup)
unmount();
// Volver a montar (simulando el segundo render de StrictMode)
const ref2 = createRef();
render(<Timer ref={ref2} caseId={CASE_ID} />);
// Segundo start() debe funcionar (isRunningRef se reseteó a false)
act(() => {
ref2.current!.start();
});
// Verificar que avanza después de re-mount
act(() => {
vi.advanceTimersByTime(1000);
});
// Accumulated del localStorage (2s) + 1s del nuevo intervalo = 3
expect(screen.getByText('00:03')).toBeDefined();
});
it('debe persistir accumulated entre StrictMode ciclos', () => {
const ref = createRef();
const { unmount } = render(<Timer ref={ref} caseId={CASE_ID} />);
act(() => {
ref.current!.start();
});
act(() => {
vi.advanceTimersByTime(7000); // 7 segundos
});
// Unmount (cleanup llama a stop → persiste 7s)
unmount();
// Re-mount (restaura accumulated = 7s de localStorage)
const ref2 = createRef();
render(<Timer ref={ref2} caseId={CASE_ID} />);
act(() => {
ref2.current!.start();
});
act(() => {
vi.advanceTimersByTime(3000); // 3s adicionales
});
// Total: 7 + 3 = 10s
expect(screen.getByText('00:10')).toBeDefined();
});
});
// ── Regresión: clearTimerStorage y getElapsed ────────────
describe('Regresión: utilidades del timer', () => {
it('clearTimerStorage debe eliminar la key de localStorage', () => {
localStorage.setItem(STORAGE_KEY, JSON.stringify({ accumulated: 10, startTimestamp: 0 }));
clearTimerStorage(CASE_ID);
expect(localStorage.getItem(STORAGE_KEY)).toBeNull();
});
it('getElapsed debe retornar accumulated cuando está detenido', () => {
const ref = createRef();
render(<Timer ref={ref} caseId={CASE_ID} />);
expect(ref.current!.getElapsed()).toBe(0);
});
it('getElapsed debe retornar accumulated + elapsed cuando está running', () => {
const ref = createRef();
render(<Timer ref={ref} caseId={CASE_ID} />);
act(() => {
ref.current!.start();
});
act(() => {
vi.advanceTimersByTime(5000);
});
expect(ref.current!.getElapsed()).toBe(5);
act(() => {
ref.current!.stop();
});
// Detenido, getElapsed debe retornar solo accumulated
expect(ref.current!.getElapsed()).toBe(5);
});
it('stop() debe ser idempotente (no falla al llamarse dos veces)', () => {
const ref = createRef();
render(<Timer ref={ref} caseId={CASE_ID} />);
act(() => {
ref.current!.start();
});
act(() => {
vi.advanceTimersByTime(3000);
});
// Primera stop
act(() => {
ref.current!.stop();
});
const elapsedAfterFirstStop = ref.current!.getElapsed();
// Segunda stop (idempotente)
act(() => {
ref.current!.stop();
});
expect(ref.current!.getElapsed()).toBe(elapsedAfterFirstStop);
});
});
});
+18 -10
View File
@@ -104,16 +104,6 @@ const Timer = forwardRef<TimerHandle, TimerProps>(({ caseId }, ref) => {
}
}, [caseId]);
// ── Cleanup on unmount ────────────────────────────────────
useEffect(() => {
return () => {
if (intervalRef.current) {
clearInterval(intervalRef.current);
intervalRef.current = null;
}
};
}, []);
// ── Imperative API ────────────────────────────────────────
const start = useCallback(() => {
@@ -128,6 +118,9 @@ const Timer = forwardRef<TimerHandle, TimerProps>(({ caseId }, ref) => {
accumulated: accumulatedRef.current,
});
// Mostrar valor actual INMEDIATAMENTE (sin esperar 1s al primer tick)
setDisplaySeconds(accumulatedRef.current);
intervalRef.current = setInterval(() => {
if (startTimestampRef.current === null) return;
const elapsed = Math.floor(
@@ -182,6 +175,21 @@ const Timer = forwardRef<TimerHandle, TimerProps>(({ caseId }, ref) => {
getElapsed,
]);
// ── Cleanup on unmount ────────────────────────────────────
useEffect(() => {
return () => {
// 1. Primero: persistir estado vía stop() (ya es idempotente)
stop();
// 2. Segundo: anular el intervalo (por si stop no lo hizo)
if (intervalRef.current) {
clearInterval(intervalRef.current);
intervalRef.current = null;
}
// 3. Tercero: resetear bandera (para StrictMode)
isRunningRef.current = false;
};
}, [stop]);
// ── Render ────────────────────────────────────────────────
return (
+68
View File
@@ -0,0 +1,68 @@
import { useState, useEffect } from 'react';
import { auth } from '@/services/auth';
// ─────────────────────────────────────────────────────────────
// Types
// ─────────────────────────────────────────────────────────────
export type AuthStatus = 'loading' | 'authenticated' | 'anonymous' | 'expired';
const CHECK_INTERVAL_MS = 30_000;
// ─────────────────────────────────────────────────────────────
// Hook
// ─────────────────────────────────────────────────────────────
/**
* Reactive hook that exposes the current authentication status.
*
* - 'loading': initial state while verifying session on mount
* - 'authenticated': valid session exists (token + expireDate valid)
* - 'anonymous': no session stored
* - 'expired': session exists but expireDate has passed
*/
export function useAuth() {
const [status, setStatus] = useState<AuthStatus>('loading');
useEffect(() => {
function checkAuth(): void {
const session = auth.getSession();
if (!session) {
setStatus('anonymous');
return;
}
// Check expireDate
const expireMs = new Date(session.expireDate).getTime();
if (isNaN(expireMs) || Date.now() >= expireMs) {
auth.logout();
setStatus('expired');
return;
}
// Verify token is still valid
const token = auth.getToken();
if (!token) {
setStatus('expired');
return;
}
setStatus('authenticated');
}
// Initial check
checkAuth();
// Periodic re-check every 30s to detect expiry
const interval = setInterval(checkAuth, CHECK_INTERVAL_MS);
return () => clearInterval(interval);
}, []);
return {
status,
isAuthenticated: status === 'authenticated',
isLoading: status === 'loading',
};
}
+22
View File
@@ -235,6 +235,28 @@ export const handlers = [
return HttpResponse.json(caseItem);
}),
// POST /api/v1/cases/:id/start
http.post('*/api/v1/cases/:id/start', async ({ params }) => {
await delay(200);
const id = parseInt(params.id as string);
const index = mockCases.findIndex((c) => c.id === id);
if (index === -1) {
return new HttpResponse(null, { status: 404 });
}
if (mockCases[index].status !== 'PENDING') {
return new HttpResponse(null, { status: 400 });
}
mockCases[index] = {
...mockCases[index],
status: 'IN_PROGRESS',
};
return HttpResponse.json(mockCases[index]);
}),
// POST /api/v1/cases/:id/resolve
http.post('*/api/v1/cases/:id/resolve', async ({ params, request }) => {
await delay(300);
+222 -16
View File
@@ -1,11 +1,83 @@
import { useEffect, useMemo } from 'react';
import { MessageSquare } from 'lucide-react';
import { useCallback } from 'react';
import { MessageSquare, RefreshCw, Loader2 } from 'lucide-react';
import { useAppStore } from '@/store/useAppStore';
import { wsClient } from '@/services/wsClient';
import { streamBuffer } from '@/services/streamBuffer';
import ConversationCard from '@/components/monitor/ConversationCard';
import ChatFeed from '@/components/monitor/ChatFeed';
import InternalNoteBanner from '@/components/monitor/InternalNoteBanner';
import EmptyState from '@/components/shared/EmptyState';
// ─────────────────────────────────────────────────────────────
// ConnectingPlaceholder — shown when init_state hasn't arrived
// ─────────────────────────────────────────────────────────────
function ConnectingPlaceholder() {
const wsStatus = useAppStore((s) => s.wsStatus);
const handleRetry = () => {
wsClient.disconnect();
wsClient.connect();
};
return (
<div className="flex items-center justify-center h-full">
<div className="flex flex-col items-center gap-3 text-text-muted">
<Loader2 size={32} className="animate-spin text-accent-orange" />
<p className="text-[13px] font-medium">Conectando...</p>
<p className="text-[11px]">Esperando datos del servidor</p>
<span className="text-[10px] text-text-disabled">
Estado: {wsStatus}
</span>
<button
type="button"
onClick={handleRetry}
className="flex items-center gap-1.5 px-3 py-1.5 mt-1 text-[12px] font-medium
text-accent-blue border border-accent-blue/30 rounded-lg
hover:bg-accent-blue/5 transition-colors"
>
<RefreshCw size={12} />
Reintentar conexión
</button>
</div>
</div>
);
}
// ─────────────────────────────────────────────────────────────
// ConversationEndedBanner
// ─────────────────────────────────────────────────────────────
function ConversationEndedBanner({ conversationId }: { conversationId: string }) {
const conversationEndedBanner = useAppStore((s) => s.conversationEndedBanner);
if (conversationEndedBanner !== conversationId) return null;
return (
<div className="shrink-0 px-4 py-2 bg-accent-yellow/10 border-b border-accent-yellow/20">
<p className="text-[12px] font-medium text-accent-yellow-dark flex items-center gap-2">
<span className="w-1.5 h-1.5 rounded-full bg-accent-yellow" />
Conversación finalizada
</p>
</div>
);
}
// ─────────────────────────────────────────────────────────────
// LoadingConversationOverlay
// ─────────────────────────────────────────────────────────────
function LoadingConversationOverlay() {
return (
<div className="flex-1 flex items-center justify-center">
<div className="flex flex-col items-center gap-2 text-text-muted">
<Loader2 size={24} className="animate-spin text-accent-orange" />
<p className="text-[13px]">Cargando conversación...</p>
</div>
</div>
);
}
// ─────────────────────────────────────────────────────────────
// Component
// ─────────────────────────────────────────────────────────────
@@ -15,23 +87,138 @@ export default function MonitorPage() {
const conversations = useAppStore((s) => s.conversations);
const selectedConversationId = useAppStore((s) => s.selectedConversationId);
const fetchConversations = useAppStore((s) => s.fetchConversations);
const fetchConversationWithMessages = useAppStore((s) => s.fetchConversationWithMessages);
const setSelectedConversationId = useAppStore((s) => s.setSelectedConversationId);
const totalConversations = useAppStore((s) => s.totalConversations);
const conversationsOffset = useAppStore((s) => s.conversationsOffset);
const selectedConversation = useAppStore((s) => s.selectedConversation);
const loadingConversation = useAppStore((s) => s.loadingConversation);
const setConversationState = useAppStore((s) => s.setConversationState);
const initStateReceived = useAppStore((s) => s.initStateReceived);
// ── Fetch conversations on mount ────────────────────────
useEffect(() => {
fetchConversations();
}, [fetchConversations]);
// ── No fetchConversations on mount — list comes only from WS init_state (Paso 2)
// ── Selected conversation object ─────────────────────────
const selectedConversation = useMemo(
() =>
conversations.find((c) => c.id === selectedConversationId) ?? null,
[conversations, selectedConversationId],
// ── Conversation click handler (Paso 6) ──────────────────
const handleLoadMore = useCallback(() => {
fetchConversations(20, conversationsOffset);
}, [fetchConversations, conversationsOffset]);
const handleConversationClick = useCallback(
async (id: string) => {
// Generate requestId for correlation (Paso 6 — ignore stale responses)
const requestId = crypto.randomUUID();
// Update state machine: idle → hydrating (Paso 7)
setConversationState(id, 'hydrating');
// Set loading state BEFORE async fetch (no flicker — Paso 6)
useAppStore.setState({
selectedConversationId: id,
selectedConversation: null,
loadingConversation: id,
currentRequestId: requestId,
conversationEndedBanner: null,
});
try {
// 1. Cargar mensajes históricos vía REST
await fetchConversationWithMessages(id);
// Paso 6: Correlation check — if requestId changed, ignore stale response
const stateAfter = useAppStore.getState();
if (stateAfter.currentRequestId !== requestId) {
console.debug('[MonitorPage] Stale REST response ignored for', id);
return; // User switched conversation, discard
}
// 2. Verificar si hay streams en buffer para esta conversación
// getBufferEntry ahora retorna un ARRAY (multi-stream) — iterar sobre todos
const bufferedStreams = streamBuffer.getBufferEntry(id);
if (bufferedStreams && bufferedStreams.length > 0) {
const sel = useAppStore.getState().selectedConversation;
if (sel && sel.id === id) {
// Mergear cada stream completado en el store
let updatedMessages = [...sel.messages];
for (const stream of bufferedStreams) {
// Ordenar tokens por index y construir contenido completo
const sorted = [...stream.tokens].sort(
(a, b) => a.index - b.index,
);
const content = sorted.map((t) => t.token).join('');
// Verificar si ya existe un mensaje con ese messageId en el store
const existingIdx = updatedMessages.findIndex(
(m) => m.id === stream.messageId,
);
if (existingIdx >= 0) {
const existing = updatedMessages[existingIdx];
if (existing.isStreaming || !existing.content) {
updatedMessages[existingIdx] = {
...existing,
content: existing.content || content,
isStreaming: false,
};
}
} else {
// Insertar como nuevo mensaje (ya completo)
updatedMessages.push({
id: stream.messageId,
conversationId: id,
role: 'agent' as any,
content,
timestamp: new Date().toISOString(),
isStreaming: false,
});
}
}
useAppStore.setState({
selectedConversation: { ...sel, messages: updatedMessages },
});
}
}
// 3. Limpiar buffer para esta conversación
streamBuffer.clear(id);
// 4. Update state machine: check if stream is active
const currentState = useAppStore.getState().conversationStates[id];
if (currentState === 'hydrating') {
// No stream started during hydration → idle
setConversationState(id, 'idle');
}
// If stream already started (via agent_stream_started handler), state is already 'streaming'
} catch (err) {
console.error('[MonitorPage] Failed to load conversation:', err);
// Check correlation before resetting
const stateAfter = useAppStore.getState();
if (stateAfter.currentRequestId === requestId) {
setConversationState(id, 'idle');
}
} finally {
// Clear loading state if still current
const stateAfter = useAppStore.getState();
if (stateAfter.currentRequestId === requestId) {
useAppStore.setState({
loadingConversation: null,
currentRequestId: null,
});
}
}
},
[fetchConversationWithMessages, setConversationState],
);
// ── Conversation click handler ────────────────────────────
const handleConversationClick = (id: string) => {
useAppStore.setState({ selectedConversationId: id });
};
// ── Show connecting placeholder if init_state hasn't arrived ──
if (!initStateReceived) {
return <ConnectingPlaceholder />;
}
// Determine what to render in the right panel
const isCurrentlyLoading = loadingConversation !== null;
const showConversation = selectedConversation && !isCurrentlyLoading;
return (
<div className="flex h-full overflow-hidden">
@@ -68,13 +255,32 @@ export default function MonitorPage() {
/>
))
)}
{conversationsOffset < totalConversations && (
<div className="px-3 pb-3">
<button
type="button"
onClick={handleLoadMore}
className="w-full px-3 py-2 text-[12px] font-medium text-text-secondary
border border-border rounded-lg hover:bg-bg-hover
transition-colors"
>
Cargar más ({totalConversations - conversationsOffset} restantes)
</button>
</div>
)}
</div>
</aside>
{/* ── Right panel (flex-1) ──────────────────────────────── */}
<main className="flex-1 flex flex-col bg-bg-base overflow-hidden">
{selectedConversation ? (
{isCurrentlyLoading ? (
<LoadingConversationOverlay />
) : showConversation ? (
<>
{/* Conversation ended banner (Paso 3) */}
<ConversationEndedBanner conversationId={selectedConversation.id} />
{/* Chat header */}
<div className="shrink-0 px-4 py-2.5 border-b border-border bg-surface flex items-center gap-2">
<span className="text-[13px] font-semibold text-text-primary">
+56 -9
View File
@@ -1,11 +1,14 @@
import type { CaseRequest, Conversation } from '@/types';
import type { CaseRequest, Conversation, ConversationSummary } from '@/types';
import { auth } from '@/services/auth';
// ─────────────────────────────────────────────────────────────
// Configuration
// ─────────────────────────────────────────────────────────────
const API_BASE =
import.meta.env.VITE_API_BASE_URL || 'http://localhost:3000/api/v1';
import.meta.env.VITE_API_BASE_URL || 'http://localhost:5503/api/v1';
const ENABLE_MSW = import.meta.env.VITE_ENABLE_MSW === 'true';
// ─────────────────────────────────────────────────────────────
// Exported Interfaces
@@ -25,7 +28,7 @@ export interface CaseFilters {
}
// ─────────────────────────────────────────────────────────────
// HTTP Error Wrapper
// HTTP Error Wrappers
// ─────────────────────────────────────────────────────────────
export class ApiError extends Error {
@@ -40,6 +43,16 @@ export class ApiError extends Error {
}
}
/**
* Error thrown when the user is not authenticated or the session has expired.
*/
export class AuthError extends Error {
constructor(message: string) {
super(message);
this.name = 'AuthError';
}
}
// ─────────────────────────────────────────────────────────────
// Internal helpers
// ─────────────────────────────────────────────────────────────
@@ -48,16 +61,40 @@ async function request<T>(
path: string,
options?: RequestInit,
): Promise<T> {
// ── Auth interceptor ──────────────────────────────────────
// EXCLUDE: do not intercept /login (the exchange endpoint) or MSW mode
const isLoginPath = path.includes('/login');
if (!isLoginPath && !ENABLE_MSW) {
const token = auth.getToken();
if (!token) {
throw new AuthError('No autenticado');
}
}
const url = `${API_BASE}${path}`;
// Build headers: merge default Content-Type with auth headers and any custom headers
const authHeaders = !isLoginPath && !ENABLE_MSW ? auth.getAuthHeaders() : {};
const mergedHeaders: Record<string, string> = {
'Content-Type': 'application/json',
Accept: 'application/json',
...authHeaders,
...(options?.headers as Record<string, string> | undefined),
};
const response = await fetch(url, {
headers: {
'Content-Type': 'application/json',
Accept: 'application/json',
},
...options,
headers: mergedHeaders,
});
// ── 401 handling ──────────────────────────────────────────
// If the server returns 401 (unauthorized), clear session and throw AuthError.
// Only do this for non-login, non-MSW requests to avoid false positives.
if (response.status === 401 && !isLoginPath && !ENABLE_MSW) {
auth.logout();
throw new AuthError('Sesión expirada');
}
if (!response.ok) {
let errorMessage: string | undefined;
try {
@@ -134,11 +171,21 @@ export const api = {
});
},
/**
* Start a case (transition PENDING → IN_PROGRESS).
* No request body needed. Backend sets startedAt.
*/
async startCase(id: string | number): Promise<CaseRequest> {
return request<CaseRequest>(`/cases/${id}/start`, {
method: 'POST',
});
},
/**
* Fetch all active conversations.
*/
async getActiveConversations(): Promise<Conversation[]> {
return request<Conversation[]>('/conversations/active');
async getActiveConversations(limit: number = 20, offset: number = 0): Promise<{ items: ConversationSummary[]; total: number }> {
return request<{ items: ConversationSummary[]; total: number }>(`/conversations/active?limit=${limit}&offset=${offset}`);
},
/**
+438
View File
@@ -0,0 +1,438 @@
// ─────────────────────────────────────────────────────────────
// Auth Service — Okan → Linguo JWT authentication
// ─────────────────────────────────────────────────────────────
// ── Constants ─────────────────────────────────────────────────
const LINGUO_LOGIN_URL = import.meta.env.VITE_LOGIN_URL || 'https://vector.linguogpt.ai/login';
const OKAN_LOGIN_URL = 'https://apps.okan.tools/login';
const SESSION_KEY = 'claro-cases:session';
const POPUP_TIMEOUT_MS = 120_000;
const POPUP_POLL_MS = 500;
const EXP_SKEW_SEC = 60;
// ── Types ─────────────────────────────────────────────────────
export interface Session {
document: string;
fullName: string;
expireDate: string;
token: string;
storedAt: number;
}
// ── Helpers ───────────────────────────────────────────────────
/**
* Extract the payload segment of a JWT (base64url → JSON).
* Returns null on malformed input.
*/
function decodeJwtPayload(rawToken: string): Record<string, unknown> | null {
try {
const parts = rawToken.split('.');
if (parts.length !== 3) return null;
// Base64url decode (replace URL-safe chars, pad with =)
let base64 = parts[1].replace(/-/g, '+').replace(/_/g, '/');
while (base64.length % 4 !== 0) {
base64 += '=';
}
const decoded = atob(base64);
return JSON.parse(decoded) as Record<string, unknown>;
} catch {
return null;
}
}
/**
* Attempt to extract a username from common JWT claim fields.
*/
function extractUsername(payload: Record<string, unknown>): string | null {
const email = payload.email as string | undefined;
const preferredUsername = payload.preferred_username as string | undefined;
const sub = payload.sub as string | undefined;
if (email && typeof email === 'string') {
const atIndex = email.indexOf('@');
if (atIndex > 0) return email.slice(0, atIndex);
}
if (preferredUsername && typeof preferredUsername === 'string') {
return preferredUsername;
}
if (sub && typeof sub === 'string') {
// 'sub' might be a full name or email
const atIndex = sub.indexOf('@');
if (atIndex > 0) return sub.slice(0, atIndex);
return sub;
}
return null;
}
// ── Core Functions ────────────────────────────────────────────
// ───────────────────────────────────────────────────────────────
// openOkanPopup
// ───────────────────────────────────────────────────────────────
/**
* Open the Okan login page in a centered popup window (600×700).
* Returns null if the popup was blocked by the browser.
*/
function openOkanPopup(): Window | null {
const width = 600;
const height = 700;
const left = window.screenX + Math.max(0, (window.innerWidth - width) / 2);
const top = window.screenY + Math.max(0, (window.innerHeight - height) / 2);
const features = [
`width=${width}`,
`height=${height}`,
`left=${Math.round(left)}`,
`top=${Math.round(top)}`,
'menubar=no',
'toolbar=no',
'location=no',
'status=no',
'resizable=yes',
'scrollbars=yes',
].join(',');
let popup: Window | null = null;
try {
popup = window.open(OKAN_LOGIN_URL, 'okan-login', features);
} catch {
// window.open may throw in some environments
return null;
}
// If popup is null or closed immediately, it was blocked
if (!popup || popup.closed) {
return null;
}
return popup;
}
// ───────────────────────────────────────────────────────────────
// captureOkanToken
// ───────────────────────────────────────────────────────────────
/** localStorage key injected by the Linguo browser extension */
const OKAN_STORAGE_KEY = 'tokenOkan';
/**
* Check if the browser extension has already injected an Okan token
* into localStorage. Returns the raw token or null.
*/
function readExtensionToken(): string | null {
try {
const raw = localStorage.getItem(OKAN_STORAGE_KEY);
if (!raw) return null;
const parsed = JSON.parse(raw) as { value?: string };
return parsed?.value || null;
} catch {
return null;
}
}
/**
* Remove the injected token from localStorage after successful read.
*/
function clearExtensionToken(): void {
try {
localStorage.removeItem(OKAN_STORAGE_KEY);
} catch { /* ignore */ }
}
/**
* Capture the Okan token via the Linguo browser extension.
*
* Opens the Okan login popup to trigger the extension's content script,
* then polls localStorage for the injected `tokenOkan` key.
*
* The extension does the heavy lifting:
* 1. Content script runs on apps.okan.tools → captures JWT
* 2. Broadcasts to all tabs via chrome.tabs.sendMessage
* 3. Injects { value: "<token>" } into localStorage under "tokenOkan"
*
* Rejects if:
* - Popup blocked by browser
* - Popup closed before token captured
* - Timeout (120s)
*/
function captureOkanToken(timeoutMs: number = POPUP_TIMEOUT_MS): Promise<string> {
return new Promise<string>((resolve, reject) => {
// Check if token was already injected before opening popup
const existingToken = readExtensionToken();
if (existingToken) {
resolve(existingToken);
return;
}
const popup = openOkanPopup();
if (!popup) {
reject(new Error('popup_blocked'));
return;
}
let resolved = false;
// Poll localStorage for the token injected by the extension
const pollInterval = setInterval(() => {
if (popup.closed) {
// User may have completed login — check one more time before giving up
const token = readExtensionToken();
if (token) {
resolved = true;
cleanup();
resolve(token);
} else {
cleanup();
reject(new Error('cancelado'));
}
return;
}
const token = readExtensionToken();
if (token) {
resolved = true;
cleanup();
resolve(token);
}
}, POPUP_POLL_MS);
const timeoutTimer = setTimeout(() => {
if (!resolved) {
cleanup();
reject(new Error('timeout'));
}
}, timeoutMs);
function cleanup(): void {
clearInterval(pollInterval);
clearTimeout(timeoutTimer);
closePopupSafely(popup!);
}
});
}
/**
* Attempt to close a popup window safely.
*/
function closePopupSafely(popup: Window): void {
try {
if (!popup.closed) {
popup.close();
}
} catch {
// Ignore errors when trying to close
}
}
// ───────────────────────────────────────────────────────────────
// validateOkanToken
// ───────────────────────────────────────────────────────────────
/**
* Validate a raw Okan JWT token by:
* 1. Decoding the payload (base64url)
* 2. Checking exp > Date.now()/1000 + EXP_SKEW_SEC
* 3. Extracting a username from email, preferred_username, or sub
*
* Returns { valid, username?, error? }.
*/
function validateOkanToken(
rawToken: string,
): { valid: boolean; username?: string; error?: string } {
if (!rawToken || typeof rawToken !== 'string') {
return { valid: false, error: 'Token vacío o inválido' };
}
const payload = decodeJwtPayload(rawToken);
if (!payload) {
return { valid: false, error: 'No se pudo decodificar el token' };
}
// Check expiration
const exp = payload.exp as number | undefined;
if (exp === undefined || typeof exp !== 'number') {
return { valid: false, error: 'Token sin fecha de expiración (exp)' };
}
const nowWithSkew = Math.floor(Date.now() / 1000) + EXP_SKEW_SEC;
if (exp <= nowWithSkew) {
return { valid: false, error: 'Token expirado' };
}
// Extract username
const username = extractUsername(payload);
if (!username) {
return { valid: false, error: 'No se pudo extraer el usuario del token' };
}
return { valid: true, username };
}
// ───────────────────────────────────────────────────────────────
// exchangeToken
// ───────────────────────────────────────────────────────────────
/**
* Exchange an Okan token for a Linguo JWT session.
* POSTs { token_okan } to the Linguo login endpoint.
* On success (200) returns a Session object.
* On error, throws with the backend's detail message.
*/
async function exchangeToken(tokenOkan: string): Promise<Session> {
const response = await fetch(LINGUO_LOGIN_URL, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Accept: '*/*',
},
body: JSON.stringify({ token_okan: tokenOkan }),
});
if (response.ok) {
const data = (await response.json()) as {
document: string;
fullName: string;
expireDate: string;
token: string;
};
const session: Session = {
document: data.document,
fullName: data.fullName,
expireDate: data.expireDate,
token: data.token,
storedAt: Date.now(),
};
return session;
}
// Try to extract error detail from response body
let errorMessage = 'Error al intercambiar token';
try {
const errorBody = (await response.json()) as { detail?: string };
if (errorBody.detail) {
errorMessage = errorBody.detail;
}
} catch {
// Ignore parse errors
}
throw new Error(errorMessage);
}
// ───────────────────────────────────────────────────────────────
// Session Storage (sessionStorage)
// ───────────────────────────────────────────────────────────────
/**
* Persist a session object to sessionStorage.
*/
function storeSession(session: Session): void {
try {
sessionStorage.setItem(SESSION_KEY, JSON.stringify(session));
} catch {
// sessionStorage may be unavailable (private browsing, quota, etc.)
console.warn('[Auth] Could not store session — sessionStorage unavailable');
}
}
/**
* Retrieve the JWT token from sessionStorage.
* Returns null if:
* - No session is stored
* - The session's expireDate has passed (client-side validation)
*/
function getToken(): string | null {
try {
const stored = sessionStorage.getItem(SESSION_KEY);
if (!stored) return null;
const session = JSON.parse(stored) as Session;
// Validate expireDate client-side
const expireMs = new Date(session.expireDate).getTime();
if (isNaN(expireMs) || Date.now() >= expireMs) {
// Session expired — clean up
sessionStorage.removeItem(SESSION_KEY);
return null;
}
return session.token;
} catch {
return null;
}
}
/**
* Check whether a valid session exists (shortcut for getToken() !== null).
*/
function isAuthenticated(): boolean {
return getToken() !== null;
}
/**
* Retrieve the full session object from sessionStorage (without expiry check on token).
* Returns null if no session is stored.
*/
function getSession(): Session | null {
try {
const stored = sessionStorage.getItem(SESSION_KEY);
if (!stored) return null;
return JSON.parse(stored) as Session;
} catch {
return null;
}
}
/**
* Clear the session from sessionStorage.
*/
function logout(): void {
try {
sessionStorage.removeItem(SESSION_KEY);
localStorage.removeItem(OKAN_STORAGE_KEY); // also clear extension-injected token
} catch {
// Ignore errors
}
}
/**
* Build the Authorization header object.
* Returns an empty object if no valid token is available.
*/
function getAuthHeaders(): Record<string, string> {
const token = getToken();
if (!token) return {};
return { Authorization: `Bearer ${token}` };
}
// ───────────────────────────────────────────────────────────────
// Public API
// ───────────────────────────────────────────────────────────────
export const auth = {
captureOkanToken,
readExtensionToken,
clearExtensionToken,
validateOkanToken,
exchangeToken,
storeSession,
getToken,
isAuthenticated,
getSession,
logout,
getAuthHeaders,
};
+491
View File
@@ -0,0 +1,491 @@
import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest';
import { streamBuffer } from './streamBuffer';
describe('streamBuffer', () => {
beforeEach(() => {
// Clear all buffers before each test
streamBuffer.clearAll();
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
// ── CA-8: Buffer limits ──────────────────────────────────
describe('CA-8: Buffer limits (TTL 60s, max 500 tokens)', () => {
it('should store a token with valid payload', () => {
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(1);
expect(entries![0].tokens).toHaveLength(1);
expect(entries![0].tokens[0]).toEqual({ token: 'Hello', index: 0 });
expect(entries![0].messageId).toBe('msg-1');
});
it('should reject token with empty string', () => {
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
streamBuffer.addToken('conv-1', 'msg-1', '', 0);
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).toBeNull();
expect(warnSpy).toHaveBeenCalledWith(
'[streamBuffer] addToken: token must be a non-empty string',
);
warnSpy.mockRestore();
});
it('should reject token with negative index', () => {
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
streamBuffer.addToken('conv-1', 'msg-1', 'token', -1);
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).toBeNull();
expect(warnSpy).toHaveBeenCalledWith(
'[streamBuffer] addToken: index must be a non-negative integer',
);
warnSpy.mockRestore();
});
it('should reject token with non-integer index', () => {
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
streamBuffer.addToken('conv-1', 'msg-1', 'token', 1.5);
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).toBeNull();
expect(warnSpy).toHaveBeenCalledWith(
'[streamBuffer] addToken: index must be a non-negative integer',
);
warnSpy.mockRestore();
});
it('should reject token with invalid conversationId', () => {
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
streamBuffer.addToken('', 'msg-1', 'token', 0);
const entries = streamBuffer.getBufferEntry('');
expect(entries).toBeNull();
expect(warnSpy).toHaveBeenCalledWith(
'[streamBuffer] addToken: invalid conversationId',
);
warnSpy.mockRestore();
});
it('should reject token with invalid messageId', () => {
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
streamBuffer.addToken('conv-1', '', 'token', 0);
expect(warnSpy).toHaveBeenCalledWith(
'[streamBuffer] addToken: invalid messageId',
);
warnSpy.mockRestore();
});
it('should accumulate up to 500 tokens per stream', () => {
for (let i = 0; i < 500; i++) {
streamBuffer.addToken('conv-1', 'msg-1', `token-${i}`, i);
}
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(1);
expect(entries![0].tokens).toHaveLength(500);
});
it('should drop tokens beyond 500 per stream and log warning', () => {
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
for (let i = 0; i < 501; i++) {
streamBuffer.addToken('conv-1', 'msg-1', `token-${i}`, i);
}
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries![0].tokens).toHaveLength(500);
expect(warnSpy).toHaveBeenCalledWith(
expect.stringMatching(/Max tokens \(500\) reached/),
);
warnSpy.mockRestore();
});
it('should expire tokens after TTL (60s)', () => {
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
expect(streamBuffer.getBufferEntry('conv-1')).not.toBeNull();
// Advance time by 61s
vi.advanceTimersByTime(61_000);
// getBufferEntry calls removeExpired internally
expect(streamBuffer.getBufferEntry('conv-1')).toBeNull();
});
it('should refresh TTL on each addToken', () => {
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
// Advance 30s
vi.advanceTimersByTime(30_000);
// Add another token — should refresh TTL
streamBuffer.addToken('conv-1', 'msg-1', ' World', 1);
// Advance 31s more (total 61s since first, but only 31s since last)
vi.advanceTimersByTime(31_000);
expect(streamBuffer.getBufferEntry('conv-1')).not.toBeNull();
// Advance another 30s (total 61s since last)
vi.advanceTimersByTime(30_000);
expect(streamBuffer.getBufferEntry('conv-1')).toBeNull();
});
it('should expire each messageId independently by TTL', () => {
// msg-1 at t=0
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
vi.advanceTimersByTime(10_000); // t=10s
// msg-2 at t=10s
streamBuffer.addToken('conv-1', 'msg-2', 'World', 0);
// Advance 55s more → t=65s
// msg-1 is 65s old → expired, msg-2 is 55s old → still fresh
vi.advanceTimersByTime(55_000);
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(1);
expect(entries![0].messageId).toBe('msg-2');
// Advance 10s more → t=75s
// msg-2 is now 65s old (75-10) → expired, conv should be empty
vi.advanceTimersByTime(10_000);
expect(streamBuffer.getBufferEntry('conv-1')).toBeNull();
});
it('should enforce global LRU limit of 200 streams', () => {
// Add 200 streams across different conversations — should succeed
for (let i = 0; i < 200; i++) {
streamBuffer.addToken(`conv-${i}`, 'msg-1', `token-${i}`, 0);
}
// All 200 present
const firstConvStreams = streamBuffer.getBufferEntry('conv-0');
expect(firstConvStreams).not.toBeNull();
expect(firstConvStreams).toHaveLength(1);
// Add one more stream — should evict the oldest (conv-0)
streamBuffer.addToken('conv-200', 'msg-1', 'overflow', 0);
// conv-0 should have been evicted (oldest)
expect(streamBuffer.getBufferEntry('conv-0')).toBeNull();
// conv-200 should be present
const newConvStreams = streamBuffer.getBufferEntry('conv-200');
expect(newConvStreams).not.toBeNull();
});
it('should keep most recent streams when LRU limit exceeded', () => {
// Add 150 streams to conv-1 (multi-msg) and 50 to others
for (let i = 0; i < 150; i++) {
streamBuffer.addToken('conv-big', `msg-${i}`, `token-${i}`, 0);
vi.advanceTimersByTime(1); // stagger timestamps
}
for (let i = 0; i < 50; i++) {
streamBuffer.addToken(`conv-small-${i}`, 'msg-1', `token-${i}`, 0);
vi.advanceTimersByTime(1);
}
// Total: 200 streams — at limit
expect(streamBuffer.getBufferEntry('conv-big')).not.toBeNull();
// Add one more — oldest in conv-big (msg-0) should be evicted
streamBuffer.addToken('conv-last', 'msg-1', 'last', 0);
const bigConv = streamBuffer.getBufferEntry('conv-big');
expect(bigConv).not.toBeNull();
// msg-0 was the oldest, should be gone
const msg0 = bigConv!.find((e) => e.messageId === 'msg-0');
expect(msg0).toBeUndefined();
// Newest streams should remain
expect(streamBuffer.getBufferEntry('conv-last')).not.toBeNull();
});
});
// ── Clear operations ─────────────────────────────────────
describe('clear operations', () => {
it('should clear a specific conversation buffer (all streams)', () => {
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
streamBuffer.addToken('conv-2', 'msg-2', 'World', 0);
expect(streamBuffer.getBufferEntry('conv-1')).not.toBeNull();
expect(streamBuffer.getBufferEntry('conv-2')).not.toBeNull();
streamBuffer.clear('conv-1');
expect(streamBuffer.getBufferEntry('conv-1')).toBeNull();
expect(streamBuffer.getBufferEntry('conv-2')).not.toBeNull();
});
it('should clear a specific message stream via clearMessage', () => {
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
streamBuffer.addToken('conv-1', 'msg-2', 'World', 0);
let entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(2);
streamBuffer.clearMessage('conv-1', 'msg-1');
entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(1);
expect(entries![0].messageId).toBe('msg-2');
});
it('should remove conversation when last stream is cleared via clearMessage', () => {
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
streamBuffer.clearMessage('conv-1', 'msg-1');
expect(streamBuffer.getBufferEntry('conv-1')).toBeNull();
});
it('should clear all buffers', () => {
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
streamBuffer.addToken('conv-2', 'msg-2', 'World', 0);
streamBuffer.clearAll();
expect(streamBuffer.getBufferEntry('conv-1')).toBeNull();
expect(streamBuffer.getBufferEntry('conv-2')).toBeNull();
});
});
// ── Multi-stream: múltiples messageId en la misma conversación ──
describe('multi-stream handling (CA-10)', () => {
it('should keep BOTH streams when messageId changes (no discard)', () => {
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
streamBuffer.addToken('conv-1', 'msg-1', ' World', 1);
let entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(1);
expect(entries![0].tokens).toHaveLength(2);
// New streaming message for same conversation — should NOT discard msg-1
streamBuffer.addToken('conv-1', 'msg-2', 'New message', 0);
entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(2); // Both streams present
expect(entries![0].messageId).toBe('msg-1');
expect(entries![0].tokens).toHaveLength(2);
expect(entries![1].messageId).toBe('msg-2');
expect(entries![1].tokens).toHaveLength(1);
expect(entries![1].tokens[0].token).toBe('New message');
});
it('should accumulate tokens for three concurrent streams independently', () => {
// Simulate TRIAGE, COORDINATOR, SPECIALIST streams
streamBuffer.addToken('conv-1', 'msg-triage', 'Triage ', 0);
streamBuffer.addToken('conv-1', 'msg-triage', 'analysis', 1);
streamBuffer.addToken('conv-1', 'msg-coord', 'Coord ', 0);
streamBuffer.addToken('conv-1', 'msg-spec', 'Specialist ', 0);
streamBuffer.addToken('conv-1', 'msg-spec', 'response', 1);
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(3);
const triage = entries!.find((e) => e.messageId === 'msg-triage');
const coord = entries!.find((e) => e.messageId === 'msg-coord');
const spec = entries!.find((e) => e.messageId === 'msg-spec');
expect(triage).toBeDefined();
expect(coord).toBeDefined();
expect(spec).toBeDefined();
expect(triage!.tokens).toHaveLength(2);
expect(coord!.tokens).toHaveLength(1);
expect(spec!.tokens).toHaveLength(2);
});
it('should NOT overwrite or corrupt tokens between interleaved streams (CA-10 isolation)', () => {
// Interleave tokens from two streams to verify isolation
streamBuffer.addToken('conv-1', 'msg-alpha', 'Alpha-0', 0);
streamBuffer.addToken('conv-1', 'msg-beta', 'Beta-0', 0);
streamBuffer.addToken('conv-1', 'msg-alpha', 'Alpha-1', 1);
streamBuffer.addToken('conv-1', 'msg-beta', 'Beta-1', 1);
streamBuffer.addToken('conv-1', 'msg-alpha', 'Alpha-2', 2);
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(2);
const alpha = entries!.find((e) => e.messageId === 'msg-alpha');
const beta = entries!.find((e) => e.messageId === 'msg-beta');
expect(alpha).toBeDefined();
expect(beta).toBeDefined();
// Alpha should have exactly 3 tokens: Alpha-0, Alpha-1, Alpha-2 (in that order)
expect(alpha!.tokens).toHaveLength(3);
expect(alpha!.tokens[0].token).toBe('Alpha-0');
expect(alpha!.tokens[1].token).toBe('Alpha-1');
expect(alpha!.tokens[2].token).toBe('Alpha-2');
// Beta should have exactly 2 tokens: Beta-0, Beta-1 (in that order)
expect(beta!.tokens).toHaveLength(2);
expect(beta!.tokens[0].token).toBe('Beta-0');
expect(beta!.tokens[1].token).toBe('Beta-1');
// Verify indices are preserved per-stream (no cross-contamination)
expect(alpha!.tokens[0].index).toBe(0);
expect(alpha!.tokens[1].index).toBe(1);
expect(alpha!.tokens[2].index).toBe(2);
expect(beta!.tokens[0].index).toBe(0);
expect(beta!.tokens[1].index).toBe(1);
});
it('should handle streams across different conversations without interference', () => {
streamBuffer.addToken('conv-a', 'msg-1', 'A1', 0);
streamBuffer.addToken('conv-b', 'msg-1', 'B1', 0);
streamBuffer.addToken('conv-a', 'msg-2', 'A2', 0);
streamBuffer.addToken('conv-b', 'msg-2', 'B2', 0);
const convAEntries = streamBuffer.getBufferEntry('conv-a');
const convBEntries = streamBuffer.getBufferEntry('conv-b');
expect(convAEntries).not.toBeNull();
expect(convBEntries).not.toBeNull();
expect(convAEntries).toHaveLength(2);
expect(convBEntries).toHaveLength(2);
expect(convAEntries![0].messageId).toBe('msg-1');
expect(convAEntries![0].tokens[0].token).toBe('A1');
expect(convAEntries![1].messageId).toBe('msg-2');
expect(convAEntries![1].tokens[0].token).toBe('A2');
expect(convBEntries![0].messageId).toBe('msg-1');
expect(convBEntries![0].tokens[0].token).toBe('B1');
expect(convBEntries![1].messageId).toBe('msg-2');
expect(convBEntries![1].tokens[0].token).toBe('B2');
});
});
// ── CA-11: getBufferEntry returns array of all streams ────
describe('getBufferEntry returns full array (CA-11)', () => {
it('should return array with all streams for merge in handleConversationClick', () => {
streamBuffer.addToken('conv-1', 'msg-a', 'TokenA', 0);
streamBuffer.addToken('conv-1', 'msg-a', 'TokenA2', 1);
streamBuffer.addToken('conv-1', 'msg-b', 'TokenB', 0);
streamBuffer.addToken('conv-1', 'msg-c', 'TokenC', 0);
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(3); // 3 streams: msg-a, msg-b, msg-c
// Verify each stream has its own tokens intact
const msgA = entries!.find((e) => e.messageId === 'msg-a');
const msgB = entries!.find((e) => e.messageId === 'msg-b');
const msgC = entries!.find((e) => e.messageId === 'msg-c');
expect(msgA).toBeDefined();
expect(msgB).toBeDefined();
expect(msgC).toBeDefined();
expect(msgA!.tokens).toHaveLength(2);
expect(msgB!.tokens).toHaveLength(1);
expect(msgC!.tokens).toHaveLength(1);
});
it('should return null when no streams exist for conversation', () => {
expect(streamBuffer.getBufferEntry('nonexistent')).toBeNull();
});
it('should return null after all streams are cleared via clearMessage', () => {
streamBuffer.addToken('conv-1', 'msg-a', 'A', 0);
streamBuffer.addToken('conv-1', 'msg-b', 'B', 0);
streamBuffer.clearMessage('conv-1', 'msg-a');
streamBuffer.clearMessage('conv-1', 'msg-b');
expect(streamBuffer.getBufferEntry('conv-1')).toBeNull();
});
});
// ── CA-9: Buffer retiene tokens cuando NO se limpia (simula loadingConversation) ──
describe('CA-9: Buffer retention when clear is gated (loadingConversation)', () => {
it('should retain tokens when not explicitly cleared (simulating agent_stream_completed during loading)', () => {
// Simulate: tokens arrive while conversation is loading (selectedConversation = null)
streamBuffer.addToken('conv-1', 'msg-1', 'Token ', 0);
streamBuffer.addToken('conv-1', 'msg-1', 'retained', 1);
// Simulate: agent_stream_completed arrives but buffer is NOT cleared
// (because selectedConversation is null — Fix #1 gate)
// NOTE: We intentionally do NOT call clear() or clearMessage()
// This is what the AppShell fix does
// Buffer should still have the tokens for later merge
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(1);
expect(entries![0].tokens).toHaveLength(2);
expect(entries![0].tokens[0].token).toBe('Token ');
expect(entries![0].tokens[1].token).toBe('retained');
});
it('should retain tokens through multiple agent_stream_completed events (no clears)', () => {
// Simulate: multiple streams arrive while conversation is loading
streamBuffer.addToken('conv-1', 'msg-triage', 'Triage ', 0);
streamBuffer.addToken('conv-1', 'msg-triage', 'result', 1);
// Simulate: agent_stream_completed for TRIAGE — NOT cleared (Fix #1 gate)
// (No clearMessage call)
streamBuffer.addToken('conv-1', 'msg-spec', 'Specialist ', 0);
streamBuffer.addToken('conv-1', 'msg-spec', 'response', 1);
// Simulate: agent_stream_completed for SPECIALIST — NOT cleared (Fix #1 gate)
// (No clearMessage call)
// All streams should still be in buffer when handleConversationClick runs
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(2); // Both streams survived
const triage = entries!.find((e) => e.messageId === 'msg-triage');
const spec = entries!.find((e) => e.messageId === 'msg-spec');
expect(triage).toBeDefined();
expect(spec).toBeDefined();
expect(triage!.tokens).toHaveLength(2);
expect(spec!.tokens).toHaveLength(2);
});
it('should still allow selective clearMessage when conversation IS selected', () => {
// Simulate: conversation IS selected — clearMessage IS called for completed stream
streamBuffer.addToken('conv-1', 'msg-triage', 'Triage', 0);
streamBuffer.addToken('conv-1', 'msg-spec', 'Specialist', 0);
// Simulate: agent_stream_completed for TRIAGE — conversation selected, so clear
streamBuffer.clearMessage('conv-1', 'msg-triage');
// msg-triage should be gone, but msg-spec should remain
const entries = streamBuffer.getBufferEntry('conv-1');
expect(entries).not.toBeNull();
expect(entries).toHaveLength(1);
expect(entries![0].messageId).toBe('msg-spec');
});
});
// ── getTokens backwards compatibility ────────────────────
describe('getTokens (backwards compat)', () => {
it('should return tokens of the most recent stream', () => {
streamBuffer.addToken('conv-1', 'msg-1', 'World', 1);
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
const tokens = streamBuffer.getTokens('conv-1');
expect(tokens).not.toBeNull();
expect(tokens).toHaveLength(2);
// Returned as stored (not sorted internally — sorting done in MonitorPage)
expect(tokens![0].token).toBe('World');
expect(tokens![1].token).toBe('Hello');
});
it('should return tokens of the most recent stream among multiple', () => {
streamBuffer.addToken('conv-1', 'msg-old', 'Old ', 0);
vi.advanceTimersByTime(100);
streamBuffer.addToken('conv-1', 'msg-new', 'New', 0);
const tokens = streamBuffer.getTokens('conv-1');
expect(tokens).not.toBeNull();
expect(tokens).toHaveLength(1);
expect(tokens![0].token).toBe('New'); // Most recent stream
});
it('should return null for non-existent conversation', () => {
expect(streamBuffer.getTokens('nonexistent')).toBeNull();
});
});
// ── cleanup() removes expired ────────────────────────────
describe('cleanup', () => {
it('should remove expired entries', () => {
streamBuffer.addToken('conv-1', 'msg-1', 'Hello', 0);
vi.advanceTimersByTime(61_000);
streamBuffer.cleanup();
expect(streamBuffer.getBufferEntry('conv-1')).toBeNull();
});
});
});
+254
View File
@@ -0,0 +1,254 @@
// ─────────────────────────────────────────────────────────────
// streamBuffer.ts — Buffer transitorio externo al store
// Almacena tokens de conversaciones NO seleccionadas con TTL,
// SOPORTANDO MÚLTIPLES STREAMS (messageId) por conversación.
// NO debe ser importado por el store; solo por AppShell y MonitorPage.
// ─────────────────────────────────────────────────────────────
//
// Estructura interna:
// Map<conversationId, Map<messageId, StreamData>>
//
// Cada stream (messageId) tiene su propio TTL (60s) y límite de 500 tokens.
// Límite global: 200 streams. LRU: se eliminan los más antiguos al exceder.
// ─────────────────────────────────────────────────────────────
const TOKEN_TTL_MS = 60_000; // 1 minuto por messageId
const MAX_TOKENS_PER_STREAM = 500;
const MAX_STREAMS_GLOBAL = 200;
interface StreamData {
tokens: { token: string; index: number }[];
timestamp: number; // TTL por stream individual
}
// Map<conversationId, Map<messageId, StreamData>>
const buffers = new Map<string, Map<string, StreamData>>();
// ─────────────────────────────────────────────────────────────
// Internal helpers
// ─────────────────────────────────────────────────────────────
/**
* Elimina streams expirados por TTL (>60s).
* También limpia conversaciones sin streams activos.
*/
function removeExpired(): void {
const now = Date.now();
for (const [convId, convMap] of buffers.entries()) {
for (const [msgId, stream] of convMap.entries()) {
if (now - stream.timestamp > TOKEN_TTL_MS) {
convMap.delete(msgId);
}
}
if (convMap.size === 0) {
buffers.delete(convId);
}
}
}
/**
* Cuenta el total de streams (messageId) en todos los niveles.
*/
function totalStreams(): number {
let count = 0;
for (const convMap of buffers.values()) {
count += convMap.size;
}
return count;
}
/**
* LRU global: si se excede MAX_STREAMS_GLOBAL, elimina los más antiguos.
*/
function enforceGlobalLimit(): void {
const currentTotal = totalStreams();
if (currentTotal <= MAX_STREAMS_GLOBAL) return;
// Recolectar todos los streams con su timestamp
const allStreams: { convId: string; msgId: string; timestamp: number }[] = [];
for (const [convId, convMap] of buffers.entries()) {
for (const [msgId, stream] of convMap.entries()) {
allStreams.push({ convId, msgId, timestamp: stream.timestamp });
}
}
// Ordenar por timestamp ascendente (más antiguos primero)
allStreams.sort((a, b) => a.timestamp - b.timestamp);
const toEvict = currentTotal - MAX_STREAMS_GLOBAL;
for (let i = 0; i < toEvict && i < allStreams.length; i++) {
const convMap = buffers.get(allStreams[i].convId);
if (convMap) {
convMap.delete(allStreams[i].msgId);
if (convMap.size === 0) {
buffers.delete(allStreams[i].convId);
}
}
}
}
// ─────────────────────────────────────────────────────────────
// Public API
// ─────────────────────────────────────────────────────────────
export const streamBuffer = {
/**
* Agrega un token al buffer para una conversación y messageId específicos.
* A diferencia de la versión anterior, NO descarta streams previos al
* cambiar messageId — cada stream es independiente.
*
* Validaciones:
* - conversationId y messageId: strings no vacíos
* - token: string no vacío
* - index: entero >= 0
* Límites:
* - 500 tokens por stream (messageId)
* - 200 streams en total (LRU global)
*/
addToken(
conversationId: string,
messageId: string,
token: string,
index: number,
): void {
// ── Validate payload ──────────────────────────────────
if (!conversationId || typeof conversationId !== 'string') {
console.warn('[streamBuffer] addToken: invalid conversationId');
return;
}
if (!messageId || typeof messageId !== 'string') {
console.warn('[streamBuffer] addToken: invalid messageId');
return;
}
if (typeof token !== 'string' || token.length === 0) {
console.warn('[streamBuffer] addToken: token must be a non-empty string');
return;
}
if (typeof index !== 'number' || index < 0 || !Number.isInteger(index)) {
console.warn('[streamBuffer] addToken: index must be a non-negative integer');
return;
}
// ── Limpieza previa de expirados ──────────────────────
removeExpired();
// ── Obtener o crear Map de messageId para esta conversación ──
let convMap = buffers.get(conversationId);
if (!convMap) {
convMap = new Map<string, StreamData>();
buffers.set(conversationId, convMap);
}
// ── Obtener o crear StreamData para este messageId ──────────
let stream = convMap.get(messageId);
if (!stream) {
stream = {
tokens: [],
timestamp: Date.now(),
};
convMap.set(messageId, stream);
}
// ── Límite de 500 tokens por stream ─────────────────────────
if (stream.tokens.length >= MAX_TOKENS_PER_STREAM) {
console.warn(
`[streamBuffer] Max tokens (${MAX_TOKENS_PER_STREAM}) reached for stream ${conversationId}/${messageId} — dropping token`,
);
return;
}
// ── Agregar token y refrescar TTL ──────────────────────────
stream.tokens.push({ token, index });
stream.timestamp = Date.now();
// ── LRU global ─────────────────────────────────────────────
enforceGlobalLimit();
},
/**
* Obtiene TODOS los streams almacenados para una conversación.
* Retorna un ARRAY de objetos { messageId, tokens } — uno por
* cada stream vivo en esa conversación.
* Retorna null si no hay ningún stream activo.
*/
getBufferEntry(
conversationId: string,
): { messageId: string; tokens: { token: string; index: number }[] }[] | null {
removeExpired();
const convMap = buffers.get(conversationId);
if (!convMap || convMap.size === 0) return null;
// Construir array con todos los streams vivos
const entries: { messageId: string; tokens: { token: string; index: number }[] }[] = [];
for (const [msgId, stream] of convMap.entries()) {
entries.push({
messageId: msgId,
tokens: stream.tokens,
});
}
return entries.length > 0 ? entries : null;
},
/**
* Retrocompatibilidad: retorna los tokens del stream MÁS RECIENTE
* (por timestamp) para una conversación, o null si no hay streams.
*/
getTokens(
conversationId: string,
): { token: string; index: number }[] | null {
removeExpired();
const convMap = buffers.get(conversationId);
if (!convMap || convMap.size === 0) return null;
// Encontrar el stream más reciente por timestamp
let latestMsgId: string | null = null;
let latestTimestamp = 0;
for (const [msgId, stream] of convMap.entries()) {
if (stream.timestamp > latestTimestamp) {
latestTimestamp = stream.timestamp;
latestMsgId = msgId;
}
}
if (!latestMsgId) return null;
const stream = convMap.get(latestMsgId);
return stream ? stream.tokens : null;
},
/**
* NUEVO: Limpia un stream específico (messageId) de una conversación.
*/
clearMessage(conversationId: string, messageId: string): void {
const convMap = buffers.get(conversationId);
if (!convMap) return;
convMap.delete(messageId);
if (convMap.size === 0) {
buffers.delete(conversationId);
}
},
/**
* Limpia TODOS los streams de una conversación específica.
*/
clear(conversationId: string): void {
buffers.delete(conversationId);
},
/**
* Limpia streams expirados y conversaciones sin streams activos.
*/
cleanup(): void {
removeExpired();
},
/**
* Vacía todo el buffer (útil al desconectar WS o reinicio).
*/
clearAll(): void {
buffers.clear();
},
};
+303
View File
@@ -0,0 +1,303 @@
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
// We need to mock auth BEFORE importing wsClient
vi.mock('@/services/auth', () => ({
auth: {
getToken: vi.fn(() => 'mock-jwt-token'),
isAuthenticated: vi.fn(() => true),
},
}));
// Mock crypto.randomUUID
const mockUUID = vi.fn(() => '00000000-0000-0000-0000-000000000001');
vi.stubGlobal('crypto', {
randomUUID: mockUUID,
});
import { wsClient } from './wsClient';
import { auth } from '@/services/auth';
// ── Proper Mock WebSocket Class ──────────────────────────────
let mockWsInstance: any = null;
let lastWsUrl: string = '';
class MockWebSocket {
static CONNECTING = 0;
static OPEN = 1;
static CLOSING = 2;
static CLOSED = 3;
readyState: number = MockWebSocket.OPEN;
onopen: ((event: any) => void) | null = null;
onclose: ((event: any) => void) | null = null;
onmessage: ((event: any) => void) | null = null;
onerror: ((event: any) => void) | null = null;
send: any = vi.fn();
close: any = vi.fn().mockImplementation(() => {
this.readyState = MockWebSocket.CLOSING;
// Simulate close event
if (this.onclose) {
this.onclose({ code: 1000, reason: 'Normal closure' });
}
this.readyState = MockWebSocket.CLOSED;
});
url: string;
constructor(url: string) {
this.url = url;
lastWsUrl = url;
this.readyState = MockWebSocket.OPEN;
mockWsInstance = this;
}
}
let mockWebSocket: any;
describe('wsClient — In-Band Auth (CA-6)', () => {
beforeEach(() => {
mockWsInstance = null;
lastWsUrl = '';
// Clear all mocks
vi.clearAllMocks();
try { vi.unstubAllGlobals(); } catch { /* OK */ }
// Stub globals
vi.stubGlobal('crypto', { randomUUID: mockUUID });
vi.useFakeTimers();
vi.clearAllTimers();
// Create a fresh mock class each test
// Must include static WebSocket constants so wsClient can compare readyState
mockWebSocket = vi.fn().mockImplementation((url: string) => new MockWebSocket(url));
mockWebSocket.CONNECTING = 0;
mockWebSocket.OPEN = 1;
mockWebSocket.CLOSING = 2;
mockWebSocket.CLOSED = 3;
vi.stubGlobal('WebSocket', mockWebSocket);
});
afterEach(() => {
wsClient.disconnect();
vi.useRealTimers();
vi.unstubAllGlobals();
vi.clearAllMocks();
});
// ── CA-6: In-Band Auth ─────────────────────────────────────
describe('CA-6: In-Band Auth — no query params', () => {
it('should connect WITHOUT token in URL', () => {
wsClient.connect();
expect(lastWsUrl).not.toContain('?token=');
expect(lastWsUrl).not.toContain('token');
expect(lastWsUrl).toBe('ws://localhost:5503/ws/dashboard');
});
it('should send auth message on open', () => {
wsClient.connect();
expect(mockWsInstance).not.toBeNull();
// Trigger onopen
mockWsInstance.onopen({});
// Verify auth message was sent (send called once with auth message)
expect(mockWsInstance.send).toHaveBeenCalledWith(
JSON.stringify({ action: 'auth', token: 'mock-jwt-token' }),
);
});
it('should call onAuthenticated when auth response received', () => {
const onAuth = vi.fn();
wsClient.onAuthenticated = onAuth;
wsClient.connect();
mockWsInstance.onopen({});
// Simulate auth response
const authResponse = { status: 'authenticated', user_id: 'user-123' };
mockWsInstance.onmessage({ data: JSON.stringify(authResponse) });
expect(onAuth).toHaveBeenCalledOnce();
});
it('should delegate business messages only after auth', () => {
const onMsg = vi.fn();
wsClient.onMessage = onMsg;
wsClient.connect();
mockWsInstance.onopen({});
// Try sending business message before auth
const businessMsg = { type: 'init_state', eventId: 'evt-1', payload: {} };
mockWsInstance.onmessage({ data: JSON.stringify(businessMsg) });
// Should NOT be delegated because auth not yet received
expect(onMsg).not.toHaveBeenCalled();
// Send auth response
const authResponse = { status: 'authenticated', user_id: 'user-123' };
mockWsInstance.onmessage({ data: JSON.stringify(authResponse) });
// Now send business message
mockWsInstance.onmessage({ data: JSON.stringify(businessMsg) });
// Should be delegated
expect(onMsg).toHaveBeenCalledTimes(1);
expect(onMsg).toHaveBeenCalledWith(businessMsg);
});
it('should timeout auth after 5 seconds and close with code 1008', () => {
wsClient.connect();
mockWsInstance.onopen({});
// Advance time by 5s
vi.advanceTimersByTime(5000);
expect(mockWsInstance.close).toHaveBeenCalledWith(1008, 'Auth timeout');
});
it('should NOT timeout if auth received within 5s', () => {
wsClient.connect();
mockWsInstance.onopen({});
// Send auth at 3s
vi.advanceTimersByTime(3000);
const authResponse = { status: 'authenticated', user_id: 'user-123' };
mockWsInstance.onmessage({ data: JSON.stringify(authResponse) });
// Advance to 6s
vi.advanceTimersByTime(3000);
// Should NOT have closed (close was not called by timer)
// Note: close may have been called in the mock's close() fn for cleanup
// We check the auth timeout specifically by looking at close(1008)
const closeCalls = mockWsInstance.close.mock.calls.filter(
(call: any[]) => call[0] === 1008,
);
expect(closeCalls).toHaveLength(0);
});
it('should handle close code 1008 and set authState to failed', () => {
wsClient.connect();
mockWsInstance.onopen({});
// Simulate close with code 1008
mockWsInstance.onclose({ code: 1008 });
expect(wsClient.getAuthState()).toBe('failed');
});
});
// ── Token missing ───────────────────────────────────────────
describe('auth token missing', () => {
it('should skip connection if no token available', () => {
vi.mocked(auth.getToken).mockReturnValueOnce(null);
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
wsClient.connect();
expect(warnSpy).toHaveBeenCalledWith('[WS] No auth token — skipping connection');
// No WebSocket should be created
expect(mockWsInstance).toBeNull();
warnSpy.mockRestore();
});
});
// ── Malformed messages ─────────────────────────────────────
describe('malformed messages', () => {
it('should silently ignore malformed JSON messages', () => {
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
wsClient.connect();
mockWsInstance.onopen({});
// Send auth to enable business messages
const authResponse = { status: 'authenticated', user_id: 'user-123' };
mockWsInstance.onmessage({ data: JSON.stringify(authResponse) });
const onMsg = vi.fn();
wsClient.onMessage = onMsg;
// Malformed message
mockWsInstance.onmessage({ data: 'not-json' });
// Should not throw and not call callback
expect(onMsg).not.toHaveBeenCalled();
warnSpy.mockRestore();
});
});
// ── Reconnection ────────────────────────────────────────────
describe('reconnection', () => {
it('should schedule reconnect on unexpected close', () => {
wsClient.connect();
mockWsInstance.onopen({});
// Simulate unexpected close — set readyState to CLOSED first (real WS behavior)
mockWsInstance.readyState = 3; // WebSocket.CLOSED
mockWsInstance.onclose({ code: 1006 }); // Abnormal closure
// Status should be reconnecting
expect(wsClient.getStatus()).toBe('reconnecting');
// Advance backoff (1s initial)
vi.advanceTimersByTime(1000);
// A new WebSocket should be created (second call to WebSocket constructor)
expect(mockWebSocket).toHaveBeenCalledTimes(2);
});
it('should not reconnect if disconnect() was called', () => {
wsClient.connect();
mockWsInstance.onopen({});
// Capture reference before disconnect
const wsBeforeDisconnect = mockWsInstance;
wsClient.disconnect();
// After disconnect, this.ws is null, so onclose is nullified on the instance
// The mock close() in our class triggers onclose, but disconnect sets
// this.ws.onclose = null so it won't fire reconnect
expect(wsBeforeDisconnect.onclose).toBeNull();
// Advance timer — no new connection should be created
vi.advanceTimersByTime(5000);
// Only 1 WebSocket was created (the original connect)
expect(mockWebSocket).toHaveBeenCalledTimes(1);
});
});
// ── Send ────────────────────────────────────────────────────
describe('send()', () => {
it('should send a properly formatted envelope', () => {
wsClient.connect();
mockWsInstance.onopen({});
wsClient.send('test_event', { key: 'value' });
// send() is called first for auth, then for the test message
const sentData = JSON.parse(mockWsInstance.send.mock.calls[1][0]);
expect(sentData.type).toBe('test_event');
expect(sentData.eventId).toBe('00000000-0000-0000-0000-000000000001');
expect(sentData.payload).toEqual({ key: 'value' });
});
it('should warn if socket not open', () => {
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
// Don't connect — socket is null
wsClient.send('test', {});
expect(warnSpy).toHaveBeenCalledWith(
'[WS] Cannot send — socket is not open. Status:',
'disconnected',
);
warnSpy.mockRestore();
});
});
});
+107 -8
View File
@@ -1,10 +1,11 @@
import type { WSEnvelope } from '@/types/wsProtocol';
import { auth } from '@/services/auth';
// ─────────────────────────────────────────────────────────────
// Configuration
// ─────────────────────────────────────────────────────────────
const DEFAULT_WS_URL = 'ws://localhost:3000/ws/dashboard';
const DEFAULT_WS_URL = 'ws://localhost:5503/ws/dashboard';
const WS_URL = import.meta.env.VITE_WS_URL || DEFAULT_WS_URL;
@@ -18,12 +19,21 @@ const BACKOFF_FACTOR = 2;
export type WsConnectionStatus = 'connected' | 'disconnected' | 'reconnecting';
export type WsAuthState = 'pending' | 'authenticated' | 'failed';
// ─────────────────────────────────────────────────────────────
// Event callback types
// ─────────────────────────────────────────────────────────────
export type MessageCallback = (envelope: WSEnvelope) => void;
export type StatusChangeCallback = (status: WsConnectionStatus) => void;
export type AuthenticatedCallback = () => void;
// ─────────────────────────────────────────────────────────────
// Constants
// ─────────────────────────────────────────────────────────────
const AUTH_TIMEOUT_MS = 5_000;
// ─────────────────────────────────────────────────────────────
// WebSocket Client
@@ -34,8 +44,11 @@ class WsClient {
private status: WsConnectionStatus = 'disconnected';
private onMessageCallback: MessageCallback | null = null;
private onStatusChangeCallback: StatusChangeCallback | null = null;
private onAuthenticatedCallback: AuthenticatedCallback | null = null;
private reconnectAttempts = 0;
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
private authTimer: ReturnType<typeof setTimeout> | null = null;
private authState: WsAuthState = 'pending';
private destroyFlag = false;
// ── Connection ───────────────────────────────────────────
@@ -45,12 +58,22 @@ class WsClient {
* If already connected, it will close and reconnect.
*/
connect(): void {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
return; // already connected
// Guard: skip if already connected or connecting (prevents double-connect in StrictMode)
if (this.ws && (this.ws.readyState === WebSocket.OPEN || this.ws.readyState === WebSocket.CONNECTING)) {
return;
}
// Check auth token before attempting connection
const token = auth.getToken();
if (!token) {
console.warn('[WS] No auth token — skipping connection');
return;
}
this.destroyFlag = false;
this.authState = 'pending';
// Paso 0: Connect WITHOUT token in URL — clean WebSocket URL
try {
this.ws = new WebSocket(WS_URL);
} catch (err) {
@@ -62,20 +85,71 @@ class WsClient {
this.ws.onopen = () => {
this.reconnectAttempts = 0;
this.setStatus('connected');
// Send In-Band Auth as first message
const authMessage = {
action: 'auth',
token: token,
};
this.ws?.send(JSON.stringify(authMessage));
// Start auth timeout: 5s to receive { status: "authenticated" }
this.authTimer = setTimeout(() => {
if (this.authState !== 'authenticated') {
console.warn('[WS] Auth timeout — no auth response within 5s');
this.authState = 'failed';
this.ws?.close(1008, 'Auth timeout');
}
}, AUTH_TIMEOUT_MS);
};
this.ws.onmessage = (event: MessageEvent) => {
if (!this.onMessageCallback) return;
try {
const envelope: WSEnvelope = JSON.parse(event.data as string);
this.onMessageCallback(envelope);
const data = JSON.parse(event.data as string);
// Handle auth response first
if (data.status === 'authenticated') {
this.authState = 'authenticated';
if (this.authTimer) {
clearTimeout(this.authTimer);
this.authTimer = null;
}
// Notify listeners that auth is complete
this.onAuthenticatedCallback?.();
return;
}
// If not yet authenticated, drop business messages
if (this.authState !== 'authenticated') {
console.warn('[WS] Dropping message — auth not yet complete');
return;
}
// Delegate business events to registered callback
if (this.onMessageCallback) {
const envelope: WSEnvelope = data;
this.onMessageCallback(envelope);
}
} catch {
// Malformed message — silently ignore
}
};
this.ws.onclose = () => {
this.ws.onclose = (event: CloseEvent) => {
// Clean up auth timer
if (this.authTimer) {
clearTimeout(this.authTimer);
this.authTimer = null;
}
// Code 1008 = auth failure — transition to failed
if (event.code === 1008) {
this.authState = 'failed';
this.setStatus('disconnected');
this.scheduleReconnect();
return;
}
// Only transition to reconnecting if we didn't intentionally close
if (!this.destroyFlag) {
this.setStatus('reconnecting');
@@ -99,6 +173,11 @@ class WsClient {
this.reconnectTimer = null;
}
if (this.authTimer !== null) {
clearTimeout(this.authTimer);
this.authTimer = null;
}
if (this.ws) {
this.ws.onclose = null; // prevent reconnect trigger
this.ws.close();
@@ -166,6 +245,26 @@ class WsClient {
return this.onStatusChangeCallback;
}
// ── Auth ──────────────────────────────────────────────────
/**
* Register a callback for when In-Band Auth completes successfully.
*/
set onAuthenticated(cb: AuthenticatedCallback | null) {
this.onAuthenticatedCallback = cb;
}
get onAuthenticated(): AuthenticatedCallback | null {
return this.onAuthenticatedCallback;
}
/**
* Get the current auth state.
*/
getAuthState(): WsAuthState {
return this.authState;
}
// ── Private helpers ───────────────────────────────────────
private setStatus(status: WsConnectionStatus): void {
+322
View File
@@ -0,0 +1,322 @@
import { describe, it, expect, beforeEach } from 'vitest';
import { useAppStore } from './useAppStore';
import type { ConversationSummary, Conversation, Message } from '@/types';
// Helper to reset store between tests
function resetStore() {
useAppStore.setState({
// Cases
cases: [],
selectedCaseId: null,
totalCases: 0,
// Conversations
conversations: [],
totalConversations: 0,
conversationsOffset: 0,
selectedConversation: null,
selectedConversationId: null,
// Idempotency
processedEventIds: [],
// Connection
initStateReceived: false,
// Loading & correlation
loadingConversation: null,
currentRequestId: null,
// State machine
conversationStates: {},
// Banner
conversationEndedBanner: null,
// UI
sidebarTab: 'all',
searchQuery: '',
applicativeFilter: null,
isDarkMode: false,
wsStatus: 'disconnected',
resolvedCaseAlert: null,
});
}
// ── Mock Conversation Summary ────────────────────────────────
function makeConvSummary(id: string, overrides: Partial<ConversationSummary> = {}): ConversationSummary {
return {
id,
clientId: `client-${id}`,
agentId: `agent-${id}`,
status: 'active',
createdAt: new Date().toISOString(),
...overrides,
};
}
function makeConversation(id: string): Conversation {
return {
id,
clientId: `client-${id}`,
agentId: `agent-${id}`,
status: 'active',
createdAt: new Date().toISOString(),
messages: [],
};
}
function makeMessage(id: string, convId: string, overrides: Partial<Message> = {}): Message {
return {
id,
conversationId: convId,
role: 'agent',
content: 'test content',
timestamp: new Date().toISOString(),
isStreaming: false,
...overrides,
};
}
describe('useAppStore', () => {
beforeEach(() => {
resetStore();
});
// ── CA-7: Idempotency ──────────────────────────────────────
describe('CA-7: Idempotency (eventId dedup)', () => {
it('should accept new eventId', () => {
const result = useAppStore.getState().addProcessedEventId('evt-1');
expect(result).toBe(true);
expect(useAppStore.getState().processedEventIds).toContain('evt-1');
});
it('should reject duplicate eventId', () => {
useAppStore.getState().addProcessedEventId('evt-1');
const result = useAppStore.getState().addProcessedEventId('evt-1');
expect(result).toBe(false);
expect(useAppStore.getState().processedEventIds).toHaveLength(1);
});
it('should accept different eventIds', () => {
useAppStore.getState().addProcessedEventId('evt-1');
const result = useAppStore.getState().addProcessedEventId('evt-2');
expect(result).toBe(true);
expect(useAppStore.getState().processedEventIds).toHaveLength(2);
});
it('should clear all processed eventIds', () => {
useAppStore.getState().addProcessedEventId('evt-1');
useAppStore.getState().addProcessedEventId('evt-2');
useAppStore.getState().clearProcessedEventIds();
expect(useAppStore.getState().processedEventIds).toHaveLength(0);
});
it('should enforce LRU eviction at 1000 entries', () => {
// Add 1000 entries
for (let i = 0; i < 1000; i++) {
useAppStore.getState().addProcessedEventId(`evt-${i}`);
}
expect(useAppStore.getState().processedEventIds).toHaveLength(1000);
// Add one more — should evict the oldest
useAppStore.getState().addProcessedEventId('evt-1000');
expect(useAppStore.getState().processedEventIds).toHaveLength(1000);
// The oldest (evt-0) should be gone
expect(useAppStore.getState().processedEventIds).not.toContain('evt-0');
// The newest should be present
expect(useAppStore.getState().processedEventIds).toContain('evt-1000');
});
});
// ── CA-3: setConversations (atomic replace via init_state) ─
describe('CA-3: setConversations atomic replace', () => {
it('should replace conversations atomically', () => {
const convs = [makeConvSummary('conv-1'), makeConvSummary('conv-2')];
useAppStore.getState().setConversations(convs);
const state = useAppStore.getState();
expect(state.conversations).toHaveLength(2);
expect(state.totalConversations).toBe(2);
expect(state.conversations[0].id).toBe('conv-1');
});
it('should replace stale conversations', () => {
const oldConvs = [makeConvSummary('conv-old')];
useAppStore.getState().setConversations(oldConvs);
expect(useAppStore.getState().conversations).toHaveLength(1);
const newConvs = [makeConvSummary('conv-new')];
useAppStore.getState().setConversations(newConvs);
expect(useAppStore.getState().conversations).toHaveLength(1);
expect(useAppStore.getState().conversations[0].id).toBe('conv-new');
});
it('should set empty array', () => {
useAppStore.getState().setConversations([]);
expect(useAppStore.getState().conversations).toHaveLength(0);
expect(useAppStore.getState().totalConversations).toBe(0);
});
});
// ── CA-4: initStateReceived ─────────────────────────────────
describe('CA-4: initStateReceived flag', () => {
it('should default to false', () => {
expect(useAppStore.getState().initStateReceived).toBe(false);
});
it('should be settable to true', () => {
useAppStore.getState().setInitStateReceived(true);
expect(useAppStore.getState().initStateReceived).toBe(true);
});
it('should be resettable to false', () => {
useAppStore.getState().setInitStateReceived(true);
useAppStore.getState().setInitStateReceived(false);
expect(useAppStore.getState().initStateReceived).toBe(false);
});
});
// ── CA-1: appendToken & completeStream ─────────────────────
describe('CA-1: appendToken / completeStream', () => {
it('should append token to an existing message', () => {
const conv = makeConversation('conv-1');
conv.messages = [makeMessage('msg-1', 'conv-1', { content: 'Hel', isStreaming: true })];
useAppStore.setState({ selectedConversation: conv, selectedConversationId: 'conv-1' });
useAppStore.getState().appendToken('conv-1', 'msg-1', 'lo', 1);
const msg = useAppStore.getState().selectedConversation!.messages[0];
expect(msg.content).toBe('Hello'); // Hel + lo = Hello (concatenation, no space)
expect(msg.isStreaming).toBe(true);
});
it('should create placeholder message if messageId does not exist', () => {
const conv = makeConversation('conv-1');
conv.messages = [];
useAppStore.setState({ selectedConversation: conv, selectedConversationId: 'conv-1' });
useAppStore.getState().appendToken('conv-1', 'msg-new', 'Hello', 0);
const msgs = useAppStore.getState().selectedConversation!.messages;
expect(msgs).toHaveLength(1);
expect(msgs[0].id).toBe('msg-new');
expect(msgs[0].content).toBe('Hello');
expect(msgs[0].isStreaming).toBe(true);
});
it('should NOT append token if selectedConversation is null', () => {
const conv = makeConversation('conv-1');
useAppStore.setState({ selectedConversation: conv, selectedConversationId: 'conv-1' });
// Nullify selectedConversation but keep id
useAppStore.setState({ selectedConversation: null });
useAppStore.getState().appendToken('conv-1', 'msg-1', 'token', 0);
// Should not crash and store should not have changed
expect(useAppStore.getState().selectedConversation).toBeNull();
});
it('should NOT append token if conversationId differs', () => {
const conv = makeConversation('conv-1');
useAppStore.setState({ selectedConversation: conv, selectedConversationId: 'conv-1' });
useAppStore.getState().appendToken('conv-other', 'msg-1', 'token', 0);
// Should not mutate selectedConversation
expect(useAppStore.getState().selectedConversation!.messages).toHaveLength(0);
});
it('should complete stream and set isStreaming to false', () => {
const conv = makeConversation('conv-1');
conv.messages = [makeMessage('msg-1', 'conv-1', { content: 'Partial', isStreaming: true })];
useAppStore.setState({ selectedConversation: conv, selectedConversationId: 'conv-1' });
useAppStore.getState().completeStream('conv-1', 'msg-1', 'Full content');
const msg = useAppStore.getState().selectedConversation!.messages[0];
expect(msg.content).toBe('Full content');
expect(msg.isStreaming).toBe(false);
});
it('should no-op completeStream if conversation not selected', () => {
useAppStore.setState({ selectedConversation: null });
// Should not throw
useAppStore.getState().completeStream('conv-1', 'msg-1', 'content');
expect(useAppStore.getState().selectedConversation).toBeNull();
});
});
// ── State machine ───────────────────────────────────────────
describe('Conversation state machine', () => {
it('should default to empty conversationStates', () => {
expect(useAppStore.getState().conversationStates).toEqual({});
});
it('should set conversation state', () => {
useAppStore.getState().setConversationState('conv-1', 'hydrating');
expect(useAppStore.getState().conversationStates['conv-1']).toBe('hydrating');
});
it('should transition through states', () => {
useAppStore.getState().setConversationState('conv-1', 'hydrating');
expect(useAppStore.getState().conversationStates['conv-1']).toBe('hydrating');
useAppStore.getState().setConversationState('conv-1', 'streaming');
expect(useAppStore.getState().conversationStates['conv-1']).toBe('streaming');
useAppStore.getState().setConversationState('conv-1', 'completed');
expect(useAppStore.getState().conversationStates['conv-1']).toBe('completed');
});
it('should handle multiple conversations independently', () => {
useAppStore.getState().setConversationState('conv-1', 'streaming');
useAppStore.getState().setConversationState('conv-2', 'idle');
expect(useAppStore.getState().conversationStates['conv-1']).toBe('streaming');
expect(useAppStore.getState().conversationStates['conv-2']).toBe('idle');
});
});
// ── Loading conversation / request correlation ──────────────
describe('loadingConversation / currentRequestId', () => {
it('should set loadingConversation', () => {
useAppStore.getState().setLoadingConversation('conv-1');
expect(useAppStore.getState().loadingConversation).toBe('conv-1');
});
it('should clear loadingConversation', () => {
useAppStore.getState().setLoadingConversation('conv-1');
useAppStore.getState().setLoadingConversation(null);
expect(useAppStore.getState().loadingConversation).toBeNull();
});
it('should set currentRequestId', () => {
useAppStore.getState().setCurrentRequestId('req-1');
expect(useAppStore.getState().currentRequestId).toBe('req-1');
});
});
// ── Conversation ended banner ───────────────────────────────
describe('conversationEndedBanner', () => {
it('should set banner', () => {
useAppStore.getState().setConversationEndedBanner('conv-1');
expect(useAppStore.getState().conversationEndedBanner).toBe('conv-1');
});
it('should clear banner', () => {
useAppStore.getState().setConversationEndedBanner('conv-1');
useAppStore.getState().setConversationEndedBanner(null);
expect(useAppStore.getState().conversationEndedBanner).toBeNull();
});
});
// ── setCases (atomic replace via init_state) ────────────────
describe('setCases atomic replace', () => {
it('should replace cases array', () => {
useAppStore.getState().setCases([{ id: 1, title: 'Test' } as any]);
expect(useAppStore.getState().cases).toHaveLength(1);
expect(useAppStore.getState().cases[0].id).toBe(1);
});
it('should set empty cases', () => {
useAppStore.getState().setCases([]);
expect(useAppStore.getState().cases).toHaveLength(0);
});
});
});
+183 -239
View File
@@ -1,5 +1,5 @@
import { create } from 'zustand';
import type { CaseRequest, Conversation, Message } from '@/types';
import type { CaseRequest, Conversation, ConversationSummary, Message } from '@/types';
import { api, type CaseFilters } from '@/services/api';
// ─────────────────────────────────────────────────────────────
@@ -8,6 +8,12 @@ import { api, type CaseFilters } from '@/services/api';
export type SidebarTab = 'all' | 'pending' | 'resolved';
export type WsStatus = 'connected' | 'disconnected' | 'reconnecting';
export type ConversationState = 'idle' | 'hydrating' | 'streaming' | 'completed';
export interface ResolvedCaseAlert {
caseId: string | number;
caseTitle: string;
}
interface AppState {
// ── Cases slice ──────────────────────────────────────────
@@ -20,17 +26,46 @@ interface AppState {
id: string | number,
data: { action: string; payload: Record<string, unknown>; note?: string },
) => Promise<void>;
startCase: (id: string | number) => Promise<void>;
setCases: (list: CaseRequest[]) => void;
// ── Conversations slice ──────────────────────────────────
conversations: Conversation[];
conversations: ConversationSummary[];
totalConversations: number;
conversationsOffset: number;
selectedConversation: Conversation | null;
selectedConversationId: string | null;
fetchConversations: () => Promise<void>;
upsertConversation: (c: Conversation) => void;
addMessage: (convId: string, msg: Message) => void;
fetchConversations: (limit?: number, offset?: number) => Promise<void>;
fetchConversationWithMessages: (id: string) => Promise<void>;
upsertConversation: (c: ConversationSummary) => void;
addMessage: (convId: string, _msg: Message) => void;
appendToken: (convId: string, msgId: string, token: string, index: number) => void;
completeStream: (convId: string, msgId: string, fullContent: string) => void;
setSelectedConversationId: (convId: string | null) => void;
removeConversation: (convId: string) => void;
setConversations: (list: ConversationSummary[]) => void;
// ── Idempotency & event dedup ────────────────────────────
processedEventIds: string[];
addProcessedEventId: (id: string) => boolean;
clearProcessedEventIds: () => void;
// ── Connection state ─────────────────────────────────────
initStateReceived: boolean;
setInitStateReceived: (v: boolean) => void;
// ── Conversation loading / request correlation ──────────
loadingConversation: string | null;
setLoadingConversation: (id: string | null) => void;
currentRequestId: string | null;
setCurrentRequestId: (id: string | null) => void;
// ── Conversation state machine ──────────────────────────
conversationStates: Record<string, ConversationState>;
setConversationState: (id: string, state: ConversationState) => void;
// ── Conversation ended banner ───────────────────────────
conversationEndedBanner: string | null;
setConversationEndedBanner: (id: string | null) => void;
// ── UI slice ─────────────────────────────────────────────
sidebarTab: SidebarTab;
@@ -38,11 +73,13 @@ interface AppState {
applicativeFilter: string | null;
isDarkMode: boolean;
wsStatus: WsStatus;
resolvedCaseAlert: ResolvedCaseAlert | null;
setSidebarTab: (tab: SidebarTab) => void;
setSearchQuery: (q: string) => void;
setApplicativeFilter: (app: string | null) => void;
toggleDarkMode: () => void;
setWsStatus: (status: WsStatus) => void;
setResolvedCaseAlert: (alert: ResolvedCaseAlert | null) => void;
}
// ─────────────────────────────────────────────────────────────
@@ -73,149 +110,11 @@ function persistDarkMode(value: boolean): void {
}
// ─────────────────────────────────────────────────────────────
// Token Streaming Buffer (Regla 4 — 50ms throttling, 20 fps)
// Token Streaming — directo sin buffer
// Cada chunk actualiza selectedConversation.messages directamente
// con mutación inmutable validada contra conversationId/messageId.
// ─────────────────────────────────────────────────────────────
interface PendingToken {
msgId: string;
token: string;
index: number;
}
interface ConversationBufferEntry {
pending: PendingToken[];
timer: ReturnType<typeof setTimeout> | null;
}
/**
* External buffer map — NOT stored in Zustand state to avoid
* triggering re-renders on every chunk. Each conversation gets
* its own entry with a pending queue and a 50ms flush timer.
*/
const conversationBuffers = new Map<string, ConversationBufferEntry>();
/**
* Flush all pending tokens for a given conversation into the store
* with a SINGLE `set()` call. Only updates the store if this
* conversation is the actively selected one (Regla 4: solo
* re-renderizar conversación seleccionada).
*/
function flushBuffer(
convId: string,
get: () => AppState,
set: (partial: AppState | ((state: AppState) => Partial<AppState>)) => void,
): void {
const entry = conversationBuffers.get(convId);
if (!entry) return;
// Clear the timer reference first
entry.timer = null;
// If the conversation no longer exists in the store, clean up the buffer
const currentState = get();
const convExists = currentState.conversations.some((c) => c.id === convId);
if (!convExists) {
conversationBuffers.delete(convId);
return;
}
// If nothing is pending, delete the entry and bail out
if (entry.pending.length === 0) {
conversationBuffers.delete(convId);
return;
}
// Only update the store for the selected conversation (Regla 4)
if (currentState.selectedConversationId !== convId) {
// Keep tokens in buffer — they'll be flushed when this conversation
// becomes selected, or cleared by completeStream.
return;
}
// Atomically take and clear the pending queue
const pendingToProcess = entry.pending;
entry.pending = [];
// Sort by index to guarantee correct order even with out-of-order delivery
pendingToProcess.sort((a, b) => a.index - b.index);
// Single batched set() call — ALL accumulated chunks in one update
set((state) => {
const convIndex = state.conversations.findIndex((c) => c.id === convId);
if (convIndex < 0) return state;
const conv = state.conversations[convIndex];
const messages = [...conv.messages];
let hasChanges = false;
for (const pending of pendingToProcess) {
const msgIndex = messages.findIndex((m) => m.id === pending.msgId);
if (msgIndex < 0) continue;
const msg = { ...messages[msgIndex] };
const existingChunks: Array<{ token: string; index: number }> =
(msg.metadata?._chunks as Array<{ token: string; index: number }>) ?? [];
const newChunks = [
...existingChunks,
{ token: pending.token, index: pending.index },
];
newChunks.sort((a, b) => a.index - b.index);
messages[msgIndex] = {
...msg,
content: newChunks.map((ch) => ch.token).join(''),
metadata: { ...msg.metadata, _chunks: newChunks },
isStreaming: true,
};
hasChanges = true;
}
if (!hasChanges) return state;
return {
conversations: state.conversations.map((c, i) =>
i === convIndex ? { ...conv, messages } : c,
),
};
});
}
/**
* Schedule a flush for the given conversation in ~50ms.
* Does nothing if a timer is already pending for this conversation.
*/
function scheduleBufferFlush(
convId: string,
get: () => AppState,
set: (partial: AppState | ((state: AppState) => Partial<AppState>)) => void,
): void {
const entry = conversationBuffers.get(convId);
if (!entry || entry.timer !== null) return;
entry.timer = setTimeout(() => {
flushBuffer(convId, get, set);
}, 50);
}
/**
* Immediately flush all pending tokens for the given conversation.
* Used when switching to a conversation mid-stream.
*/
function forceFlushBuffer(
convId: string,
get: () => AppState,
set: (partial: AppState | ((state: AppState) => Partial<AppState>)) => void,
): void {
const entry = conversationBuffers.get(convId);
if (!entry) return;
if (entry.timer !== null) {
clearTimeout(entry.timer);
}
flushBuffer(convId, get, set);
}
// ─────────────────────────────────────────────────────────────
// Store
// ─────────────────────────────────────────────────────────────
@@ -252,6 +151,22 @@ export const useAppStore = create<AppState>((set, get) => ({
return { cases: [c, ...state.cases] };
}),
setCases: (list: CaseRequest[]) => set({ cases: list }),
startCase: async (id: string | number) => {
try {
const updated = await api.startCase(id);
const index = get().cases.findIndex((c) => c.id === id);
if (index >= 0) {
const cases = [...get().cases];
cases[index] = updated as any;
set({ cases });
}
} catch (err) {
console.error('[Store] startCase failed:', err);
}
},
resolveCase: async (id, data) => {
try {
const updatedCase = await api.resolveCase(id, data);
@@ -274,19 +189,51 @@ export const useAppStore = create<AppState>((set, get) => ({
// ── Conversations initial state ──────────────────────────
conversations: [],
totalConversations: 0,
conversationsOffset: 0,
selectedConversationId: null,
selectedConversation: null,
fetchConversations: async () => {
fetchConversations: async (limit = 20, offset = 0) => {
try {
const conversations = await api.getActiveConversations();
set({ conversations });
const data = await api.getActiveConversations(limit, offset);
set((state) => ({
conversations: offset === 0 ? data.items : [...state.conversations, ...data.items],
totalConversations: data.total,
conversationsOffset: offset + data.items.length,
}));
} catch (err) {
console.error('[Store] fetchConversations failed:', err);
set({ conversations: [] });
if (offset === 0) set({ conversations: [], totalConversations: 0, conversationsOffset: 0 });
}
},
upsertConversation: (c: Conversation) =>
fetchConversationWithMessages: async (id: string) => {
try {
const conversation = await api.getConversation(id);
set((state) => {
// Regla 2: si ya hay un stream activo, merge en lugar de sobrescribir
const current = state.selectedConversation;
if (current && current.id === id) {
const streamingMsg = current.messages.find((m) => m.isStreaming);
if (streamingMsg) {
// Mantener el mensaje en streaming, mergear el resto
const backendMsgs = conversation.messages || [];
const merged = backendMsgs.map((bm) => {
const streamMatch = current.messages.find((cm) => cm.id === bm.id && cm.isStreaming);
return streamMatch || bm;
});
return { selectedConversation: { ...conversation, messages: merged } as any };
}
}
return { selectedConversation: conversation as any };
});
} catch (err) {
console.error('[Store] fetchConversationWithMessages failed:', err);
}
},
upsertConversation: (c: ConversationSummary) =>
set((state) => {
const index = state.conversations.findIndex(
(existing) => existing.id === c.id,
@@ -299,7 +246,7 @@ export const useAppStore = create<AppState>((set, get) => ({
return { conversations: [...state.conversations, c] };
}),
addMessage: (convId: string, msg: Message) =>
addMessage: (convId: string, _msg: Message) =>
set((state) => {
const convIndex = state.conversations.findIndex(
(c) => c.id === convId,
@@ -307,115 +254,110 @@ export const useAppStore = create<AppState>((set, get) => ({
if (convIndex < 0) return state;
const updated = [...state.conversations];
updated[convIndex] = {
...updated[convIndex],
messages: [...updated[convIndex].messages, msg],
};
// Messages updated via selectedConversation on demand
return { conversations: updated };
}),
appendToken: (convId: string, msgId: string, token: string, index: number) => {
// Step 1: Add chunk to the conversation's external buffer
let entry = conversationBuffers.get(convId);
if (!entry) {
entry = { pending: [], timer: null };
conversationBuffers.set(convId, entry);
}
entry.pending.push({ msgId, token, index });
appendToken: (convId, msgId, token, _index) => {
set((state) => {
const sel = state.selectedConversation;
if (!sel || sel.id !== convId) return {}; // guard: conversación correcta
// Step 2: Schedule a flush only if this is the selected conversation
// (non-selected conversations accumulate in buffer without triggering re-renders)
const state = get();
if (state.selectedConversationId === convId) {
scheduleBufferFlush(convId, get, set);
}
let msgIdx = sel.messages.findIndex((m) => m.id === msgId);
if (msgIdx < 0) {
// Crear placeholder si no existe
const messages = [...sel.messages, {
id: msgId,
conversationId: convId,
role: 'agent' as any,
content: token,
timestamp: new Date().toISOString(),
isStreaming: true,
}];
return { selectedConversation: { ...sel, messages } };
}
const messages = [...sel.messages];
messages[msgIdx] = {
...messages[msgIdx],
content: messages[msgIdx].content + token,
isStreaming: true,
};
return { selectedConversation: { ...sel, messages } };
});
},
completeStream: (convId: string, msgId: string, fullContent: string) => {
// Step 1: Clear the conversation's buffer — no more tokens expected
const entry = conversationBuffers.get(convId);
if (entry) {
if (entry.timer !== null) {
clearTimeout(entry.timer);
}
conversationBuffers.delete(convId);
}
// Step 2: Perform a single store update to set the final content
completeStream: (convId, msgId, fullContent) => {
set((state) => {
const convIndex = state.conversations.findIndex(
(c) => c.id === convId,
);
if (convIndex < 0) return state;
const conv = state.conversations[convIndex];
const msgIndex = conv.messages.findIndex((m) => m.id === msgId);
if (msgIndex < 0) return state;
const messages = [...conv.messages];
const msg = { ...messages[msgIndex] };
// Clear chunk buffer — rebuild metadata without _chunks
const cleanMetadata: Record<string, unknown> = {};
if (msg.metadata) {
for (const [key, value] of Object.entries(msg.metadata)) {
if (key !== '_chunks') {
cleanMetadata[key] = value;
}
}
}
messages[msgIndex] = {
...msg,
content: fullContent,
isStreaming: false,
metadata: cleanMetadata,
};
return {
conversations: state.conversations.map((c, i) =>
i === convIndex ? { ...conv, messages } : c,
),
};
const sel = state.selectedConversation;
if (!sel || sel.id !== convId) return {};
const msgIdx = sel.messages.findIndex((m) => m.id === msgId);
if (msgIdx < 0) return {};
const messages = [...sel.messages];
messages[msgIdx] = { ...messages[msgIdx], content: fullContent, isStreaming: false };
return { selectedConversation: { ...sel, messages } };
});
},
setSelectedConversationId: (convId: string | null) => {
// Force-flush any pending buffer for the newly selected conversation
const prevSelected = get().selectedConversationId;
set({ selectedConversationId: convId });
if (convId !== null && convId !== prevSelected) {
// If switching to a conversation that has buffered tokens, flush them immediately
forceFlushBuffer(convId, get, set);
}
},
removeConversation: (convId: string) => {
// Clear the buffer for this conversation
const entry = conversationBuffers.get(convId);
if (entry) {
if (entry.timer !== null) {
clearTimeout(entry.timer);
}
conversationBuffers.delete(convId);
}
// ── Atomic replacements (WS init_state) ────────────────
set((state) => ({
conversations: state.conversations.filter((c) => c.id !== convId),
selectedConversationId:
state.selectedConversationId === convId
? null
: state.selectedConversationId,
}));
setConversations: (list: ConversationSummary[]) =>
set({ conversations: list, totalConversations: list.length }),
// ── Idempotency & event dedup ──────────────────────────
processedEventIds: [],
addProcessedEventId: (id: string) => {
const current = get().processedEventIds;
// If already present, reject duplicate
if (current.includes(id)) return false;
// LRU eviction: max 1000 entries, drop oldest if full
const updated = current.length >= 1000 ? current.slice(1) : current;
set({ processedEventIds: [...updated, id] });
return true;
},
clearProcessedEventIds: () => set({ processedEventIds: [] }),
// ── Connection state ───────────────────────────────────
initStateReceived: false,
setInitStateReceived: (v: boolean) => set({ initStateReceived: v }),
// ── Conversation loading / request correlation ─────────
loadingConversation: null,
setLoadingConversation: (id: string | null) => set({ loadingConversation: id }),
currentRequestId: null,
setCurrentRequestId: (id: string | null) => set({ currentRequestId: id }),
// ── Conversation state machine ─────────────────────────
conversationStates: {},
setConversationState: (id: string, state: ConversationState) =>
set((prev) => ({
conversationStates: { ...prev.conversationStates, [id]: state },
})),
// ── Conversation ended banner ──────────────────────────
conversationEndedBanner: null,
setConversationEndedBanner: (id: string | null) =>
set({ conversationEndedBanner: id }),
// ── UI initial state ──────────────────────────────────
sidebarTab: 'all',
searchQuery: '',
applicativeFilter: null,
isDarkMode: readDarkMode(),
wsStatus: 'disconnected',
resolvedCaseAlert: null,
setSidebarTab: (tab) => set({ sidebarTab: tab }),
@@ -430,5 +372,7 @@ export const useAppStore = create<AppState>((set, get) => ({
return { isDarkMode: next };
}),
setResolvedCaseAlert: (alert) => set({ resolvedCaseAlert: alert }),
setWsStatus: (status) => set({ wsStatus: status }),
}));
+5 -2
View File
@@ -58,15 +58,18 @@ export interface Message {
metadata?: Record<string, unknown>;
}
export interface Conversation {
export interface ConversationSummary {
id: string;
clientId: string;
agentId: string;
status: 'active' | 'paused' | 'ended';
messages: Message[];
createdAt: string;
}
export interface Conversation extends ConversationSummary {
messages: Message[];
}
export interface FormField {
key: string;
label: string;
+48 -4
View File
@@ -43,7 +43,8 @@ export type InitStatePayload = z.infer<typeof InitStatePayloadSchema>;
// 2.2 conversation_started — Nueva conversación
export const ConversationStartedPayloadSchema = z.object({
conversation: z.record(z.unknown()),
conversationId: z.string(),
agentId: z.string().optional(),
});
export type ConversationStartedPayload = z.infer<typeof ConversationStartedPayloadSchema>;
@@ -68,6 +69,8 @@ export type UserMessagePayload = z.infer<typeof UserMessagePayloadSchema>;
export const AgentStreamStartedPayloadSchema = z.object({
conversationId: z.string(),
messageId: z.string(),
agentName: z.string().optional(),
agentType: z.string().optional(),
});
export type AgentStreamStartedPayload = z.infer<typeof AgentStreamStartedPayloadSchema>;
@@ -101,15 +104,21 @@ export type AgentStatusUpdatePayload = z.infer<typeof AgentStatusUpdatePayloadSc
// 2.9 hitl_request — Se requiere intervención humana
export const HITLRequestPayloadSchema = z.object({
case: z.record(z.unknown()),
conversationId: z.string(),
id: z.number(),
title: z.string(),
tipoSolicitud: z.string(),
uiPattern: z.string(),
conversationId: z.string().optional(),
correlationId: z.string().optional(),
status: z.string(),
});
export type HITLRequestPayload = z.infer<typeof HITLRequestPayloadSchema>;
// 2.10 hitl_resolved — Caso resuelto (broadcast)
// caseId puede venir como número o string desde el backend
export const HITLResolvedPayloadSchema = z.object({
caseId: z.string(),
caseId: z.union([z.number(), z.string()]),
resolution: z.record(z.unknown()),
});
@@ -124,6 +133,38 @@ export const ErrorPayloadSchema = z.object({
export type ErrorPayload = z.infer<typeof ErrorPayloadSchema>;
// 2.12 heartbeat — Señal de salud de la conexión (no requiere acción en UI)
export const HeartbeatPayloadSchema = z.object({
timestamp: z.string(),
});
export type HeartbeatPayload = z.infer<typeof HeartbeatPayloadSchema>;
// 2.13 conversation_assigned — Conversación asignada a un asesor
export const ConversationAssignedPayloadSchema = z.object({
conversationId: z.string(),
advisorId: z.string(),
assignedAt: z.string(),
leaseExpiresAt: z.string().optional(),
});
export type ConversationAssignedPayload = z.infer<typeof ConversationAssignedPayloadSchema>;
// 2.14 internal_note — Nota interna redifundida por el servidor
export const InternalNoteServerPayloadSchema = z.object({
conversationId: z.string(),
message: z.object({
id: z.string(),
conversationId: z.string(),
role: z.literal('internal'),
content: z.string(),
advisorId: z.string().optional(),
timestamp: z.string(),
}),
});
export type InternalNoteServerPayload = z.infer<typeof InternalNoteServerPayloadSchema>;
// ═══════════════════════════════════════════════════════════════
// 3. Eventos cliente → servidor (Sección 8.4)
// ═══════════════════════════════════════════════════════════════
@@ -155,6 +196,9 @@ export const serverEventPayloadSchemas: Record<string, z.ZodType<unknown>> = {
agent_status_update: AgentStatusUpdatePayloadSchema,
hitl_request: HITLRequestPayloadSchema,
hitl_resolved: HITLResolvedPayloadSchema,
heartbeat: HeartbeatPayloadSchema,
conversation_assigned: ConversationAssignedPayloadSchema,
internal_note: InternalNoteServerPayloadSchema,
error: ErrorPayloadSchema,
};
+1
View File
@@ -3,6 +3,7 @@
interface ImportMetaEnv {
readonly VITE_API_BASE_URL: string;
readonly VITE_WS_URL: string;
readonly VITE_LOGIN_URL: string;
readonly VITE_ENABLE_MSW: string;
}