bugfixing

Author: networmix

docs/reference/api-full.md

@@ -12,7 +12,7 @@ Quick links:
 - [CLI Reference](cli.md)
 - [DSL Reference](dsl.md)
 
-Generated from source code on: August 13, 2025 at 21:26 UTC
+Generated from source code on: August 14, 2025 at 01:51 UTC
 
 Modules auto-discovered: 68
 
@@ -1436,7 +1436,7 @@ are present, traversal may not terminate.
 - `add(self, other: 'PathBundle') -> 'PathBundle'` - Concatenate this bundle with another bundle (end-to-start).
 - `contains(self, other: 'PathBundle') -> 'bool'` - Check if this bundle's edge set contains all edges of `other`.
 - `from_path(path: 'Path', resolve_edges: 'bool' = False, graph: 'Optional[StrictMultiDiGraph]' = None, edge_select: 'Optional[EdgeSelect]' = None, cost_attr: 'str' = 'cost', capacity_attr: 'str' = 'capacity') -> 'PathBundle'` - Construct a PathBundle from a single `Path` object.
-- `get_sub_path_bundle(self, new_dst_node: 'NodeID', graph: 'StrictMultiDiGraph', cost_attr: 'str' = 'cost') -> 'PathBundle'` - Create a sub-bundle ending at `new_dst_node` (which must appear in this bundle).
+- `get_sub_path_bundle(self, new_dst_node: 'NodeID', graph: 'StrictMultiDiGraph', cost_attr: 'str' = 'cost') -> 'PathBundle'` - Create a sub-bundle ending at `new_dst_node` with correct minimal cost.
 - `is_disjoint_from(self, other: 'PathBundle') -> 'bool'` - Check if this bundle shares no edges with `other`.
 - `is_subset_of(self, other: 'PathBundle') -> 'bool'` - Check if this bundle's edge set is contained in `other`'s edge set.
 - `resolve_to_paths(self, split_parallel_edges: 'bool' = False) -> 'Iterator[Path]'` - Generate all concrete `Path` objects contained in this PathBundle.

ngraph/paths/bundle.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 from collections import deque
+from heapq import heappop, heappush
 from typing import Dict, Iterator, List, Optional, Set, Tuple
 
 from ngraph.algorithms.base import Cost, EdgeSelect
@@ -77,11 +78,13 @@ def __init__(
         while queue:
             node = queue.popleft()
             self.nodes.add(node)
+            # Ensure key exists even if `node` has no predecessors.
+            self.pred.setdefault(node, {})
 
             # Traverse all predecessors of `node`
             for prev_node, edges_list in pred[node].items():
                 # Record these edges in our local `pred` structure
-                self.pred.setdefault(node, {})[prev_node] = edges_list
+                self.pred[node][prev_node] = edges_list
                 # Update the set of all edges seen in this bundle
                 self.edges.update(edges_list)
                 # Store the tuple form for quick equality checks on parallel edges
@@ -284,11 +287,14 @@ def get_sub_path_bundle(
         graph: StrictMultiDiGraph,
         cost_attr: str = "cost",
     ) -> PathBundle:
-        """Create a sub-bundle ending at `new_dst_node` (which must appear in this bundle).
+        """Create a sub-bundle ending at `new_dst_node` with correct minimal cost.
 
-        This method performs a reverse traversal (BFS) from `new_dst_node` up to
-        `self.src_node`, collecting edges and recalculating the cost along the way
-        using the minimal edge attribute found.
+        The returned bundle contains the predecessor subgraph that reaches from
+        `self.src_node` to `new_dst_node` using only relations present in this
+        bundle's `pred`. The `cost` of the returned bundle is recomputed as the
+        minimal sum of per-hop costs along any valid path from `self.src_node`
+        to `new_dst_node`, where each hop cost is the minimum of `cost_attr`
+        across the parallel edges recorded for that hop.
 
         Args:
             new_dst_node: The new destination node, which must be present in `pred`.
@@ -305,30 +311,55 @@ def get_sub_path_bundle(
             raise ValueError(f"{new_dst_node} not in this PathBundle's pred")
 
         edges_dict = graph.get_edges()
-        new_pred: Dict[NodeID, Dict[NodeID, List[EdgeID]]] = {self.src_node: {}}
-        new_cost: float = 0.0
-
-        visited: Set[NodeID] = set()
-        queue: deque[Tuple[float, NodeID]] = deque([(0.0, new_dst_node)])
-        visited.add(new_dst_node)
 
-        while queue:
-            cost_to_node, node = queue.popleft()
-            # For each predecessor of `node`, add them to new_pred
+        # 1) Build the restricted predecessor subgraph reachable from new_dst_node
+        #    back to self.src_node. This preserves all allowed predecessors without
+        #    collapsing to a single path.
+        new_pred: Dict[NodeID, Dict[NodeID, List[EdgeID]]] = {self.src_node: {}}
+        visited_build: Set[NodeID] = set([new_dst_node])
+        queue_build: deque[NodeID] = deque([new_dst_node])
+        while queue_build:
+            node = queue_build.popleft()
             for prev_node, edges_list in self.pred[node].items():
                 new_pred.setdefault(node, {})[prev_node] = edges_list
-                # Recompute the cost increment of traveling from prev_node to node
-                cost_increment = min(
-                    edges_dict[eid][3][cost_attr] for eid in edges_list
-                )
-                updated_cost = cost_to_node + cost_increment
-
-                # Enqueue predecessor if not source and not yet visited
-                if prev_node != self.src_node and prev_node not in visited:
-                    visited.add(prev_node)
-                    queue.append((updated_cost, prev_node))
-                else:
-                    # Once we reach the src_node, record the final cost
-                    new_cost = updated_cost
+                if prev_node != self.src_node and prev_node not in visited_build:
+                    visited_build.add(prev_node)
+                    queue_build.append(prev_node)
+
+        # 2) Compute minimal cost from self.src_node to new_dst_node over new_pred
+        #    using Dijkstra on the reversed edges (from dst backwards to src).
+        def hop_cost(u: NodeID, v: NodeID) -> float:
+            # cost to go from u<-v (i.e., v -> u in forward direction)
+            edges_list = new_pred[u][v]
+            return float(min(edges_dict[eid][3][cost_attr] for eid in edges_list))
+
+        # Trivial case: src == dst
+        if new_dst_node == self.src_node:
+            return PathBundle(self.src_node, new_dst_node, new_pred, 0.0)
+
+        dist: Dict[NodeID, float] = {new_dst_node: 0.0}
+        heap: List[Tuple[float, NodeID]] = [(0.0, new_dst_node)]
+        best_cost: float = float("inf")
+
+        while heap:
+            cost_to_node, node = heappop(heap)
+            if cost_to_node > dist.get(node, float("inf")):
+                continue
+            if node == self.src_node:
+                best_cost = cost_to_node
+                break
+            # Relax predecessors of `node` (reverse traversal)
+            for prev_node in new_pred.get(node, {}):
+                c = cost_to_node + hop_cost(node, prev_node)
+                if c < dist.get(prev_node, float("inf")):
+                    dist[prev_node] = c
+                    heappush(heap, (c, prev_node))
+
+        # If src_node was not reached, this subgraph does not connect src->new_dst.
+        # Treat as an error to avoid silent mis-reporting.
+        if best_cost == float("inf"):
+            raise ValueError(
+                f"No path from '{self.src_node}' to '{new_dst_node}' within this PathBundle."
+            )
 
-        return PathBundle(self.src_node, new_dst_node, new_pred, new_cost)
+        return PathBundle(self.src_node, new_dst_node, new_pred, float(best_cost))

tests/paths/test_bundle.py

@@ -1,9 +1,9 @@
-from typing import List, Set
+from typing import Dict, List, Set
 
 import pytest
 
 from ngraph.algorithms.base import EdgeSelect
-from ngraph.graph.strict_multidigraph import StrictMultiDiGraph
+from ngraph.graph.strict_multidigraph import EdgeID, NodeID, StrictMultiDiGraph
 from ngraph.paths.bundle import Path, PathBundle
 
 
@@ -148,6 +148,51 @@ def test_get_sub_bundle_2(self, triangle1):
         }
         assert sub_bundle.cost == 0
 
+    def test_get_sub_bundle_min_cost_across_alternatives(self):
+        g = StrictMultiDiGraph()
+        for n in ("A", "B", "C", "D"):
+            g.add_node(n)
+
+        # Two routes A->D: A->C->D cost 2 (1+1), A->B->D cost 20 (10+10)
+        e_ac = g.add_edge("A", "C", cost=1)
+        e_cd = g.add_edge("C", "D", cost=1)
+        e_ab = g.add_edge("A", "B", cost=10)
+        e_bd = g.add_edge("B", "D", cost=10)
+
+        pred: Dict[NodeID, Dict[NodeID, List[EdgeID]]] = {
+            "A": {},
+            "C": {"A": [e_ac]},
+            "B": {"A": [e_ab]},
+            "D": {"C": [e_cd], "B": [e_bd]},
+        }
+
+        bundle = PathBundle("A", "D", pred, cost=2)
+        sub = bundle.get_sub_path_bundle("D", g, cost_attr="cost")
+        assert sub.cost == 2
+
+    def test_get_sub_bundle_src_equals_dst_zero_cost(self):
+        g = StrictMultiDiGraph()
+        g.add_node("A")
+        pred: Dict[NodeID, Dict[NodeID, List[EdgeID]]] = {"A": {}}
+        bundle = PathBundle("A", "A", pred, cost=0)
+        sub = bundle.get_sub_path_bundle("A", g)
+        assert sub.cost == 0
+
+    def test_get_sub_bundle_raises_when_src_unreachable_in_subgraph(self):
+        g = StrictMultiDiGraph()
+        for n in ("A", "B", "C"):
+            g.add_node(n)
+        # pred missing any chain from A to C (only B->C exists)
+        e_bc = g.add_edge("B", "C", cost=1)
+        pred: Dict[NodeID, Dict[NodeID, List[EdgeID]]] = {
+            "A": {},
+            "B": {},
+            "C": {"B": [e_bc]},
+        }
+        bundle = PathBundle("A", "C", pred, cost=0)
+        with pytest.raises(ValueError, match="No path from 'A' to 'C'"):
+            _ = bundle.get_sub_path_bundle("C", g)
+
     def test_add_method(self):
         """Test concatenating two PathBundles with matching src/dst."""
         pb1 = PathBundle(

tests/paths/test_path.py

@@ -165,3 +165,18 @@ def test_get_sub_path_empty_parallel_edges():
     sub = p.get_sub_path("N2", g)
     assert sub.cost == 10
     assert len(sub) == 2
+
+
+def test_get_sub_path_uses_min_cost_among_parallel_edges():
+    g = StrictMultiDiGraph()
+    for n in ("A", "B", "C"):
+        g.add_node(n)
+
+    e1 = g.add_edge("A", "B", cost=5)
+    e2 = g.add_edge("A", "B", cost=7)
+    e3 = g.add_edge("B", "C", cost=1)
+
+    path_tuple: PathTuple = (("A", (e1, e2)), ("B", (e3,)), ("C", ()))
+    p = Path(path_tuple, cost=0)
+    sub = p.get_sub_path("B", g, cost_attr="cost")
+    assert sub.cost == 5