Certify max-flow optimality instead of trusting recorded values

tests/py/conftest.py

@@ -164,17 +164,138 @@ def _assert(
 
 @pytest.fixture
 def assert_valid_min_cut():
-    """Return an assertion helper to validate MinCut edge ids are unique and valid."""
+    """Return an assertion helper to validate a MinCut.
 
-    def _assert(g: ngc.StrictMultiDiGraph, min_cut) -> None:
+    Always checks that edge ids are unique and in range. Note that an empty cut
+    satisfies those two checks vacuously, so pass ``total_flow`` whenever the call
+    site computed a true maximum flow: that enables the max-flow/min-cut duality
+    check (``total_flow == sum(capacity[cut])``), which is what actually pins the
+    result down.
+
+    Duality only holds for a genuine maximum flow. Do NOT pass ``total_flow`` for
+    ``EQUAL_BALANCED`` placement (an ECMP admission model that deliberately places
+    less than the maximum) or for ``shortest_path=True`` (a single augmentation).
+    """
+
+    def _assert(
+        g: ngc.StrictMultiDiGraph, min_cut, total_flow: float | None = None
+    ) -> None:
         edges = [int(e) for e in getattr(min_cut, "edges", [])]
-        assert len(edges) == len(set(edges))
+        assert len(edges) == len(set(edges)), "min-cut has duplicate edge ids"
         for e in edges:
             assert 0 <= e < g.num_edges()
 
+        if total_flow is None:
+            return
+
+        cap = np.asarray(g.capacity_view(), dtype=float)
+        cut_cap = float(cap[edges].sum()) if edges else 0.0
+        assert cut_cap == pytest.approx(total_flow), (
+            f"max-flow/min-cut duality violated: flow={total_flow} "
+            f"but cut capacity={cut_cap} (cut={edges})"
+        )
+        if total_flow > 0:
+            assert edges, "positive flow reported with an empty min-cut"
+
     return _assert
 
 
+@pytest.fixture
+def certify_max_flow():
+    """Return a helper that *proves* a max-flow result correct without an oracle.
+
+    Checks four properties. Together they are a certificate of optimality: by weak
+    duality any feasible flow is bounded by any s-t cut, so a feasible flow whose
+    value equals the capacity of a genuine cut is necessarily maximum.
+
+    1. feasibility   -- ``0 <= flow(e) <= capacity(e)``, and masked-out edges idle
+    2. conservation  -- inflow == outflow at interior nodes; net at sink == total
+    3. genuine cut   -- deleting the reported cut edges disconnects src from dst
+    4. tightness     -- ``total == sum(capacity[cut])``
+
+    Pass ``maximal=False`` for configurations that deliberately place less than the
+    maximum (``EQUAL_BALANCED`` placement, ``shortest_path=True``); only checks 1-2
+    apply there, and they still hold.
+
+    Requires ``with_edge_flows=True`` on the ``max_flow`` call.
+    """
+
+    def _certify(
+        g: ngc.StrictMultiDiGraph,
+        summary,
+        total: float,
+        src: int,
+        dst: int,
+        *,
+        maximal: bool = True,
+        node_mask=None,
+        edge_mask=None,
+        tol: float = 1e-6,
+    ) -> None:
+        flow = np.asarray(summary.edge_flows, dtype=float)
+        assert flow.size == g.num_edges(), (
+            "certify_max_flow needs edge flows; pass with_edge_flows=True"
+        )
+        cap = np.asarray(g.capacity_view(), dtype=float)
+        e_src = np.asarray(g.edge_src_view())
+        e_dst = np.asarray(g.edge_dst_view())
+        n = g.num_nodes()
+
+        # 1. feasibility
+        assert np.all(flow >= -tol), "negative flow on some edge"
+        assert np.all(flow <= cap + tol), "flow exceeds capacity on some edge"
+
+        usable = np.ones(g.num_edges(), dtype=bool)
+        if edge_mask is not None:
+            usable &= np.asarray(edge_mask, dtype=bool)
+        if node_mask is not None:
+            nm = np.asarray(node_mask, dtype=bool)
+            usable &= nm[e_src] & nm[e_dst]
+        assert np.all(np.abs(flow[~usable]) <= tol), "flow placed on a masked-out edge"
+
+        # 2. conservation
+        net = np.zeros(n)
+        np.add.at(net, e_src, -flow)
+        np.add.at(net, e_dst, flow)
+        interior = [v for v in range(n) if v != src and v != dst]
+        if interior:
+            worst = float(np.max(np.abs(net[interior])))
+            assert worst <= tol, f"flow not conserved at an interior node (net={worst})"
+        assert net[dst] == pytest.approx(total, abs=tol), (
+            f"net inflow at sink ({net[dst]}) != reported total ({total})"
+        )
+
+        if not maximal:
+            return
+
+        # 3. the reported cut must actually separate src from dst
+        cut = {int(e) for e in np.asarray(summary.min_cut.edges)}
+        adj: list[list[int]] = [[] for _ in range(n)]
+        for e in range(g.num_edges()):
+            if e in cut or not usable[e] or cap[e] <= 0:
+                continue
+            adj[int(e_src[e])].append(int(e_dst[e]))
+        seen = {int(src)}
+        stack = [int(src)]
+        while stack:
+            u = stack.pop()
+            for v in adj[u]:
+                if v not in seen:
+                    seen.add(v)
+                    stack.append(v)
+        assert int(dst) not in seen, (
+            f"reported min-cut {sorted(cut)} does not separate {src} from {dst}"
+        )
+
+        # 4. tightness -- with (1)-(3) this proves the flow is maximum
+        cut_cap = float(cap[sorted(cut)].sum()) if cut else 0.0
+        assert cut_cap == pytest.approx(total, abs=tol), (
+            f"flow ({total}) != min-cut capacity ({cut_cap}); flow is not maximal"
+        )
+
+    return _certify
+
+
 @pytest.fixture
 def build_graph():
     """Build a StrictMultiDiGraph from edge tuples.

tests/py/test_max_flow.py

@@ -33,7 +33,7 @@ def test_max_flow_square1_proportional_with_edge_flows(
     # Multi-tier: cost 2 path carries 1, then cost 4 path carries 2 => total 3
     assert np.isclose(total, 3.0)
     assert_edge_flows_shape(g, summary, expected_present=True)
-    assert_valid_min_cut(g, summary.min_cut)
+    assert_valid_min_cut(g, summary.min_cut, total)
     fb = flows_by_eid(g, summary.edge_flows)
     # Check totals on 4 edges in this small graph (order is deterministic by compaction)
     assert len(fb) == g.num_edges()
@@ -61,6 +61,7 @@ def test_square1_equal_balanced_min_cut_and_distribution(
     # Min-cut returns EdgeIds; ensure it has size 2 and corresponds to cut around source or sink
     mc = set(map(int, summary.min_cut.edges))
     assert len(mc) == 2
+    # EQUAL_BALANCED places less than the maximum, so duality does not apply.
     assert_valid_min_cut(g, summary.min_cut)
     assert_edge_flows_shape(g, summary, expected_present=True)
     # Cost distribution checks live in test_max_flow_cost_distribution.py
@@ -133,6 +134,7 @@ def test_max_flow_square1_shortest_path_single_augmentation(
     )
     assert np.isclose(total, 1.0)
     assert_edge_flows_shape(g, summary, expected_present=True)
+    # shortest_path=True is a single augmentation, not a maximum flow.
     assert_valid_min_cut(g, summary.min_cut)
     fb = flows_by_eid(g, summary.edge_flows)
     # Should augment along A->B->C only once
@@ -162,6 +164,7 @@ def test_max_flow_line1_equal_balanced(
     # Equal-balanced across tiers: limited by A->B capacity => total 5
     assert np.isclose(total, 5.0)
     assert_edge_flows_shape(g, summary, expected_present=True)
+    # EQUAL_BALANCED places less than the maximum, so duality does not apply.
     assert_valid_min_cut(g, summary.min_cut)
     fb = flows_by_eid(g, summary.edge_flows)
     # Expect 2 across min-cost tier (1 + 1), then remaining 3 on min/higher edges by successive tiers
@@ -238,7 +241,7 @@ def test_max_flow_graph3_proportional_parallel_distribution(
     # Incoming to C via min-cost parents: from B (cap 1+2+3=6) and from E (cap 4) => total 10
     assert np.isclose(total, 10.0)
     assert_edge_flows_shape(g, summary, expected_present=True)
-    assert_valid_min_cut(g, summary.min_cut)
+    assert_valid_min_cut(g, summary.min_cut, total)
     fb = flows_by_eid(g, summary.edge_flows)
     # B->C parallels proportional to capacity: 1:2:3 over total 6 => 1,2,3
     assert np.isclose(fb[4], 1.0)
@@ -276,7 +279,7 @@ def test_max_flow_two_disjoint_shortest_routes_proportional(
     # Bottlenecks along S->A->T = min(3,2)=2 and S->B->T = min(4,1)=1 => total 3
     assert np.isclose(total, 3.0)
     assert_edge_flows_shape(g, summary, expected_present=True)
-    assert_valid_min_cut(g, summary.min_cut)
+    assert_valid_min_cut(g, summary.min_cut, total)
     fb = flows_by_eid(g, summary.edge_flows)
     assert np.isclose(fb[0], 2.0)  # S->A
     assert np.isclose(fb[2], 2.0)  # A->T

tests/py/test_max_flow_summary.py

@@ -33,7 +33,7 @@ def test_flow_summary_shapes_and_min_cut_valid(
     )
     assert np.isclose(total, 8.0)
     assert_edge_flows_shape(g, summary, expected_present=True)
-    assert_valid_min_cut(g, summary.min_cut)
+    assert_valid_min_cut(g, summary.min_cut, total)
     # Cost distribution checked in test_max_flow_cost_distribution.py
 
 

tests/py/test_maxflow_certificate.py

@@ -0,0 +1,197 @@
+"""Oracle-free correctness certificates for max-flow.
+
+The suite historically checked max-flow results against hand-computed values on a
+handful of fixed topologies. That misses defects whose trigger is a graph nobody
+thought to write down -- the 0.7.2 under-reporting bug (fixed in 0.8.0) was not
+exposed by any fixture in the suite, at any src/dst pair.
+
+These tests take the opposite approach: generate graphs in the region where such
+defects live (dispersed edge costs, full-duplex links) and check a *certificate of
+optimality* rather than a recorded number. See the ``certify_max_flow`` fixture for
+the four properties; together they prove maximality, so no external oracle and no
+stored expected value is required.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+import netgraph_core as ngc
+
+# A fixed seed keeps these deterministic, so a given defect is either caught on every
+# run or never -- which means the corpus has to be shown to have teeth, not assumed to.
+# This seed was picked because its corpus trips the 0.7.2 under-reporting bug within the
+# first few graphs; both sweeps below were confirmed to fail against 0.7.2 and pass
+# against 0.8.0. An arbitrary seed needed ~5750 graphs to hit the same defect, so this
+# is purely a cheaper corpus, not a narrower one.
+SEED = 198
+QUICK_GRAPHS = 2000
+THOROUGH_GRAPHS = 20000
+
+
+def _random_graph(rng: np.random.Generator) -> tuple[ngc.StrictMultiDiGraph, int, int]:
+    """Build a random full-duplex graph with dispersed costs.
+
+    Cost dispersion is the essential ingredient: with a single cost tier there is
+    nothing for shortest-path tier ordering to get wrong, and randomized checks over
+    equal-cost graphs find nothing.
+    """
+    n = int(rng.integers(4, 9))
+    m = int(rng.integers(n, 2 * n))
+    links = []
+    for _ in range(m):
+        u, v = rng.choice(n, 2, replace=False)
+        cap = float(rng.integers(1, 6))
+        cost = int(rng.integers(1, 21))
+        links.append((int(u), int(v), cap, cost))
+
+    edges = [e for u, v, c, k in links for e in ((u, v, c, k), (v, u, c, k))]
+    src = np.array([e[0] for e in edges], dtype=np.int32)
+    dst = np.array([e[1] for e in edges], dtype=np.int32)
+    cap = np.array([e[2] for e in edges], dtype=np.float64)
+    cost = np.array([e[3] for e in edges], dtype=np.int64)
+    g = ngc.StrictMultiDiGraph.from_arrays(
+        n, src, dst, cap, cost, np.arange(len(edges), dtype=np.int64)
+    )
+    return g, 0, n - 1
+
+
+def _sweep(count: int, algs, certify_max_flow) -> None:
+    rng = np.random.default_rng(SEED)
+    for i in range(count):
+        g, s, t = _random_graph(rng)
+        total, summary = algs.max_flow(algs.build_graph(g), s, t, with_edge_flows=True)
+        try:
+            certify_max_flow(g, summary, total, s, t)
+        except AssertionError as exc:  # pragma: no cover - only on a real defect
+            pytest.fail(f"certificate failed on random graph #{i} (seed {SEED}): {exc}")
+
+
+def test_random_graphs_certify(algs, certify_max_flow):
+    """Every generated max-flow result must carry a valid optimality certificate."""
+    _sweep(QUICK_GRAPHS, algs, certify_max_flow)
+
+
+@pytest.mark.slow
+def test_random_graphs_certify_thorough(algs, certify_max_flow):
+    """Wider sweep of the same generator."""
+    _sweep(THOROUGH_GRAPHS, algs, certify_max_flow)
+
+
+FIXTURES = [
+    "line1_graph",
+    "square1_graph",
+    "square2_graph",
+    "graph3",
+    "square4_graph",
+    "triangle1_graph",
+]
+
+
+@pytest.mark.parametrize("fixture_name", FIXTURES)
+def test_named_fixtures_certify_all_pairs(
+    fixture_name, request, algs, certify_max_flow
+):
+    """Certify the shared fixtures for every src/dst pair, not just the tested one.
+
+    A fixture is usually queried for a single pair; the remaining pairs are free
+    coverage of the same topology.
+    """
+    g = request.getfixturevalue(fixture_name)
+    handle = algs.build_graph(g)
+    n = g.num_nodes()
+    for s in range(n):
+        for t in range(n):
+            if s == t:
+                continue
+            total, summary = algs.max_flow(handle, s, t, with_edge_flows=True)
+            certify_max_flow(g, summary, total, s, t)
+
+
+def test_certificate_rejects_a_non_maximal_flow(square1_graph, algs, certify_max_flow):
+    """The certificate must have teeth.
+
+    ``shortest_path=True`` performs a single augmentation and is deliberately not
+    maximal, so it is a genuine non-maximal result reachable through the public API.
+    Certifying it as maximal must fail. Without this guard the certificate could
+    silently degrade into an assertion that cannot fail -- which is exactly how the
+    previous ``assert_valid_min_cut`` helper missed the 0.7.2 bug.
+    """
+    g = square1_graph
+    total, summary = algs.max_flow(
+        algs.build_graph(g), 0, 2, shortest_path=True, with_edge_flows=True
+    )
+    assert total == pytest.approx(1.0)
+
+    # Feasibility and conservation still hold for a partial placement.
+    certify_max_flow(g, summary, total, 0, 2, maximal=False)
+
+    with pytest.raises(AssertionError):
+        certify_max_flow(g, summary, total, 0, 2, maximal=True)
+
+
+def test_equal_balanced_is_feasible_but_not_certified_maximal(
+    line1_graph, algs, certify_max_flow
+):
+    """EQUAL_BALANCED is an admission model, so only feasibility/conservation apply."""
+    g = line1_graph
+    total, summary = algs.max_flow(
+        algs.build_graph(g),
+        0,
+        2,
+        flow_placement=ngc.FlowPlacement.EQUAL_BALANCED,
+        with_edge_flows=True,
+    )
+    certify_max_flow(g, summary, total, 0, 2, maximal=False)
+
+
+def test_masked_runs_certify(algs, certify_max_flow):
+    """Masking must not break the certificate, and must not leak flow onto masked edges.
+
+    Masks are a good place for a future defect of this kind to hide: no external
+    solver models them, so an oracle-based test cannot reach them at all.
+
+    Node and edge masks are checked separately and together, because they are
+    separate branches in ``calc_max_flow`` -- including inside the residual
+    completion phase added in 0.8.0, whose forward and reverse arc loops each
+    consult ``edge_mask`` independently. A node-mask-only sweep leaves those
+    branches, and the certificate's own ``edge_mask`` handling, unexercised.
+    """
+    rng = np.random.default_rng(SEED + 1)
+    saw_masked_node = False
+    saw_masked_edge = False
+
+    for i in range(300):
+        g, s, t = _random_graph(rng)
+        n = g.num_nodes()
+
+        node_mask = None
+        edge_mask = None
+        mode = i % 3  # 0: nodes only, 1: edges only, 2: both
+        if mode in (0, 2):
+            node_mask = np.ones(n, dtype=bool)
+            victim = int(rng.integers(0, n))
+            if victim not in (s, t):
+                node_mask[victim] = False
+                saw_masked_node = True
+        if mode in (1, 2):
+            edge_mask = rng.random(g.num_edges()) > 0.25
+            if not edge_mask.all():
+                saw_masked_edge = True
+
+        total, summary = algs.max_flow(
+            algs.build_graph(g),
+            s,
+            t,
+            node_mask=node_mask,
+            edge_mask=edge_mask,
+            with_edge_flows=True,
+        )
+        certify_max_flow(
+            g, summary, total, s, t, node_mask=node_mask, edge_mask=edge_mask
+        )
+
+    # Guard against the sweep silently degrading into an unmasked one.
+    assert saw_masked_node, "node-mask branch was never exercised"
+    assert saw_masked_edge, "edge-mask branch was never exercised"