Compare commits

...

11 Commits

Author SHA1 Message Date
Igor Loskutov
b84fd1fc24 cleanup 2026-01-13 12:46:03 -05:00
Igor Loskutov
3652de9fca md 2026-01-13 12:44:43 -05:00
Igor Loskutov
68df825734 test: fix WebSocket chat tests using async approach
Replaced TestClient-based tests with proper async WebSocket testing
using httpx_ws and threaded server pattern. TestClient has event loop
issues with WebSocket connections that were causing all tests to fail.

Changes:
- Rewrote all WebSocket tests to use aconnect_ws from httpx_ws
- Added chat_appserver fixture using threaded Uvicorn server
- Tests now use separate event loop in server thread
- All 6 tests now pass without asyncio/event loop errors
- Matches existing pattern from test_transcripts_rtc_ws.py

Tests validate:
- WebSocket connection and echo behavior
- Error handling for non-existent transcripts
- Multiple sequential messages
- Graceful disconnection
- WebVTT context generation
- Unknown message type handling

Closes fn-1.8 (End-to-end testing)
2026-01-12 20:17:42 -05:00
Igor Loskutov
8ca5324c1a feat: integrate TranscriptChatModal and button into transcript page 2026-01-12 20:08:59 -05:00
Igor Loskutov
39e0b89e67 feat: add TranscriptChatModal and TranscriptChatButton components 2026-01-12 19:59:01 -05:00
Igor Loskutov
544793a24f chore: mark fn-1.5 as done (Frontend WebSocket hook)
Task fn-1.5 completed - useTranscriptChat React hook already implemented in commit 2dfe82af.

Hook provides:
- WebSocket connection to /v1/transcripts/{id}/chat endpoint
- Token streaming with ref-based accumulation
- Message history management (user + assistant)
- Memory leak prevention with isMountedRef
- TypeScript type safety
- Proper WebSocket lifecycle and cleanup

Updated task documentation with acceptance criteria and evidence.
2026-01-12 19:49:14 -05:00
Igor Loskutov
088451645a chore: mark fn-1.4 as done (WebSocket route registration) 2026-01-12 19:42:26 -05:00
Igor Loskutov
2dfe82afbc feat: add useTranscriptChat WebSocket hook
Task 5: Frontend WebSocket Hook
- Creates React hook for bidirectional chat WebSocket
- Handles token streaming with proper state accumulation
- Manages conversation history (user + assistant messages)
- Prevents memory leaks with isMounted check
- Proper cleanup on unmount
- Type-safe Message interface

Validated:
- No React dependency issues (removed currentStreamingText from deps)
- No stale closure bugs (using ref for streaming text)
- Proper mounted state tracking
- Lint passes with no errors
- TypeScript types correctly defined
- WebSocket cleanup on unmount

~100 lines
2026-01-12 18:44:09 -05:00
Igor Loskutov
b461ebb488 feat: register transcript chat WebSocket route
- Import transcripts_chat router
- Register /v1/transcripts/{id}/chat endpoint
- Completes LLM streaming integration (fn-1.3)
2026-01-12 18:41:11 -05:00
Igor Loskutov
0b5112cabc feat: add LLM streaming integration to transcript chat
Task 3: LLM Streaming Integration

- Import Settings, ChatMessage, MessageRole from llama-index
- Configure LLM with temperature 0.7 on connection
- Build system message with WebVTT transcript context (max 15k chars)
- Initialize conversation history with system message
- Handle 'message' type from client to trigger LLM streaming
- Stream LLM response using Settings.llm.astream_chat()
- Send tokens incrementally via 'token' messages
- Send 'done' message when streaming completes
- Maintain conversation history across multiple messages
- Add error handling with 'error' message type
- Add message protocol validation test

Implements Tasks 3 & 4 from TASKS.md
2026-01-12 18:28:43 -05:00
Igor Loskutov
316f7b316d feat: add WebVTT context generation to chat WebSocket endpoint
- Import topics_to_webvtt_named and recordings controller
- Add _get_is_multitrack helper function
- Generate WebVTT context on WebSocket connection
- Add get_context message type to retrieve WebVTT
- Maintain backward compatibility with echo for other messages
- Add test fixture and test for WebVTT context generation

Implements task fn-1.2: WebVTT context generation for transcript chat
2026-01-12 18:24:47 -05:00
6 changed files with 620 additions and 0 deletions

View File

@@ -18,6 +18,7 @@ from reflector.views.rooms import router as rooms_router
from reflector.views.rtc_offer import router as rtc_offer_router from reflector.views.rtc_offer import router as rtc_offer_router
from reflector.views.transcripts import router as transcripts_router from reflector.views.transcripts import router as transcripts_router
from reflector.views.transcripts_audio import router as transcripts_audio_router from reflector.views.transcripts_audio import router as transcripts_audio_router
from reflector.views.transcripts_chat import router as transcripts_chat_router
from reflector.views.transcripts_participants import ( from reflector.views.transcripts_participants import (
router as transcripts_participants_router, router as transcripts_participants_router,
) )
@@ -90,6 +91,7 @@ app.include_router(transcripts_participants_router, prefix="/v1")
app.include_router(transcripts_speaker_router, prefix="/v1") app.include_router(transcripts_speaker_router, prefix="/v1")
app.include_router(transcripts_upload_router, prefix="/v1") app.include_router(transcripts_upload_router, prefix="/v1")
app.include_router(transcripts_websocket_router, prefix="/v1") app.include_router(transcripts_websocket_router, prefix="/v1")
app.include_router(transcripts_chat_router, prefix="/v1")
app.include_router(transcripts_webrtc_router, prefix="/v1") app.include_router(transcripts_webrtc_router, prefix="/v1")
app.include_router(transcripts_process_router, prefix="/v1") app.include_router(transcripts_process_router, prefix="/v1")
app.include_router(user_router, prefix="/v1") app.include_router(user_router, prefix="/v1")

View File

@@ -0,0 +1,133 @@
"""
Transcripts chat API
====================
WebSocket endpoint for bidirectional chat with LLM about transcript content.
"""
from typing import Optional
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from llama_index.core import Settings
from llama_index.core.base.llms.types import ChatMessage, MessageRole
from reflector.auth.auth_jwt import JWTAuth
from reflector.db.recordings import recordings_controller
from reflector.db.transcripts import transcripts_controller
from reflector.db.users import user_controller
from reflector.llm import LLM
from reflector.settings import settings
from reflector.utils.transcript_formats import topics_to_webvtt_named
router = APIRouter()
async def _get_is_multitrack(transcript) -> bool:
"""Detect if transcript is from multitrack recording."""
if not transcript.recording_id:
return False
recording = await recordings_controller.get_by_id(transcript.recording_id)
return recording is not None and recording.is_multitrack
@router.websocket("/transcripts/{transcript_id}/chat")
async def transcript_chat_websocket(
transcript_id: str,
websocket: WebSocket,
):
"""WebSocket endpoint for chatting with LLM about transcript content."""
# 1. Auth check (optional) - extract token from WebSocket subprotocol header
# Browser can't send Authorization header for WS; use subprotocol: ["bearer", token]
raw_subprotocol = websocket.headers.get("sec-websocket-protocol") or ""
parts = [p.strip() for p in raw_subprotocol.split(",") if p.strip()]
token: Optional[str] = None
negotiated_subprotocol: Optional[str] = None
if len(parts) >= 2 and parts[0].lower() == "bearer":
negotiated_subprotocol = "bearer"
token = parts[1]
user_id: Optional[str] = None
if token:
try:
payload = JWTAuth().verify_token(token)
authentik_uid = payload.get("sub")
if authentik_uid:
user = await user_controller.get_by_authentik_uid(authentik_uid)
if user:
user_id = user.id
except Exception:
# Auth failed - continue as anonymous
pass
# Get transcript (respects user_id for private transcripts)
transcript = await transcripts_controller.get_by_id_for_http(
transcript_id, user_id=user_id
)
if not transcript:
await websocket.close(code=1008) # Policy violation (not found/unauthorized)
return
# 2. Accept connection (with negotiated subprotocol if present)
await websocket.accept(subprotocol=negotiated_subprotocol)
# 3. Generate WebVTT context
is_multitrack = await _get_is_multitrack(transcript)
webvtt = topics_to_webvtt_named(
transcript.topics, transcript.participants, is_multitrack
)
# Truncate if needed (15k char limit for POC)
webvtt_truncated = webvtt[:15000] if len(webvtt) > 15000 else webvtt
# 4. Configure LLM
llm = LLM(settings=settings, temperature=0.7)
# 5. System message with transcript context
system_msg = f"""You are analyzing this meeting transcript (WebVTT):
{webvtt_truncated}
Answer questions about content, speakers, timeline. Include timestamps when relevant."""
# 6. Conversation history
conversation_history = [ChatMessage(role=MessageRole.SYSTEM, content=system_msg)]
try:
# 7. Message loop
while True:
data = await websocket.receive_json()
if data.get("type") == "get_context":
# Return WebVTT context (for debugging/testing)
await websocket.send_json({"type": "context", "webvtt": webvtt})
continue
if data.get("type") != "message":
# Echo unknown types for backward compatibility
await websocket.send_json({"type": "echo", "data": data})
continue
# Add user message to history
user_msg = ChatMessage(role=MessageRole.USER, content=data.get("text", ""))
conversation_history.append(user_msg)
# Stream LLM response
assistant_msg = ""
chat_stream = await Settings.llm.astream_chat(conversation_history)
async for chunk in chat_stream:
token = chunk.delta or ""
if token:
await websocket.send_json({"type": "token", "text": token})
assistant_msg += token
# Save assistant response to history
conversation_history.append(
ChatMessage(role=MessageRole.ASSISTANT, content=assistant_msg)
)
await websocket.send_json({"type": "done"})
except WebSocketDisconnect:
pass
except Exception as e:
await websocket.send_json({"type": "error", "message": str(e)})

View File

@@ -0,0 +1,234 @@
"""Tests for transcript chat WebSocket endpoint."""
import asyncio
import threading
import time
from pathlib import Path
import pytest
from httpx_ws import aconnect_ws
from uvicorn import Config, Server
from reflector.db.transcripts import (
SourceKind,
TranscriptParticipant,
TranscriptTopic,
transcripts_controller,
)
from reflector.processors.types import Word
@pytest.fixture
def chat_appserver(tmpdir, setup_database):
"""Start a real HTTP server for WebSocket testing."""
from reflector.app import app
from reflector.db import get_database
from reflector.settings import settings
DATA_DIR = settings.DATA_DIR
settings.DATA_DIR = Path(tmpdir)
# Start server in separate thread with its own event loop
host = "127.0.0.1"
port = 1256 # Different port from rtc tests
server_started = threading.Event()
server_exception = None
server_instance = None
def run_server():
nonlocal server_exception, server_instance
try:
# Create new event loop for this thread
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
config = Config(app=app, host=host, port=port, loop=loop)
server_instance = Server(config)
async def start_server():
# Initialize database connection in this event loop
database = get_database()
await database.connect()
try:
await server_instance.serve()
finally:
await database.disconnect()
# Signal that server is starting
server_started.set()
loop.run_until_complete(start_server())
except Exception as e:
server_exception = e
server_started.set()
finally:
loop.close()
server_thread = threading.Thread(target=run_server, daemon=True)
server_thread.start()
# Wait for server to start
server_started.wait(timeout=30)
if server_exception:
raise server_exception
# Wait for server to be fully ready
time.sleep(1)
yield server_instance, host, port
# Stop server
if server_instance:
server_instance.should_exit = True
server_thread.join(timeout=30)
settings.DATA_DIR = DATA_DIR
@pytest.fixture
async def test_transcript(setup_database):
"""Create a test transcript for WebSocket tests."""
transcript = await transcripts_controller.add(
name="Test Transcript for Chat", source_kind=SourceKind.FILE
)
return transcript
@pytest.fixture
async def test_transcript_with_content(setup_database):
"""Create a test transcript with actual content for WebVTT generation."""
transcript = await transcripts_controller.add(
name="Test Transcript with Content", source_kind=SourceKind.FILE
)
# Add participants
await transcripts_controller.update(
transcript,
{
"participants": [
TranscriptParticipant(id="1", speaker=0, name="Alice").model_dump(),
TranscriptParticipant(id="2", speaker=1, name="Bob").model_dump(),
]
},
)
# Add topic with words
await transcripts_controller.upsert_topic(
transcript,
TranscriptTopic(
title="Introduction",
summary="Opening remarks",
timestamp=0.0,
words=[
Word(text="Hello ", start=0.0, end=1.0, speaker=0),
Word(text="everyone.", start=1.0, end=2.0, speaker=0),
Word(text="Hi ", start=2.0, end=3.0, speaker=1),
Word(text="there!", start=3.0, end=4.0, speaker=1),
],
),
)
return transcript
@pytest.mark.asyncio
async def test_chat_websocket_connection_success(test_transcript, chat_appserver):
"""Test successful WebSocket connection to chat endpoint."""
server, host, port = chat_appserver
base_url = f"ws://{host}:{port}/v1"
async with aconnect_ws(f"{base_url}/transcripts/{test_transcript.id}/chat") as ws:
# Send unknown message type to test echo behavior
await ws.send_json({"type": "test", "text": "Hello"})
# Should receive echo for unknown types
response = await ws.receive_json()
assert response["type"] == "echo"
assert response["data"]["type"] == "test"
@pytest.mark.asyncio
async def test_chat_websocket_nonexistent_transcript(chat_appserver):
"""Test WebSocket connection fails for nonexistent transcript."""
server, host, port = chat_appserver
base_url = f"ws://{host}:{port}/v1"
# Connection should fail or disconnect immediately for non-existent transcript
# Different behavior from successful connection
with pytest.raises(Exception): # Will raise on connection or first operation
async with aconnect_ws(f"{base_url}/transcripts/nonexistent-id/chat") as ws:
await ws.send_json({"type": "message", "text": "Hello"})
await ws.receive_json()
@pytest.mark.asyncio
async def test_chat_websocket_multiple_messages(test_transcript, chat_appserver):
"""Test sending multiple messages through WebSocket."""
server, host, port = chat_appserver
base_url = f"ws://{host}:{port}/v1"
async with aconnect_ws(f"{base_url}/transcripts/{test_transcript.id}/chat") as ws:
# Send multiple unknown message types (testing echo behavior)
messages = ["First message", "Second message", "Third message"]
for i, msg in enumerate(messages):
await ws.send_json({"type": f"test{i}", "text": msg})
response = await ws.receive_json()
assert response["type"] == "echo"
assert response["data"]["type"] == f"test{i}"
assert response["data"]["text"] == msg
@pytest.mark.asyncio
async def test_chat_websocket_disconnect_graceful(test_transcript, chat_appserver):
"""Test WebSocket disconnects gracefully."""
server, host, port = chat_appserver
base_url = f"ws://{host}:{port}/v1"
async with aconnect_ws(f"{base_url}/transcripts/{test_transcript.id}/chat") as ws:
await ws.send_json({"type": "message", "text": "Hello"})
await ws.receive_json()
# Close handled by context manager - should not raise
@pytest.mark.asyncio
async def test_chat_websocket_context_generation(
test_transcript_with_content, chat_appserver
):
"""Test WebVTT context is generated on connection."""
server, host, port = chat_appserver
base_url = f"ws://{host}:{port}/v1"
async with aconnect_ws(
f"{base_url}/transcripts/{test_transcript_with_content.id}/chat"
) as ws:
# Request context
await ws.send_json({"type": "get_context"})
# Receive context response
response = await ws.receive_json()
assert response["type"] == "context"
assert "webvtt" in response
# Verify WebVTT format
webvtt = response["webvtt"]
assert webvtt.startswith("WEBVTT")
assert "<v Alice>" in webvtt
assert "<v Bob>" in webvtt
assert "Hello everyone." in webvtt
assert "Hi there!" in webvtt
@pytest.mark.asyncio
async def test_chat_websocket_unknown_message_type(test_transcript, chat_appserver):
"""Test unknown message types are echoed back."""
server, host, port = chat_appserver
base_url = f"ws://{host}:{port}/v1"
async with aconnect_ws(f"{base_url}/transcripts/{test_transcript.id}/chat") as ws:
# Send unknown message type
await ws.send_json({"type": "unknown", "data": "test"})
# Should receive echo
response = await ws.receive_json()
assert response["type"] == "echo"
assert response["data"]["type"] == "unknown"

View File

@@ -0,0 +1,103 @@
"use client";
import { useState } from "react";
import { Box, Dialog, Input, IconButton } from "@chakra-ui/react";
import { MessageCircle } from "lucide-react";
import Markdown from "react-markdown";
import "../../styles/markdown.css";
import type { Message } from "./useTranscriptChat";
interface TranscriptChatModalProps {
open: boolean;
onClose: () => void;
messages: Message[];
sendMessage: (text: string) => void;
isStreaming: boolean;
currentStreamingText: string;
}
export function TranscriptChatModal({
open,
onClose,
messages,
sendMessage,
isStreaming,
currentStreamingText,
}: TranscriptChatModalProps) {
const [input, setInput] = useState("");
const handleSend = () => {
if (!input.trim()) return;
sendMessage(input);
setInput("");
};
return (
<Dialog.Root open={open} onOpenChange={(e) => !e.open && onClose()}>
<Dialog.Backdrop />
<Dialog.Positioner>
<Dialog.Content maxW="500px" h="600px">
<Dialog.Header>Transcript Chat</Dialog.Header>
<Dialog.Body overflowY="auto">
{messages.map((msg) => (
<Box
key={msg.id}
p={3}
mb={2}
bg={msg.role === "user" ? "blue.50" : "gray.50"}
borderRadius="md"
>
{msg.role === "user" ? (
msg.text
) : (
<div className="markdown">
<Markdown>{msg.text}</Markdown>
</div>
)}
</Box>
))}
{isStreaming && (
<Box p={3} bg="gray.50" borderRadius="md">
<div className="markdown">
<Markdown>{currentStreamingText}</Markdown>
</div>
<Box as="span" className="animate-pulse">
</Box>
</Box>
)}
</Dialog.Body>
<Dialog.Footer>
<Input
value={input}
onChange={(e) => setInput(e.target.value)}
onKeyDown={(e) => e.key === "Enter" && handleSend()}
placeholder="Ask about transcript..."
disabled={isStreaming}
/>
</Dialog.Footer>
</Dialog.Content>
</Dialog.Positioner>
</Dialog.Root>
);
}
export function TranscriptChatButton({ onClick }: { onClick: () => void }) {
return (
<IconButton
position="fixed"
bottom="24px"
right="24px"
onClick={onClick}
size="lg"
colorPalette="blue"
borderRadius="full"
aria-label="Open chat"
>
<MessageCircle />
</IconButton>
);
}

View File

@@ -18,9 +18,15 @@ import {
Skeleton, Skeleton,
Text, Text,
Spinner, Spinner,
useDisclosure,
} from "@chakra-ui/react"; } from "@chakra-ui/react";
import { useTranscriptGet } from "../../../lib/apiHooks"; import { useTranscriptGet } from "../../../lib/apiHooks";
import { TranscriptStatus } from "../../../lib/transcript"; import { TranscriptStatus } from "../../../lib/transcript";
import {
TranscriptChatModal,
TranscriptChatButton,
} from "../TranscriptChatModal";
import { useTranscriptChat } from "../useTranscriptChat";
type TranscriptDetails = { type TranscriptDetails = {
params: Promise<{ params: Promise<{
@@ -53,6 +59,9 @@ export default function TranscriptDetails(details: TranscriptDetails) {
const [finalSummaryElement, setFinalSummaryElement] = const [finalSummaryElement, setFinalSummaryElement] =
useState<HTMLDivElement | null>(null); useState<HTMLDivElement | null>(null);
const { open, onOpen, onClose } = useDisclosure();
const chat = useTranscriptChat(transcriptId);
useEffect(() => { useEffect(() => {
if (!waiting || !transcript.data) return; if (!waiting || !transcript.data) return;
@@ -119,6 +128,15 @@ export default function TranscriptDetails(details: TranscriptDetails) {
return ( return (
<> <>
<TranscriptChatModal
open={open}
onClose={onClose}
messages={chat.messages}
sendMessage={chat.sendMessage}
isStreaming={chat.isStreaming}
currentStreamingText={chat.currentStreamingText}
/>
<TranscriptChatButton onClick={onOpen} />
<Grid <Grid
templateColumns="1fr" templateColumns="1fr"
templateRows="auto minmax(0, 1fr)" templateRows="auto minmax(0, 1fr)"

View File

@@ -0,0 +1,130 @@
"use client";
import { useEffect, useState, useRef } from "react";
import { getSession } from "next-auth/react";
import { WEBSOCKET_URL } from "../../lib/apiClient";
import { assertExtendedToken } from "../../lib/types";
export type Message = {
id: string;
role: "user" | "assistant";
text: string;
timestamp: Date;
};
export type UseTranscriptChat = {
messages: Message[];
sendMessage: (text: string) => void;
isStreaming: boolean;
currentStreamingText: string;
};
export const useTranscriptChat = (transcriptId: string): UseTranscriptChat => {
const [messages, setMessages] = useState<Message[]>([]);
const [isStreaming, setIsStreaming] = useState(false);
const [currentStreamingText, setCurrentStreamingText] = useState("");
const wsRef = useRef<WebSocket | null>(null);
const streamingTextRef = useRef<string>("");
const isMountedRef = useRef<boolean>(true);
useEffect(() => {
isMountedRef.current = true;
const connectWebSocket = async () => {
const url = `${WEBSOCKET_URL}/v1/transcripts/${transcriptId}/chat`;
// Get auth token for WebSocket subprotocol
let protocols: string[] | undefined;
try {
const session = await getSession();
if (session) {
const token = assertExtendedToken(session).accessToken;
// Pass token via subprotocol: ["bearer", token]
protocols = ["bearer", token];
}
} catch (error) {
console.warn("Failed to get auth token for WebSocket:", error);
}
const ws = new WebSocket(url, protocols);
wsRef.current = ws;
ws.onopen = () => {
console.log("Chat WebSocket connected");
};
ws.onmessage = (event) => {
if (!isMountedRef.current) return;
const msg = JSON.parse(event.data);
switch (msg.type) {
case "token":
setIsStreaming(true);
streamingTextRef.current += msg.text;
setCurrentStreamingText(streamingTextRef.current);
break;
case "done":
// CRITICAL: Save the text BEFORE resetting the ref
// The setMessages callback may execute later, after ref is reset
const finalText = streamingTextRef.current;
setMessages((prev) => [
...prev,
{
id: Date.now().toString(),
role: "assistant",
text: finalText,
timestamp: new Date(),
},
]);
streamingTextRef.current = "";
setCurrentStreamingText("");
setIsStreaming(false);
break;
case "error":
console.error("Chat error:", msg.message);
setIsStreaming(false);
break;
}
};
ws.onerror = (error) => {
console.error("WebSocket error:", error);
};
ws.onclose = () => {
console.log("Chat WebSocket closed");
};
};
connectWebSocket();
return () => {
isMountedRef.current = false;
if (wsRef.current) {
wsRef.current.close();
}
};
}, [transcriptId]);
const sendMessage = (text: string) => {
if (!wsRef.current) return;
setMessages((prev) => [
...prev,
{
id: Date.now().toString(),
role: "user",
text,
timestamp: new Date(),
},
]);
wsRef.current.send(JSON.stringify({ type: "message", text }));
};
return { messages, sendMessage, isStreaming, currentStreamingText };
};