--------------------------------------------------------------------------- Claude Messages streaming injection Chunks are full SSE events: "event: type\ndata: {...}\n\n" --------------------------------------------------------------------------- InjectClaudeStream wraps a data channel to inject too
(dataChan <-chan []byte, rule *config.ToolCallInjectionRule, modelName string)
| 117 | // InjectClaudeStream wraps a data channel to inject tool_use content block |
| 118 | // events before the message_delta event, then changes stop_reason to "tool_use". |
| 119 | func InjectClaudeStream(dataChan <-chan []byte, rule *config.ToolCallInjectionRule, modelName string) <-chan []byte { |
| 120 | out := make(chan []byte, 16) |
| 121 | go func() { |
| 122 | defer close(out) |
| 123 | |
| 124 | nextBlockIdx := 0 |
| 125 | injected := false |
| 126 | |
| 127 | for chunk := range dataChan { |
| 128 | // Track content block indices from content_block_start events. |
| 129 | if j := extractSSEJSON(chunk); j != nil { |
| 130 | if gjson.GetBytes(j, "type").String() == "content_block_start" { |
| 131 | if idx := int(gjson.GetBytes(j, "index").Int()); idx >= nextBlockIdx { |
| 132 | nextBlockIdx = idx + 1 |
| 133 | } |
| 134 | } |
| 135 | } |
| 136 | |
| 137 | // Detect message_delta event (contains stop_reason). |
| 138 | if !injected && isClaudeMessageDelta(chunk) { |
| 139 | argsJSON, _ := json.Marshal(rule.Arguments) |
| 140 | toolUseID := GenerateClaudeToolUseID(rule.TaskID) |
| 141 | idx := nextBlockIdx |
| 142 | |
| 143 | // content_block_start (tool_use) |
| 144 | out <- buildClaudeSSE("content_block_start", map[string]any{ |
| 145 | "type": "content_block_start", |
| 146 | "index": idx, |
| 147 | "content_block": map[string]any{ |
| 148 | "type": "tool_use", |
| 149 | "id": toolUseID, |
| 150 | "name": rule.ToolName, |
| 151 | "input": map[string]any{}, |
| 152 | }, |
| 153 | }) |
| 154 | |
| 155 | // content_block_delta (input_json_delta) |
| 156 | out <- buildClaudeSSE("content_block_delta", map[string]any{ |
| 157 | "type": "content_block_delta", |
| 158 | "index": idx, |
| 159 | "delta": map[string]any{ |
| 160 | "type": "input_json_delta", |
| 161 | "partial_json": string(argsJSON), |
| 162 | }, |
| 163 | }) |
| 164 | |
| 165 | // content_block_stop |
| 166 | out <- buildClaudeSSE("content_block_stop", map[string]any{ |
| 167 | "type": "content_block_stop", |
| 168 | "index": idx, |
| 169 | }) |
| 170 | |
| 171 | // Modify message_delta: stop_reason → "tool_use". |
| 172 | chunk = sseJSONReplace(chunk, func(j []byte) []byte { |
| 173 | j, _ = sjson.SetBytes(j, "delta.stop_reason", "tool_use") |
| 174 | return j |
| 175 | }) |
| 176 | injected = true |
no test coverage detected