fix(runtime): reduce memory growth in long-running multi-simulation workloads

assume/common/outputs.py

@@ -658,6 +658,13 @@ async def on_stop(self):
         # insert left records into db
         await self.store_dfs()
 
+        # Dispose the SQLAlchemy engine to release pooled connections
+        if self.db is not None:
+            try:
+                self.db.dispose()
+            except Exception:
+                pass
+
         if self.db is None:
             return
 

assume/common/units_operator.py

@@ -234,7 +234,9 @@ def handle_market_feedback(self, content: ClearingMessage, meta: MetaDict) -> No
             order["market_id"] = content["market_id"]
 
         marketconfig = self.registered_markets[content["market_id"]]
-        self.valid_orders[marketconfig.product_type].extend(orderbook)
+        # Only keep accepted orders for dispatch tracking —
+        # rejected orders are not needed and cause unbounded growth.
+        self.valid_orders[marketconfig.product_type].extend(accepted_orders)
         self.set_unit_dispatch(orderbook, marketconfig)
         self.write_actual_dispatch(marketconfig.product_type)
 

assume/common/utils.py

@@ -737,16 +737,52 @@ def parse_duration(duration_str):
 
 def calculate_content_size(content: list | dict) -> int:
     """
-    Calculate the size of a content in bytes.
+    Calculate the deep memory size of content in bytes.
+
+    Uses recursive traversal to account for nested structures,
+    DataFrames, numpy arrays, and torch tensors — unlike sys.getsizeof
+    which only measures the shallow container size.
     """
-    if isinstance(content, dict):  # For dictionaries
+    try:
+        import pandas as pd
+
+        if isinstance(content, pd.DataFrame):
+            return int(content.memory_usage(deep=True).sum())
+        if isinstance(content, pd.Series):
+            return int(content.memory_usage(deep=True))
+    except Exception:
+        pass
+
+    try:
+        import numpy as np
+
+        if isinstance(content, np.ndarray):
+            return content.nbytes
+    except Exception:
+        pass
+
+    try:
+        import torch
+
+        if isinstance(content, torch.Tensor):
+            return content.nelement() * content.element_size()
+    except Exception:
+        pass
+
+    if isinstance(content, dict):
         return sys.getsizeof(content) + sum(
-            sys.getsizeof(value) for value in content.values()
+            calculate_content_size(k) + calculate_content_size(v)
+            for k, v in content.items()
         )
-    elif isinstance(content, list):  # For lists, including lists of dicts
+    elif isinstance(content, (list, tuple)):
         return sys.getsizeof(content) + sum(
             calculate_content_size(item) for item in content
         )
+    elif isinstance(content, str):
+        return sys.getsizeof(content)
+    elif isinstance(content, (int, float, bool, type(None))):
+        return sys.getsizeof(content)
+    # Fallback for unknown types
     return sys.getsizeof(content)
 
 

assume/markets/base_market.py

@@ -670,7 +670,7 @@ async def clear_market(self, market_products: list[MarketProduct]):
                         0,
                     )
 
-        self.open_auctions - set(market_products)
+        self.open_auctions -= set(market_products)
 
         accepted_orderbook = sorted(
             accepted_orderbook, key=lambda x: str(x["agent_addr"])
@@ -725,6 +725,12 @@ async def clear_market(self, market_products: list[MarketProduct]):
 
         await self.store_market_results(market_meta)
 
+        # Prevent unbounded growth of in-memory results.
+        # Keep only results that data-request handlers may still need
+        # (those whose product_start has not yet passed).
+        now = timestamp2datetime(self.context.current_timestamp)
+        self.results = [r for r in self.results if r.get("product_start", now) >= now]
+
         if flows is not None and len(flows) > 0:
             await self.store_flows(flows)
 

assume/reinforcement_learning/learning_role.py

@@ -616,6 +616,14 @@ def init_logging(
             train_start (str): The start time of simulation.
         """
 
+        # Explicitly close the previous logger to release its DB engine
+        # and SummaryWriter file handles before creating a new one.
+        if (
+            hasattr(self, "tensor_board_logger")
+            and self.tensor_board_logger is not None
+        ):
+            self.tensor_board_logger.close()
+
         self.tensor_board_logger = TensorBoardLogger(
             simulation_id=simulation_id,
             db_uri=db_uri,

assume/reinforcement_learning/tensorboard_logger.py

@@ -363,10 +363,28 @@ def update_tensorboard(self):
             logger.error(f"Unexpected error in update_tensorboard: {e}")
             return
 
+    def close(self):
+        """
+        Explicitly release resources held by this logger.
+        Call this before discarding the instance to avoid leaked
+        file handles (SummaryWriter) and DB connection pools.
+        """
+        if hasattr(self, "writer") and self.writer is not None:
+            try:
+                self.writer.flush()
+                self.writer.close()
+            except Exception:
+                pass
+            self.writer = None
+        if hasattr(self, "db") and self.db is not None:
+            try:
+                self.db.dispose()
+            except Exception:
+                pass
+            self.db = None
+
     def __del__(self):
         """
         Deletes the WriteOutput instance.
         """
-        if hasattr(self, "writer") and self.writer is not None:
-            self.writer.flush()
-            self.writer.close()
+        self.close()

assume/world.py

@@ -864,6 +864,7 @@ def run(self):
     def reset(self):
         """
         Reset the market operators, markets, unit operators, and forecast providers to empty dictionaries.
+        Also clears caches and large data structures to prevent memory leaks across episodes.
 
         Returns:
             None
@@ -874,6 +875,26 @@ def reset(self):
         self.units = {}
         self.forecast_providers = {}
 
+        # Clear forecast algorithm lru_caches that retain full DataFrames
+        # and references to old unit/config objects from previous episodes.
+        from assume.common.forecast_algorithms import (
+            calculate_naive_congestion_signal,
+            calculate_naive_price,
+            calculate_naive_price_elastic,
+            calculate_naive_price_inelastic,
+            calculate_naive_renewable_utilisation,
+            calculate_naive_residual_load,
+            sort_units,
+        )
+
+        sort_units.cache_clear()
+        calculate_naive_price_inelastic.cache_clear()
+        calculate_naive_price_elastic.cache_clear()
+        calculate_naive_price.cache_clear()
+        calculate_naive_residual_load.cache_clear()
+        calculate_naive_congestion_signal.cache_clear()
+        calculate_naive_renewable_utilisation.cache_clear()
+
     def add_unit(
         self,
         id: str,

docs/source/release_notes.rst

@@ -25,6 +25,7 @@ Upcoming Release
   - **Fix bug in forecasts**, that occurred when using complex clearing
   - **Fix infeasible power output in PowerPlant**: ``calculate_min_max_power`` now correctly accounts for base load, positive/negative capacity reserves, and heat demand when computing additional power. If reduced availability makes the unit infeasible to run, both min and max power are set to 0. A warning is issued if previous dispatch exceeded available power.
   - **Fix upward redispatch potential**, so that availabilities are now correctly considered instead of the nominal power output of the unit
+  - **Reduce memory growth in long-running workflows**: Several likely object-retention points were cleaned up so that running many simulations in parallel for long periods no longer accumulates memory as aggressively. Functional behaviour is intended to remain unchanged.
 
 0.6.1 - (25th March 2026)
 =========================