From fd4db563fe7aacb898b3b46b68c5f1ea9d6e199d Mon Sep 17 00:00:00 2001 From: Andrey Golovanov Date: Mon, 24 Aug 2026 01:52:22 +0100 Subject: [PATCH 1/2] Certify max-flow optimality instead of trusting recorded values The 0.7.2 under-reporting bug (fixed in 0.8.0) survived ~400 tests because the suite could not reach it, not because it checked too loosely. Measured: none of the shared fixtures expose it, at any src/dst pair (0 of 78 combinations), so no amount of assertion strengthening would have caught it. The binding constraint was input generation - every test graph had near-uniform costs, and the defect needs dispersed costs to trigger. Add an oracle-free optimality certificate and generate graphs in the region where such defects live: - conftest: certify_max_flow checks feasibility, conservation, that the reported min-cut genuinely separates src from dst, and tightness. By weak duality those four together *prove* the flow is maximum, so no external solver and no stored expected value is needed. Masked runs are supported, and maximal=False covers EQUAL_BALANCED and shortest_path=True, where only the first two properties hold. - test_maxflow_certificate: seeded sweeps over full-duplex graphs with dispersed costs, plus certification of every shared fixture across all src/dst pairs. The seed was chosen so its corpus trips the 0.7.2 defect within the first few graphs; both sweeps were confirmed to fail against 0.7.2 and pass against 0.8.0. Includes a guard that the certificate rejects a known non-maximal flow, so it cannot decay into an assertion that always passes. - conftest: assert_valid_min_cut previously checked only that edge ids were unique and in range, both of which an empty cut satisfies vacuously - so it was guaranteed to pass on exactly the output the bug produced. It now takes an optional total_flow and asserts duality. Wired into the four call sites that compute a true maximum; the EQUAL_BALANCED and shortest_path sites are annotated with why duality does not apply there. - test_review_regressions: the batch/serial parity test generated uniform costs and compared two paths that both call calc_max_flow, so it could only ever prove consistency, never correctness. Costs are now dispersed and results certified; this makes the test fail against 0.7.2, which it did not before. 444 Python tests pass (from 433), suite 2.4s -> 3.8s. Co-Authored-By: Claude Opus 5 --- tests/py/conftest.py | 127 +++++++++++++++++++- tests/py/test_max_flow.py | 9 +- tests/py/test_max_flow_summary.py | 2 +- tests/py/test_maxflow_certificate.py | 166 +++++++++++++++++++++++++++ tests/py/test_review_regressions.py | 13 ++- 5 files changed, 308 insertions(+), 9 deletions(-) create mode 100644 tests/py/test_maxflow_certificate.py diff --git a/tests/py/conftest.py b/tests/py/conftest.py index ed33a8e..17422e5 100644 --- a/tests/py/conftest.py +++ b/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. diff --git a/tests/py/test_max_flow.py b/tests/py/test_max_flow.py index 512aa7b..645730a 100644 --- a/tests/py/test_max_flow.py +++ b/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 diff --git a/tests/py/test_max_flow_summary.py b/tests/py/test_max_flow_summary.py index 5e2bee3..609b3e3 100644 --- a/tests/py/test_max_flow_summary.py +++ b/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 diff --git a/tests/py/test_maxflow_certificate.py b/tests/py/test_maxflow_certificate.py new file mode 100644 index 0000000..d3e7e10 --- /dev/null +++ b/tests/py/test_maxflow_certificate.py @@ -0,0 +1,166 @@ +"""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. + """ + rng = np.random.default_rng(SEED + 1) + for _ in range(200): + g, s, t = _random_graph(rng) + n = g.num_nodes() + node_mask = np.ones(n, dtype=bool) + victim = int(rng.integers(0, n)) + if victim not in (s, t): + node_mask[victim] = False + total, summary = algs.max_flow( + algs.build_graph(g), s, t, node_mask=node_mask, with_edge_flows=True + ) + certify_max_flow(g, summary, total, s, t, node_mask=node_mask) diff --git a/tests/py/test_review_regressions.py b/tests/py/test_review_regressions.py index ea83104..778763d 100644 --- a/tests/py/test_review_regressions.py +++ b/tests/py/test_review_regressions.py @@ -182,14 +182,20 @@ def new(): class TestBatchMaxFlowParity: """Finding 4: batch results must equal per-pair results (now parallel).""" - def test_batch_equals_serial(self, algs): + def test_batch_equals_serial(self, algs, certify_max_flow): rng = np.random.default_rng(7) n = 12 edges = [] for _ in range(40): u, v = rng.integers(0, n, size=2) if u != v: - edges.append((int(u), int(v), float(rng.integers(1, 8)), 1)) + # Costs must be dispersed. With a single cost tier there is nothing + # for shortest-path tier ordering to get wrong, so a uniform-cost + # generator only ever samples the region where max-flow defects of + # the Finding 1 kind cannot occur. + cap = float(rng.integers(1, 8)) + cost = int(rng.integers(1, 21)) + edges.append((int(u), int(v), cap, cost)) g = _graph(n, edges) pg = algs.build_graph(g) pairs = np.array([[i, (i + 5) % n] for i in range(8)], dtype=np.int32) @@ -200,6 +206,9 @@ def test_batch_equals_serial(self, algs): np.testing.assert_array_equal( np.asarray(batch[i].edge_flows), np.asarray(summary.edge_flows) ) + # Agreement between two code paths that share calc_max_flow only proves + # consistency; both could be wrong together. Certify the result too. + certify_max_flow(g, summary, total, int(a), int(b)) class TestFlowPolicySrcDstGuard: From a2eab9e58a09a7c595c7cfe0eb0a19f65b0a3b81 Mon Sep 17 00:00:00 2001 From: Andrey Golovanov Date: Mon, 24 Aug 2026 01:59:11 +0100 Subject: [PATCH 2/2] Exercise edge masks in the masked certificate sweep The sweep supplied only node_mask, so it never reached calc_max_flow's edge_mask branches -- including the two inside the 0.8.0 residual completion phase, whose forward and reverse arc loops consult edge_mask independently (src/max_flow.cpp:190, :202) and compute_min_cut's own edge-mask path. The certificate helper's edge_mask handling was likewise unexercised despite the test claiming coverage for masked-out edges. Rotate through node-only, edge-only, and combined masking, and assert both branches were actually reached so the sweep cannot silently degrade into an unmasked one. This is prospective branch coverage, not retroactive detection: the masked sweep passes against 0.7.2 either way, since the under-reporting defect is not mask-specific. Verified separately that the masked-edge assertion does fire on a violating input, so it is not another check that cannot fail. Reported by Codex review on #6. Co-Authored-By: Claude Opus 5 --- tests/py/test_maxflow_certificate.py | 45 +++++++++++++++++++++++----- 1 file changed, 38 insertions(+), 7 deletions(-) diff --git a/tests/py/test_maxflow_certificate.py b/tests/py/test_maxflow_certificate.py index d3e7e10..da58792 100644 --- a/tests/py/test_maxflow_certificate.py +++ b/tests/py/test_maxflow_certificate.py @@ -151,16 +151,47 @@ def test_masked_runs_certify(algs, certify_max_flow): 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) - for _ in range(200): + 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 = np.ones(n, dtype=bool) - victim = int(rng.integers(0, n)) - if victim not in (s, t): - node_mask[victim] = False + + 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, with_edge_flows=True + 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 ) - certify_max_flow(g, summary, total, s, t, node_mask=node_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"