from pathlib import Path from . import Entry from ..io_buffer import IOBuffer from .background_entry import BackgroundEntry from .parse_error_entry import ParseErrorEntry from .read_entry import ReadEntry from .reasoning_entry import ReasoningEntry from .repeat_entry import RepeatEntry from .single_entry import SingleEntry from .write_entry import WriteEntry class EntryFactory: def create_entry(data: dict, work_dir: Path, io_buffer: IOBuffer) -> Entry: entry_type = data.get("type") entry_id = data.get("id") if entry_type == "background": return BackgroundEntry(entry_id, work_dir, data["script"]) elif entry_type == "read_stdin": return ReadEntry(entry_id, io_buffer) elif entry_type == "reasoning": return ReasoningEntry(entry_id, data["content"]) elif entry_type == "repeat": return RepeatEntry( entry_id, work_dir, data["script"], data.get("timeout"), data.get("limit") ) elif entry_type == "single": return SingleEntry( entry_id, work_dir, data["script"], data.get("timeout"), data.get("limit") ) elif entry_type == "write": return WriteEntry(entry_id, data["content"], io_buffer) raise ValueError(f"Unknown entry type: {entry_type}") def update_entry(entry: Entry, data: dict): if isinstance(entry, Entry): if "id" in data: entry.id = data["id"] if isinstance(entry, SingleEntry): if "work_dir" in data: entry.work_dir = Path(data["work_dir"]) if "script" in data: entry.script = data["script"] if "timeout" in data: entry.timeout = float(data["timeout"]) if data["timeout"] is not None else None if "limit" in data: entry.limit = int(data["limit"]) if data["limit"] else None if "stdout" in data: entry.stdout = data["stdout"] if "stderr" in data: entry.stderr = data["stderr"] if "exit_code" in data: entry.exit_code = int(data["exit_code"]) if data["exit_code"] is not None else None if "executed" in data: entry.executed = bool(data["executed"]) if "timed_out" in data: entry.timed_out = bool(data["timed_out"]) if isinstance(entry, RepeatEntry): if "work_dir" in data: entry.work_dir = Path(data["work_dir"]) if "script" in data: entry.script = data["script"] if "timeout" in data: entry.timeout = float(data["timeout"]) if data["timeout"] is not None else None if "limit" in data: entry.limit = int(data["limit"]) if data["limit"] else None if "stdout" in data: entry.stdout = data["stdout"] if "stderr" in data: entry.stderr = data["stderr"] if "exit_code" in data: entry.exit_code = int(data["exit_code"]) if data["exit_code"] is not None else None if "executed" in data: entry.executed = bool(data["executed"]) if "timed_out" in data: entry.timed_out = bool(data["timed_out"]) elif isinstance(entry, BackgroundEntry): if "work_dir" in data: entry.work_dir = Path(data["work_dir"]) if "script" in data: entry.script = data["script"] if "stdout" in data: entry.stdout = data["stdout"] if "stderr" in data: entry.stderr = data["stderr"] if "exit_code" in data: entry.exit_code = int(data["exit_code"]) if data["exit_code"] is not None else None elif isinstance(entry, ParseErrorEntry): if "content" in data: entry.content = data["content"] if "error" in data: entry.error = data["error"] elif isinstance(entry, ReadEntry): if "content" in data: entry.content = data["content"] if "read" in data: entry.read = bool(data["read"]) elif isinstance(entry, ReasoningEntry): if "content" in data: entry.content = data["content"] elif isinstance(entry, WriteEntry): if "content" in data: entry.content = data["content"] if "written" in data: entry.written = bool(data["written"])