# -*- coding: utf-8 -*- """ EXECUTION ENGINE — TaskExecutor ================================ Fuehrt Tasks TATSAECHLICH aus (nicht nur planen). Task Queue (JSON) -> Executor -> Action -> Audit -> Next Task Unterstützte Actions: - send_email: SMTP-E-Mail tatsächlich versenden - export_ffg_form: Formular via FormFiller exportieren - query_esa_api: ESA openEO API tatsächlich abfragen - query_cern_api: CERN OpenData API tatsächlich abfragen - run_guardian_check: Guardian Risk-Check tatsächlich ausführen - write_file: Datei tatsächlich schreiben - run_script: Python-Skript tatsächlich ausführen Jede Aktion erzeugt: - SHA-256 Audit-Eintrag - EpistemicState: known (erledigt) vs estimated (geplant) """ from __future__ import annotations import hashlib import json import logging import os import smtplib import subprocess import sys import time import traceback from dataclasses import dataclass, field from datetime import datetime, timezone from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from enum import Enum from pathlib import Path from typing import Any, Dict, List, Optional # Projekt-Root ins Path ROOT = Path(__file__).resolve().parent.parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from autonomous.audit_chain import AuditChain from autonomous.epistemic_state import EpistemicState, EpistemicStatus logger = logging.getLogger("execution_engine") # ─── ENUMS ──────────────────────────────────────────────────────────────────── class TaskStatus(Enum): PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" SKIPPED = "skipped" class EpistemicActionState(Enum): KNOWN = "known" # Tatsaechlich erledigt, verifiziert ESTIMATED = "estimated" # Geplant, noch nicht ausgefuehrt # ─── DATA CLASSES ───────────────────────────────────────────────────────────── @dataclass class TaskResult: """Ergebnis einer Task-Ausführung.""" task_id: str task_type: str status: TaskStatus output: Dict[str, Any] = field(default_factory=dict) error: str = "" sha256: str = "" epistemic_state: str = EpistemicActionState.ESTIMATED.value started_at: str = "" completed_at: str = "" duration_ms: float = 0.0 def to_dict(self) -> dict: return { "task_id": self.task_id, "task_type": self.task_type, "status": self.status.value, "output": self.output, "error": self.error, "sha256": self.sha256, "epistemic_state": self.epistemic_state, "started_at": self.started_at, "completed_at": self.completed_at, "duration_ms": self.duration_ms, } # ─── ACTION HANDLERS ────────────────────────────────────────────────────────── class ActionHandler: """ Basisklasse fuer konkrete Action-Handler. Jeder Handler fuehrt eine tatsaechliche Aktion aus. """ def __init__(self, audit_chain: AuditChain, epistemic: EpistemicState): self.audit = audit_chain self.epistemic = epistemic def execute(self, params: Dict[str, Any]) -> TaskResult: raise NotImplementedError class SendEmailHandler(ActionHandler): """Versendet E-Mails tatsächlich via SMTP.""" ACTION_TYPE = "send_email" def execute(self, params: Dict[str, Any]) -> TaskResult: to_addr = params.get("to", "") subject = params.get("subject", "") body = params.get("body", "") html = params.get("html", False) if not to_addr or not subject: return TaskResult( task_id=params.get("_task_id", "email"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error="Missing 'to' or 'subject' parameter", ) smtp_host = os.getenv("SMTP_HOST", "smtp.gmail.com") smtp_port = int(os.getenv("SMTP_PORT", "587")) smtp_user = os.getenv("SMTP_USER", "") smtp_pass = os.getenv("SMTP_PASS", "") if not smtp_user or not smtp_pass: return TaskResult( task_id=params.get("_task_id", "email"), task_type=self.ACTION_TYPE, status=TaskStatus.SKIPPED, error="SMTP credentials not configured (SMTP_USER/SMTP_PASS missing in .env)", output={"to": to_addr, "subject": subject}, ) start = time.monotonic() try: msg = MIMEMultipart("alternative") msg["From"] = smtp_user msg["To"] = to_addr msg["Subject"] = subject if html: msg.attach(MIMEText(body, "html", "utf-8")) else: msg.attach(MIMEText(body, "plain", "utf-8")) with smtplib.SMTP(smtp_host, smtp_port, timeout=30) as server: server.starttls() server.login(smtp_user, smtp_pass) server.sendmail(smtp_user, [to_addr], msg.as_string()) duration = (time.monotonic() - start) * 1000 result_sha = hashlib.sha256( json.dumps({"to": to_addr, "subject": subject}, sort_keys=True).encode() ).hexdigest() self.audit.record("email_sent", {"to": to_addr, "subject": subject, "sha256": result_sha}) self.epistemic.add_belief( f"Email sent to {to_addr}: {subject}", EpistemicStatus.CERTAIN, 0.95, source="execution_engine", ttl_hours=720, ) return TaskResult( task_id=params.get("_task_id", "email"), task_type=self.ACTION_TYPE, status=TaskStatus.COMPLETED, output={"to": to_addr, "subject": subject, "sent_at": datetime.now(timezone.utc).isoformat()}, sha256=result_sha, epistemic_state=EpistemicActionState.KNOWN.value, duration_ms=duration, ) except Exception as exc: duration = (time.monotonic() - start) * 1000 self.audit.record("email_failed", {"to": to_addr, "error": str(exc)}) return TaskResult( task_id=params.get("_task_id", "email"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error=str(exc), output={"to": to_addr, "subject": subject}, duration_ms=duration, ) class ExportFFGFormHandler(ActionHandler): """Exportiert FFG-Formulare tatsächlich via FormFiller.""" ACTION_TYPE = "export_ffg_form" def execute(self, params: Dict[str, Any]) -> TaskResult: from space.form_filler import FormFiller application_path = params.get("application", "") output_path = params.get("output", "") portal = params.get("portal", "ffg_ecall") if not application_path: return TaskResult( task_id=params.get("_task_id", "ffg_form"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error="Missing 'application' parameter (path to markdown file)", ) start = time.monotonic() try: filler = FormFiller() app_data = filler.load_markdown_application(application_path) if output_path: out = Path(output_path) out.parent.mkdir(parents=True, exist_ok=True) if portal == "ffg_ecall": xml_str = filler.export_ffg_ecall(app_data) out.write_text(xml_str, encoding="utf-8") output_info = {"format": "ffg_ecall_xml", "output_file": str(out), "size_bytes": out.stat().st_size} elif portal == "esa_sysnova": esa_dict = filler.export_esa_sysnova(app_data) json_str = json.dumps(esa_dict, indent=2, ensure_ascii=False) out.write_text(json_str, encoding="utf-8") output_info = {"format": "esa_sysnova_json", "output_file": str(out), "size_bytes": out.stat().st_size} elif portal == "eu_funding": json_ld = filler.export_eu_funding(app_data) out.write_text(json_ld, encoding="utf-8") output_info = {"format": "eu_funding_jsonld", "output_file": str(out), "size_bytes": out.stat().st_size} else: return TaskResult( task_id=params.get("_task_id", "ffg_form"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error=f"Unknown portal: {portal}", ) else: # Kein Output-Pfad — nur validieren errors = filler.validate_form(app_data, portal) output_info = {"format": portal, "validation_errors": len(errors), "errors": errors} duration = (time.monotonic() - start) * 1000 result_sha = hashlib.sha256( json.dumps(app_data["project"], sort_keys=True, ensure_ascii=False).encode() ).hexdigest() self.audit.record("form_exported", {"application": application_path, "output": output_path or "validation_only", "portal": portal, "sha256": result_sha}) self.epistemic.add_belief( f"Form exported: {application_path} -> {output_path or 'validation'} ({portal})", EpistemicStatus.CERTAIN, 0.9, source="execution_engine", ttl_hours=720, ) return TaskResult( task_id=params.get("_task_id", "ffg_form"), task_type=self.ACTION_TYPE, status=TaskStatus.COMPLETED, output=output_info, sha256=result_sha, epistemic_state=EpistemicActionState.KNOWN.value, duration_ms=duration, ) except Exception as exc: duration = (time.monotonic() - start) * 1000 self.audit.record("form_export_failed", {"application": application_path, "error": str(exc)}) return TaskResult( task_id=params.get("_task_id", "ffg_form"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error=f"{exc}\n{traceback.format_exc()}", duration_ms=duration, ) class QueryESAAPIHandler(ActionHandler): """Führt ESA API Query tatsächlich aus.""" ACTION_TYPE = "query_esa_api" def execute(self, params: Dict[str, Any]) -> TaskResult: backend = params.get("backend", "openeo-dev.vito.be") save_path = params.get("save", "") start = time.monotonic() try: # HTTP-Anfrage ohne externe Abhaengigkeiten from urllib.request import Request, urlopen import ssl # SSL-Kontext: Erst mit verify, bei Fallback ohne ssl_context = ssl.create_default_context() use_ssl_verify = True url = f"https://{backend}/openeo/1.2" req = Request(url, headers={"User-Agent": "ORION-DDGK-ExecutionEngine/1.0"}) try: with urlopen(req, timeout=20, context=ssl_context) as resp: backend_data = json.loads(resp.read()) except Exception as ssl_exc: if "CERTIFICATE_VERIFY_FAILED" in str(ssl_exc): logger.warning("SSL verify failed for %s, retrying without verification", backend) ssl_context = ssl._create_unverified_context() use_ssl_verify = False with urlopen(req, timeout=20, context=ssl_context) as resp: backend_data = json.loads(resp.read()) else: raise # Collections abfragen coll_url = f"https://{backend}/openeo/1.2/collections" coll_req = Request(coll_url, headers={"User-Agent": "ORION-DDGK-ExecutionEngine/1.0"}) with urlopen(coll_req, timeout=20, context=ssl_context) as resp: coll_data = json.loads(resp.read()) collections = coll_data.get("collections", []) result_data = { "backend": backend, "backend_info": backend_data, "collections_count": len(collections), "collection_ids": [c.get("id", "") for c in collections[:20]], "ssl_verified": use_ssl_verify, "queried_at": datetime.now(timezone.utc).isoformat(), } if save_path: out = Path(save_path) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(result_data, indent=2, ensure_ascii=False), encoding="utf-8") result_data["saved_to"] = str(out) duration = (time.monotonic() - start) * 1000 result_sha = hashlib.sha256(json.dumps(result_data, sort_keys=True).encode()).hexdigest() self.audit.record("esa_api_queried", {"backend": backend, "collections": len(collections), "sha256": result_sha}) self.epistemic.add_belief( f"ESA API queried: {backend} — {len(collections)} collections", EpistemicStatus.CERTAIN, 0.9, source="execution_engine", ttl_hours=24, ) return TaskResult( task_id=params.get("_task_id", "esa_api"), task_type=self.ACTION_TYPE, status=TaskStatus.COMPLETED, output=result_data, sha256=result_sha, epistemic_state=EpistemicActionState.KNOWN.value, duration_ms=duration, ) except Exception as exc: duration = (time.monotonic() - start) * 1000 self.audit.record("esa_api_failed", {"backend": backend, "error": str(exc)}) return TaskResult( task_id=params.get("_task_id", "esa_api"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error=str(exc), output={"backend": backend, "queried_at": datetime.now(timezone.utc).isoformat()}, duration_ms=duration, ) class QueryCERNAPIHandler(ActionHandler): """Führt CERN API Query tatsächlich aus.""" ACTION_TYPE = "query_cern_api" def execute(self, params: Dict[str, Any]) -> TaskResult: save_path = params.get("save", "") query = params.get("query", "") size = params.get("size", 10) start = time.monotonic() try: from urllib.request import Request, urlopen url = f"https://opendata.cern.ch/api/records/?size={size}" if query: url += f"&q={query}" req = Request(url, headers={"User-Agent": "ORION-DDGK-ExecutionEngine/1.0"}) with urlopen(req, timeout=20) as resp: data = json.loads(resp.read()) hits = data.get("hits", {}) records = hits.get("hits", []) total = hits.get("total", 0) record_summaries = [] for rec in records[:size]: md = rec.get("metadata", {}) record_summaries.append({ "id": rec.get("id", ""), "title": md.get("title", "Untitled"), "collection": md.get("collection", md.get("collections", "")), "experiment": md.get("experiment", ""), }) result_data = { "query": query or "all", "records_returned": len(records), "records_total": total, "records": record_summaries, "queried_at": datetime.now(timezone.utc).isoformat(), } if save_path: out = Path(save_path) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(result_data, indent=2, ensure_ascii=False), encoding="utf-8") result_data["saved_to"] = str(out) duration = (time.monotonic() - start) * 1000 result_sha = hashlib.sha256(json.dumps(result_data, sort_keys=True).encode()).hexdigest() self.audit.record("cern_api_queried", {"query": query, "records": len(records), "sha256": result_sha}) self.epistemic.add_belief( f"CERN API queried: {len(records)} records (total: {total})", EpistemicStatus.CERTAIN, 0.9, source="execution_engine", ttl_hours=24, ) return TaskResult( task_id=params.get("_task_id", "cern_api"), task_type=self.ACTION_TYPE, status=TaskStatus.COMPLETED, output=result_data, sha256=result_sha, epistemic_state=EpistemicActionState.KNOWN.value, duration_ms=duration, ) except Exception as exc: duration = (time.monotonic() - start) * 1000 self.audit.record("cern_api_failed", {"query": query, "error": str(exc)}) return TaskResult( task_id=params.get("_task_id", "cern_api"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error=str(exc), output={"query": query, "queried_at": datetime.now(timezone.utc).isoformat()}, duration_ms=duration, ) class RunGuardianCheckHandler(ActionHandler): """Führt Guardian Risk-Check tatsächlich aus.""" ACTION_TYPE = "run_guardian_check" def execute(self, params: Dict[str, Any]) -> TaskResult: from orion_guardian_v2 import ORIONGuardianV2, AssessmentContext target = params.get("target", "all_forms") check_actions = params.get("actions", []) start = time.monotonic() guardian = ORIONGuardianV2() # Wenn keine konkreten Aktionen gegeben, Standard-Checks if not check_actions: if target == "all_forms": check_actions = [ {"action": "export_ffg_ecall", "tool": "FormFiller", "args": {"portal": "ffg_ecall"}}, {"action": "export_esa_sysnova", "tool": "FormFiller", "args": {"portal": "esa_sysnova"}}, {"action": "export_eu_funding", "tool": "FormFiller", "args": {"portal": "eu_funding"}}, ] elif target == "api_calls": check_actions = [ {"action": "query_esa_api", "tool": "ESAAdapter", "args": {"backend": "openeo-dev.vito.be"}}, {"action": "query_cern_api", "tool": "CERNAdapter", "args": {}}, ] elif target == "email": check_actions = [ {"action": "send_email_via_smtp", "tool": "SMTP", "args": {"host": os.getenv("SMTP_HOST", "")}}, ] else: check_actions = [ {"action": f"guardian_check_target_{target}", "tool": "generic", "args": {}}, ] assessments = [] overall_risk = 0 for ca in check_actions: ctx = AssessmentContext( action=ca.get("action", ""), tool_name=ca.get("tool", ""), tool_args=ca.get("args", {}), user_approved=params.get("user_approved", False), ) assessment = guardian.assess(ctx) assessments.append({ "action": ca.get("action", ""), "score": assessment.score, "level": assessment.level, "decision": assessment.decision, "reasons": assessment.reasons, "chain_hash": assessment.chain_hash, }) overall_risk = max(overall_risk, assessment.score) duration = (time.monotonic() - start) * 1000 result_sha = hashlib.sha256( json.dumps(assessments, sort_keys=True).encode() ).hexdigest() self.audit.record("guardian_check", {"target": target, "overall_risk": overall_risk, "checks": len(assessments), "sha256": result_sha}) self.epistemic.add_belief( f"Guardian check completed: target={target}, risk={overall_risk}, checks={len(assessments)}", EpistemicStatus.CERTAIN, 0.95, source="execution_engine", ttl_hours=168, ) return TaskResult( task_id=params.get("_task_id", "guardian"), task_type=self.ACTION_TYPE, status=TaskStatus.COMPLETED, output={ "target": target, "overall_risk_score": overall_risk, "assessments": assessments, "checks_run": len(assessments), }, sha256=result_sha, epistemic_state=EpistemicActionState.KNOWN.value, duration_ms=duration, ) class WriteFileHandler(ActionHandler): """Schreibt Dateien tatsächlich.""" ACTION_TYPE = "write_file" def execute(self, params: Dict[str, Any]) -> TaskResult: file_path = params.get("path", "") content = params.get("content", "") encoding = params.get("encoding", "utf-8") mode = params.get("mode", "write") # write | append if not file_path: return TaskResult( task_id=params.get("_task_id", "write_file"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error="Missing 'path' parameter", ) start = time.monotonic() try: out = Path(file_path) out.parent.mkdir(parents=True, exist_ok=True) if mode == "append" and out.exists(): out.write_text(out.read_text(encoding=encoding) + content, encoding=encoding) else: out.write_text(content, encoding=encoding) duration = (time.monotonic() - start) * 1000 file_sha = hashlib.sha256(content.encode(encoding)).hexdigest() self.audit.record("file_written", {"path": str(out), "size": out.stat().st_size, "sha256": file_sha}) self.epistemic.add_belief( f"File written: {file_path} ({out.stat().st_size} bytes)", EpistemicStatus.CERTAIN, 0.95, source="execution_engine", ) return TaskResult( task_id=params.get("_task_id", "write_file"), task_type=self.ACTION_TYPE, status=TaskStatus.COMPLETED, output={"path": str(out), "size_bytes": out.stat().st_size, "mode": mode}, sha256=file_sha, epistemic_state=EpistemicActionState.KNOWN.value, duration_ms=duration, ) except Exception as exc: duration = (time.monotonic() - start) * 1000 self.audit.record("file_write_failed", {"path": file_path, "error": str(exc)}) return TaskResult( task_id=params.get("_task_id", "write_file"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error=str(exc), duration_ms=duration, ) class RunScriptHandler(ActionHandler): """Führt Python-Skripte tatsächlich aus.""" ACTION_TYPE = "run_script" def execute(self, params: Dict[str, Any]) -> TaskResult: script = params.get("script", "") args = params.get("args", []) cwd = params.get("cwd", str(ROOT)) capture_output = params.get("capture_output", True) timeout = params.get("timeout", 300) if not script: return TaskResult( task_id=params.get("_task_id", "run_script"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error="Missing 'script' parameter", ) start = time.monotonic() try: cmd = [sys.executable, script] + args result = subprocess.run( cmd, cwd=cwd, capture_output=capture_output, text=True, timeout=timeout, ) duration = (time.monotonic() - start) * 1000 result_sha = hashlib.sha256( json.dumps({"script": script, "returncode": result.returncode}, sort_keys=True).encode() ).hexdigest() output_data = { "script": script, "returncode": result.returncode, "stdout": result.stdout[-5000:] if capture_output and result.stdout else "", "stderr": result.stderr[-5000:] if capture_output and result.stderr else "", } status = TaskStatus.COMPLETED if result.returncode == 0 else TaskStatus.FAILED self.audit.record( "script_executed", {"script": script, "returncode": result.returncode, "sha256": result_sha}, ) self.epistemic.add_belief( f"Script executed: {script} (rc={result.returncode})", EpistemicStatus.CERTAIN if result.returncode == 0 else EpistemicStatus.LIKELY, 0.9 if result.returncode == 0 else 0.5, source="execution_engine", ) return TaskResult( task_id=params.get("_task_id", "run_script"), task_type=self.ACTION_TYPE, status=status, output=output_data, sha256=result_sha, epistemic_state=EpistemicActionState.KNOWN.value, duration_ms=duration, ) except subprocess.TimeoutExpired as exc: duration = (time.monotonic() - start) * 1000 self.audit.record("script_timeout", {"script": script, "timeout": timeout}) return TaskResult( task_id=params.get("_task_id", "run_script"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error=f"Timeout after {timeout}s: {exc}", duration_ms=duration, ) except Exception as exc: duration = (time.monotonic() - start) * 1000 self.audit.record("script_failed", {"script": script, "error": str(exc)}) return TaskResult( task_id=params.get("_task_id", "run_script"), task_type=self.ACTION_TYPE, status=TaskStatus.FAILED, error=str(exc), duration_ms=duration, ) # ─── TASK EXECUTOR ──────────────────────────────────────────────────────────── HANDLER_REGISTRY: Dict[str, type] = { SendEmailHandler.ACTION_TYPE: SendEmailHandler, ExportFFGFormHandler.ACTION_TYPE: ExportFFGFormHandler, QueryESAAPIHandler.ACTION_TYPE: QueryESAAPIHandler, QueryCERNAPIHandler.ACTION_TYPE: QueryCERNAPIHandler, RunGuardianCheckHandler.ACTION_TYPE: RunGuardianCheckHandler, WriteFileHandler.ACTION_TYPE: WriteFileHandler, RunScriptHandler.ACTION_TYPE: RunScriptHandler, } class TaskExecutor: """ Haupt-Executor: Liest Task-Queue, fuehrt Tasks aus, schreibt Audit. """ def __init__( self, queue_path: Optional[str | Path] = None, audit_chain: Optional[AuditChain] = None, epistemic: Optional[EpistemicState] = None, agent_id: int = 1, agent_name: str = "TaskExecutor", ): self.queue_path = Path(queue_path) if queue_path else ROOT / "autonomous" / "task_queue.json" self.agent_id = agent_id self.agent_name = agent_name self.audit = audit_chain or AuditChain(agent_id=agent_id, agent_name=agent_name) self.epistemic = epistemic or EpistemicState(agent_id=agent_id, agent_name=agent_name) self.handlers: Dict[str, ActionHandler] = {} self.results: List[TaskResult] = [] # Handler registrieren for action_type, handler_cls in HANDLER_REGISTRY.items(): self.handlers[action_type] = handler_cls(self.audit, self.epistemic) # ── Queue I/O ───────────────────────────────────────────────────────── def load_queue(self) -> List[Dict[str, Any]]: """Laedt die Task-Queue aus JSON.""" if not self.queue_path.exists(): raise FileNotFoundError(f"Task queue not found: {self.queue_path}") data = json.loads(self.queue_path.read_text(encoding="utf-8")) if isinstance(data, dict): tasks = data.get("tasks", []) elif isinstance(data, list): tasks = data else: raise ValueError(f"Invalid queue format: expected list or dict with 'tasks'") return tasks def save_queue(self, tasks: List[Dict[str, Any]]) -> None: """Speichert die aktualisierte Task-Queue.""" queue_data = { "version": "1.0.0", "updated_at": datetime.now(timezone.utc).isoformat(), "total_tasks": len(tasks), "tasks": tasks, } self.queue_path.write_text( json.dumps(queue_data, indent=2, ensure_ascii=False), encoding="utf-8" ) # ── Execution ───────────────────────────────────────────────────────── def execute_task(self, task: Dict[str, Any]) -> TaskResult: """Fuehrt einen einzelnen Task aus.""" task_type = task.get("type", task.get("action", "")) params = task.get("params", task.get("args", task.get("config", {}))) task_id = task.get("id", task.get("task_id", f"task_{int(time.time())}")) # Task-ID in Params speichern params["_task_id"] = task_id ts_start = datetime.now(timezone.utc).isoformat() # Status auf RUNNING setzen task["status"] = TaskStatus.RUNNING.value task["started_at"] = ts_start handler = self.handlers.get(task_type) if not handler: result = TaskResult( task_id=task_id, task_type=task_type, status=TaskStatus.FAILED, error=f"Unknown task type: {task_type}. Available: {list(self.handlers.keys())}", started_at=ts_start, completed_at=datetime.now(timezone.utc).isoformat(), ) else: result = handler.execute(params) result.started_at = ts_start result.completed_at = datetime.now(timezone.utc).isoformat() # Task in Queue aktualisieren task["status"] = result.status.value task["result"] = result.to_dict() task["completed_at"] = result.completed_at self.results.append(result) # Audit self.audit.record( f"task_{result.status.value}", {"task_id": task_id, "type": task_type, "sha256": result.sha256}, ) return result def execute_all(self, stop_on_error: bool = False) -> List[TaskResult]: """Fuehrt alle Tasks in der Queue aus.""" tasks = self.load_queue() # Sicherstellen dass jeder Task einen ID hat for i, task in enumerate(tasks): if "id" not in task and "task_id" not in task: task["id"] = f"task_{i:03d}_{task.get('type', task.get('action', 'unknown'))}" if "status" not in task: task["status"] = TaskStatus.PENDING.value print(f"\n{'='*60}") print(f" EXECUTION PIPELINE START") print(f" Tasks: {len(tasks)}") print(f" Queue: {self.queue_path}") print(f"{'='*60}\n") for i, task in enumerate(tasks): task_id = task.get("id", task.get("task_id", "?")) task_type = task.get("type", task.get("action", "?")) status = task.get("status", TaskStatus.PENDING.value) # Ueberspringe bereits erledigte Tasks if status in (TaskStatus.COMPLETED.value, TaskStatus.SKIPPED.value): print(f" [{i+1}/{len(tasks)}] SKIP {task_id} ({task_type}) — already {status}") continue print(f" [{i+1}/{len(tasks)}] EXEC {task_id} ({task_type}) ...", end=" ", flush=True) result = self.execute_task(task) icon = { TaskStatus.COMPLETED: "OK", TaskStatus.FAILED: "FAIL", TaskStatus.SKIPPED: "SKIP", TaskStatus.RUNNING: "RUN", TaskStatus.PENDING: "PEND", }.get(result.status, "??") print(f"[{icon}] {result.duration_ms:.0f}ms") if result.error: print(f" ERROR: {result.error[:120]}") if result.sha256: print(f" SHA-256: {result.sha256[:32]}...") if result.status == TaskStatus.FAILED and stop_on_error: print(f"\n STOP on error at task {task_id}") break # Queue aktualisieren self.save_queue(tasks) # Zusammenfassung completed = sum(1 for r in self.results if r.status == TaskStatus.COMPLETED) failed = sum(1 for r in self.results if r.status == TaskStatus.FAILED) skipped = sum(1 for r in self.results if r.status == TaskStatus.SKIPPED) print(f"\n{'='*60}") print(f" EXECUTION PIPELINE END") print(f" Completed: {completed} | Failed: {failed} | Skipped: {skipped}") print(f" Audit entries: {self.audit.count()}") print(f" Epistemic beliefs: {len(self.epistemic.beliefs)}") print(f"{'='*60}\n") # Epistemic State speichern epistemic_path = ROOT / "autonomous" / "epistemic_executor.json" self.epistemic.save(epistemic_path) return self.results def execute_single(self, task_id: str) -> TaskResult: """Fuehrt einen spezifischen Task aus der Queue aus.""" tasks = self.load_queue() for task in tasks: tid = task.get("id", task.get("task_id", "")) if tid == task_id: return self.execute_task(task) raise ValueError(f"Task not found: {task_id}") def get_report(self) -> Dict[str, Any]: """Erstellt Ausfuehrungsbericht.""" return { "executed_at": datetime.now(timezone.utc).isoformat(), "total_tasks": len(self.results), "completed": sum(1 for r in self.results if r.status == TaskStatus.COMPLETED), "failed": sum(1 for r in self.results if r.status == TaskStatus.FAILED), "skipped": sum(1 for r in self.results if r.status == TaskStatus.SKIPPED), "results": [r.to_dict() for r in self.results], "audit_entries": self.audit.count(), "epistemic_beliefs": len(self.epistemic.beliefs), "audit_integrity": self.audit.verify_chain()[0], }