Skip to content

Handle chunk SSE events for token-by-token streaming #10

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Nauxie merged 3 commits into from
Jan 14, 2026
Merged

Handle chunk SSE events for token-by-token streaming #10

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e5e7617
Handle chunk SSE events for token-by-token streaming
Nauxie Jan 14, 2026
6e398ad
typing
Nauxie Jan 14, 2026
585603a
streaming full support
Nauxie Jan 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/api/client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@ import type {
SendMessageParams,
} from '../types';

export const DEFAULT_ENGINE_URL = 'https://chat-embed-deployment.onrender.com';
// export const DEFAULT_ENGINE_URL = 'http://localhost:8003';
//export const DEFAULT_ENGINE_URL = 'https://chat-embed-deployment.onrender.com';
export const DEFAULT_ENGINE_URL = 'http://localhost:8003';

export function createAPIClient(
engineBaseUrl: string = DEFAULT_ENGINE_URL
Expand Down
106 changes: 71 additions & 35 deletions src/hooks/useChat.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {

const sessionStartTime = useRef<number>(0);
const isInitialized = useRef(false);
const streamBuffers = useRef<Record<string, string>>({});

// Initialize or restore session
useEffect(() => {
Expand Down Expand Up @@ -99,6 +100,31 @@ export function useChat(options: UseChatOptions): UseChatReturn {
return '';
}, [config.embedId]);

const upsertAssistantMessage = useCallback((messageId: string, content: string) => {
setMessages((prev) => {
let found = false;
const next = prev.map((m) => {
if (m.id === messageId) {
found = true;
return { ...m, content, status: 'streaming' as const };
}
return m;
});

if (!found) {
next.push({
id: messageId,
role: 'assistant',
content,
timestamp: Date.now(),
status: 'streaming',
});
}

return next;
});
}, []);

const handleSSEEvent = useCallback(
(event: SSEEvent, messageId: string, updateSessionId: (id: string) => void) => {
switch (event.type) {
Expand All @@ -115,16 +141,22 @@ export function useChat(options: UseChatOptions): UseChatReturn {
break;
}
case 'message': {
// Agent message
// Agent message (complete message, e.g., from say())
const data = event.data as { content: string; role?: string };
if (data.content) {
setMessages((prev) =>
prev.map((m) =>
m.id === messageId
? { ...m, content: data.content, status: 'streaming' as const }
: m
)
);
streamBuffers.current[messageId] = data.content;
upsertAssistantMessage(messageId, data.content);
}
break;
}
case 'chunk': {
// Streaming chunk - append to existing message content
const data = event.data as { content: string };
if (data.content) {
const current = streamBuffers.current[messageId] ?? '';
const next = current + data.content;
streamBuffers.current[messageId] = next;
upsertAssistantMessage(messageId, next);
}
break;
}
Comment on lines +152 to +162
Copy link

@greptile-apps greptile-apps bot Jan 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

logic: 'chunk' is not defined in the SSEEventType union in src/types/index.ts. TypeScript will allow this code but it breaks type safety.

Suggested change
case 'chunk': {
// Streaming chunk - append to existing message content
const data = event.data as { content: string };
if (data.content) {
setMessages((prev) =>
prev.map((m) =>
m.id === messageId
? { ...m, content: m.content + data.content, status: 'streaming' as const }
: m
)
);
}
break;
}
case 'chunk': {
// Streaming chunk - append to existing message content
const data = event.data as { content: string };
if (data.content) {
setMessages((prev) =>
prev.map((m) =>
m.id === messageId
? { ...m, content: m.content + data.content, status: 'streaming' as const }
: m
)
);
}
break;
}

Add 'chunk' to the type definition:

export type SSEEventType =
  | 'session'
  | 'message'
  | 'chunk' // Token-by-token streaming
  | 'tool_call'
  | 'waiting'
  | 'done'
  | 'completed'
  | 'error';
Prompt To Fix With AI 
This is a comment left during a code review.
Path: src/hooks/useChat.ts
Line: 131:144

Comment:
**logic:** `'chunk'` is not defined in the `SSEEventType` union in `src/types/index.ts`. TypeScript will allow this code but it breaks type safety.

```suggestion
        case 'chunk': {
          // Streaming chunk - append to existing message content
          const data = event.data as { content: string };
          if (data.content) {
            setMessages((prev) =>
              prev.map((m) =>
                m.id === messageId
                  ? { ...m, content: m.content + data.content, status: 'streaming' as const }
                  : m
              )
            );
          }
          break;
        }
```

Add `'chunk'` to the type definition:
```typescript
export type SSEEventType =
  | 'session'
  | 'message'
  | 'chunk' // Token-by-token streaming
  | 'tool_call'
  | 'waiting'
  | 'done'
  | 'completed'
  | 'error';
```

How can I resolve this? If you propose a fix, please make it concise.

Expand Down Expand Up @@ -171,13 +203,9 @@ export function useChat(options: UseChatOptions): UseChatReturn {

// If tool call has content, also update the message
if (data.content) {
setMessages((prev) =>
prev.map((m) =>
m.id === messageId
? { ...m, content: data.content as string, status: 'streaming' as const }
: m
)
);
const content = data.content as string;
streamBuffers.current[messageId] = content;
upsertAssistantMessage(messageId, content);
}
break;
}
Expand All @@ -190,26 +218,31 @@ export function useChat(options: UseChatOptions): UseChatReturn {
setMessages((prev) =>
prev.map((m) => (m.id === messageId ? { ...m, status: 'sent' as const } : m))
);
delete streamBuffers.current[messageId];
break;
}
case 'completed': {
// Conversation ended by agent
setMessages((prev) =>
prev.map((m) => (m.id === messageId ? { ...m, status: 'sent' as const } : m))
);
// Mark session as completed
if (sessionId) {
storeSession(config.embedId, {
sessionId,
deploymentId: config.deploymentId,
workerId: config.workerId,
flowId: config.flowId,
startTime: sessionStartTime.current,
messages,
toolCalls,
status: 'completed',
});
}
setMessages((prev) => {
const updatedMessages = prev.map((m) =>
m.id === messageId ? { ...m, status: 'sent' as const } : m
);
// Mark session as completed with current messages
if (sessionId) {
storeSession(config.embedId, {
sessionId,
deploymentId: config.deploymentId,
workerId: config.workerId,
flowId: config.flowId,
startTime: sessionStartTime.current,
messages: updatedMessages,
toolCalls: [], // Tool calls stored separately
status: 'completed',
});
}
return updatedMessages;
});
delete streamBuffers.current[messageId];
break;
}
case 'error': {
Expand All @@ -229,7 +262,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
}
}
},
[config, sessionId, messages, toolCalls, onSessionStart]
[config, sessionId, onSessionStart, upsertAssistantMessage]
);

const processSSEStream = useCallback(
Expand All @@ -246,9 +279,12 @@ export function useChat(options: UseChatOptions): UseChatReturn {
buffer += decoder.decode(value, { stream: true });

// SSE messages are separated by double newlines
while (buffer.includes('\n\n')) {
const [message, rest] = buffer.split('\n\n', 2);
buffer = rest;
// Note: We use indexOf instead of split with limit because split(str, 2)
// only returns 2 elements and discards the rest, causing dropped chunks
let idx: number;
while ((idx = buffer.indexOf('\n\n')) !== -1) {
const message = buffer.slice(0, idx);
buffer = buffer.slice(idx + 2);

// Parse SSE data lines
for (const line of message.split('\n')) {
Expand Down
1 change: 1 addition & 0 deletions src/types/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ export interface ToolCall {
export type SSEEventType =
| 'session' // Session info (session_id, is_new)
| 'message' // Agent text response (content, role)
| 'chunk' // Streaming chunk (token-by-token content)
| 'tool_call' // Tool/function execution
| 'waiting' // Agent is processing
| 'done' // Stream complete
Expand Down
Loading