enabled multipath for TE_UCMP policy, improved CI script

Author: networmix

.github/workflows/python-test.yml

@@ -20,16 +20,6 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           python -m pip install '.[dev]'
-      - name: Lint with Ruff
+      - name: CI checks (lint + validate + tests)
         run: |
-          ruff check .
-          ruff format --check .
-      - name: Type check with Pyright
-        run: |
-          pyright
-      - name: Validate YAML schemas
-        run: |
-          make validate
-      - name: Test with pytest and check coverage
-        run: |
-          pytest
+          make check-ci

Makefile

@@ -15,7 +15,8 @@ help:
 	@echo ""
 	@echo "Code Quality & Testing:"
 	@echo "  make check         - Run all pre-commit checks and tests (includes slow and benchmark)"
-	@echo "  make lint          - Run only linting (ruff + pyright)"
+	@echo "  make check-ci      - Run non-mutating checks and tests (CI entrypoint)"
+	@echo "  make lint          - Run only linting (non-mutating: ruff + pyright)"
 	@echo "  make format        - Auto-format code with ruff"
 	@echo "  make test          - Run tests with coverage (includes slow and benchmark)"
 	@echo "  make qt            - Run quick tests only (excludes slow and benchmark)"
@@ -56,10 +57,17 @@ check:
 	@echo "🔍 Running complete code quality checks and tests..."
 	@bash dev/run-checks.sh
 
+check-ci:
+	@echo "🔍 Running CI checks (non-mutating lint + schema validation + tests)..."
+	@$(MAKE) lint
+	@$(MAKE) validate
+	@$(MAKE) test
+
 lint:
-	@echo "🧹 Running linting checks..."
-	@pre-commit run ruff --all-files
-	@pre-commit run pyright --all-files
+	@echo "🧹 Running linting checks (non-mutating)..."
+	@ruff format --check .
+	@ruff check .
+	@pyright
 
 format:
 	@echo "✨ Auto-formatting code..."

dev/run-checks.sh

@@ -23,17 +23,31 @@ if [ ! -f .git/hooks/pre-commit ]; then
     echo ""
 fi
 
-# Run pre-commit checks
-echo "🏃 Running pre-commit on all files..."
+# Run pre-commit with fixers (first pass), do not fail if files were modified
+echo "🏃 Running pre-commit (first pass: apply auto-fixes if needed)..."
+set +e
 pre-commit run --all-files
+first_pass_status=$?
+set -e
 
-if [ $? -ne 0 ]; then
+if [ $first_pass_status -ne 0 ]; then
+    echo "ℹ️  Some hooks modified files or reported issues. Re-running checks..."
+fi
+
+# Re-run to verify all checks pass after fixes; fail on any remaining issues
+echo "🏃 Running pre-commit (second pass: verify all checks)..."
+if ! pre-commit run --all-files; then
     echo ""
-    echo "❌ Pre-commit checks failed. Please fix the issues above before running tests."
-    echo "💡 Tip: Most formatting issues can be auto-fixed by running the checks again."
+    echo "❌ Pre-commit checks failed after applying fixes. Please address the issues above."
     exit 1
 fi
 
+# Track whether auto-fixes were applied and resolved issues
+autofixed=0
+if [ $first_pass_status -ne 0 ]; then
+    autofixed=1
+fi
+
 echo ""
 echo "✅ Pre-commit checks passed!"
 echo ""
@@ -66,9 +80,17 @@ pytest
 
 if [ $? -eq 0 ]; then
     echo ""
-    echo "🎉 All checks and tests passed! Your code is ready for commit."
+    if [ $autofixed -eq 1 ]; then
+        echo "🎉 All checks and tests passed. Auto-fixes were applied by pre-commit."
+    else
+        echo "🎉 All checks and tests passed."
+    fi
 else
     echo ""
-    echo "❌ Some tests failed. Please fix the issues above and try again."
+    if [ $autofixed -eq 1 ]; then
+        echo "❌ Some tests failed. Note: auto-fixes were applied earlier by pre-commit."
+    else
+        echo "❌ Some tests failed."
+    fi
     exit 1
 fi

ngraph/flows/policy.py

@@ -827,7 +827,7 @@ def get_flow_policy(flow_policy_config: FlowPolicyConfig) -> FlowPolicy:
             path_alg=base.PathAlg.SPF,
             flow_placement=FlowPlacement.PROPORTIONAL,
             edge_select=base.EdgeSelect.ALL_MIN_COST_WITH_CAP_REMAINING,
-            multipath=False,
+            multipath=True,
         )
     elif flow_policy_config == FlowPolicyConfig.TE_ECMP_UP_TO_256_LSP:
         # TE with up to 256 LSPs using ECMP flow placement.

ngraph/graph/strict_multidigraph.py

@@ -109,7 +109,7 @@ def remove_node(self, n: NodeID) -> None:
     #
     # Edge management
     #
-    def add_edge(
+    def add_edge(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
         u_for_edge: NodeID,
         v_for_edge: NodeID,

ngraph/monte_carlo/functions.py

@@ -133,21 +133,29 @@ def demand_placement_analysis(
 ) -> FlowIterationResult:
     """Analyze traffic demand placement success rates.
 
-    Returns a structured dictionary per iteration containing per-demand offered
-    and placed volumes (in Gbit/s) and an iteration-level summary. This shape
-    is designed for downstream computation of delivered bandwidth percentiles
-    without having to reconstruct per-iteration joint distributions.
+    Produces per-demand FlowEntry records and an iteration-level summary suitable
+    for downstream statistics (e.g., delivered percentiles) without reconstructing
+    joint distributions.
+
+    Additionally exposes placement engine counters to aid performance analysis:
+    - Per-demand: ``FlowEntry.data.policy_metrics`` (dict) with totals collected by
+      the active FlowPolicy (e.g., ``spf_calls_total``, ``flows_created_total``,
+      ``reopt_calls_total``, ``place_iterations_total``, ``placed_total``).
+    - Per-iteration: ``FlowIterationResult.data.iteration_metrics`` aggregating the
+      same counters across all demands in the iteration.
 
     Args:
         network_view: NetworkView with potential exclusions applied.
         demands_config: List of demand configurations (serializable dicts).
         placement_rounds: Number of placement optimization rounds.
         include_flow_details: When True, include cost_distribution per flow.
-        include_used_edges: When True, include set of used edges per demand in entry data.
+        include_used_edges: When True, include set of used edges per demand in entry data
+            as ``FlowEntry.data.edges`` with ``edges_kind='used'``.
         **kwargs: Ignored. Accepted for interface compatibility.
 
     Returns:
-        FlowIterationResult describing this iteration.
+        FlowIterationResult describing this iteration. The ``data`` field contains
+        ``{"iteration_metrics": { ... }}``.
     """
     # Reconstruct demands from config to avoid passing complex objects
     demands = []
@@ -179,6 +187,15 @@ def demand_placement_analysis(
     total_demand = 0.0
     total_placed = 0.0
 
+    # Aggregate iteration-level engine metrics across all demands
+    iteration_metrics: dict[str, float] = {
+        "spf_calls_total": 0.0,
+        "flows_created_total": 0.0,
+        "reopt_calls_total": 0.0,
+        "place_iterations_total": 0.0,
+        "placed_total": 0.0,
+    }
+
     for dmd in tm.demands:
         offered = float(getattr(dmd, "volume", 0.0))
         placed = float(getattr(dmd, "placed_demand", 0.0))
@@ -212,6 +229,24 @@ def demand_placement_analysis(
                 extra["edges"] = sorted(edge_strings)
                 extra["edges_kind"] = "used"
 
+        # Always expose per-demand FlowPolicy metrics when available
+        fp = getattr(dmd, "flow_policy", None)
+        if fp is not None:
+            try:
+                # Cumulative totals over the policy's lifetime within this iteration
+                totals: dict[str, float] = fp.get_metrics()  # type: ignore[assignment]
+            except Exception:
+                totals = {}
+            if totals:
+                extra["policy_metrics"] = {k: float(v) for k, v in totals.items()}
+                # Accumulate iteration-level totals across demands on known keys
+                for key in iteration_metrics.keys():
+                    if key in totals:
+                        try:
+                            iteration_metrics[key] += float(totals[key])
+                        except Exception:
+                            pass
+
         entry = FlowEntry(
             source=str(getattr(dmd, "src_node", "")),
             destination=str(getattr(dmd, "dst_node", "")),
@@ -235,7 +270,11 @@ def demand_placement_analysis(
         dropped_flows=dropped_flows,
         num_flows=len(flow_entries),
     )
-    return FlowIterationResult(flows=flow_entries, summary=summary)
+    return FlowIterationResult(
+        flows=flow_entries,
+        summary=summary,
+        data={"iteration_metrics": iteration_metrics},
+    )
 
 
 def sensitivity_analysis(

ngraph/workflow/analysis/cost_power_analysis.py

@@ -69,12 +69,7 @@ def analyze(self, results: Dict[str, Any], **kwargs: Any) -> Dict[str, Any]:
         data_obj = step_data.get("data", {}) if isinstance(step_data, dict) else {}
 
         ctx = data_obj.get("context", {}) if isinstance(data_obj, dict) else {}
-        try:
-            agg_level = int(ctx.get("aggregation_level", 2))
-        except Exception as exc:  # pragma: no cover - defensive
-            raise ValueError(
-                f"Invalid aggregation_level in CostPower data: {exc}"
-            ) from exc
+        agg_level = int(ctx.get("aggregation_level", 2))
 
         levels = data_obj.get("levels", {}) if isinstance(data_obj, dict) else {}
         level_key = str(agg_level)
@@ -99,12 +94,9 @@ def analyze(self, results: Dict[str, Any], **kwargs: Any) -> Dict[str, Any]:
             if isinstance(workflow_meta, dict) and workflow_meta:
                 candidates: list[tuple[int, str]] = []
                 for name, meta in workflow_meta.items():
-                    try:
-                        if str(meta.get("step_type")) == "TrafficMatrixPlacement":
-                            order = int(meta.get("execution_order", 1_000_000))
-                            candidates.append((order, name))
-                    except Exception:
-                        continue
+                    if str(meta.get("step_type")) == "TrafficMatrixPlacement":
+                        order = int(meta.get("execution_order", 1_000_000))
+                        candidates.append((order, name))
                 if candidates:
                     traffic_step_name = sorted(candidates)[0][1]
 
@@ -153,12 +145,9 @@ def path_at_level(name: str, level: int) -> str:
 
         # Attribute delivered volumes to both endpoints' sites (count once if same)
         for rec in base_flows:
-            try:
-                src = str(rec.get("source", ""))
-                dst = str(rec.get("destination", ""))
-                placed = float(rec.get("placed", 0.0))
-            except Exception:
-                continue
+            src = str(rec.get("source", ""))
+            dst = str(rec.get("destination", ""))
+            placed = float(rec.get("placed", 0.0))
             if not src or not dst or placed <= 0.0:
                 continue
 

ngraph/workflow/base.py

@@ -144,14 +144,24 @@ def execute(self, scenario: "Scenario") -> None:
             self.run(scenario)
             end_time = time.time()
             duration = end_time - start_time
+            # Persist step duration into step-scoped metadata for downstream analysis
+            existing_md = scenario.results.get("metadata", {})
+            if not isinstance(existing_md, dict):
+                raise TypeError("Results metadata must be a dict")
+            updated_md = dict(existing_md)
+            updated_md["duration_sec"] = float(duration)
+            scenario.results.put("metadata", updated_md)
             logger.info(
                 f"Completed workflow step: {display_name} ({step_type}) "
                 f"in {duration:.3f} seconds"
             )
             try:
                 store = getattr(scenario.results, "_store", {})
                 keys = ", ".join(sorted(list(store.get(step_name, {}).keys())))
-            except Exception:
+            except Exception as exc:
+                logger.debug(
+                    "Failed to read results keys for step %s: %s", display_name, exc
+                )
                 keys = "-"
             logger.debug(
                 "Step %s finished: duration=%.3fs, results_keys=%s",
@@ -171,8 +181,10 @@ def execute(self, scenario: "Scenario") -> None:
             # Always exit step scope
             try:
                 scenario.results.exit_step()
-            except Exception:
-                pass
+            except Exception as exc:
+                logger.warning(
+                    "Failed to exit step scope cleanly for %s: %s", display_name, exc
+                )
 
     @abstractmethod
     def run(self, scenario: "Scenario") -> None:

ngraph/workflow/cost_power.py

@@ -226,20 +226,17 @@ def add_values(
         )
 
         # Log root summary
-        try:
-            root_items = levels_payload.get(0, [])
-            root = root_items[0] if root_items else {}
-            logger.info(
-                "CostPower complete: name=%s capex=%.3f power=%.3f platform_capex=%.3f optics_capex=%.3f duration=%.3fs",
-                self.name or self.__class__.__name__,
-                float(root.get("capex_total", 0.0)),
-                float(root.get("power_total_watts", 0.0)),
-                float(root.get("platform_capex", 0.0)),
-                float(root.get("optics_capex", 0.0)),
-                time.perf_counter() - t0,
-            )
-        except Exception:
-            pass
+        root_items = levels_payload.get(0, [])
+        root = root_items[0] if root_items else {}
+        logger.info(
+            "CostPower complete: name=%s capex=%.3f power=%.3f platform_capex=%.3f optics_capex=%.3f duration=%.3fs",
+            self.name or self.__class__.__name__,
+            float(root.get("capex_total", 0.0)),
+            float(root.get("power_total_watts", 0.0)),
+            float(root.get("platform_capex", 0.0)),
+            float(root.get("optics_capex", 0.0)),
+            time.perf_counter() - t0,
+        )
 
 
 register_workflow_step("CostPower")(CostPower)

ngraph/workflow/traffic_matrix_placement_step.py

@@ -104,17 +104,17 @@ def run(self, scenario: "Scenario") -> None:
             ) from exc
 
         def _serialize_policy(cfg: Any) -> Any:
-            try:
-                from ngraph.flows.policy import FlowPolicyConfig  # local import
-            except Exception:  # pragma: no cover - defensive
-                return str(cfg) if cfg is not None else None
+            from ngraph.flows.policy import FlowPolicyConfig  # local import
+
             if cfg is None:
                 return None
             if isinstance(cfg, FlowPolicyConfig):
                 return cfg.name
+            # Fall back to string when it cannot be coerced to enum
             try:
                 return FlowPolicyConfig(int(cfg)).name
-            except Exception:
+            except Exception as exc:
+                logger.debug("Unrecognized flow_policy_config value: %r (%s)", cfg, exc)
                 return str(cfg)
 
         base_demands: list[dict[str, Any]] = [
@@ -133,15 +133,12 @@ def _serialize_policy(cfg: Any) -> Any:
 
         # Resolve alpha
         effective_alpha = self._resolve_alpha(scenario)
-        try:
-            alpha_src = getattr(self, "_alpha_source", None) or "explicit"
-            logger.info(
-                "Using alpha: value=%.6g source=%s",
-                float(effective_alpha),
-                str(alpha_src),
-            )
-        except Exception:
-            pass
+        alpha_src = getattr(self, "_alpha_source", None) or "explicit"
+        logger.info(
+            "Using alpha: value=%.6g source=%s",
+            float(effective_alpha),
+            str(alpha_src),
+        )
 
         demands_config: list[dict[str, Any]] = []
         for td in td_list: