(self, messages: List[Message])
| 408 | f'But the checking limit has reached.') |
| 409 | |
| 410 | async def after_tool_call(self, messages: List[Message]): |
| 411 | is_prepare = len(messages[-1].tool_calls |
| 412 | or []) > 0 or messages[-1].role != 'assistant' |
| 413 | is_code_finish = '<result>' in messages[ |
| 414 | -1].content and '</result>' in messages[ |
| 415 | -1].content and not is_prepare |
| 416 | is_import = ( |
| 417 | self.is_stop_imports() and not is_code_finish and not is_prepare |
| 418 | and '<result>' in messages[-1].content |
| 419 | and '</result>' not in messages[-1].content) |
| 420 | is_check = messages[-1].role == 'assistant' and len( |
| 421 | messages[-1].tool_calls or []) == 0 and not is_import |
| 422 | message = messages[-1] |
| 423 | all_issues = [] |
| 424 | |
| 425 | if is_import: |
| 426 | self._before_import_check(messages) |
| 427 | |
| 428 | if is_code_finish: |
| 429 | |
| 430 | # Saving code |
| 431 | result, remaining_text = extract_code_blocks(message.content) |
| 432 | if result: |
| 433 | _response = remaining_text |
| 434 | saving_result = '' |
| 435 | for r in result: |
| 436 | path = r['filename'] |
| 437 | code = r['code'] |
| 438 | |
| 439 | path = os.path.join(self.output_dir, path) |
| 440 | |
| 441 | lock_dir = os.path.join(self.output_dir, DEFAULT_LOCK_DIR) |
| 442 | |
| 443 | with file_lock(lock_dir, r['filename']): |
| 444 | new_file = not os.path.exists(path) |
| 445 | if new_file: |
| 446 | os.makedirs(os.path.dirname(path), exist_ok=True) |
| 447 | with open(path, 'w') as f: |
| 448 | f.write(code) |
| 449 | self.add_unchecked_file(r['filename']) |
| 450 | else: |
| 451 | with open(path, 'r') as f: |
| 452 | code = f.read() |
| 453 | _response += f'\n<result>{path.split(".")[-1]}: {r["filename"]}\n{code}\n</result>\n' |
| 454 | saving_result += f'Save file <{r["filename"]}> successfully\n' |
| 455 | message.content = _response |
| 456 | messages.append(Message(role='user', content=saving_result)) |
| 457 | |
| 458 | if is_check: |
| 459 | # After checking when fix ended or write ended |
| 460 | for uncheck_file in list(self.unchecked_files.keys()): |
| 461 | with open(os.path.join(self.output_dir, uncheck_file), |
| 462 | 'r') as f: |
| 463 | _code = f.read() |
| 464 | lsp_feedback = await self._incremental_check( |
| 465 | uncheck_file, _code) |
| 466 | lsp_feedback = lsp_feedback.strip() |
| 467 | if lsp_feedback: |
nothing calls this directly
no test coverage detected