(
graph_context: GraphContext,
prompt: str,
attachments: Sequence[Union[str, Path]],
)
| 51 | |
| 52 | |
| 53 | def _build_task_input( |
| 54 | graph_context: GraphContext, |
| 55 | prompt: str, |
| 56 | attachments: Sequence[Union[str, Path]], |
| 57 | ) -> Union[str, list[Message]]: |
| 58 | if not attachments: |
| 59 | return prompt |
| 60 | |
| 61 | attachments_dir = graph_context.directory / "code_workspace" / "attachments" |
| 62 | attachments_dir.mkdir(parents=True, exist_ok=True) |
| 63 | store = AttachmentStore(attachments_dir) |
| 64 | builder = TaskInputBuilder(store) |
| 65 | normalized_paths = [str(Path(path).expanduser()) for path in attachments] |
| 66 | return builder.build_from_file_paths(prompt, normalized_paths) |
| 67 | |
| 68 | |
| 69 | def _run_workflow_with_logger( |
no test coverage detected