From 3259e038fb205832612c4a1f2f45b4d3cde66fb5 Mon Sep 17 00:00:00 2001 From: Ajax Davis Date: Fri, 23 Jan 2026 09:03:10 +1000 Subject: [PATCH] feat: add Omega AI agent chat feature - Add Prisma models for conversations, messages, participants, tool runs, and user settings - Create API endpoints for conversation CRUD and SSE message streaming - Build landing page with sample prompts at /omega - Build chat interface with real-time streaming at /omega/[conversationId] - Integrate @tpmjs/registry-search and @tpmjs/registry-execute packages - Use OpenAI GPT-4.1 Mini as the default model --- apps/web/package.json | 2 + .../omega/conversations/[id]/cancel/route.ts | 100 + .../conversations/[id]/messages/route.ts | 476 +++++ .../app/api/omega/conversations/[id]/route.ts | 202 ++ .../src/app/api/omega/conversations/route.ts | 140 ++ .../src/app/omega/[conversationId]/page.tsx | 672 +++++++ apps/web/src/app/omega/layout.tsx | 21 + apps/web/src/app/omega/page.tsx | 221 +++ apps/web/src/lib/omega/system-prompt.ts | 78 + packages/db/prisma/schema.prisma | 145 ++ pnpm-lock.yaml | 1667 +---------------- 11 files changed, 2074 insertions(+), 1650 deletions(-) create mode 100644 apps/web/src/app/api/omega/conversations/[id]/cancel/route.ts create mode 100644 apps/web/src/app/api/omega/conversations/[id]/messages/route.ts create mode 100644 apps/web/src/app/api/omega/conversations/[id]/route.ts create mode 100644 apps/web/src/app/api/omega/conversations/route.ts create mode 100644 apps/web/src/app/omega/[conversationId]/page.tsx create mode 100644 apps/web/src/app/omega/layout.tsx create mode 100644 apps/web/src/app/omega/page.tsx create mode 100644 apps/web/src/lib/omega/system-prompt.ts diff --git a/apps/web/package.json b/apps/web/package.json index d4be2fe..ca3ce25 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -33,6 +33,8 @@ "@tpmjs/env": "workspace:*", "@tpmjs/npm-client": "workspace:*", "@tpmjs/package-executor": "workspace:*", + "@tpmjs/registry-execute": "workspace:*", + "@tpmjs/registry-search": "workspace:*", "@tpmjs/types": "workspace:*", "@tpmjs/ui": "workspace:*", "@tpmjs/utils": "workspace:*", diff --git a/apps/web/src/app/api/omega/conversations/[id]/cancel/route.ts b/apps/web/src/app/api/omega/conversations/[id]/cancel/route.ts new file mode 100644 index 0000000..89f923e --- /dev/null +++ b/apps/web/src/app/api/omega/conversations/[id]/cancel/route.ts @@ -0,0 +1,100 @@ +/** + * Omega Cancel Endpoint + * + * POST: Cancel a running conversation execution + */ + +import { prisma } from '@tpmjs/db'; +import type { NextRequest } from 'next/server'; + +import { authenticateRequest } from '~/lib/api-keys/middleware'; +import { + apiForbidden, + apiInternalError, + apiNotFound, + apiSuccess, + apiUnauthorized, + apiValidationError, +} from '~/lib/api-response'; + +export const runtime = 'nodejs'; +export const dynamic = 'force-dynamic'; +export const maxDuration = 10; + +type RouteContext = { + params: Promise<{ id: string }>; +}; + +/** + * POST /api/omega/conversations/[id]/cancel + * Cancel a running conversation execution + */ +export async function POST(_request: NextRequest, context: RouteContext) { + const requestId = crypto.randomUUID(); + + try { + const authResult = await authenticateRequest(); + if (!authResult.authenticated || !authResult.userId) { + return apiUnauthorized('Authentication required', requestId); + } + + const { id } = await context.params; + + // Fetch conversation + const conversation = await prisma.omegaConversation.findUnique({ + where: { id }, + select: { + ownerId: true, + executionState: true, + participants: { + select: { userId: true }, + }, + }, + }); + + if (!conversation) { + return apiNotFound('Conversation', requestId); + } + + // Check if user is owner or participant + const isOwner = authResult.userId === conversation.ownerId; + const isParticipant = conversation.participants.some((p) => p.userId === authResult.userId); + + if (!isOwner && !isParticipant) { + return apiForbidden('Access denied', requestId); + } + + // Check if conversation is running + if (conversation.executionState !== 'running') { + return apiValidationError( + 'Conversation is not running', + { currentState: conversation.executionState }, + requestId + ); + } + + // Update conversation state to cancelled + await prisma.omegaConversation.update({ + where: { id }, + data: { executionState: 'cancelled' }, + }); + + // Mark any running tool runs as cancelled + await prisma.omegaToolRun.updateMany({ + where: { + conversationId: id, + status: 'running', + }, + data: { + status: 'error', + error: 'Cancelled by user', + completedAt: new Date(), + }, + }); + + return apiSuccess({ cancelled: true }, { requestId }); + } catch (error) { + console.error('Failed to cancel Omega conversation:', error); + return apiInternalError('Failed to cancel conversation', requestId); + } +} diff --git a/apps/web/src/app/api/omega/conversations/[id]/messages/route.ts b/apps/web/src/app/api/omega/conversations/[id]/messages/route.ts new file mode 100644 index 0000000..3ecc071 --- /dev/null +++ b/apps/web/src/app/api/omega/conversations/[id]/messages/route.ts @@ -0,0 +1,476 @@ +/** + * Omega Messages Endpoint + * + * POST: Send a message and stream the AI response via SSE + * + * This endpoint implements the core Omega chat functionality with: + * - Registry search tool for discovering tools + * - Tool executor for running discovered tools + * - SSE streaming for real-time updates + */ + +import { Prisma, prisma } from '@tpmjs/db'; +import { registryExecuteTool } from '@tpmjs/registry-execute'; +import { registrySearchTool } from '@tpmjs/registry-search'; +import type { ModelMessage } from 'ai'; +import { type NextRequest, NextResponse } from 'next/server'; +import { z } from 'zod'; +import { authenticateRequest } from '~/lib/api-keys/middleware'; +import { buildSystemPrompt } from '~/lib/omega/system-prompt'; +import { checkRateLimit, type RateLimitConfig } from '~/lib/rate-limit'; + +/** + * Rate limit for Omega chat: 20 requests per minute + * Stricter limit because this involves multiple tool executions + */ +const OMEGA_RATE_LIMIT: RateLimitConfig = { + limit: 20, + windowSeconds: 60, +}; + +export const runtime = 'nodejs'; +export const dynamic = 'force-dynamic'; +export const maxDuration = 300; // 5 minutes for complex tool chains + +type RouteContext = { + params: Promise<{ id: string }>; +}; + +const SendMessageSchema = z.object({ + message: z.string().min(1).max(10000), +}); + +/** + * POST /api/omega/conversations/[id]/messages + * Send a message and stream the AI response via SSE + */ +// biome-ignore lint/complexity/noExcessiveCognitiveComplexity: Complex streaming logic required +export async function POST(request: NextRequest, context: RouteContext): Promise { + const startTime = Date.now(); + + // Authenticate request + const authResult = await authenticateRequest(); + if (!authResult.authenticated || !authResult.userId) { + return NextResponse.json({ error: 'Unauthorized' }, { status: 401 }); + } + + // Check rate limit + const rateLimitResponse = checkRateLimit(request, OMEGA_RATE_LIMIT); + if (rateLimitResponse) { + return rateLimitResponse; + } + + const { id: conversationId } = await context.params; + + try { + const body = await request.json(); + const parsed = SendMessageSchema.safeParse(body); + if (!parsed.success) { + return NextResponse.json( + { success: false, error: 'Invalid request', details: parsed.error.flatten() }, + { status: 400 } + ); + } + + // Fetch conversation and verify access + const conversation = await prisma.omegaConversation.findUnique({ + where: { id: conversationId }, + include: { + participants: { + select: { userId: true }, + }, + }, + }); + + if (!conversation) { + return NextResponse.json( + { success: false, error: 'Conversation not found' }, + { status: 404 } + ); + } + + // Check if user is owner or participant + const isOwner = authResult.userId === conversation.ownerId; + const isParticipant = conversation.participants.some((p) => p.userId === authResult.userId); + + if (!isOwner && !isParticipant) { + return NextResponse.json({ success: false, error: 'Access denied' }, { status: 403 }); + } + + // Get user info + const user = await prisma.user.findUnique({ + where: { id: authResult.userId }, + select: { id: true, name: true, email: true }, + }); + + // Get user settings for pinned/blocked tools and custom prompt + const userSettings = await prisma.omegaUserSettings.findUnique({ + where: { userId: authResult.userId }, + }); + + // Update conversation state to running + await prisma.omegaConversation.update({ + where: { id: conversationId }, + data: { executionState: 'running' }, + }); + + // Save user message + await prisma.omegaMessage.create({ + data: { + conversationId, + role: 'USER', + content: parsed.data.message, + authorId: user?.id, + authorEmail: user?.email, + authorName: user?.name, + }, + }); + + // Fetch recent messages for context + const recentMessages = await prisma.omegaMessage.findMany({ + where: { conversationId }, + orderBy: { createdAt: 'desc' }, + take: 20, // Last 20 messages for context + }); + recentMessages.reverse(); + + // Build AI SDK messages + const messages: ModelMessage[] = []; + + // Add system prompt + const systemPrompt = buildSystemPrompt({ + customSystemPrompt: userSettings?.customSystemPrompt, + pinnedToolIds: userSettings?.pinnedToolIds || [], + }); + messages.push({ role: 'system', content: systemPrompt }); + + // Add conversation history + for (const msg of recentMessages.slice(0, -1)) { + // Exclude the message we just added + if (msg.role === 'USER') { + messages.push({ role: 'user', content: msg.content }); + } else if (msg.role === 'ASSISTANT') { + if (msg.toolCalls && Array.isArray(msg.toolCalls) && msg.toolCalls.length > 0) { + const toolCallParts = ( + msg.toolCalls as Array<{ toolCallId: string; toolName: string; args: unknown }> + ).map((tc) => ({ + type: 'tool-call' as const, + toolCallId: tc.toolCallId, + toolName: tc.toolName, + input: tc.args, + })); + + const content: Array< + | { type: 'text'; text: string } + | { type: 'tool-call'; toolCallId: string; toolName: string; input: unknown } + > = []; + + if (msg.content) { + content.push({ type: 'text', text: msg.content }); + } + content.push(...toolCallParts); + messages.push({ role: 'assistant', content }); + } else { + messages.push({ role: 'assistant', content: msg.content }); + } + } else if (msg.role === 'TOOL') { + // Handle tool results (stored in toolCalls as a workaround) + const toolResults = msg.toolCalls as Array<{ + toolCallId: string; + toolName: string; + output: unknown; + }> | null; + if (toolResults && toolResults.length > 0) { + for (const tr of toolResults) { + messages.push({ + role: 'tool', + content: [ + { + type: 'tool-result' as const, + toolCallId: tr.toolCallId, + toolName: tr.toolName, + output: { + type: 'json' as const, + value: tr.output as Parameters[0], + }, + }, + ], + }); + } + } + } + } + + // Add new user message + messages.push({ role: 'user', content: parsed.data.message }); + + // Build Omega tools using published TPMJS packages + const tools = { + registrySearch: registrySearchTool, + registryExecute: registryExecuteTool, + }; + + // Get the provider model (using OpenAI by default) + const { createOpenAI } = await import('@ai-sdk/openai'); + const apiKey = process.env.OPENAI_API_KEY; + + if (!apiKey) { + await prisma.omegaConversation.update({ + where: { id: conversationId }, + data: { executionState: 'idle' }, + }); + return NextResponse.json( + { success: false, error: 'Omega is not configured. Missing OPENAI_API_KEY.' }, + { status: 500 } + ); + } + + const openai = createOpenAI({ apiKey }); + const model = openai('gpt-4.1-mini'); + + // Create SSE stream + const stream = new ReadableStream({ + // biome-ignore lint/complexity/noExcessiveCognitiveComplexity: Complex streaming logic + async start(controller) { + const encoder = new TextEncoder(); + + const sendEvent = (event: string, data: unknown) => { + const message = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; + controller.enqueue(encoder.encode(message)); + }; + + try { + const { streamText, stepCountIs } = await import('ai'); + + let fullContent = ''; + const toolCallsMap: Map = + new Map(); + const pendingToolResults: Array<{ + toolCallId: string; + toolName: string; + output: unknown; + }> = []; + let inputTokens = 0; + let outputTokens = 0; + + // Stream the response with up to 10 tool call iterations + const result = await streamText({ + model, + messages, + tools, + stopWhen: stepCountIs(10), + onChunk: async ({ chunk }) => { + if (chunk.type === 'tool-call') { + const input = 'args' in chunk ? chunk.args : chunk.input; + + console.log('[Omega] Tool call:', { + toolName: chunk.toolName, + toolCallId: chunk.toolCallId, + }); + + toolCallsMap.set(chunk.toolCallId, { + toolCallId: chunk.toolCallId, + toolName: chunk.toolName, + args: input, + }); + + // Create tool run record + await prisma.omegaToolRun.create({ + data: { + conversationId, + toolName: chunk.toolName, + input: input as Prisma.InputJsonValue, + status: 'running', + }, + }); + + sendEvent('run.step.tool.started', { + toolCallId: chunk.toolCallId, + toolName: chunk.toolName, + input, + }); + } + }, + // biome-ignore lint/complexity/noExcessiveCognitiveComplexity: Complex tool result handling + onStepFinish: async ({ toolCalls, toolResults, usage }) => { + // Capture tool calls + if (toolCalls && Array.isArray(toolCalls)) { + for (const tc of toolCalls) { + if (!toolCallsMap.has(tc.toolCallId)) { + const args = + 'input' in tc ? tc.input : 'args' in tc ? (tc as { args: unknown }).args : {}; + toolCallsMap.set(tc.toolCallId, { + toolCallId: tc.toolCallId, + toolName: tc.toolName, + args, + }); + } + } + } + + // Process tool results + if (toolResults && toolResults.length > 0) { + for (const tr of toolResults) { + const isError = + tr.output && typeof tr.output === 'object' && 'error' in tr.output; + + // Build update data for tool run + const toolRunUpdateData: Prisma.OmegaToolRunUpdateManyMutationInput = { + output: tr.output as Prisma.InputJsonValue, + status: isError ? 'error' : 'success', + completedAt: new Date(), + executionTimeMs: Date.now() - startTime, + }; + + // Add error message if the tool execution failed + if (isError) { + toolRunUpdateData.error = + typeof tr.output === 'object' && tr.output && 'error' in tr.output + ? String((tr.output as { error: unknown }).error) + : 'Unknown error'; + } + + // Update tool run record + await prisma.omegaToolRun.updateMany({ + where: { + conversationId, + toolName: tr.toolName, + status: 'running', + }, + data: toolRunUpdateData, + }); + + sendEvent('run.step.tool.completed', { + toolCallId: tr.toolCallId, + toolName: tr.toolName, + output: tr.output, + isError, + }); + + pendingToolResults.push({ + toolCallId: tr.toolCallId, + toolName: tr.toolName, + output: tr.output, + }); + } + } + + if (usage) { + inputTokens += usage.inputTokens ?? 0; + outputTokens += usage.outputTokens ?? 0; + } + }, + }); + + // Stream text chunks + for await (const chunk of result.textStream) { + fullContent += chunk; + sendEvent('message.delta', { content: chunk }); + } + + // Get final usage + const finalUsage = await result.usage; + if (finalUsage) { + inputTokens = finalUsage.inputTokens ?? inputTokens; + outputTokens = finalUsage.outputTokens ?? outputTokens; + } + + const allToolCalls = Array.from(toolCallsMap.values()); + + // Save assistant message + const assistantMessage = await prisma.omegaMessage.create({ + data: { + conversationId, + role: 'ASSISTANT', + content: fullContent, + toolCalls: + allToolCalls.length > 0 + ? (allToolCalls as unknown as Prisma.InputJsonValue) + : Prisma.JsonNull, + inputTokens, + outputTokens, + }, + }); + + // Save tool results as TOOL messages + if (pendingToolResults.length > 0) { + await prisma.omegaMessage.create({ + data: { + conversationId, + role: 'TOOL', + content: 'Tool results', + toolCalls: pendingToolResults as unknown as Prisma.InputJsonValue, + }, + }); + } + + // Update conversation + const updateData: Prisma.OmegaConversationUpdateInput = { + executionState: 'idle', + inputTokensTotal: { increment: inputTokens }, + outputTokensTotal: { increment: outputTokens }, + updatedAt: new Date(), + }; + + // Auto-generate title from first message if not set + if (conversation.title === null) { + updateData.title = parsed.data.message.slice(0, 100); + } + + await prisma.omegaConversation.update({ + where: { id: conversationId }, + data: updateData, + }); + + const executionTimeMs = Date.now() - startTime; + + sendEvent('run.completed', { + messageId: assistantMessage.id, + conversationId, + inputTokens, + outputTokens, + executionTimeMs, + }); + } catch (error) { + console.error('[Omega] Stream error:', error); + + // Update conversation state + await prisma.omegaConversation.update({ + where: { id: conversationId }, + data: { executionState: 'idle' }, + }); + + sendEvent('run.failed', { + error: error instanceof Error ? error.message : 'Unknown error', + }); + } finally { + controller.close(); + } + }, + }); + + return new NextResponse(stream, { + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }, + }); + } catch (error) { + console.error('[Omega] Failed to process message:', error); + + // Reset conversation state + await prisma.omegaConversation.update({ + where: { id: conversationId }, + data: { executionState: 'idle' }, + }); + + return NextResponse.json( + { + success: false, + error: error instanceof Error ? error.message : 'Failed to process message', + }, + { status: 500 } + ); + } +} diff --git a/apps/web/src/app/api/omega/conversations/[id]/route.ts b/apps/web/src/app/api/omega/conversations/[id]/route.ts new file mode 100644 index 0000000..b6e1c65 --- /dev/null +++ b/apps/web/src/app/api/omega/conversations/[id]/route.ts @@ -0,0 +1,202 @@ +/** + * Omega Single Conversation Endpoint + * + * GET: Fetch conversation with messages + * DELETE: Delete a conversation + */ + +import { prisma } from '@tpmjs/db'; +import type { NextRequest } from 'next/server'; + +import { authenticateRequest } from '~/lib/api-keys/middleware'; +import { + apiForbidden, + apiInternalError, + apiNotFound, + apiSuccess, + apiUnauthorized, +} from '~/lib/api-response'; + +export const runtime = 'nodejs'; +export const dynamic = 'force-dynamic'; +export const maxDuration = 30; + +type RouteContext = { + params: Promise<{ id: string }>; +}; + +/** + * GET /api/omega/conversations/[id] + * Fetch conversation with messages (paginated) + */ +export async function GET(request: NextRequest, context: RouteContext) { + const requestId = crypto.randomUUID(); + + try { + const authResult = await authenticateRequest(); + const { id } = await context.params; + + // Fetch conversation + const conversation = await prisma.omegaConversation.findUnique({ + where: { id }, + include: { + participants: { + select: { + id: true, + userId: true, + displayName: true, + email: true, + role: true, + joinedAt: true, + }, + }, + _count: { + select: { + messages: true, + toolRuns: true, + }, + }, + }, + }); + + if (!conversation) { + return apiNotFound('Conversation', requestId); + } + + // Check if user is owner or participant + const isOwner = authResult.userId === conversation.ownerId; + const isParticipant = conversation.participants.some((p) => p.userId === authResult.userId); + + if (!isOwner && !isParticipant) { + return apiForbidden('Access denied', requestId); + } + + // Fetch messages with pagination + const { searchParams } = new URL(request.url); + const limit = Math.min(Number.parseInt(searchParams.get('limit') || '50', 10), 100); + const before = searchParams.get('before'); + const after = searchParams.get('after'); + + const whereClause: { + conversationId: string; + createdAt?: { lt?: Date; gt?: Date }; + } = { conversationId: id }; + + if (before) { + whereClause.createdAt = { lt: new Date(before) }; + } else if (after) { + whereClause.createdAt = { gt: new Date(after) }; + } + + const shouldFetchDesc = !after; + + const messages = await prisma.omegaMessage.findMany({ + where: whereClause, + orderBy: { createdAt: shouldFetchDesc ? 'desc' : 'asc' }, + take: limit + 1, + }); + + const hasMoreMessages = messages.length > limit; + let paginatedMessages = hasMoreMessages ? messages.slice(0, limit) : messages; + + if (shouldFetchDesc) { + paginatedMessages = paginatedMessages.reverse(); + } + + // Fetch recent tool runs + const toolRuns = await prisma.omegaToolRun.findMany({ + where: { conversationId: id }, + orderBy: { startedAt: 'desc' }, + take: 20, + }); + + return apiSuccess( + { + id: conversation.id, + title: conversation.title, + executionState: conversation.executionState, + inputTokensTotal: conversation.inputTokensTotal, + outputTokensTotal: conversation.outputTokensTotal, + createdAt: conversation.createdAt, + updatedAt: conversation.updatedAt, + isOwner, + participants: conversation.participants, + messageCount: conversation._count.messages, + toolRunCount: conversation._count.toolRuns, + messages: paginatedMessages.map((m) => ({ + id: m.id, + role: m.role, + content: m.content, + authorId: m.authorId, + authorName: m.authorName, + toolCalls: m.toolCalls, + inputTokens: m.inputTokens, + outputTokens: m.outputTokens, + createdAt: m.createdAt, + })), + toolRuns: toolRuns.map((tr) => ({ + id: tr.id, + toolName: tr.toolName, + status: tr.status, + startedAt: tr.startedAt, + completedAt: tr.completedAt, + executionTimeMs: tr.executionTimeMs, + error: tr.error, + })), + }, + { + requestId, + pagination: { + limit, + hasMore: hasMoreMessages, + ...(before && { before }), + ...(after && { after }), + }, + } + ); + } catch (error) { + console.error('Failed to fetch Omega conversation:', error); + return apiInternalError('Failed to fetch conversation', requestId); + } +} + +/** + * DELETE /api/omega/conversations/[id] + * Delete a conversation (owner only) + */ +export async function DELETE(_request: NextRequest, context: RouteContext) { + const requestId = crypto.randomUUID(); + + try { + const authResult = await authenticateRequest(); + if (!authResult.authenticated || !authResult.userId) { + return apiUnauthorized('Authentication required', requestId); + } + + const { id } = await context.params; + + // Check ownership + const conversation = await prisma.omegaConversation.findUnique({ + where: { id }, + select: { ownerId: true }, + }); + + if (!conversation) { + return apiNotFound('Conversation', requestId); + } + + if (conversation.ownerId !== authResult.userId) { + return apiForbidden('Only the owner can delete this conversation', requestId); + } + + // Delete conversation (cascades to messages, participants, tool runs) + await prisma.omegaConversation.delete({ + where: { id }, + }); + + return apiSuccess({ deleted: true }, { requestId }); + } catch (error) { + console.error('Failed to delete Omega conversation:', error); + return apiInternalError('Failed to delete conversation', requestId); + } +} diff --git a/apps/web/src/app/api/omega/conversations/route.ts b/apps/web/src/app/api/omega/conversations/route.ts new file mode 100644 index 0000000..6d6d669 --- /dev/null +++ b/apps/web/src/app/api/omega/conversations/route.ts @@ -0,0 +1,140 @@ +/** + * Omega Conversations Endpoint + * + * POST: Create a new Omega conversation + * GET: List user's Omega conversations + */ + +import { prisma } from '@tpmjs/db'; +import type { NextRequest } from 'next/server'; + +import { authenticateRequest } from '~/lib/api-keys/middleware'; +import { apiForbidden, apiInternalError, apiSuccess, apiUnauthorized } from '~/lib/api-response'; + +export const runtime = 'nodejs'; +export const dynamic = 'force-dynamic'; +export const maxDuration = 30; + +/** + * GET /api/omega/conversations + * List all Omega conversations for the authenticated user + */ +export async function GET(request: NextRequest) { + const requestId = crypto.randomUUID(); + + try { + const authResult = await authenticateRequest(); + if (!authResult.authenticated || !authResult.userId) { + return apiUnauthorized('Authentication required', requestId); + } + + const { searchParams } = new URL(request.url); + const limit = Math.min(Number.parseInt(searchParams.get('limit') || '20', 10), 50); + const offset = Number.parseInt(searchParams.get('offset') || '0', 10); + + const conversations = await prisma.omegaConversation.findMany({ + where: { ownerId: authResult.userId }, + orderBy: { updatedAt: 'desc' }, + take: limit + 1, + skip: offset, + include: { + _count: { + select: { + messages: true, + toolRuns: true, + }, + }, + }, + }); + + const hasMore = conversations.length > limit; + const data = hasMore ? conversations.slice(0, limit) : conversations; + + return apiSuccess( + data.map((c) => ({ + id: c.id, + title: c.title, + executionState: c.executionState, + inputTokensTotal: c.inputTokensTotal, + outputTokensTotal: c.outputTokensTotal, + messageCount: c._count.messages, + toolRunCount: c._count.toolRuns, + createdAt: c.createdAt, + updatedAt: c.updatedAt, + })), + { + requestId, + pagination: { + limit, + offset, + hasMore, + }, + } + ); + } catch (error) { + console.error('Failed to list Omega conversations:', error); + return apiInternalError('Failed to list conversations', requestId); + } +} + +/** + * POST /api/omega/conversations + * Create a new Omega conversation + */ +export async function POST(_request: NextRequest) { + const requestId = crypto.randomUUID(); + + try { + const authResult = await authenticateRequest(); + if (!authResult.authenticated || !authResult.userId) { + return apiUnauthorized('Authentication required', requestId); + } + + // Get user info for participant + const user = await prisma.user.findUnique({ + where: { id: authResult.userId }, + select: { id: true, name: true, email: true }, + }); + + if (!user) { + return apiForbidden('User not found', requestId); + } + + const userId = authResult.userId; + + // Create conversation with owner as participant + const conversation = await prisma.$transaction(async (tx) => { + const conv = await tx.omegaConversation.create({ + data: { + ownerId: userId, + }, + }); + + // Add owner as participant + await tx.omegaParticipant.create({ + data: { + conversationId: conv.id, + userId: user.id, + displayName: user.name, + email: user.email, + role: 'owner', + }, + }); + + return conv; + }); + + return apiSuccess( + { + id: conversation.id, + title: conversation.title, + executionState: conversation.executionState, + createdAt: conversation.createdAt, + }, + { requestId, status: 201 } + ); + } catch (error) { + console.error('Failed to create Omega conversation:', error); + return apiInternalError('Failed to create conversation', requestId); + } +} diff --git a/apps/web/src/app/omega/[conversationId]/page.tsx b/apps/web/src/app/omega/[conversationId]/page.tsx new file mode 100644 index 0000000..a57ed19 --- /dev/null +++ b/apps/web/src/app/omega/[conversationId]/page.tsx @@ -0,0 +1,672 @@ +'use client'; + +import { Badge } from '@tpmjs/ui/Badge/Badge'; +import { Button } from '@tpmjs/ui/Button/Button'; +import { Icon } from '@tpmjs/ui/Icon/Icon'; +import { Textarea } from '@tpmjs/ui/Textarea/Textarea'; +import Link from 'next/link'; +import { useParams, useRouter } from 'next/navigation'; +import { useCallback, useEffect, useRef, useState } from 'react'; +import { Virtuoso, type VirtuosoHandle } from 'react-virtuoso'; +import { Streamdown } from 'streamdown'; +import { AppHeader } from '~/components/AppHeader'; + +interface Message { + id: string; + role: 'USER' | 'ASSISTANT' | 'TOOL'; + content: string; + toolCalls?: Array<{ + toolCallId: string; + toolName: string; + args: unknown; + }>; + inputTokens?: number; + outputTokens?: number; + createdAt: string; +} + +interface ToolCall { + toolCallId: string; + toolName: string; + input?: unknown; + output?: unknown; + status: 'pending' | 'running' | 'success' | 'error'; + isError?: boolean; +} + +interface Conversation { + id: string; + title: string | null; + executionState: string; + inputTokensTotal: number; + outputTokensTotal: number; + createdAt: string; + updatedAt: string; +} + +/** + * Tool call debug card component + */ +function ToolCallCard({ + toolCall, + isExpanded, + onToggle, +}: { + toolCall: ToolCall; + isExpanded: boolean; + onToggle: () => void; +}) { + const statusColors = { + pending: 'bg-warning/10 text-warning border-warning/30', + running: 'bg-info/10 text-info border-info/30', + success: 'bg-success/10 text-success border-success/30', + error: 'bg-error/10 text-error border-error/30', + }; + + const statusIcons: Record = { + pending: 'info', + running: 'loader', + success: 'check', + error: 'alertCircle', + }; + + const formatJson = (data: unknown): React.ReactNode => { + try { + return JSON.stringify(data, null, 2); + } catch { + return String(data); + } + }; + + return ( +
+ {/* Header */} + + + {/* Expanded Content */} + {isExpanded && ( +
+ {/* Input Section */} + {toolCall.input !== undefined && toolCall.input !== null ? ( +
+
+ + Input + +
+
+
+                {formatJson(toolCall.input)}
+              
+
+ ) : null} + + {/* Output Section */} + {toolCall.output !== undefined && toolCall.output !== null ? ( +
+
+ + Output + +
+
+
+                {formatJson(toolCall.output)}
+              
+
+ ) : null} + + {/* Status indicator for running */} + {toolCall.status === 'running' && !toolCall.output && ( +
+ + Executing... +
+ )} +
+ )} +
+ ); +} + +/** + * Omega Chat Page + */ +export default function OmegaChatPage(): React.ReactElement { + const params = useParams(); + const router = useRouter(); + const conversationId = params.conversationId as string; + + const [conversation, setConversation] = useState(null); + const [messages, setMessages] = useState([]); + const [input, setInput] = useState(''); + const [isLoading, setIsLoading] = useState(true); + const [isSending, setIsSending] = useState(false); + const [streamingContent, setStreamingContent] = useState(''); + const [error, setError] = useState(null); + const [toolCalls, setToolCalls] = useState([]); + const [expandedToolCalls, setExpandedToolCalls] = useState>(new Set()); + const [viewMode, setViewMode] = useState<'chat' | 'debug'>('chat'); + + // Track first item index for prepending (Virtuoso pattern) + const [firstItemIndex, setFirstItemIndex] = useState(10000); + + const virtuosoRef = useRef(null); + const inputRef = useRef(null); + + const toggleToolCall = (toolCallId: string) => { + setExpandedToolCalls((prev) => { + const next = new Set(prev); + if (next.has(toolCallId)) { + next.delete(toolCallId); + } else { + next.add(toolCallId); + } + return next; + }); + }; + + // Fetch conversation details + const fetchConversation = useCallback(async () => { + try { + const response = await fetch(`/api/omega/conversations/${conversationId}`); + + if (response.status === 404) { + setError('Conversation not found'); + setIsLoading(false); + return; + } + + if (!response.ok) { + const data = await response.json(); + throw new Error(data.error || 'Failed to fetch conversation'); + } + + const data = await response.json(); + // API returns conversation data directly with messages nested + const { messages: messageList, ...conversationData } = data.data; + setConversation(conversationData); + setMessages(messageList || []); + setFirstItemIndex(10000); + } catch (err) { + console.error('Failed to fetch conversation:', err); + setError(err instanceof Error ? err.message : 'Failed to fetch conversation'); + } finally { + setIsLoading(false); + } + }, [conversationId]); + + useEffect(() => { + fetchConversation(); + }, [fetchConversation]); + + // Check for initial prompt from landing page + // biome-ignore lint/correctness/useExhaustiveDependencies: Only run on mount and when messages load + useEffect(() => { + const initialPrompt = sessionStorage.getItem(`omega_prompt_${conversationId}`); + if (initialPrompt && messages.length === 0 && !isSending) { + sessionStorage.removeItem(`omega_prompt_${conversationId}`); + setInput(initialPrompt); + // Auto-send after a brief delay + const timer = setTimeout(() => { + handleSendWithContent(initialPrompt); + }, 500); + return () => clearTimeout(timer); + } + }, [conversationId, messages.length]); + + // biome-ignore lint/complexity/noExcessiveCognitiveComplexity: Chat send handler with streaming and tool calls + const handleSendWithContent = async (messageContent: string) => { + if (!messageContent.trim() || isSending) return; + + setInput(''); + setIsSending(true); + setStreamingContent(''); + setError(null); + setToolCalls([]); + + // Optimistically add user message + const userMessage: Message = { + id: `temp-${Date.now()}`, + role: 'USER', + content: messageContent, + createdAt: new Date().toISOString(), + }; + setMessages((prev) => [...prev, userMessage]); + + try { + const response = await fetch(`/api/omega/conversations/${conversationId}/messages`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ message: messageContent }), + }); + + if (!response.ok) { + const errorData = await response.json(); + throw new Error(errorData.error || 'Failed to send message'); + } + + // Handle SSE stream + const reader = response.body?.getReader(); + if (!reader) throw new Error('No response body'); + + const decoder = new TextDecoder(); + let buffer = ''; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + + // Parse SSE events + const lines = buffer.split('\n'); + buffer = lines.pop() || ''; // Keep incomplete line in buffer + + let eventType = ''; + for (const line of lines) { + if (line.startsWith('event: ')) { + eventType = line.slice(7); + } else if (line.startsWith('data: ')) { + const data = JSON.parse(line.slice(6)); + + switch (eventType) { + case 'message.delta': + setStreamingContent((prev) => prev + data.content); + break; + case 'run.step.tool.started': + // Add tool call to tracking + setToolCalls((prev) => [ + ...prev, + { + toolCallId: data.toolCallId, + toolName: data.toolName, + input: data.input, + status: 'running', + }, + ]); + // Auto-expand new tool calls + setExpandedToolCalls((prev) => new Set([...prev, data.toolCallId])); + break; + case 'run.step.tool.completed': + // Update tool call with result + setToolCalls((prev) => + // biome-ignore lint/complexity/noExcessiveCognitiveComplexity: Tool call update logic + prev.map((tc) => + tc.toolCallId === data.toolCallId + ? { + ...tc, + output: data.output, + status: data.isError ? ('error' as const) : ('success' as const), + isError: data.isError, + } + : tc + ) + ); + break; + case 'run.completed': + // Refresh messages + await fetchConversation(); + setStreamingContent(''); + setToolCalls([]); + break; + case 'run.failed': + throw new Error(data.error); + } + } + } + } + } catch (err) { + console.error('Failed to send message:', err); + setError(err instanceof Error ? err.message : 'Failed to send message'); + // Remove optimistic message on error + setMessages((prev) => prev.filter((m) => m.id !== userMessage.id)); + } finally { + setIsSending(false); + setStreamingContent(''); + inputRef.current?.focus(); + } + }; + + const handleSend = async () => { + if (!input.trim() || isSending) return; + await handleSendWithContent(input.trim()); + }; + + const handleKeyDown = (e: React.KeyboardEvent) => { + if (e.key === 'Enter' && !e.shiftKey) { + e.preventDefault(); + handleSend(); + } + }; + + const startNewConversation = async () => { + try { + const response = await fetch('/api/omega/conversations', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + }); + + if (!response.ok) { + const data = await response.json(); + throw new Error(data.error || 'Failed to create conversation'); + } + + const data = await response.json(); + router.push(`/omega/${data.data.id}`); + } catch (err) { + console.error('Failed to create conversation:', err); + setError(err instanceof Error ? err.message : 'Failed to create conversation'); + } + }; + + if (isLoading) { + return ( +
+ +
+ +
+
+ ); + } + + if (error && !conversation) { + return ( +
+ +
+
+ +

Unable to Load Chat

+

{error}

+ + + +
+
+
+ ); + } + + return ( +
+ + +
+ {/* Chat Header */} +
+
+
+ + + +
+

+ {conversation?.title || 'New Conversation'} +

+
+ + Omega + + GPT-4.1 Mini + {conversation && ( + + {conversation.inputTokensTotal + conversation.outputTokensTotal} tokens + + )} +
+
+
+
+ +
+
+ {/* View Mode Tabs */} +
+ + +
+
+ + {/* Debug JSON View */} + {viewMode === 'debug' && ( +
+
+
+
+

+ Raw Messages Array ({messages.length} messages) +

+

+ Messages are ordered by createdAt. Each message includes role, content, and tool + call data. +

+
+ +
+
+                {JSON.stringify(messages, null, 2)}
+              
+
+
+ )} + + {/* Chat View */} + {viewMode === 'chat' && ( + <> + {/* Messages */} +
+ {messages.length === 0 && !streamingContent ? ( +
+
+
+ +
+

+ Start a conversation with Omega +

+

+ Describe what you need, and Omega will find and use the right tools from the + TPMJS registry. +

+
+
+ ) : ( + ( +
+ {/* Live tool calls during streaming */} + {toolCalls.length > 0 && ( +
+ {toolCalls.map((tc) => ( +
+
+ toggleToolCall(tc.toolCallId)} + /> +
+
+ ))} +
+ )} + + {streamingContent && ( +
+
+
+ {streamingContent} +
+ +
+
+ )} + + {isSending && !streamingContent && toolCalls.length === 0 && ( +
+
+
+ + Omega is thinking... +
+
+
+ )} +
+ ), + }} + itemContent={(_index, message) => ( +
+ {/* USER message */} + {message.role === 'USER' && ( +
+
+
{message.content}
+
+
+ )} + + {/* ASSISTANT message */} + {message.role === 'ASSISTANT' && message.content && ( +
+
+
+ {message.content} +
+ {/* Token usage for debugging */} + {(message.inputTokens || message.outputTokens) && ( +
+ {message.inputTokens && In: {message.inputTokens}} + {message.inputTokens && message.outputTokens && | } + {message.outputTokens && Out: {message.outputTokens}} +
+ )} +
+
+ )} + + {/* TOOL message - show as collapsed tool result */} + {message.role === 'TOOL' && ( +
+
+
+ Tool Results +
+
+                              {message.content}
+                            
+
+
+ )} +
+ )} + /> + )} +
+ + {/* Error Message */} + {error && ( +
+

{error}

+
+ )} + + {/* Input Area */} +
+
+