--------------------------------------------------------------------------- OpenAI Responses API streaming injection Chunks are SSE events without outer newlines: "event: type\ndata: {...}" The handler adds surrounding newlines. -----------------------------------------------------------------------
(dataChan <-chan []byte, rule *config.ToolCallInjectionRule, modelName string)
| 206 | // InjectResponsesStream wraps a data channel to inject function_call events |
| 207 | // before the response.completed event. |
| 208 | func InjectResponsesStream(dataChan <-chan []byte, rule *config.ToolCallInjectionRule, modelName string) <-chan []byte { |
| 209 | out := make(chan []byte, 16) |
| 210 | go func() { |
| 211 | defer close(out) |
| 212 | |
| 213 | lastSeqNum := 0 |
| 214 | nextOutputIdx := 0 |
| 215 | injected := false |
| 216 | |
| 217 | for chunk := range dataChan { |
| 218 | // Track sequence numbers and output indices. |
| 219 | if j := extractSSEJSON(chunk); j != nil { |
| 220 | if seq := int(gjson.GetBytes(j, "sequence_number").Int()); seq > lastSeqNum { |
| 221 | lastSeqNum = seq |
| 222 | } |
| 223 | if gjson.GetBytes(j, "output_index").Exists() { |
| 224 | if oi := int(gjson.GetBytes(j, "output_index").Int()); oi >= nextOutputIdx { |
| 225 | nextOutputIdx = oi + 1 |
| 226 | } |
| 227 | } |
| 228 | } |
| 229 | |
| 230 | // Detect response.completed event. |
| 231 | if !injected && isResponsesCompleted(chunk) { |
| 232 | argsJSON, _ := json.Marshal(rule.Arguments) |
| 233 | callID := GenerateOpenAIToolCallID(rule.TaskID) |
| 234 | fcID := "fc_" + callID |
| 235 | oi := nextOutputIdx |
| 236 | seq := lastSeqNum + 1 |
| 237 | |
| 238 | fcItem := map[string]any{ |
| 239 | "id": fcID, |
| 240 | "type": "function_call", |
| 241 | "status": "completed", |
| 242 | "name": rule.ToolName, |
| 243 | "arguments": string(argsJSON), |
| 244 | "call_id": callID, |
| 245 | } |
| 246 | |
| 247 | // response.output_item.added |
| 248 | out <- buildResponsesSSE("response.output_item.added", map[string]any{ |
| 249 | "type": "response.output_item.added", |
| 250 | "sequence_number": seq, |
| 251 | "output_index": oi, |
| 252 | "item": map[string]any{ |
| 253 | "id": fcID, |
| 254 | "type": "function_call", |
| 255 | "status": "in_progress", |
| 256 | "name": rule.ToolName, |
| 257 | "arguments": "", |
| 258 | "call_id": callID, |
| 259 | }, |
| 260 | }) |
| 261 | seq++ |
| 262 | |
| 263 | // response.function_call_arguments.delta |
| 264 | out <- buildResponsesSSE("response.function_call_arguments.delta", map[string]any{ |
| 265 | "type": "response.function_call_arguments.delta", |
no test coverage detected