| 1 | from __future__ import annotations |
| 2 | |
| 3 | from typing import Any |
| 4 | |
| 5 | |
| 6 | _TIMEOUT_KEYS = ("first_output_timeout", "between_output_timeout", "max_exec_timeout", "dialog_timeout") |
| 7 | |
| 8 | |
| 9 | def _parse_patterns(value: object) -> list[str]: |
| 10 | if isinstance(value, str): |
| 11 | items = value.splitlines() |
| 12 | elif isinstance(value, (list, tuple)): |
| 13 | items = value |
| 14 | else: |
| 15 | return [] |
| 16 | return [str(item).strip() for item in items if str(item).strip()] |
| 17 | |
| 18 | |
| 19 | def _parse_timeouts(cfg: dict[str, Any], prefix: str, defaults: tuple[int, ...]) -> dict[str, int]: |
| 20 | return { |
| 21 | key: int(cfg.get(f"{prefix}_{key}", default)) |
| 22 | for key, default in zip(_TIMEOUT_KEYS, defaults) |
| 23 | } |
| 24 | |
| 25 | |
| 26 | def build_exec_config(*, agent: object | None = None) -> dict[str, Any]: |
| 27 | from helpers import plugins |
| 28 | |
| 29 | cfg = plugins.get_plugin_config("_code_execution", agent=agent) or {} |
| 30 | return { |
| 31 | "version": 1, |
| 32 | "code_exec_timeouts": _parse_timeouts(cfg, "code_exec", (30, 15, 240, 5)), |
| 33 | "output_timeouts": _parse_timeouts(cfg, "output", (120, 60, 600, 5)), |
| 34 | "prompt_patterns": _parse_patterns(cfg.get("prompt_patterns", "")), |
| 35 | "dialog_patterns": _parse_patterns(cfg.get("dialog_patterns", "")), |
| 36 | } |