beam: Dinic max-flow implementation with randomized cross-check vs Edmonds-Karp (8 tests pass)

wizard-work/beam/dinic_maxflow.py

@@ -0,0 +1,74 @@
+"""Dinic's maximum flow algorithm (integer capacities).
+
+Correctness hinges on three invariants:
+  1. BFS builds a level graph from s using only edges with residual > 0.
+  2. DFS sends blocking flow respecting levels (level[v] + 1 == level[u]
+     along an edge).
+  3. Current-arc (iter) pointers skip saturated edges within a phase.
+
+Verified against brute-force (Edmonds-Karp / Ford-Fulkerson) on random
+small graphs including multi-edges, zero capacities, disconnected graphs.
+"""
+from collections import deque
+
+
+class Dinic:
+    def __init__(self, n):
+        self.n = n
+        self.adj = [[] for _ in range(n)]
+
+    def add_edge(self, u, v, cap):
+        # forward edge index; backward is its pair
+        fwd = [v, cap, None]  # placeholder; store as dict-like list
+        # We'll store edges as [to, cap, rev_index]
+        fi = len(self.adj[u])
+        bi = len(self.adj[v])
+        self.adj[u].append([v, cap, bi])
+        self.adj[v].append([u, 0, fi])
+        return fi
+
+    def bfs(self, s, t):
+        self.level = [-1] * self.n
+        self.level[s] = 0
+        q = deque([s])
+        while q:
+            u = q.popleft()
+            for v, cap, _ in self.adj[u]:
+                if cap > 0 and self.level[v] < 0:
+                    self.level[v] = self.level[u] + 1
+                    q.append(v)
+        return self.level[t] >= 0
+
+    def dfs(self, u, t, f):
+        if u == t:
+            return f
+        for i in range(self.it[u], len(self.adj[u])):
+            self.it[u] = i
+            v, cap, rev = self.adj[u][i]
+            if cap > 0 and self.level[v] == self.level[u] + 1:
+                pushed = self.dfs(v, t, min(f, cap))
+                if pushed > 0:
+                    self.adj[u][i][1] -= pushed
+                    self.adj[v][rev][1] += pushed
+                    return pushed
+        return 0
+
+    def max_flow(self, s, t):
+        flow = 0
+        INF = 10 ** 18
+        while self.bfs(s, t):
+            self.it = [0] * self.n
+            while True:
+                pushed = self.dfs(s, t, INF)
+                if pushed == 0:
+                    break
+                flow += pushed
+        return flow
+
+
+def max_flow(n, edges, s, t):
+    """edges: list of (u, v, cap). Returns max flow value."""
+    d = Dinic(n)
+    for u, v, c in edges:
+        d.add_edge(u, v, c)
+    return d.max_flow(s, t)

wizard-work/beam/test_dinic_maxflow.py

@@ -0,0 +1,84 @@
+import random
+from dinic_maxflow import max_flow as dinic_flow
+
+
+def ek_flow(n, edges, s, t):
+    # Edmonds-Karp (BFS augmenting paths) reference
+    from collections import deque
+    cap = [[0]*n for _ in range(n)]
+    for u, v, c in edges:
+        cap[u][v] += c
+    flow = 0
+    while True:
+        parent = [-1]*n
+        parent[s] = s
+        q = deque([s])
+        while q and parent[t] == -1:
+            u = q.popleft()
+            for v in range(n):
+                if parent[v] == -1 and cap[u][v] > 0:
+                    parent[v] = u
+                    q.append(v)
+        if parent[t] == -1:
+            break
+        # bottleneck
+        aug = 10**18
+        v = t
+        while v != s:
+            u = parent[v]
+            aug = min(aug, cap[u][v])
+            v = u
+        v = t
+        while v != s:
+            u = parent[v]
+            cap[u][v] -= aug
+            cap[v][u] += aug
+            v = u
+        flow += aug
+    return flow
+
+
+def test_simple():
+    assert dinic_flow(4, [(0,1,10),(1,3,10),(0,2,10),(2,3,10)], 0, 3) == 20
+
+
+def test_single():
+    assert dinic_flow(2, [(0,1,5)], 0, 1) == 5
+
+
+def test_zero_cap():
+    assert dinic_flow(2, [(0,1,0)], 0, 1) == 0
+
+
+def test_disconnected():
+    assert dinic_flow(4, [(0,1,5)], 0, 3) == 0
+
+
+def test_bottleneck():
+    assert dinic_flow(3, [(0,1,10),(1,2,3)], 0, 2) == 3
+
+
+def test_multi_edge():
+    assert dinic_flow(2, [(0,1,3),(0,1,4)], 0, 1) == 7
+
+
+def test_random_vs_ek():
+    rng = random.Random(42)
+    for _ in range(300):
+        n = rng.randint(2, 7)
+        s, t = 0, n-1
+        m = rng.randint(0, 12)
+        edges = []
+        for _ in range(m):
+            u = rng.randrange(n)
+            v = rng.randrange(n)
+            c = rng.randint(0, 6)
+            edges.append((u, v, c))
+        a = dinic_flow(n, edges, s, t)
+        b = ek_flow(n, edges, s, t)
+        assert a == b, (edges, a, b)
+
+
+def test_self_loop_ignored():
+    # self-loops shouldn't inflate flow
+    assert dinic_flow(2, [(0,0,100),(0,1,5)], 0, 1) == 5