from typing import Dict, List, Any, Optional from dataclasses import dataclass @dataclass class Segment: segment_id: str code: str namespace: Dict[str, Any] class SegmentRuntimeError(Exception): pass class SegmentRuntime: @staticmethod def stitch_segments(segments: List[Segment]) -> str: if not segments: raise SegmentRuntimeError("No segments to stitch") lines = [] for segment in segments: lines.append(f"# --- Segment {segment.segment_id} ---") lines.append(segment.code) lines.append("") return "\n".join(lines) @staticmethod def merge_namespaces(namespaces: List[Dict[str, Any]]) -> Dict[str, Any]: merged = {} for ns in namespaces: for key, value in ns.items(): if key in merged and merged[key] != value: raise SegmentRuntimeError(f"Namespace conflict on key: {key}") merged[key] = value return merged @staticmethod def execute_segments(segments: List[Segment], debug: bool = False) -> Dict[str, Any]: if not segments: raise SegmentRuntimeError("No segments to execute") merged_ns = SegmentRuntime.merge_namespaces([s.namespace for s in segments]) stitched_code = SegmentRuntime.stitch_segments(segments) globals_dict = { "__name__": "__main__", "__segments__": [s.segment_id for s in segments], } globals_dict.update(merged_ns) try: exec(compile(stitched_code, "", "exec"), globals_dict) except Exception as e: if debug: raise raise SegmentRuntimeError(f"Execution failed: {e}") return globals_dict