mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-05-25 18:55:19 +02:00
Add conversations and turns foundation + DDD (#188)
- Store conversations and turns for: - playground chat - api - New DDD code organisation with container dependency injection - sdk update - streaming api support
This commit is contained in:
Ramnique Singh
GitHub
parent
659b23ae2b
commit
51a33ab2df
39 changed files with 1474 additions and 525 deletions
|
|
@ -16,6 +16,7 @@ export function App({
|
|||
messageSubscriber,
|
||||
onPanelClick,
|
||||
triggerCopilotChat,
|
||||
isLiveWorkflow,
|
||||
}: {
|
||||
hidden?: boolean;
|
||||
projectId: string;
|
||||
|
|
@ -23,6 +24,7 @@ export function App({
|
|||
messageSubscriber?: (messages: z.infer<typeof Message>[]) => void;
|
||||
onPanelClick?: () => void;
|
||||
triggerCopilotChat?: (message: string) => void;
|
||||
isLiveWorkflow: boolean;
|
||||
}) {
|
||||
const [counter, setCounter] = useState<number>(0);
|
||||
const [showDebugMessages, setShowDebugMessages] = useState<boolean>(true);
|
||||
|
|
@ -118,6 +120,7 @@ export function App({
|
|||
onCopyClick={(fn) => { getCopyContentRef.current = fn; }}
|
||||
showDebugMessages={showDebugMessages}
|
||||
triggerCopilotChat={triggerCopilotChat}
|
||||
isLiveWorkflow={isLiveWorkflow}
|
||||
/>
|
||||
</div>
|
||||
</Panel>
|
||||
|
|
|
|||
|
|
@ -1,8 +1,8 @@
|
|||
'use client';
|
||||
import { useEffect, useRef, useState, useCallback } from "react";
|
||||
import { getAssistantResponseStreamId } from "@/app/actions/actions";
|
||||
import { createCachedTurn, createConversation } from "@/app/actions/playground-chat.actions";
|
||||
import { Messages } from "./messages";
|
||||
import z from "zod";
|
||||
import { z } from "zod";
|
||||
import { Message, ToolMessage } from "@/app/lib/types/types";
|
||||
import { Workflow } from "@/app/lib/types/workflow_types";
|
||||
import { ComposeBoxPlayground } from "@/components/common/compose-box-playground";
|
||||
|
|
@ -11,6 +11,7 @@ import { BillingUpgradeModal } from "@/components/common/billing-upgrade-modal";
|
|||
import { ChevronDownIcon } from "@heroicons/react/24/outline";
|
||||
import { FeedbackModal } from "./feedback-modal";
|
||||
import { FIX_WORKFLOW_PROMPT, FIX_WORKFLOW_PROMPT_WITH_FEEDBACK, EXPLAIN_WORKFLOW_PROMPT_ASSISTANT, EXPLAIN_WORKFLOW_PROMPT_TOOL, EXPLAIN_WORKFLOW_PROMPT_TRANSITION } from "../copilot-prompts";
|
||||
import { TurnEvent } from "@/src/entities/models/turn";
|
||||
|
||||
export function Chat({
|
||||
projectId,
|
||||
|
|
@ -20,6 +21,7 @@ export function Chat({
|
|||
showDebugMessages = true,
|
||||
showJsonMode = false,
|
||||
triggerCopilotChat,
|
||||
isLiveWorkflow,
|
||||
}: {
|
||||
projectId: string;
|
||||
workflow: z.infer<typeof Workflow>;
|
||||
|
|
@ -28,10 +30,12 @@ export function Chat({
|
|||
showDebugMessages?: boolean;
|
||||
showJsonMode?: boolean;
|
||||
triggerCopilotChat?: (message: string) => void;
|
||||
isLiveWorkflow: boolean;
|
||||
}) {
|
||||
const conversationId = useRef<string | null>(null);
|
||||
const [messages, setMessages] = useState<z.infer<typeof Message>[]>([]);
|
||||
const [loadingAssistantResponse, setLoadingAssistantResponse] = useState<boolean>(false);
|
||||
const [fetchResponseError, setFetchResponseError] = useState<string | null>(null);
|
||||
const [loading, setLoading] = useState<boolean>(false);
|
||||
const [error, setError] = useState<string | null>(null);
|
||||
const [billingError, setBillingError] = useState<string | null>(null);
|
||||
const [lastAgenticRequest, setLastAgenticRequest] = useState<unknown | null>(null);
|
||||
const [lastAgenticResponse, setLastAgenticResponse] = useState<unknown | null>(null);
|
||||
|
|
@ -142,7 +146,7 @@ export function Chat({
|
|||
if (eventSourceRef.current) {
|
||||
eventSourceRef.current.close();
|
||||
eventSourceRef.current = null;
|
||||
setLoadingAssistantResponse(false);
|
||||
setLoading(false);
|
||||
}
|
||||
}, []);
|
||||
|
||||
|
|
@ -152,7 +156,7 @@ export function Chat({
|
|||
content: prompt,
|
||||
}];
|
||||
setMessages(updatedMessages);
|
||||
setFetchResponseError(null);
|
||||
setError(null);
|
||||
setIsLastInteracted(true);
|
||||
}
|
||||
|
||||
|
|
@ -165,7 +169,7 @@ export function Chat({
|
|||
} else {
|
||||
setShowUnreadBubble(true);
|
||||
}
|
||||
}, [optimisticMessages, loadingAssistantResponse, autoScroll]);
|
||||
}, [optimisticMessages, loading, autoScroll]);
|
||||
|
||||
// Expose copy function to parent
|
||||
useEffect(() => {
|
||||
|
|
@ -190,148 +194,175 @@ export function Chat({
|
|||
}
|
||||
}, [messages, messageSubscriber]);
|
||||
|
||||
// get assistant response
|
||||
// get agent response
|
||||
useEffect(() => {
|
||||
let ignore = false;
|
||||
let eventSource: EventSource | null = null;
|
||||
let msgs: z.infer<typeof Message>[] = [];
|
||||
|
||||
async function process() {
|
||||
setLoadingAssistantResponse(true);
|
||||
setFetchResponseError(null);
|
||||
|
||||
// Reset request/response state before making new request
|
||||
setLastAgenticRequest(null);
|
||||
setLastAgenticResponse(null);
|
||||
|
||||
let streamId: string | null = null;
|
||||
try {
|
||||
const response = await getAssistantResponseStreamId(
|
||||
projectId,
|
||||
workflow,
|
||||
messages,
|
||||
);
|
||||
// first, if there is no conversation id, create it
|
||||
if (!conversationId.current) {
|
||||
const response = await createConversation({
|
||||
projectId,
|
||||
workflow,
|
||||
isLiveWorkflow,
|
||||
});
|
||||
conversationId.current = response.id;
|
||||
}
|
||||
|
||||
// set up a cached turn
|
||||
const response = await createCachedTurn({
|
||||
conversationId: conversationId.current,
|
||||
messages: messages.slice(-1), // only send the last message
|
||||
});
|
||||
if (ignore) {
|
||||
return;
|
||||
}
|
||||
if ('billingError' in response) {
|
||||
setBillingError(response.billingError);
|
||||
setFetchResponseError(response.billingError);
|
||||
setLoadingAssistantResponse(false);
|
||||
console.log('returning from getAssistantResponseStreamId due to billing error');
|
||||
return;
|
||||
}
|
||||
streamId = response.streamId;
|
||||
} catch (err) {
|
||||
if (!ignore) {
|
||||
setFetchResponseError(`Failed to get assistant response: ${err instanceof Error ? err.message : 'Unknown error'}`);
|
||||
setLoadingAssistantResponse(false);
|
||||
}
|
||||
}
|
||||
// if ('billingError' in response) {
|
||||
// setBillingError(response.billingError);
|
||||
// setError(response.billingError);
|
||||
// setLoading(false);
|
||||
// console.log('returning from createRun due to billing error');
|
||||
// return;
|
||||
// }
|
||||
|
||||
if (ignore || !streamId) {
|
||||
return;
|
||||
}
|
||||
// stream events
|
||||
eventSource = new EventSource(`/api/stream-response/${response.key}`);
|
||||
eventSourceRef.current = eventSource;
|
||||
|
||||
console.log(`chat.tsx: got streamid: ${streamId}`);
|
||||
eventSource = new EventSource(`/api/stream-response/${streamId}`);
|
||||
eventSourceRef.current = eventSource;
|
||||
// handle events
|
||||
eventSource.addEventListener("message", (event) => {
|
||||
console.log(`chat.tsx: got message: ${JSON.stringify(event.data)}`);
|
||||
if (ignore) {
|
||||
return;
|
||||
}
|
||||
|
||||
eventSource.addEventListener("message", (event) => {
|
||||
console.log(`chat.tsx: got message: ${event.data}`);
|
||||
if (ignore) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const data = JSON.parse(event.data);
|
||||
const turnEvent = TurnEvent.parse(data);
|
||||
console.log(`chat.tsx: got event: ${turnEvent}`);
|
||||
|
||||
try {
|
||||
const data = JSON.parse(event.data);
|
||||
const parsedMsg = Message.parse(data);
|
||||
msgs.push(parsedMsg);
|
||||
// Update optimistic messages immediately for real-time streaming UX
|
||||
setOptimisticMessages(prev => [...prev, parsedMsg]);
|
||||
} catch (err) {
|
||||
console.error('Failed to parse SSE message:', err);
|
||||
setFetchResponseError(`Failed to parse SSE message: ${err instanceof Error ? err.message : 'Unknown error'}`);
|
||||
// Rollback to last known good state on parsing errors
|
||||
setOptimisticMessages(messages);
|
||||
}
|
||||
});
|
||||
switch (turnEvent.type) {
|
||||
case "message": {
|
||||
// Handle regular message events
|
||||
const generatedMessage = turnEvent.data;
|
||||
// Update optimistic messages immediately for real-time streaming UX
|
||||
setOptimisticMessages(prev => [...prev, generatedMessage]);
|
||||
break;
|
||||
}
|
||||
case "done": {
|
||||
// Handle completion event
|
||||
if (eventSource) {
|
||||
eventSource.close();
|
||||
eventSourceRef.current = null;
|
||||
}
|
||||
|
||||
eventSource.addEventListener('done', (event) => {
|
||||
console.log(`chat.tsx: got done event: ${event.data}`);
|
||||
if (eventSource) {
|
||||
eventSource.close();
|
||||
eventSourceRef.current = null;
|
||||
}
|
||||
// Combine state and collected messages in the response
|
||||
setLastAgenticResponse({
|
||||
turn: turnEvent.turn,
|
||||
messages: turnEvent.turn.output,
|
||||
});
|
||||
|
||||
const parsed = JSON.parse(event.data);
|
||||
// Commit all streamed messages atomically to the source of truth
|
||||
setMessages([...messages, ...turnEvent.turn.output]);
|
||||
setLoading(false);
|
||||
break;
|
||||
}
|
||||
case "error": {
|
||||
// Handle error event
|
||||
if (eventSource) {
|
||||
eventSource.close();
|
||||
eventSourceRef.current = null;
|
||||
}
|
||||
|
||||
// Combine state and collected messages in the response
|
||||
setLastAgenticResponse({
|
||||
...parsed,
|
||||
messages: msgs
|
||||
console.error('Turn Error:', turnEvent.error);
|
||||
if (!ignore) {
|
||||
setLoading(false);
|
||||
setError('Error: ' + turnEvent.error);
|
||||
// Rollback to last known good state on stream errors
|
||||
setOptimisticMessages(messages);
|
||||
|
||||
// check if billing error
|
||||
if (turnEvent.isBillingError) {
|
||||
setBillingError(turnEvent.error);
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
console.error('Failed to parse SSE message:', err);
|
||||
setError(`Failed to parse SSE message: ${err instanceof Error ? err.message : 'Unknown error'}`);
|
||||
// Rollback to last known good state on parsing errors
|
||||
setOptimisticMessages(messages);
|
||||
}
|
||||
});
|
||||
|
||||
// Commit all streamed messages atomically to the source of truth
|
||||
setMessages([...messages, ...msgs]);
|
||||
setLoadingAssistantResponse(false);
|
||||
});
|
||||
eventSource.addEventListener('stream_error', (event) => {
|
||||
console.log(`chat.tsx: got stream_error event: ${event.data}`);
|
||||
if (eventSource) {
|
||||
eventSource.close();
|
||||
eventSourceRef.current = null;
|
||||
}
|
||||
|
||||
console.error('SSE Error:', event);
|
||||
if (!ignore) {
|
||||
setLoading(false);
|
||||
setError('Error: ' + JSON.parse(event.data).error);
|
||||
// Rollback to last known good state on stream errors
|
||||
setOptimisticMessages(messages);
|
||||
}
|
||||
});
|
||||
|
||||
eventSource.addEventListener('stream_error', (event) => {
|
||||
console.log(`chat.tsx: got stream_error event: ${event.data}`);
|
||||
if (eventSource) {
|
||||
eventSource.close();
|
||||
eventSourceRef.current = null;
|
||||
}
|
||||
|
||||
console.error('SSE Error:', event);
|
||||
eventSource.onerror = (error) => {
|
||||
console.error('SSE Error:', error);
|
||||
if (!ignore) {
|
||||
setLoading(false);
|
||||
setError('Stream connection failed');
|
||||
// Rollback to last known good state on connection errors
|
||||
setOptimisticMessages(messages);
|
||||
}
|
||||
};
|
||||
} catch (err) {
|
||||
if (!ignore) {
|
||||
setLoadingAssistantResponse(false);
|
||||
setFetchResponseError('Error: ' + JSON.parse(event.data).error);
|
||||
// Rollback to last known good state on stream errors
|
||||
setOptimisticMessages(messages);
|
||||
setError(`Failed to create run: ${err instanceof Error ? err.message : 'Unknown error'}`);
|
||||
setLoading(false);
|
||||
}
|
||||
});
|
||||
|
||||
eventSource.onerror = (error) => {
|
||||
console.error('SSE Error:', error);
|
||||
if (!ignore) {
|
||||
setLoadingAssistantResponse(false);
|
||||
setFetchResponseError('Stream connection failed');
|
||||
// Rollback to last known good state on connection errors
|
||||
setOptimisticMessages(messages);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
// if last message is not a user message, return
|
||||
if (messages.length > 0) {
|
||||
const last = messages[messages.length - 1];
|
||||
if (last.role !== 'user') {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// if there is an error, return
|
||||
if (fetchResponseError) {
|
||||
// if there are no messages yet, return
|
||||
if (messages.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
console.log(`executing response process: fetchresponseerr: ${fetchResponseError}`);
|
||||
// if last message is not a user message, return
|
||||
const last = messages[messages.length - 1];
|
||||
if (last.role !== 'user') {
|
||||
return;
|
||||
}
|
||||
|
||||
// if there is an error, return
|
||||
if (error) {
|
||||
return;
|
||||
}
|
||||
|
||||
console.log(`chat.tsx: fetching agent response`);
|
||||
setLoading(true);
|
||||
setError(null);
|
||||
process();
|
||||
|
||||
return () => {
|
||||
ignore = true;
|
||||
if (eventSource) {
|
||||
eventSource.close();
|
||||
eventSourceRef.current = null;
|
||||
}
|
||||
};
|
||||
}, [
|
||||
conversationId,
|
||||
messages,
|
||||
projectId,
|
||||
workflow,
|
||||
fetchResponseError,
|
||||
isLiveWorkflow,
|
||||
error,
|
||||
]);
|
||||
|
||||
return (
|
||||
|
|
@ -349,9 +380,17 @@ export function Chat({
|
|||
>
|
||||
<Messages
|
||||
projectId={projectId}
|
||||
messages={optimisticMessages}
|
||||
messages={[
|
||||
{
|
||||
role: 'assistant',
|
||||
content: 'Hi, how can I help you today?',
|
||||
agentName: 'assistant',
|
||||
responseType: 'external',
|
||||
},
|
||||
...optimisticMessages,
|
||||
]}
|
||||
toolCallResults={toolCallResults}
|
||||
loadingAssistantResponse={loadingAssistantResponse}
|
||||
loadingAssistantResponse={loading}
|
||||
workflow={workflow}
|
||||
showDebugMessages={showDebugMessages}
|
||||
showJsonMode={showJsonMode}
|
||||
|
|
@ -403,15 +442,15 @@ export function Chat({
|
|||
</Button>
|
||||
</div>
|
||||
)}
|
||||
{fetchResponseError && (
|
||||
{error && (
|
||||
<div className="mb-4 p-3 bg-red-50 dark:bg-red-900/20 border border-red-200 dark:border-red-800
|
||||
rounded-lg flex gap-2 justify-between items-center">
|
||||
<p className="text-red-600 dark:text-red-400 text-sm">{fetchResponseError}</p>
|
||||
<p className="text-red-600 dark:text-red-400 text-sm">{error}</p>
|
||||
<Button
|
||||
size="sm"
|
||||
color="danger"
|
||||
onPress={() => {
|
||||
setFetchResponseError(null);
|
||||
setError(null);
|
||||
setBillingError(null);
|
||||
}}
|
||||
>
|
||||
|
|
@ -423,7 +462,7 @@ export function Chat({
|
|||
<ComposeBoxPlayground
|
||||
handleUserMessage={handleUserMessage}
|
||||
messages={messages.filter(msg => msg.content !== undefined) as any}
|
||||
loading={loadingAssistantResponse}
|
||||
loading={loading}
|
||||
shouldAutoFocus={isLastInteracted}
|
||||
onFocus={() => setIsLastInteracted(true)}
|
||||
onCancel={handleStop}
|
||||
|
|
|
|||
|
|
@ -1160,6 +1160,7 @@ export function WorkflowEditor({
|
|||
messageSubscriber={updateChatMessages}
|
||||
onPanelClick={handlePlaygroundClick}
|
||||
triggerCopilotChat={triggerCopilotChat}
|
||||
isLiveWorkflow={isLive}
|
||||
/>
|
||||
{state.present.selection?.type === "agent" && <AgentConfig
|
||||
key={`agent-${state.present.workflow.agents.findIndex(agent => agent.name === state.present.selection!.name)}`}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue