from pathlib import Path from typing import List import xml.etree.ElementTree as ET from .entry import Entry from .entry.background_entry import BackgroundEntry from .entry.parse_error_entry import ParseErrorEntry from .entry.read_entry import ReadEntry from .entry.reasoning_entry import ReasoningEntry from .entry.repeat_entry import RepeatEntry from .entry.single_entry import SingleEntry from .entry.write_entry import WriteEntry class IterationParser: """Parses iteration XML files into entries""" @staticmethod def parse_iteration(content: str, work_dir: Path, io_buffer) -> tuple[str, str, List[Entry]]: """ Parse iteration XML content into context, response and entries Args: content: Raw XML content as string io_buffer: Buffer to use for IO operations Returns: Tuple of (context_str, response_str, entry_list) """ root = ET.fromstring(content) context_elem = root.find("context") response_elem = root.find("response") if context_elem is None: raise ValueError("Invalid iteration file - missing context") if response_elem is None: raise ValueError("Invalid iteration file - missing response") context = context_elem.text.strip() response = response_elem.text.strip() entries = [] for elem in ET.fromstring(context): if elem.tag == "background": entries.append(IterationParser._parse_background(elem, work_dir)) elif elem.tag == "parse_error": entries.append(IterationParser._parse_parse_error(elem)) elif elem.tag == "read_stdin": entries.append(IterationParser._parse_read(elem, io_buffer)) elif elem.tag == "reasoning": entries.append(IterationParser._parse_reasoning(elem)) elif elem.tag == "repeat": entries.append(IterationParser._parse_repeat(elem, work_dir)) elif elem.tag == "single": entries.append(IterationParser._parse_single(elem, work_dir)) elif elem.tag == "write_stdout": entries.append(IterationParser._parse_write(elem, io_buffer)) return (context, response, entries) @staticmethod def _parse_background(elem: ET.Element, work_dir: Path) -> BackgroundEntry: entry = BackgroundEntry( id=elem.get("id"), script=(elem.text or "").strip(), work_dir=work_dir ) if elem.get("exit_code"): entry.exit_code = int(elem.get("exit_code")) stdout = elem.find("stdout") if stdout is not None: entry.stdout = (stdout.text or "").strip() stderr = elem.find("stderr") if stderr is not None: entry.stderr = (stderr.text or "").strip() return entry @staticmethod def _parse_parse_error(elem: ET.Element) -> ParseErrorEntry: error = elem.find("error") content = elem.find("content") return ParseErrorEntry( id=elem.get("id"), content=(content.text or "").strip(), error=(error.text or "").strip() ) @staticmethod def _parse_read(elem: ET.Element, io_buffer) -> ReadEntry: entry = ReadEntry( id=elem.get("id"), io_buffer=io_buffer ) entry.content = (elem.text or "").strip() entry.read = True return entry @staticmethod def _parse_reasoning(elem: ET.Element) -> ReasoningEntry: return ReasoningEntry( id=elem.get("id"), content=(elem.text or "").strip() ) @staticmethod def _parse_repeat(elem: ET.Element, work_dir: Path) -> RepeatEntry: entry = RepeatEntry( id=elem.get("id"), work_dir=work_dir, script=(elem.text or "").strip(), timeout=float(elem.get("timeout")) if elem.get("timeout") else None, limit=int(elem.get("limit")) if elem.get("limit") else None ) if elem.get("exit_code"): entry.exit_code = int(elem.get("exit_code")) entry.timed_out = elem.get("timed_out") == "true" stdout = elem.find("stdout") if stdout is not None: entry.stdout = (stdout.text or "").strip() stderr = elem.find("stderr") if stderr is not None: entry.stderr = (stderr.text or "").strip() return entry @staticmethod def _parse_single(elem: ET.Element, work_dir: Path) -> SingleEntry: entry = SingleEntry( id=elem.get("id"), work_dir=work_dir, script=(elem.text or "").strip(), timeout=float(elem.get("timeout")) if elem.get("timeout") else None, limit=int(elem.get("limit")) if elem.get("limit") else None ) if elem.get("exit_code"): entry.exit_code = int(elem.get("exit_code")) entry.executed = True entry.timed_out = elem.get("timed_out") == "true" stdout = elem.find("stdout") if stdout is not None: entry.stdout = (stdout.text or "").strip() stderr = elem.find("stderr") if stderr is not None: entry.stderr = (stderr.text or "").strip() return entry @staticmethod def _parse_write(elem: ET.Element, io_buffer) -> WriteEntry: entry = WriteEntry( id=elem.get("id"), content=(elem.text or "").strip(), io_buffer=io_buffer ) entry.written = True return entry