#!/usr/bin/env python3 """AgentFlow-Plan Execute — Tier-2 Gap G2-3. Universeller Plan-Runner fuer Aria. Liest Plan-JSON (Schema siehe 02-Wissen/aria-agentflow-plan-standard.md), validiert Schema, fuehrt Step-by-Step aus mit Core-Plan-Tools. Pattern aus BMW Business-Manual 2.3.5 + Workflow-Orchestrator-Agent-Prompt. Usage: python3 aria-agentflow-execute.py validate python3 aria-agentflow-execute.py run [--dry-run] Adopt-Items: B1 (Plan-JSON-Standard), B2 (Workflow-Orchestrator-Prompt) """ from __future__ import annotations import argparse import json import sys import time from pathlib import Path from datetime import datetime, timezone # Core-Plan-Tools (universell) _PLAN_DATA = {} _CURRENT_STEP = None _AUDIT_LOG = [] def core_read_plan() -> dict: """Plan-Overview ohne Step-Details.""" return { "plan_id": _PLAN_DATA.get("plan_id"), "title": _PLAN_DATA.get("title"), "process_goal": _PLAN_DATA.get("process_goal"), "termination_criterion": _PLAN_DATA.get("termination_criterion"), "self_recovery": _PLAN_DATA.get("self_recovery"), "step_count": len(_PLAN_DATA.get("steps", [])), "step_ids": [s["id"] for s in _PLAN_DATA.get("steps", [])], } def core_get_step(step_id: str) -> dict | None: for s in _PLAN_DATA.get("steps", []): if s["id"] == step_id: return s return None def core_get_self_recovery() -> str: return _PLAN_DATA.get("self_recovery", "no self_recovery defined") def core_wait_tool(seconds: float): s = min(max(float(seconds), 0.1), 60.0) time.sleep(s) return f"waited {s}s" # Tool-Registry: erlaubte Domain-Tools werden hier registriert _TOOL_REGISTRY = { "core.read_plan": core_read_plan, "core.get_step": core_get_step, "core.get_self_recovery": core_get_self_recovery, "core.wait_tool": core_wait_tool, } def register_tool(name: str, fn): _TOOL_REGISTRY[name] = fn def log_event(event: str, **kwargs): entry = {"ts": datetime.now(timezone.utc).isoformat(), "event": event, **kwargs} _AUDIT_LOG.append(entry) def validate(plan: dict) -> list[str]: """Return list of validation-errors. Empty list = valid.""" errors = [] required = ["plan_id", "title", "process_goal", "termination_criterion", "steps"] for k in required: if k not in plan: errors.append(f"missing top-level key: {k}") if "termination_criterion" in plan: tc = plan["termination_criterion"] if not isinstance(tc, dict): errors.append("termination_criterion must be object") else: if "check_tool" not in tc or "expected_state" not in tc: errors.append("termination_criterion needs check_tool + expected_state") if "steps" in plan: if not isinstance(plan["steps"], list) or not plan["steps"]: errors.append("steps must be non-empty array") else: step_ids = set() for i, s in enumerate(plan["steps"]): if not isinstance(s, dict): errors.append(f"step[{i}] not object") continue if "id" not in s: errors.append(f"step[{i}] missing id") else: if s["id"] in step_ids: errors.append(f"step[{i}] duplicate id: {s['id']}") step_ids.add(s["id"]) for k in ("title", "context", "actions", "expected_transitions"): if k not in s: errors.append(f"step '{s.get('id','?')}' missing: {k}") if "actions" in s and isinstance(s["actions"], list): for ai, action in enumerate(s["actions"]): if not isinstance(action, dict): errors.append(f"step '{s.get('id','?')}' action[{ai}] not object") continue for k in ("name", "tool", "description"): if k not in action: errors.append(f"step '{s.get('id','?')}' action[{ai}] missing: {k}") if "expected_transitions" in s and isinstance(s["expected_transitions"], list): for ti, trans in enumerate(s["expected_transitions"]): if not isinstance(trans, dict) or "on" not in trans or "to" not in trans: errors.append(f"step '{s.get('id','?')}' transition[{ti}] needs on+to") return errors def execute_action(action: dict, allowed_tools: set, dry_run: bool = False) -> dict: """Run a single action. Tool must be in allowed_tools whitelist.""" name = action.get("name", "unnamed") tool = action.get("tool") if isinstance(tool, list): tool = tool[0] # use first if multiple options listed if not tool: return {"status": "skipped", "reason": "no tool specified"} if tool not in allowed_tools: return {"status": "denied", "reason": f"tool '{tool}' not in step-whitelist"} if dry_run: return {"status": "dry-run", "tool": tool} fn = _TOOL_REGISTRY.get(tool) if not fn: return {"status": "missing", "reason": f"tool '{tool}' not registered"} args = action.get("args", {}) or {} try: if isinstance(args, dict): result = fn(**args) if args else fn() else: result = fn(*args) return {"status": "ok", "tool": tool, "result": str(result)[:500]} except TypeError: # Try no-args fallback for tools that don't accept args try: result = fn() return {"status": "ok", "tool": tool, "result": str(result)[:500]} except Exception as e: return {"status": "error", "tool": tool, "error": str(e)[:200]} except Exception as e: return {"status": "error", "tool": tool, "error": str(e)[:200]} def run_step(step: dict, dry_run: bool = False) -> dict: global _CURRENT_STEP _CURRENT_STEP = step["id"] log_event("step_start", step=step["id"]) print(f"[step] {step['id']}: {step['title']}") allowed_tools = set() for a in step.get("general_tools", []): allowed_tools.add(a if isinstance(a, str) else (a[0] if a else "")) for action in step.get("actions", []): t = action.get("tool") if isinstance(t, list): for x in t: allowed_tools.add(x) elif t: allowed_tools.add(t) # Always allow core tools allowed_tools |= {"core.read_plan", "core.get_step", "core.get_self_recovery", "core.wait_tool"} results = [] for action in step.get("actions", []): print(f" action: {action.get('name')} (tool: {action.get('tool')})") r = execute_action(action, allowed_tools, dry_run=dry_run) log_event("action", step=step["id"], action=action.get("name"), result=r) results.append(r) if r.get("status") == "error": return {"status": "step_error", "step": step["id"], "results": results} # Take first transition (default) transitions = step.get("expected_transitions", []) next_step = transitions[0]["to"] if transitions else None log_event("step_done", step=step["id"], next=next_step) return {"status": "ok", "step": step["id"], "next_step": next_step, "results": results} def run_plan(plan_path: Path, dry_run: bool = False): global _PLAN_DATA text = plan_path.read_text() plan = json.loads(text) _PLAN_DATA = plan errors = validate(plan) if errors: print("VALIDATION ERRORS:") for e in errors: print(f" - {e}") return 1 print(f"Plan: {plan['plan_id']} - {plan['title']}") print(f"Steps: {len(plan['steps'])}") if dry_run: print("\n=== DRY-RUN ===") for step in plan["steps"]: r = run_step(step, dry_run=dry_run) if r.get("status") != "ok": print(f"FAILED at step {r.get('step')}: {r}") return 2 print(f"\nDONE. Audit log entries: {len(_AUDIT_LOG)}") return 0 def main(): ap = argparse.ArgumentParser() sub = ap.add_subparsers(dest="cmd") p_val = sub.add_parser("validate") p_val.add_argument("plan_path") p_run = sub.add_parser("run") p_run.add_argument("plan_path") p_run.add_argument("--dry-run", action="store_true") args = ap.parse_args() if args.cmd == "validate": plan = json.loads(Path(args.plan_path).read_text()) errs = validate(plan) if errs: for e in errs: print(f"ERROR: {e}") return 1 print(f"OK: {plan['plan_id']} is valid ({len(plan.get('steps',[]))} steps)") elif args.cmd == "run": return run_plan(Path(args.plan_path), dry_run=args.dry_run) else: ap.print_help() return 1 if __name__ == "__main__": sys.exit(main() or 0)