diff --git a/.github/workflows/build-base-image.yml b/.github/workflows/build-base-image.yml index aff9897..b9f101e 100644 --- a/.github/workflows/build-base-image.yml +++ b/.github/workflows/build-base-image.yml @@ -7,6 +7,7 @@ on: branches: - main - deep-agent + - feat/rhitaif-221 workflow_dispatch: # Allow manual trigger for testing permissions: diff --git a/src/frontend/hooks/useStreamingAPI.ts b/src/frontend/hooks/useStreamingAPI.ts index 9526e3b..f0a01d4 100644 --- a/src/frontend/hooks/useStreamingAPI.ts +++ b/src/frontend/hooks/useStreamingAPI.ts @@ -548,6 +548,9 @@ export function useStreamingAPI(threadId: string) { onMetadata(data) { setTraceId(data.trace_id); }, + onCodeOutput(content) { + console.log('[code_execution_output]', content); + }, }; manager.stream(streamRequest, callbacks).then(() => { @@ -704,6 +707,9 @@ export function useStreamingAPI(threadId: string) { onMetadata(data) { setTraceId(data.trace_id); }, + onCodeOutput(content) { + console.log('[code_execution_output]', content); + }, }; await manager.stream(resumeRequest, callbacks); diff --git a/src/frontend/lib/streaming/SSEProcessor.ts b/src/frontend/lib/streaming/SSEProcessor.ts index 3106374..e22e2cc 100644 --- a/src/frontend/lib/streaming/SSEProcessor.ts +++ b/src/frontend/lib/streaming/SSEProcessor.ts @@ -4,7 +4,8 @@ import type { HITLInterruptValue } from '@/types/deep-agent'; export type SSEChunk = | { type: 'token'; content: string; chunk_id: number } | { type: 'message'; content: Message; chunk_id: number } - | { type: 'interrupt'; content: { value: HITLInterruptValue; resumable: boolean }; chunk_id: number }; + | { type: 'interrupt'; content: { value: HITLInterruptValue; resumable: boolean }; chunk_id: number } + | { type: 'code_output'; content: string; chunk_id: number }; export type McpStatusData = { tool: string; @@ -32,7 +33,7 @@ function isRecord(value: unknown): value is Record { function parseSSEChunkPayload(parsed: unknown): SSEChunk | null { if (!isRecord(parsed)) return null; const type = parsed.type; - if (type !== 'token' && type !== 'message' && type !== 'interrupt') return null; + if (type !== 'token' && type !== 'message' && type !== 'interrupt' && type !== 'code_output') return null; const chunkIdRaw = parsed.chunk_id; if (typeof chunkIdRaw !== 'number' || !Number.isFinite(chunkIdRaw)) { return null; @@ -44,6 +45,11 @@ function parseSSEChunkPayload(parsed: unknown): SSEChunk | null { return { type: 'token', content: contentUnknown, chunk_id: chunkIdRaw }; } + if (type === 'code_output') { + if (typeof contentUnknown !== 'string') return null; + return { type: 'code_output', content: contentUnknown, chunk_id: chunkIdRaw }; + } + if (type === 'interrupt') { if (!isRecord(contentUnknown)) return null; let rawValue: unknown = contentUnknown.value; diff --git a/src/frontend/lib/streaming/StreamingManager.ts b/src/frontend/lib/streaming/StreamingManager.ts index e5a46c4..72dfa1b 100644 --- a/src/frontend/lib/streaming/StreamingManager.ts +++ b/src/frontend/lib/streaming/StreamingManager.ts @@ -47,6 +47,7 @@ export type StreamCallback = { onDone: () => void; onMcpStatus?: (event: McpStreamStatusEvent) => void; onMetadata?: (data: StreamMetadataPayload) => void; + onCodeOutput?: (content: string) => void; }; export class StreamingManager { @@ -93,6 +94,8 @@ export class StreamingManager { callbacks.onToken(event.data.content); } else if (event.data.type === 'interrupt') { callbacks.onInterrupt(event.data.content); + } else if (event.data.type === 'code_output') { + callbacks.onCodeOutput?.(event.data.content); } else { callbacks.onMessage(event.data.content); } diff --git a/src/server/router/proxy.router.ts b/src/server/router/proxy.router.ts index e0cb1b5..e43945c 100644 --- a/src/server/router/proxy.router.ts +++ b/src/server/router/proxy.router.ts @@ -381,7 +381,7 @@ async function proxyRoutes(fastify: FastifyInstance) { const runBody: Record = { assistant_id: 'agent', - stream_mode: ['messages', 'updates'], + stream_mode: ['messages', 'updates', 'custom'], }; if (isResume) { runBody.command = { resume: message }; @@ -529,6 +529,25 @@ async function proxyRoutes(fastify: FastifyInstance) { continue; } + // Code execution output streaming (stream_mode="custom") + // Skip ALL custom events to avoid corrupting partial text state + if (sseType === 'custom') { + if ( + typeof parsed === 'object' && + parsed !== null && + (parsed as Record).type === 'code_output' + ) { + const codeChunk = { + type: 'code_output', + content: (parsed as Record).content ?? '', + chunk_id: chunkId, + }; + reply.raw.write(`event: chunk\ndata: ${JSON.stringify(codeChunk)}\n\n`); + chunkId++; + } + continue; + } + const [uiChunks, nextPartial] = translateMessageEvent( sseType, parsed,