MCPcopy Create free account
hub / github.com/Noumena-Network/code / handleMessageFromStream

Function handleMessageFromStream

src/utils/messages.ts:2999–3195  ·  view source on GitHub ↗
(
  message:
    | Message
    | TombstoneMessage
    | StreamEvent
    | RequestStartEvent
    | ToolUseSummaryMessage,
  onMessage: (message: Message) => void,
  onUpdateLength: (newContent: string) => void,
  onSetStreamMode: (mode: SpinnerMode) => void,
  onStreamingToolUses: (
    f: (streamingToolUse: StreamingToolUse[]) => StreamingToolUse[],
  ) => void,
  onTombstone?: (message: Message) => void,
  onStreamingThinking?: (
    f: (current: StreamingThinking | null) => StreamingThinking | null,
  ) => void,
  onApiMetrics?: (metrics: { ttftMs: number }) => void,
  onStreamingText?: (f: (current: string | null) => string | null) => void,
)

Source from the content-addressed store, hash-verified

2997 * Handles messages from a stream, updating response length for deltas and appending completed messages
2998 */
2999export function handleMessageFromStream(
3000 message:
3001 | Message
3002 | TombstoneMessage
3003 | StreamEvent
3004 | RequestStartEvent
3005 | ToolUseSummaryMessage,
3006 onMessage: (message: Message) => void,
3007 onUpdateLength: (newContent: string) => void,
3008 onSetStreamMode: (mode: SpinnerMode) => void,
3009 onStreamingToolUses: (
3010 f: (streamingToolUse: StreamingToolUse[]) => StreamingToolUse[],
3011 ) => void,
3012 onTombstone?: (message: Message) => void,
3013 onStreamingThinking?: (
3014 f: (current: StreamingThinking | null) => StreamingThinking | null,
3015 ) => void,
3016 onApiMetrics?: (metrics: { ttftMs: number }) => void,
3017 onStreamingText?: (f: (current: string | null) => string | null) => void,
3018): void {
3019 if (
3020 message.type !== 'stream_event' &&
3021 message.type !== 'stream_request_start'
3022 ) {
3023 // Handle tombstone messages - remove the targeted message instead of adding
3024 if (message.type === 'tombstone') {
3025 onTombstone?.(message.message)
3026 return
3027 }
3028 // Tool use summary messages are SDK-only, ignore them in stream handling
3029 if (message.type === 'tool_use_summary') {
3030 return
3031 }
3032 // Capture complete thinking blocks for real-time display in transcript mode
3033 if (message.type === 'assistant') {
3034 const thinkingBlock = message.message.content.find(
3035 block => block.type === 'thinking',
3036 )
3037 if (thinkingBlock && thinkingBlock.type === 'thinking') {
3038 onStreamingThinking?.(() => ({
3039 thinking: thinkingBlock.thinking,
3040 isStreaming: false,
3041 streamingEndedAt: Date.now(),
3042 }))
3043 }
3044
3045 // When a committed assistant message contains a tool_use block, remove the
3046 // matching streaming preview entry immediately. This prevents the synthetic
3047 // streaming message and the real committed message from rendering together
3048 // during the window between content_block_stop and message_stop.
3049 const toolUseIds = message.message.content
3050 .filter(block => block.type === 'tool_use')
3051 .map(block => (block as { id: string }).id)
3052 if (toolUseIds.length > 0) {
3053 onStreamingToolUses(prev =>
3054 prev.filter(
3055 streamingToolUse =>
3056 !toolUseIds.includes(streamingToolUse.contentBlock.id),

Callers 3

execAgentHookFunction · 0.85
useRemoteSessionFunction · 0.85

Calls 3

onStreamingThinkingFunction · 0.85
onStreamingTextFunction · 0.85
isConnectorTextBlockFunction · 0.85

Tested by

no test coverage detected