Fix memory leak in appsi LegacySolverInterface wrapper

pyomo/contrib/appsi/base.py

@@ -1601,10 +1601,8 @@ def solve(
         symbol_map.bySymbol = dict(self.symbol_map.bySymbol)
         symbol_map.aliases = dict(self.symbol_map.aliases)
         symbol_map.default_labeler = self.symbol_map.default_labeler
-        model.solutions.add_symbol_map(symbol_map)
-        legacy_results._smap_id = id(symbol_map)
+        legacy_results._smap_id = None
 
-        delete_legacy_soln = True
         if load_solutions:
             if hasattr(model, 'dual') and model.dual.import_enabled():
                 for c, val in results.solution_loader.get_duals().items():
@@ -1616,7 +1614,6 @@ def solve(
                 for v, val in results.solution_loader.get_reduced_costs().items():
                     model.rc[v] = val
         elif results.best_feasible_objective is not None:
-            delete_legacy_soln = False
             for v, val in results.solution_loader.get_primals().items():
                 legacy_soln.variable[symbol_map.getSymbol(v)] = {'Value': val}
             if hasattr(model, 'dual') and model.dual.import_enabled():
@@ -1631,9 +1628,9 @@ def solve(
                 for v, val in results.solution_loader.get_reduced_costs().items():
                     legacy_soln.variable['Rc'] = val
 
-        legacy_results.solution.insert(legacy_soln)
-        if delete_legacy_soln:
-            legacy_results.solution.delete(0)
+            legacy_results.solution.insert(legacy_soln)
+            legacy_results._smap = symbol_map
+            legacy_results._smap_id = id(symbol_map)
 
         self.config = original_config
         self.options = original_options

pyomo/contrib/appsi/tests/test_legacy_leak.py

@@ -0,0 +1,79 @@
+# ____________________________________________________________________________________
+#
+# Pyomo: Python Optimization Modeling Objects
+# Copyright (c) 2008-2026 National Technology and Engineering Solutions of Sandia, LLC
+# Under the terms of Contract DE-NA0003525 with National Technology and Engineering
+# Solutions of Sandia, LLC, the U.S. Government retains certain rights in this
+# software.  This software is distributed under the 3-clause BSD License.
+# ____________________________________________________________________________________
+
+import pyomo.environ as pyo
+import pyomo.common.unittest as unittest
+import gc
+from pyomo.contrib.appsi.cmodel import cmodel_available
+
+from pyomo.common.dependencies import attempt_import
+
+# Note: tracemalloc is not always available, e.g., under PyPy
+tracemalloc, tracemalloc_available = attempt_import('tracemalloc')
+
+
+class TestAppsiLegacyLeak(unittest.TestCase):
+    @unittest.skipIf(not tracemalloc_available, "tracemalloc not available")
+    @unittest.skipIf(not cmodel_available, "APPSI C-extension not available")
+    def test_legacy_solver_wrapper_memory_leak(self):
+        tracemalloc.start()
+        # 1. Create a minimal structure
+        model = pyo.ConcreteModel()
+        model.I = pyo.RangeSet(100)
+        model.x = pyo.Var(model.I)
+        model.obj = pyo.Objective(expr=sum(model.x[i] for i in model.I))
+        model.c = pyo.ConstraintList()
+        for i in model.I:
+            model.c.add(model.x[i] >= 0)
+
+        # 2. Instantiate Legacy Wrapper
+        solver = pyo.SolverFactory('appsi_cbc')
+        if not solver.available():
+            raise unittest.SkipTest("appsi_cbc solver is not available")
+
+        solver.set_instance(model)
+
+        # Warm-up solve
+        solver.solve(model)
+        gc.collect()
+
+        # 3. Take initial memory snapshot
+
+        s1 = tracemalloc.take_snapshot()
+
+        # 4. Perform iterative solves
+        iterations = 10
+        for _ in range(iterations):
+            solver.solve(model)
+
+        gc.collect()
+        s2 = tracemalloc.take_snapshot()
+        tracemalloc.stop()
+
+        # 5. Measure memory delta
+        stats = s2.compare_to(s1, 'lineno')
+        total_leak = sum(stat.size_diff for stat in stats)
+
+        initial_size = sum(stat.size for stat in s1.statistics('lineno'))
+        final_size = sum(stat.size for stat in s2.statistics('lineno'))
+        percentage_increase_per_solve = (total_leak / iterations) / initial_size * 100
+
+        # We allow a small tolerance for memory use growth, set here
+        threshold_pct = 3
+        print(f"Percentage increase per solve: {percentage_increase_per_solve}%")
+        # Check if the leak is substantial
+        self.assertLess(
+            percentage_increase_per_solve,
+            threshold_pct,
+            f"More than {threshold_pct}% memory leak detected across iterations",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

pyomo/contrib/solver/common/base.py

@@ -559,16 +559,8 @@ def _solution_handler(
         """Method to handle the preferred action for the solution"""
         symbol_map = legacy_soln.symbol_map = SymbolMap()
         symbol_map.default_labeler = NumericLabeler('x')
-        if not hasattr(model, 'solutions'):
-            # This logic gets around Issue #2130 in which
-            # solutions is not an attribute on Blocks
-            from pyomo.core.base.PyomoModel import ModelSolutions
-
-            setattr(model, 'solutions', ModelSolutions(model))
-        model.solutions.add_symbol_map(symbol_map)
-        legacy_results._smap = symbol_map
-        legacy_results._smap_id = id(symbol_map)
-        delete_legacy_soln = True
+        legacy_results._smap_id = None
+
         if load_solutions:
             if hasattr(model, 'dual') and model.dual.import_enabled():
                 for con, val in results.solution_loader.get_duals().items():
@@ -577,7 +569,6 @@ def _solution_handler(
                 for var, val in results.solution_loader.get_reduced_costs().items():
                     model.rc[var] = val
         elif results.incumbent_objective is not None:
-            delete_legacy_soln = False
             for var, val in results.solution_loader.get_primals().items():
                 legacy_soln.variable[symbol_map.getSymbol(var)] = {'Value': val}
             if hasattr(model, 'dual') and model.dual.import_enabled():
@@ -587,15 +578,16 @@ def _solution_handler(
                 for var, val in results.solution_loader.get_reduced_costs().items():
                     legacy_soln.variable['Rc'] = val
 
-        legacy_results.solution.insert(legacy_soln)
+            legacy_results.solution.insert(legacy_soln)
+            legacy_results._smap_id = id(symbol_map)
+            legacy_results._smap = symbol_map
+
         # Timing info was not originally on the legacy results, but we
         # want to make it accessible to folks who are utilizing the
         # backwards compatible version.  Note that embedding the
         # ConfigDict broke pickling the legacy_results, so we will only
         # return raw nested dicts
         legacy_results.timing_info = results.timing_info.value()
-        if delete_legacy_soln:
-            legacy_results.solution.delete(0)
         return legacy_results
 
     def solve(