processStepStream consumes a fantasy StreamResponse and accumulates all content into a stepResult. Callbacks fire inline and their errors propagate directly.
( ctx context.Context, stream fantasy.StreamResponse, publishMessagePart func(codersdk.ChatMessageRole, codersdk.ChatMessagePart), )
| 926 | // accumulates all content into a stepResult. Callbacks fire |
| 927 | // inline and their errors propagate directly. |
| 928 | func processStepStream( |
| 929 | ctx context.Context, |
| 930 | stream fantasy.StreamResponse, |
| 931 | publishMessagePart func(codersdk.ChatMessageRole, codersdk.ChatMessagePart), |
| 932 | ) (stepResult, error) { |
| 933 | var result stepResult |
| 934 | |
| 935 | activeToolCalls := make(map[string]*fantasy.ToolCallContent) |
| 936 | activeTextContent := make(map[string]string) |
| 937 | activeReasoningContent := make(map[string]reasoningState) |
| 938 | // Track tool names by ID for input delta publishing. |
| 939 | toolNames := make(map[string]string) |
| 940 | |
| 941 | for part := range stream { |
| 942 | switch part.Type { |
| 943 | case fantasy.StreamPartTypeTextStart: |
| 944 | activeTextContent[part.ID] = "" |
| 945 | |
| 946 | case fantasy.StreamPartTypeTextDelta: |
| 947 | if _, exists := activeTextContent[part.ID]; exists { |
| 948 | activeTextContent[part.ID] += part.Delta |
| 949 | } |
| 950 | publishMessagePart(codersdk.ChatMessageRoleAssistant, codersdk.ChatMessageText(part.Delta)) |
| 951 | |
| 952 | case fantasy.StreamPartTypeTextEnd: |
| 953 | if text, exists := activeTextContent[part.ID]; exists { |
| 954 | result.content = append(result.content, fantasy.TextContent{ |
| 955 | Text: text, |
| 956 | ProviderMetadata: part.ProviderMetadata, |
| 957 | }) |
| 958 | delete(activeTextContent, part.ID) |
| 959 | } |
| 960 | |
| 961 | case fantasy.StreamPartTypeReasoningStart: |
| 962 | activeReasoningContent[part.ID] = reasoningState{ |
| 963 | text: part.Delta, |
| 964 | options: part.ProviderMetadata, |
| 965 | startedAt: dbtime.Now(), |
| 966 | } |
| 967 | |
| 968 | case fantasy.StreamPartTypeReasoningDelta: |
| 969 | if active, exists := activeReasoningContent[part.ID]; exists { |
| 970 | active.text += part.Delta |
| 971 | if len(part.ProviderMetadata) > 0 { |
| 972 | active.options = part.ProviderMetadata |
| 973 | } |
| 974 | activeReasoningContent[part.ID] = active |
| 975 | } |
| 976 | publishMessagePart(codersdk.ChatMessageRoleAssistant, codersdk.ChatMessageReasoning(part.Delta)) |
| 977 | |
| 978 | case fantasy.StreamPartTypeReasoningEnd: |
| 979 | if active, exists := activeReasoningContent[part.ID]; exists { |
| 980 | if len(part.ProviderMetadata) > 0 { |
| 981 | active.options = part.ProviderMetadata |
| 982 | } |
| 983 | content := fantasy.ReasoningContent{ |
| 984 | Text: active.text, |
| 985 | ProviderMetadata: active.options, |