""" 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 6. On success with BTC enabled: verify/apply forensic stamps, emit btc_stamp Emits events on the bus so the UI / WS stream / cause graph / timeline store all pick them up. """ import asyncio import os 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 # BTC integration (best-effort -- module may not be present in minimal installs) try: from backend.toolchain.btc import probe_btc, verify_btc_stamp, apply_btc_stamp HAS_BTC_INTEGRATION = True except Exception: HAS_BTC_INTEGRATION = False probe_btc = None verify_btc_stamp = None apply_btc_stamp = None # Shared CAS integration — provides DAG-aware content-addressable caching # across sorcery-go and Fester. Eliminates redundant rebuilds when the # same artifact (by SHA-256) was already produced by any node/runtime. try: from backend.storage.cas_api import STORE as CAS_STORE HAS_CAS = True except Exception: HAS_CAS = False CAS_STORE = None def _find_output_binary(action: Dict[str, Any]) -> Optional[str]: """Return the first output path from *action* that looks like an ELF binary. Checks the action's ``outputs`` list and the ``dir`` working directory. Returns ``None`` if no plausible binary is found. """ outputs = action.get("outputs") or [] work_dir = action.get("dir") or "/tmp" for out in outputs: if not isinstance(out, str): continue # Make relative paths absolute against the work directory if not os.path.isabs(out): path = os.path.join(work_dir, out) else: path = out if os.path.isfile(path): # Quick ELF magic check try: with open(path, "rb") as f: magic = f.read(4) if magic == b"\x7fELF": return path except OSError: continue return None 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 (MinIO) self.cas = CAS_STORE # shared CAS (always available if module loaded) 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 # BTC config cache (read once per engine instance) self._btc_enabled = False self._btc_probe: Dict[str, Any] = {} self._init_btc_config() def _init_btc_config(self): """Read BTC configuration and probe the local node once.""" try: from backend.config import CONFIG btc_cfg = CONFIG.get("btc", {}) self._btc_enabled = btc_cfg.get("enabled", True) if self._btc_enabled and probe_btc is not None: btc_root = btc_cfg.get("root", "/opt/BTC") self._btc_probe = probe_btc(btc_root) except Exception: self._btc_enabled = False # ------------------------------------------------- # DEBUGGER HOOKS # ------------------------------------------------- def pause(self) -> None: self._paused = True def resume(self) -> None: self._paused = False if self._step_event: self._step_event.set() def step(self) -> None: """Advance one action when paused.""" self._step_mode = True if self._step_event: self._step_event.set() # ------------------------------------------------- # MAIN ENTRYPOINT # ------------------------------------------------- async def run(self, project: Dict[str, Any]) -> 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: try shared CAS first, then MinIO cache. # The CAS is DAG-aware — if sorcery-go already built this # target (any runtime, any node), we skip it entirely. cache_hit = False action_hash = action.get("hash", "") # 1. Check shared CAS (cross-runtime, cross-node) if HAS_CAS and self.cas and action_hash: try: cas_meta = self.cas.check_action(action_hash) if cas_meta is not None: cache_hit = True self.bus.emit( EventType.CACHE_UPDATE, action=action["name"], node=node_name, state="hit", meta={ "build_id": self.build_id, "cache_source": "cas", "original_node": cas_meta.get("node"), "original_runtime": cas_meta.get("runtime"), }, ) except Exception: pass # CAS errors are non-fatal # 2. Fall back to MinIO cache if not cache_hit and self.cache and action_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, "cache_source": "minio"}, ) except Exception: pass # Cache errors are non-fatal # Emit running # Capture temp_before for the feedback learning loop (F-04). temp_before = self._node_temp(node_name) action_started = time.time() 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, }, ) # --- Feedback learning loop (F-04) ---------------------- # Feed execution outcome + thermal delta into the policy # engine so the scheduler can learn which nodes succeed. if not cache_hit and node_name: temp_after = self._node_temp(node_name) duration = time.time() - action_started try: from backend.pipeline.feedback import report_execution report_execution( node=node_name, action=action["name"], success=(state == "done"), duration=duration, temp_before=temp_before, temp_after=temp_after, ) except Exception: # Feedback must never break the pipeline. pass # --- BTC forensic stamp (post-build, best-effort) ---------- if (state == "done" and not cache_hit and self._btc_enabled and HAS_BTC_INTEGRATION): await self._handle_btc_stamp(action, node_name) 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 # ------------------------------------------------- # BTC STAMP HANDLING # ------------------------------------------------- async def _handle_btc_stamp( self, action: Dict[str, Any], node_name: Optional[str], ): """After a successful build, verify or apply BTC forensic stamps. Runs in the executor thread pool so filesystem calls do not block the event loop. Failures are never propagated — they are logged and emitted as BTC_STAMP events with ``state="stamp_failed"``. """ binary_path = await asyncio.to_thread(_find_output_binary, action) if binary_path is None: return def _do_stamp(): stamp_info = verify_btc_stamp(binary_path) if stamp_info["has_note"] or stamp_info["has_xattr"]: # Binary already carries BTC stamps -- report them return { "state": "verified", "binary": binary_path, "action": action["name"], **stamp_info, } # No stamps found -- try to apply them sys_label = self._btc_probe.get("sys_label", "DCOSNET-UNKNOWN-UNKNOWN-UNKNOWN") arch = self._btc_probe.get("arch", "native") apply_result = apply_btc_stamp( binary_path=binary_path, forge_step=action["name"], sys_label=sys_label, arch=arch, ) if apply_result["success"]: # Re-verify after stamping stamp_info = verify_btc_stamp(binary_path) return { "state": "stamped", "binary": binary_path, "action": action["name"], "stripped": apply_result.get("stripped", False), **stamp_info, } else: return { "state": "stamp_failed", "binary": binary_path, "action": action["name"], "error": apply_result.get("error", "unknown"), "has_note": False, "has_xattr": False, "valid": False, } try: stamp_meta = await asyncio.to_thread(_do_stamp) self.bus.emit( EventType.BTC_STAMP, action=action["name"], node=node_name, state=stamp_meta.get("state", "unknown"), meta={ "build_id": self.build_id, **stamp_meta, }, ) except Exception: pass # BTC stamping must never break the pipeline # ------------------------------------------------- # INTERNAL # ------------------------------------------------- def _node_temp(self, node_name: Optional[str]) -> float: """Best-effort temperature read from the node registry. Returns 0.0 if the registry is unavailable or the node is unknown. Used by the feedback loop to compute thermal spikes. """ if not node_name or not self.node_registry: return 0.0 try: state = self.node_registry.get(node_name) if state is not None: return float(getattr(state, "temp", 0.0) or 0.0) except Exception: pass return 0.0 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