Function ApplyStreamInjection

sdk/api/handlers/injection_helpers.go:57–96

ApplyStreamInjection wraps data and error channels for streaming injection. If injection is nil, returns channels unchanged. For "replace" timing: drains upstream channels, returns fabricated stream. For "append" timing: wraps dataChan with format-specific stream injector.

(
	dataChan <-chan []byte,
	errChan <-chan *interfaces.ErrorMessage,
	injection *config.ToolCallInjectionRule,
	format, model string,
)

// For "replace" timing: drains upstream channels, returns fabricated stream.
// For "append" timing: wraps dataChan with format-specific stream injector.
func ApplyStreamInjection(
	dataChan <-chan []byte,
	errChan <-chan *interfaces.ErrorMessage,
	injection *config.ToolCallInjectionRule,
	format, model string,
) (<-chan []byte, <-chan *interfaces.ErrorMessage) {
	if injection == nil {
		return dataChan, errChan
	}
	f := toolinjection.GetFormat(format)
	if f == nil {
		return dataChan, errChan
	}

	if injection.Timing == "replace" {
		// Drain upstream channels in background.
		go func() {
			for range dataChan {
			}
		}()
		if errChan != nil {
			go func() {
				for range errChan {
				}
			}()
		}

		// Fabricate a complete stream.
		chunks := f.FabricateStream(injection, model)
		fakeChan := make(chan []byte, len(chunks))
		for _, chunk := range chunks {
			fakeChan <- chunk
		}
		close(fakeChan)
		return fakeChan, nil
	}

	// Append mode: wrap the real stream.
	return f.InjectStream(dataChan, injection, model), errChan
}