(self, tool_event_id: Optional[str], content: str)
| 103 | self._tool_preview_lengths[tool_event_id] = length |
| 104 | |
| 105 | async def _stream_code_preview(self, tool_event_id: Optional[str], content: str) -> None: |
| 106 | if not tool_event_id or not content: |
| 107 | return |
| 108 | |
| 109 | already_sent = self._tool_preview_lengths.get(tool_event_id, 0) |
| 110 | total_len = len(content) |
| 111 | if already_sent >= total_len: |
| 112 | return |
| 113 | |
| 114 | max_chunks = 18 |
| 115 | min_step = 200 |
| 116 | step = max(min_step, total_len // max_chunks) |
| 117 | start = already_sent if already_sent > 0 else 0 |
| 118 | |
| 119 | for end in range(start + step, total_len, step): |
| 120 | await self._send("setCode", content[:end]) |
| 121 | self._mark_preview_length(tool_event_id, end) |
| 122 | await asyncio.sleep(0.01) |
| 123 | |
| 124 | await self._send("setCode", content) |
| 125 | self._mark_preview_length(tool_event_id, total_len) |
| 126 | |
| 127 | async def _handle_streamed_tool_delta( |
| 128 | self, |
no test coverage detected