(chunk: str, full: str)
| 440 | await self.handle_reasoning_stream(stream_data["full"]) |
| 441 | |
| 442 | async def stream_callback(chunk: str, full: str): |
| 443 | nonlocal last_response_stream_full |
| 444 | await self.handle_intervention() |
| 445 | # output the agent response stream |
| 446 | if chunk == full: |
| 447 | printer.print("Response: ") # start of response |
| 448 | # Pass chunk and full data to extensions for processing |
| 449 | stream_data = {"chunk": chunk, "full": full} |
| 450 | stop_response: str | None = None |
| 451 | |
| 452 | snapshot = extract_tools.extract_json_root_string(full) |
| 453 | if snapshot: |
| 454 | parsed_snapshot = extract_tools.json_parse_dirty(snapshot) |
| 455 | if parsed_snapshot is not None: |
| 456 | try: |
| 457 | await self.validate_tool_request(parsed_snapshot) |
| 458 | except Exception: |
| 459 | pass |
| 460 | else: |
| 461 | previous_full = last_response_stream_full |
| 462 | stream_data["full"] = snapshot |
| 463 | if snapshot.startswith(previous_full): |
| 464 | stream_data["chunk"] = snapshot[len(previous_full) :] |
| 465 | else: |
| 466 | stream_data["chunk"] = snapshot |
| 467 | stop_response = snapshot |
| 468 | |
| 469 | await extension.call_extensions_async( |
| 470 | "response_stream_chunk", |
| 471 | self, |
| 472 | loop_data=self.loop_data, |
| 473 | stream_data=stream_data, |
| 474 | ) |
| 475 | # Stream masked chunk after extensions processed it |
| 476 | if stream_data.get("chunk"): |
| 477 | printer.stream(stream_data["chunk"]) |
| 478 | # Use the potentially modified full text for downstream processing |
| 479 | await self.handle_response_stream(stream_data["full"]) |
| 480 | last_response_stream_full = stream_data["full"] |
| 481 | if stop_response is not None: |
| 482 | return stop_response |
| 483 | |
| 484 | # call main LLM |
| 485 | llm_result = await self.call_chat_model_turn( |
nothing calls this directly
no test coverage detected