From 70bdbd74eb6440f3f9204e55a0561c2d43683f35 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 22 Oct 2025 22:38:15 +0200 Subject: [PATCH] refactor: deduplicate user messages and sync agent history on reconnection --- packages/app/src/app/index.tsx | 9 +- .../app/src/components/agent-stream-view.tsx | 293 ++---------------- packages/app/src/types/stream.ts | 103 +++++- packages/server/src/server/messages.ts | 1 + packages/server/src/server/session.ts | 25 +- 5 files changed, 142 insertions(+), 289 deletions(-) diff --git a/packages/app/src/app/index.tsx b/packages/app/src/app/index.tsx index 98077942b..d9d7fd225 100644 --- a/packages/app/src/app/index.tsx +++ b/packages/app/src/app/index.tsx @@ -48,7 +48,7 @@ import { ActiveProcesses } from "@/components/active-processes"; import { AgentStreamView } from "@/components/agent-stream-view"; import { ConversationSelector } from "@/components/conversation-selector"; import { VolumeMeter } from "@/components/volume-meter"; -import { reduceStreamUpdate, type StreamItem } from "@/types/stream"; +import { reduceStreamUpdate, generateMessageId, type StreamItem } from "@/types/stream"; import { CreateAgentModal } from "@/components/create-agent-modal"; import { Settings, @@ -1015,6 +1015,9 @@ export default function VoiceAssistantScreen() { // Realtime mode always routes to orchestrator (handled by realtime audio) ws.sendUserMessage(userInput); } else if (viewMode === "agent" && activeAgentId) { + // Generate unique message ID for deduplication + const messageId = generateMessageId(); + // Optimistically add user message to stream setAgentStreamState((prev) => { const currentStream = prev.get(activeAgentId) || []; @@ -1025,6 +1028,7 @@ export default function VoiceAssistantScreen() { update: { sessionUpdate: "user_message_chunk", content: { type: "text", text: userInput }, + messageId, }, }, new Date() @@ -1034,13 +1038,14 @@ export default function VoiceAssistantScreen() { return updated; }); - // Send to agent + // Send to agent with messageId const message: WSInboundMessage = { type: "session", message: { type: "send_agent_message", agentId: activeAgentId, text: userInput, + messageId, }, }; ws.send(message); diff --git a/packages/app/src/components/agent-stream-view.tsx b/packages/app/src/components/agent-stream-view.tsx index 5ac46cd68..5ca4059ed 100644 --- a/packages/app/src/components/agent-stream-view.tsx +++ b/packages/app/src/components/agent-stream-view.tsx @@ -3,7 +3,7 @@ import { View, Text, ScrollView, Pressable } from 'react-native'; import { useSafeAreaInsets } from 'react-native-safe-area-context'; import { StyleSheet } from 'react-native-unistyles'; import type { AgentStatus } from '@server/server/acp/types'; -import { AssistantMessage } from './message'; +import { AssistantMessage, UserMessage, ActivityLog, ToolCall } from './message'; import type { StreamItem } from '@/types/stream'; export interface AgentStreamViewProps { @@ -17,106 +17,6 @@ export interface AgentStreamViewProps { streamItems: StreamItem[]; } -function getStatusColor(status: AgentStatus): string { - switch (status) { - case 'initializing': - return '#f59e0b'; - case 'ready': - return '#3b82f6'; - case 'processing': - return '#fbbf24'; - case 'completed': - return '#22c55e'; - case 'failed': - return '#ef4444'; - case 'killed': - return '#9ca3af'; - default: - return '#9ca3af'; - } -} - -function getStatusIcon(status: AgentStatus): string { - switch (status) { - case 'initializing': - case 'processing': - return '⏳'; - case 'ready': - case 'completed': - return '✓'; - case 'failed': - case 'killed': - return '✗'; - default: - return '•'; - } -} - -function formatTimestamp(date: Date): string { - // Handle invalid dates gracefully - if (!date || !(date instanceof Date) || isNaN(date.getTime())) { - return '--:--:--'; - } - - return new Intl.DateTimeFormat('en-US', { - hour: 'numeric', - minute: '2-digit', - second: '2-digit', - hour12: true, - }).format(date); -} - -function UserMessageCard({ text, timestamp }: { text: string; timestamp: Date }) { - return ( - - - {formatTimestamp(timestamp)} - - You - {text} - - ); -} - -function ThoughtCard({ text, timestamp }: { text: string; timestamp: Date }) { - return ( - - - {formatTimestamp(timestamp)} - - Thinking - {text} - - ); -} - -function ToolCallCard({ - title, - status, - toolKind, - timestamp -}: { - title: string; - status: 'pending' | 'in_progress' | 'completed' | 'failed'; - toolKind?: string; - timestamp: Date; -}) { - return ( - - - {formatTimestamp(timestamp)} - - - {toolKind || 'Tool'} - - {status} - - - {title} - - ); -} - export function AgentStreamView({ agentId, agent, @@ -149,10 +49,10 @@ export function AgentStreamView({ switch (item.kind) { case 'user_message': return ( - ); @@ -167,23 +67,32 @@ export function AgentStreamView({ case 'thought': return ( - ); - case 'tool_call': + case 'tool_call': { + // Map status: pending/in_progress -> executing, completed -> completed, failed -> failed + const toolStatus = item.status === 'pending' || item.status === 'in_progress' + ? 'executing' as const + : item.status === 'completed' + ? 'completed' as const + : 'failed' as const; + return ( - ); + } case 'plan': // TODO: Render plan component @@ -209,164 +118,6 @@ const stylesheet = StyleSheet.create((theme) => ({ flex: 1, backgroundColor: theme.colors.background, }, - userMessageCard: { - backgroundColor: theme.colors.card, - borderRadius: theme.borderRadius.lg, - padding: theme.spacing[3], - marginBottom: theme.spacing[2], - borderLeftWidth: 3, - borderLeftColor: theme.colors.primary, - }, - userMessageLabel: { - color: theme.colors.primary, - fontSize: theme.fontSize.xs, - fontWeight: theme.fontWeight.semibold, - marginBottom: theme.spacing[1], - }, - userMessageText: { - color: theme.colors.foreground, - fontSize: theme.fontSize.sm, - lineHeight: 20, - }, - timestamp: { - color: theme.colors.mutedForeground, - fontSize: theme.fontSize.xs, - marginBottom: theme.spacing[1], - }, - thoughtCard: { - backgroundColor: theme.colors.card, - borderRadius: theme.borderRadius.lg, - padding: theme.spacing[3], - marginBottom: theme.spacing[2], - borderLeftWidth: 3, - borderLeftColor: theme.colors.palette.purple[500], - }, - thoughtLabel: { - color: theme.colors.palette.purple[500], - fontSize: theme.fontSize.xs, - fontWeight: theme.fontWeight.semibold, - marginBottom: theme.spacing[1], - }, - thoughtText: { - color: theme.colors.foreground, - fontSize: theme.fontSize.sm, - lineHeight: 20, - fontStyle: 'italic', - }, - toolCallCard: { - backgroundColor: theme.colors.card, - borderRadius: theme.borderRadius.lg, - padding: theme.spacing[3], - marginBottom: theme.spacing[2], - borderLeftWidth: 3, - borderLeftColor: theme.colors.palette.blue[500], - }, - toolCallHeader: { - flexDirection: 'row', - justifyContent: 'space-between', - alignItems: 'center', - marginBottom: theme.spacing[1], - }, - toolCallLabel: { - color: theme.colors.palette.blue[500], - fontSize: theme.fontSize.xs, - fontWeight: theme.fontWeight.semibold, - }, - toolCallStatus: { - fontSize: theme.fontSize.xs, - fontWeight: theme.fontWeight.medium, - }, - toolCallTitle: { - color: theme.colors.foreground, - fontSize: theme.fontSize.sm, - lineHeight: 20, - }, - header: { - backgroundColor: theme.colors.card, - borderBottomWidth: theme.borderWidth[1], - borderBottomColor: theme.colors.border, - paddingHorizontal: theme.spacing[4], - paddingBottom: theme.spacing[4], - }, - backButton: { - backgroundColor: theme.colors.muted, - paddingHorizontal: theme.spacing[4], - paddingVertical: theme.spacing[2], - borderRadius: theme.borderRadius.lg, - marginBottom: theme.spacing[4], - }, - backButtonText: { - color: theme.colors.foreground, - fontSize: theme.fontSize.sm, - fontWeight: theme.fontWeight.semibold, - textAlign: "center", - }, - agentInfoRow: { - flexDirection: "row", - alignItems: "center", - justifyContent: "space-between", - marginBottom: theme.spacing[3], - }, - agentIdRow: { - flexDirection: "row", - alignItems: "center", - gap: theme.spacing[2], - }, - agentLabel: { - color: theme.colors.mutedForeground, - fontSize: theme.fontSize.sm, - }, - agentId: { - color: theme.colors.foreground, - fontSize: theme.fontSize.sm, - fontFamily: "monospace", - }, - statusBadge: { - flexDirection: "row", - alignItems: "center", - gap: theme.spacing[2], - paddingHorizontal: theme.spacing[3], - paddingVertical: theme.spacing[1], - borderRadius: theme.borderRadius.full, - }, - statusIcon: { - color: theme.colors.foreground, - fontSize: theme.fontSize.xs, - }, - statusText: { - color: theme.colors.foreground, - fontSize: theme.fontSize.xs, - fontWeight: theme.fontWeight.semibold, - }, - createdText: { - color: theme.colors.mutedForeground, - fontSize: theme.fontSize.xs, - }, - controlButtons: { - flexDirection: "row", - gap: theme.spacing[2], - marginTop: theme.spacing[4], - }, - cancelButton: { - flex: 1, - backgroundColor: theme.colors.palette.orange[600], - paddingHorizontal: theme.spacing[4], - paddingVertical: theme.spacing[2], - borderRadius: theme.borderRadius.lg, - }, - killButton: { - flex: 1, - backgroundColor: theme.colors.destructive, - paddingHorizontal: theme.spacing[4], - paddingVertical: theme.spacing[2], - borderRadius: theme.borderRadius.lg, - }, - controlButtonText: { - color: theme.colors.foreground, - fontSize: theme.fontSize.sm, - fontWeight: theme.fontWeight.semibold, - textAlign: "center", - }, scrollView: { flex: 1, paddingHorizontal: theme.spacing[4], diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 24b46ccbd..a1ea69e65 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -1,9 +1,51 @@ import type { SessionNotification } from '@agentclientprotocol/sdk'; -// Simple ID generator that works in React Native -let idCounter = 0; -function generateId(): string { - return `stream_${Date.now()}_${idCounter++}`; +/** + * Simple hash function for deterministic ID generation + * Uses a basic string hash algorithm for consistency across runs + */ +function simpleHash(str: string): string { + let hash = 0; + for (let i = 0; i < str.length; i++) { + const char = str.charCodeAt(i); + hash = ((hash << 5) - hash) + char; + hash = hash & hash; // Convert to 32bit integer + } + return Math.abs(hash).toString(36); +} + +/** + * Generate a simple unique ID (timestamp + random) + */ +export function generateMessageId(): string { + return `msg_${Date.now()}_${Math.random().toString(36).substring(2, 11)}`; +} + +/** + * Derive deterministic ID for user message based on timestamp and content + * If messageId is provided from server, use that instead + */ +function deriveUserMessageId(text: string, timestamp: Date, messageId?: string): string { + if (messageId) { + return messageId; + } + return `user_${timestamp.getTime()}_${simpleHash(text)}`; +} + +/** + * Derive deterministic ID for assistant message based on index in state + */ +function deriveAssistantMessageId(state: StreamItem[]): string { + const count = state.filter(s => s.kind === 'assistant_message').length; + return `assistant_${count}`; +} + +/** + * Derive deterministic ID for thought based on index in state + */ +function deriveThoughtId(state: StreamItem[]): string { + const count = state.filter(s => s.kind === 'thought').length; + return `thought_${count}`; } /** @@ -86,7 +128,7 @@ export interface ArtifactItem { * Parsed notification types (internal representation before converting to StreamItem) */ type ParsedNotification = - | { kind: 'user_message_chunk'; text: string } + | { kind: 'user_message_chunk'; text: string; messageId?: string } | { kind: 'agent_message_chunk'; text: string } | { kind: 'agent_thought_chunk'; text: string } | { kind: 'tool_call'; toolCallId: string; title: string; status?: string; toolKind?: string; rawInput?: any; rawOutput?: any; content?: any[]; locations?: any[] } @@ -113,6 +155,7 @@ function parseNotification(notification: SessionNotification | any): ParsedNotif return { kind: 'user_message_chunk', text: update.content?.text || '', + messageId: update.messageId, }; case 'agent_message_chunk': @@ -201,9 +244,16 @@ export function reduceStreamUpdate( // User message chunks - always create new message (they're complete from server) if (parsed.kind === 'user_message_chunk') { + const id = deriveUserMessageId(parsed.text, timestamp, parsed.messageId); + + // Idempotency check - if this exact message already exists, don't add it again + if (state.some(item => item.id === id)) { + return state; + } + return [...state, { kind: 'user_message', - id: generateId(), + id, text: parsed.text, timestamp, }]; @@ -222,10 +272,11 @@ export function reduceStreamUpdate( }, ]; } - // Create new message + // Create new message with deterministic ID + const id = deriveAssistantMessageId(state); return [...state, { kind: 'assistant_message', - id: generateId(), + id, text: parsed.text, timestamp, }]; @@ -244,10 +295,11 @@ export function reduceStreamUpdate( }, ]; } - // Create new thought + // Create new thought with deterministic ID + const id = deriveThoughtId(state); return [...state, { kind: 'thought', - id: generateId(), + id, text: parsed.text, timestamp, }]; @@ -273,8 +325,33 @@ export function reduceStreamUpdate( }); } - // New tool call + // New tool call - check if exists first (server may send multiple notifications for same tool) if (parsed.kind === 'tool_call') { + const existingIndex = state.findIndex( + item => item.kind === 'tool_call' && item.toolCallId === parsed.toolCallId + ); + + if (existingIndex >= 0) { + // Update existing tool call + return state.map((item, idx) => { + if (idx === existingIndex && item.kind === 'tool_call') { + return { + ...item, + title: parsed.title, + status: (parsed.status as any) || item.status, + toolKind: (parsed.toolKind as any) || item.toolKind, + rawInput: parsed.rawInput ? { ...item.rawInput, ...parsed.rawInput } : item.rawInput, + rawOutput: parsed.rawOutput ? { ...item.rawOutput, ...parsed.rawOutput } : item.rawOutput, + content: parsed.content || item.content, + locations: parsed.locations || item.locations, + timestamp, + }; + } + return item; + }); + } + + // Create new tool call return [...state, { kind: 'tool_call', id: parsed.toolCallId, @@ -290,12 +367,12 @@ export function reduceStreamUpdate( }]; } - // Plan - replace existing or create new + // Plan - replace existing or create new (uses constant ID since only one plan exists) if (parsed.kind === 'plan') { const withoutPlan = state.filter(item => item.kind !== 'plan'); return [...withoutPlan, { kind: 'plan', - id: generateId(), + id: 'plan_latest', entries: parsed.entries as any, timestamp, }]; diff --git a/packages/server/src/server/messages.ts b/packages/server/src/server/messages.ts index 7b85c305e..ceded7bf2 100644 --- a/packages/server/src/server/messages.ts +++ b/packages/server/src/server/messages.ts @@ -48,6 +48,7 @@ export const SendAgentMessageSchema = z.object({ type: z.literal("send_agent_message"), agentId: z.string(), text: z.string(), + messageId: z.string().optional(), // Client-provided ID for deduplication }); export const SendAgentAudioSchema = z.object({ diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 929d5739e..b791907b0 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -315,7 +315,7 @@ export class Session { break; case "send_agent_message": - await this.handleSendAgentMessage(msg.agentId, msg.text); + await this.handleSendAgentMessage(msg.agentId, msg.text, msg.messageId); break; case "send_agent_audio": @@ -442,7 +442,7 @@ export class Session { /** * Handle text message to agent */ - private async handleSendAgentMessage(agentId: string, text: string): Promise { + private async handleSendAgentMessage(agentId: string, text: string, messageId?: string): Promise { console.log( `[Session ${this.clientId}] Sending text to agent ${agentId}: ${text.substring(0, 50)}...` ); @@ -463,6 +463,7 @@ export class Session { type: "text", text: text, }, + ...(messageId ? { messageId } : {}), }, } as any, }, @@ -660,9 +661,27 @@ export class Session { // Get live agents with session modes const agents = this.agentManager?.listAgents() || []; - // Subscribe to all existing agents (in case of reconnection) + // Subscribe to all existing agents and send their history (in case of reconnection) for (const agent of agents) { this.subscribeToAgent(agent.id); + + // Get historical updates from AgentManager + const history = this.agentManager.getAgentUpdates(agent.id); + + // Send each historical update to client + console.log( + `[Session ${this.clientId}] Sending ${history.length} historical updates for agent ${agent.id}` + ); + for (const update of history) { + this.emit({ + type: "agent_update", + payload: { + agentId: update.agentId, + timestamp: update.timestamp.toISOString(), + notification: update.notification, + }, + }); + } } // Get live commands from terminal manager