28 lines
1.3 KiB
Python
28 lines
1.3 KiB
Python
"""Compile a role selection plan into a small dependency graph."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
def build_task_graph(selection_plan: dict[str, Any]) -> dict[str, Any]:
|
|
selected = selection_plan.get("selected") or {}
|
|
producers = []
|
|
for group in ("owner", "contributors"):
|
|
value = selected.get(group)
|
|
items = value if isinstance(value, list) else ([value] if value else [])
|
|
for item in items:
|
|
producers.append(item["role-id"] if isinstance(item, dict) else str(item))
|
|
reviewers = [item["role-id"] if isinstance(item, dict) else str(item)
|
|
for item in selected.get("reviewers", []) or []]
|
|
nodes = []
|
|
for role_id in producers:
|
|
nodes.append({"task-id": f"produce:{role_id}", "role-id": role_id, "depends-on": []})
|
|
producer_nodes = [node["task-id"] for node in nodes]
|
|
for role_id in reviewers:
|
|
nodes.append({"task-id": f"review:{role_id}", "role-id": role_id,
|
|
"depends-on": producer_nodes, "independent": True})
|
|
if len(nodes) > 1:
|
|
nodes.append({"task-id": "synthesize", "role-id": selection_plan.get("synthesis-role") or "OPS-ORCH",
|
|
"depends-on": [node["task-id"] for node in nodes], "reads": "projection-first"})
|
|
return {"workflow-id": selection_plan.get("workflow-id"), "nodes": nodes}
|