""" PipelineEngine — executes a build DAG. For each action: 1. Schedule: pick best node via scheduler 2. Emit task_update(scheduled) with deps + critical info 3. Check cache: if hit, emit cache_update + skip execution 4. Execute: emit task_update(running), run via runtime router, emit task_update(done|failed) 5. On failure: emit failure event + stop downstream actions Emits events on the bus so the UI / WS stream / cause graph / timeline store all pick them up. """ import asyncio import time from typing import Any, Dict, List, Optional from backend.events.bus import EventBus from backend.events.schema import EventType from backend.scheduler.optimizer import choose_best_node from backend.executor.runtime_router import execute_action from backend.cache.minio_cache import MinioCache from backend.graph.plan import build_action_graph from backend.graph.critical_path import compute_critical_path class PipelineEngine: """Runs a build DAG against a cluster of nodes. Constructor: nodes: list of node dicts (config + state) node_registry: NodeStateRegistry singleton (for live state lookups) event_bus: EventBus singleton (for emitting events) cache: optional MinioCache (or compatible). If None, no cache. build_id: optional build identifier (for correlating events) """ def __init__(self, nodes, node_registry, event_bus: EventBus, cache: Optional[Any] = None, build_id: Optional[str] = None): self.nodes = nodes self.node_registry = node_registry self.bus = event_bus self.cache = cache # may be None self.build_id = build_id or f"build-{int(time.time())}" # Pause / step state (used by debugger) self._paused = False self._step_mode = False self._step_event = asyncio.Event() if asyncio.get_event_loop() else None # Results self.last_results: List[tuple] = [] self.critical_path: Optional[Dict[str, Any]] = None # ------------------------------------------------- # DEBUGGER HOOKS # ------------------------------------------------- def pause(self): self._paused = True def resume(self): self._paused = False if self._step_event: self._step_event.set() def step(self): """Advance one action when paused.""" self._step_mode = True if self._step_event: self._step_event.set() # ------------------------------------------------- # MAIN ENTRYPOINT # ------------------------------------------------- async def run(self, project) -> List[tuple]: """Build the project. Returns list of (action_name, state) tuples.""" actions = build_action_graph(project) self.critical_path = compute_critical_path(actions) critical_names = set(self.critical_path.get("score_map", {}).keys()) if self.critical_path else set() self.last_results = [] completed: Dict[str, str] = {} # name -> state failed_actions: set = set() for action in actions: # Skip if any dependency failed deps = action.get("deps", []) failed_deps = [d for d in deps if completed.get(d) == "failed"] if failed_deps: self.bus.emit( EventType.TASK_UPDATE, action=action["name"], node=None, state="skipped", reason=f"deps_failed:{','.join(failed_deps)}", meta={ "deps": deps, "critical": action["name"] in critical_names, "target": action.get("target", "native"), "build_id": self.build_id, }, ) completed[action["name"]] = "skipped" self.last_results.append((action["name"], "skipped")) continue # Wait if paused await self._wait_if_paused() # Schedule node = choose_best_node(self.nodes, action, self.node_registry) node_name = node["name"] if node else None self.bus.emit( EventType.TASK_UPDATE, action=action["name"], node=node_name, state="scheduled", meta={ "deps": deps, "critical": action["name"] in critical_names, "target": action.get("target", "native"), "build_id": self.build_id, }, ) # Cache check cache_hit = False if self.cache and action.get("hash"): try: if self.cache.exists(action["hash"]): cache_hit = True self.bus.emit( EventType.CACHE_UPDATE, action=action["name"], node=node_name, state="hit", meta={"build_id": self.build_id}, ) except Exception: # Cache errors are non-fatal pass # Emit running self.bus.emit( EventType.TASK_UPDATE, action=action["name"], node=node_name, state="running", meta={ "deps": deps, "critical": action["name"] in critical_names, "target": action.get("target", "native"), "cache": "hit" if cache_hit else None, "build_id": self.build_id, }, ) if not cache_hit: # Execute (in a thread so we don't block the event loop) # Pass action itself as workspace hint — execute_action reads # action["dir"] and action["env"] for proper cwd + env setup. try: rc = await asyncio.to_thread( execute_action, action, action.get("dir") or "/tmp", node or {"name": "localhost"} ) state = "done" if rc == 0 else "failed" except Exception as e: state = "failed" rc = -1 self.bus.emit( EventType.FAILURE, action=action["name"], node=node_name, state="failed", reason=f"execution_exception:{type(e).__name__}:{e}", meta={"build_id": self.build_id}, ) else: state = "done" rc = 0 # Emit terminal state self.bus.emit( EventType.TASK_UPDATE, action=action["name"], node=node_name, state=state, meta={ "deps": deps, "critical": action["name"] in critical_names, "target": action.get("target", "native"), "cache": "hit" if cache_hit else None, "build_id": self.build_id, "rc": rc, }, ) completed[action["name"]] = state self.last_results.append((action["name"], state)) if state == "failed": failed_actions.add(action["name"]) # Brief yield so the WS broadcast has time to fire await asyncio.sleep(0.05) return self.last_results # ------------------------------------------------- # INTERNAL # ------------------------------------------------- async def _wait_if_paused(self): if not self._paused: return # Wait until resumed (or stepped) if self._step_event is None: self._step_event = asyncio.Event() self._step_event.clear() while self._paused and not self._step_mode: await self._step_event.wait() self._step_event.clear() self._step_mode = False