Implement Partition -> IntegralFlowWithMultipliers reduction (#363)

Adds Sahni's multiplier-flow gadget: each Partition element becomes an item
vertex whose multiplier amplifies a binary source choice into either 0 or a_i
units entering a relay. A single bottleneck arc of capacity S/2 converts the
target's "net inflow at least R" condition into the exact equality needed by
Partition. Odd-S inputs reduce to a fixed infeasible 3-vertex instance.

- src/rules/partition_integralflowwithmultipliers.rs: reduction impl
  with odd-S/even-S branches and witness extraction from source arcs
- src/unit_tests/rules/partition_integralflowwithmultipliers.rs: 5 tests
  (closed-loop, structure on even-total YES instance, even-total NO
  exercises bottleneck, odd-total fixed NO target, witness extraction,
  canonical example spec)
- src/rules/mod.rs: register module and example specs
- docs/paper/reductions.typ: full theorem with construction, correctness
  proof, and worked example using the canonical fixture

Closes #363.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: isPANN

docs/paper/reductions.typ

@@ -11690,6 +11690,50 @@ where $P$ is a penalty weight large enough that any constraint violation costs m
   _Solution extraction._ Return the same binary selection vector: element $i$ is in the partition subset if and only if it is selected in the Subset Sum witness.
 ]
 
+#let part_ifwm = load-example("Partition", "IntegralFlowWithMultipliers")
+#let part_ifwm_sol = part_ifwm.solutions.at(0)
+#let part_ifwm_sizes = part_ifwm.source.instance.sizes
+#let part_ifwm_n = part_ifwm_sizes.len()
+#let part_ifwm_total = part_ifwm_sizes.fold(0, (a, b) => a + b)
+#let part_ifwm_half = part_ifwm_total / 2
+#let part_ifwm_selected = part_ifwm_sol.source_config.enumerate().filter(((i, x)) => x == 1).map(((i, x)) => i)
+#let part_ifwm_selected_sizes = part_ifwm_selected.map(i => part_ifwm_sizes.at(i))
+#let part_ifwm_source_arcs = part_ifwm_sol.target_config.slice(0, part_ifwm_n)
+#let part_ifwm_relay_arcs = part_ifwm_sol.target_config.slice(part_ifwm_n, 2 * part_ifwm_n)
+#let part_ifwm_bottleneck = part_ifwm_sol.target_config.at(2 * part_ifwm_n)
+#reduction-rule("Partition", "IntegralFlowWithMultipliers",
+  example: true,
+  example-caption: [#part_ifwm_n elements, total sum $S = #part_ifwm_total$, bottleneck $R = #part_ifwm_half$],
+  extra: [
+    #pred-commands(
+      "pred create --example " + problem-spec(part_ifwm.source) + " -o partition.json",
+      "pred reduce partition.json --to " + target-spec(part_ifwm) + " -o bundle.json",
+      "pred solve bundle.json",
+      "pred evaluate partition.json --config " + part_ifwm_sol.source_config.map(str).join(","),
+    )
+
+    *Step 1 -- Source instance.* The canonical Partition multiset is $(#part_ifwm_sizes.map(str).join(", "))$, so the total is $S = #part_ifwm_total$ and any balanced witness must sum to $S / 2 = #part_ifwm_half$.
+
+    *Step 2 -- Build the relay network.* The reduction creates vertices $s$, one item vertex $v_i$ per element, a relay vertex $w$, and sink $t$. It adds unit-capacity arcs $(s, v_i)$, item arcs $(v_i, w)$ with capacities $(#part_ifwm_sizes.map(str).join(", "))$, and one bottleneck arc $(w, t)$ with capacity $#part_ifwm_half$. The target witness therefore has $#part_ifwm_sol.target_config.len()$ arc-flow coordinates ordered as source arcs, relay arcs, then the bottleneck arc.
+
+    *Step 3 -- Verify the canonical witness.* The source witness $bold(x) = (#part_ifwm_sol.source_config.map(str).join(", "))$ selects item indices $\{#part_ifwm_selected.map(str).join(", ")\}$ with sizes $(#part_ifwm_selected_sizes.map(str).join(", "))$, summing to $#part_ifwm_half$. On the target side, the source arcs carry $(#part_ifwm_source_arcs.map(str).join(", "))$, the relay arcs carry $(#part_ifwm_relay_arcs.map(str).join(", "))$, and the bottleneck arc carries $#part_ifwm_bottleneck$. Thus the relay receives $#part_ifwm_selected_sizes.map(str).join(" + ") = #part_ifwm_half$ units and the sink inflow equals the requirement #sym.checkmark.
+
+    *Witness semantics.* The fixture stores one canonical balanced subset. Other balanced subsets may exist, but every feasible target witness still extracts by reading the first $n$ unit-capacity source arcs.
+  ],
+)[
+  This $O(n)$ reduction @sahni1974 @garey1979[ND33] implements Sahni's multiplier-flow gadget for subset selection. Each Partition element becomes an item vertex whose multiplier amplifies a binary source choice into either $0$ or $a_i$ units entering a relay. A single bottleneck arc of capacity $S / 2$ then converts the target model's sink condition "net inflow at least $R$" into the exact equality needed by Partition.
+][
+  _Construction._ Let the source multiset be $A = {a_1, dots, a_n}$ with total sum $S = sum_(i=1)^n a_i$. If $S$ is odd, return a fixed infeasible Integral Flow With Multipliers instance with vertices $(s, u, t)$, arcs $(s, u)$ and $(u, t)$ both of capacity $1$, multiplier $h(u) = 2$, and requirement $R = 1$. Otherwise set $M = S / 2$ and build a directed graph with vertices $s, v_1, dots, v_n, w, t$. Add arcs $(s, v_i)$ of capacity $1$ and arcs $(v_i, w)$ of capacity $a_i$ for each $i in {1, dots, n}$, plus one bottleneck arc $(w, t)$ of capacity $M$. Assign multipliers $h(v_i) = a_i$ and $h(w) = 1$, and set the sink requirement to $R = M$.
+
+  _Correctness._ ($arrow.r.double$) If the Partition instance is satisfiable, choose a subset $I subset.eq {1, dots, n}$ with $sum_(i in I) a_i = S / 2 = M$. Send one unit on $(s, v_i)$ for each $i in I$ and zero otherwise. Multiplier conservation at each item vertex forces $f(v_i, w) = a_i$ when $i in I$ and $0$ otherwise. The relay multiplier is $1$, so the total flow on $(w, t)$ becomes $sum_(i in I) a_i = M$, which respects the bottleneck capacity and meets the sink requirement $R = M$. When $S$ is odd, the source instance is unsatisfiable and the fixed target is also infeasible because conservation at $u$ would require $f(u, t) = 2 f(s, u)$ while the arc capacity is only $1$.
+
+  ($arrow.l.double$) Suppose the target instance is feasible. In the odd branch the fixed target is infeasible, so only the even branch can yield a witness. Every arc $(s, v_i)$ has capacity $1$, hence integrality forces $f(s, v_i) in {0, 1}$. Conservation at $v_i$ gives $f(v_i, w) = a_i f(s, v_i)$, so each item contributes either $0$ or exactly $a_i$ units to the relay. Conservation at $w$ with multiplier $1$ gives
+  $ f(w, t) = sum_(i=1)^n a_i f(s, v_i). $
+  The bottleneck capacity enforces $f(w, t) <= M$, while the sink requirement enforces $f(w, t) >= R = M$. Therefore $f(w, t) = M = S / 2$, and the indices with $f(s, v_i) = 1$ form a balanced partition subset.
+
+  _Solution extraction._ Read the first $n$ arc-flow coordinates, corresponding to the unit-capacity arcs $(s, v_1), dots, (s, v_n)$. Output bit $x_i = f(s, v_i)$. In the odd branch, return the all-zero source vector.
+]
+
 // Removed: Partition → ShortestWeightConstrainedPath (unsound reduction, #1006)
 
 #let ks_qubo = load-example("Knapsack", "QUBO")

src/rules/mod.rs

@@ -101,6 +101,7 @@ pub(crate) mod naesatisfiability_setsplitting;
 pub(crate) mod paintshop_qubo;
 pub(crate) mod partition_binpacking;
 pub(crate) mod partition_cosineproductintegration;
+pub(crate) mod partition_integralflowwithmultipliers;
 pub(crate) mod partition_knapsack;
 pub(crate) mod partition_multiprocessorscheduling;
 pub(crate) mod partition_openshopscheduling;
@@ -447,6 +448,7 @@ pub(crate) fn canonical_rule_example_specs() -> Vec<crate::example_db::specs::Ru
     specs.extend(minimummultiwaycut_qubo::canonical_rule_example_specs());
     specs.extend(paintshop_qubo::canonical_rule_example_specs());
     specs.extend(partition_cosineproductintegration::canonical_rule_example_specs());
+    specs.extend(partition_integralflowwithmultipliers::canonical_rule_example_specs());
     specs.extend(partition_knapsack::canonical_rule_example_specs());
     specs.extend(partition_openshopscheduling::canonical_rule_example_specs());
     specs.extend(partition_productionplanning::canonical_rule_example_specs());

src/rules/partition_integralflowwithmultipliers.rs

@@ -0,0 +1,126 @@
+//! Reduction from Partition to IntegralFlowWithMultipliers.
+//!
+//! For an even total sum `S`, this is Sahni's multiplier-flow gadget:
+//! items are binary source choices amplified by vertex multipliers and merged
+//! through a single bottleneck arc of capacity `S / 2`. For an odd total sum,
+//! the reduction returns a fixed infeasible target instance.
+
+use crate::models::graph::IntegralFlowWithMultipliers;
+use crate::models::misc::Partition;
+use crate::reduction;
+use crate::rules::traits::{ReduceTo, ReductionResult};
+use crate::topology::DirectedGraph;
+
+/// Result of reducing Partition to IntegralFlowWithMultipliers.
+#[derive(Debug, Clone)]
+pub struct ReductionPartitionToIntegralFlowWithMultipliers {
+    target: IntegralFlowWithMultipliers,
+    source_n: usize,
+    item_arc_count: usize,
+}
+
+impl ReductionResult for ReductionPartitionToIntegralFlowWithMultipliers {
+    type Source = Partition;
+    type Target = IntegralFlowWithMultipliers;
+
+    fn target_problem(&self) -> &Self::Target {
+        &self.target
+    }
+
+    fn extract_solution(&self, target_solution: &[usize]) -> Vec<usize> {
+        if self.item_arc_count == 0 {
+            return vec![0; self.source_n];
+        }
+
+        if target_solution.len() < self.item_arc_count {
+            return vec![0; self.source_n];
+        }
+
+        target_solution[..self.item_arc_count].to_vec()
+    }
+}
+
+#[reduction(overhead = {
+    num_vertices = "num_elements + 3",
+    num_arcs = "2 * num_elements + 1",
+    max_capacity = "total_sum",
+    requirement = "total_sum",
+})]
+impl ReduceTo<IntegralFlowWithMultipliers> for Partition {
+    type Result = ReductionPartitionToIntegralFlowWithMultipliers;
+
+    fn reduce_to(&self) -> Self::Result {
+        let total_sum = self.total_sum();
+        let source_n = self.num_elements();
+
+        if !total_sum.is_multiple_of(2) {
+            let graph = DirectedGraph::new(3, vec![(0, 1), (1, 2)]);
+            return ReductionPartitionToIntegralFlowWithMultipliers {
+                target: IntegralFlowWithMultipliers::new(graph, 0, 2, vec![1, 2, 1], vec![1, 1], 1),
+                source_n,
+                item_arc_count: 0,
+            };
+        }
+
+        let half_sum = total_sum / 2;
+        let relay = source_n + 1;
+        let sink = source_n + 2;
+
+        let mut arcs = Vec::with_capacity(2 * source_n + 1);
+        let mut capacities = Vec::with_capacity(2 * source_n + 1);
+        let mut multipliers = vec![1; source_n + 3];
+
+        for (index, &size) in self.sizes().iter().enumerate() {
+            let item_vertex = index + 1;
+            arcs.push((0, item_vertex));
+            capacities.push(1);
+            multipliers[item_vertex] = size;
+        }
+
+        for (index, &size) in self.sizes().iter().enumerate() {
+            let item_vertex = index + 1;
+            arcs.push((item_vertex, relay));
+            capacities.push(size);
+        }
+
+        arcs.push((relay, sink));
+        capacities.push(half_sum);
+        multipliers[relay] = 1;
+
+        let graph = DirectedGraph::new(source_n + 3, arcs);
+        ReductionPartitionToIntegralFlowWithMultipliers {
+            target: IntegralFlowWithMultipliers::new(
+                graph,
+                0,
+                sink,
+                multipliers,
+                capacities,
+                half_sum,
+            ),
+            source_n,
+            item_arc_count: source_n,
+        }
+    }
+}
+
+#[cfg(feature = "example-db")]
+pub(crate) fn canonical_rule_example_specs() -> Vec<crate::example_db::specs::RuleExampleSpec> {
+    use crate::export::SolutionPair;
+
+    vec![crate::example_db::specs::RuleExampleSpec {
+        id: "partition_to_integralflowwithmultipliers",
+        build: || {
+            crate::example_db::specs::rule_example_with_witness::<_, IntegralFlowWithMultipliers>(
+                Partition::new(vec![2, 3, 4, 5, 6, 4]),
+                SolutionPair {
+                    source_config: vec![1, 0, 1, 0, 1, 0],
+                    target_config: vec![1, 0, 1, 0, 1, 0, 2, 0, 4, 0, 6, 0, 12],
+                },
+            )
+        },
+    }]
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/rules/partition_integralflowwithmultipliers.rs"]
+mod tests;

src/unit_tests/rules/partition_integralflowwithmultipliers.rs

@@ -0,0 +1,104 @@
+use super::*;
+use crate::models::graph::IntegralFlowWithMultipliers;
+use crate::models::misc::Partition;
+use crate::rules::test_helpers::assert_satisfaction_round_trip_from_satisfaction_target;
+use crate::solvers::BruteForce;
+
+#[test]
+fn test_partition_to_integralflowwithmultipliers_closed_loop() {
+    let source = Partition::new(vec![1, 2, 3]);
+    let reduction = ReduceTo::<IntegralFlowWithMultipliers>::reduce_to(&source);
+
+    assert_satisfaction_round_trip_from_satisfaction_target(
+        &source,
+        &reduction,
+        "Partition -> IntegralFlowWithMultipliers closed loop",
+    );
+}
+
+#[test]
+fn test_partition_to_integralflowwithmultipliers_structure_even_total() {
+    let source = Partition::new(vec![2, 3, 4, 5, 6, 4]);
+    let reduction = ReduceTo::<IntegralFlowWithMultipliers>::reduce_to(&source);
+    let target = reduction.target_problem();
+
+    assert_eq!(target.num_vertices(), 9);
+    assert_eq!(
+        target.graph().arcs(),
+        vec![
+            (0, 1),
+            (0, 2),
+            (0, 3),
+            (0, 4),
+            (0, 5),
+            (0, 6),
+            (1, 7),
+            (2, 7),
+            (3, 7),
+            (4, 7),
+            (5, 7),
+            (6, 7),
+            (7, 8),
+        ]
+    );
+    assert_eq!(
+        target.capacities(),
+        &[1, 1, 1, 1, 1, 1, 2, 3, 4, 5, 6, 4, 12]
+    );
+    assert_eq!(target.multipliers(), &[1, 2, 3, 4, 5, 6, 4, 1, 1]);
+    assert_eq!(target.requirement(), 12);
+}
+
+#[test]
+fn test_partition_to_integralflowwithmultipliers_even_no_instance_uses_bottleneck() {
+    let source = Partition::new(vec![3, 5]);
+    let reduction = ReduceTo::<IntegralFlowWithMultipliers>::reduce_to(&source);
+    let target = reduction.target_problem();
+
+    assert_eq!(target.capacities(), &[1, 1, 3, 5, 4]);
+    assert!(BruteForce::new().find_witness(target).is_none());
+}
+
+#[test]
+fn test_partition_to_integralflowwithmultipliers_odd_total_is_fixed_no_instance() {
+    let source = Partition::new(vec![1, 2]);
+    let reduction = ReduceTo::<IntegralFlowWithMultipliers>::reduce_to(&source);
+    let target = reduction.target_problem();
+
+    assert_eq!(target.num_vertices(), 3);
+    assert_eq!(target.graph().arcs(), vec![(0, 1), (1, 2)]);
+    assert_eq!(target.multipliers(), &[1, 2, 1]);
+    assert_eq!(target.capacities(), &[1, 1]);
+    assert_eq!(target.requirement(), 1);
+    assert!(BruteForce::new().find_witness(target).is_none());
+    assert_eq!(reduction.extract_solution(&[]), vec![0, 0]);
+}
+
+#[test]
+fn test_partition_to_integralflowwithmultipliers_extract_solution() {
+    let source = Partition::new(vec![2, 3, 4, 5, 6, 4]);
+    let reduction = ReduceTo::<IntegralFlowWithMultipliers>::reduce_to(&source);
+
+    assert_eq!(
+        reduction.extract_solution(&[1, 0, 1, 0, 1, 0, 2, 0, 4, 0, 6, 0, 12]),
+        vec![1, 0, 1, 0, 1, 0]
+    );
+}
+
+#[cfg(feature = "example-db")]
+#[test]
+fn test_partition_to_integralflowwithmultipliers_canonical_example_spec() {
+    let example = (canonical_rule_example_specs()
+        .into_iter()
+        .find(|spec| spec.id == "partition_to_integralflowwithmultipliers")
+        .expect("canonical example spec should exist")
+        .build)();
+
+    assert_eq!(example.solutions.len(), 1);
+    let solution = &example.solutions[0];
+    assert_eq!(solution.source_config, vec![1, 0, 1, 0, 1, 0]);
+    assert_eq!(
+        solution.target_config,
+        vec![1, 0, 1, 0, 1, 0, 2, 0, 4, 0, 6, 0, 12]
+    );
+}