diff --git a/railiance_fabric/coordination_graph.py b/railiance_fabric/coordination_graph.py index 1067183..e439916 100644 --- a/railiance_fabric/coordination_graph.py +++ b/railiance_fabric/coordination_graph.py @@ -66,6 +66,18 @@ def _indexed_depends_targets(workplan: dict[str, Any]) -> list[str]: return targets +TASK_NODE_SIZE = 26 +WORKPLAN_BASE_SIZE = 44 +WORKPLAN_SIZE_STEP = 10 +WORKPLAN_SIZE_CAP = 8 + + +def chokepoint_size(in_degree: int) -> int: + """Grow workplan nodes with the number of open items waiting on them.""" + + return WORKPLAN_BASE_SIZE + WORKPLAN_SIZE_STEP * min(max(in_degree, 0), WORKPLAN_SIZE_CAP) + + def coordination_graph_payload( workplans: list[dict[str, Any]], tasks: list[dict[str, Any]], @@ -153,6 +165,7 @@ def coordination_graph_payload( "needsHuman": human, "displayState": "highlight" if status == "wait" or human else "show", "unresolved": status == "wait" or human, + "visualSize": TASK_NODE_SIZE, } } ) @@ -254,11 +267,16 @@ def coordination_graph_payload( indexed_pairs.add((src_wp, dst)) for el in elements: - node_id = el["data"]["id"] + data = el["data"] + if data.get("kind") == "Edge": + data.setdefault("edgeWidth", 3 if data.get("edgeType") in {"depends_on", "waits_on"} else 1.5) + continue + node_id = data["id"] if not node_id.startswith("workplan:"): continue wp_id = node_id.split(":", 1)[1] - el["data"]["chokepoint"] = int(chokepoint.get(wp_id) or 0) + data["chokepoint"] = int(chokepoint.get(wp_id) or 0) + data["visualSize"] = chokepoint_size(data["chokepoint"]) generated = datetime.now(timezone.utc).isoformat() return { diff --git a/tests/test_coordination_graph.py b/tests/test_coordination_graph.py index 14ca63c..5a3f45d 100644 --- a/tests/test_coordination_graph.py +++ b/tests/test_coordination_graph.py @@ -158,6 +158,13 @@ def test_indexed_depends_on_outranks_citation_fallback() -> None: if el["data"]["id"] == "workplan:wp-b" ) assert chokepoint == 1 + sizes = { + el["data"]["id"]: el["data"]["visualSize"] + for el in payload["elements"] + if el["data"]["id"].startswith("workplan:") + } + assert sizes["workplan:wp-b"] > sizes["workplan:wp-a"] + assert all(e["edgeWidth"] > 0 for e in edges) def test_canonical_id_depends_on_resolves_to_hub_uuid() -> None: