Fix FileAggregator resource leak

Track opened files in InterpreterState and close them in process_prog finally block.

Signed-off-by: Louis Mandel <lmandel@us.ibm.com>

Author: mandel

src/pdl/pdl_interpreter.py

@@ -280,10 +280,20 @@ def process_prog(
         {"stdlib": stdlib_dict, "pdl_usage": state.llm_usage}
     )
 
-    result, document, final_scope, trace = process_block(
-        state, stdlib_scope, block=prog.root, loc=loc
-    )
-    return result, document, final_scope, trace
+    try:
+        result, document, final_scope, trace = process_block(
+            state, stdlib_scope, block=prog.root, loc=loc
+        )
+        return result, document, final_scope, trace
+    finally:
+        # Close all opened files
+        for fp in state.opened_files:
+            try:
+                if not fp.closed:
+                    fp.close()
+            except Exception:
+                # Ignore errors during cleanup
+                pass  # nosec B110
 
 
 def process_block(
@@ -2657,6 +2667,7 @@ def process_aggregator(
             fp = open(  # pylint: disable=consider-using-with
                 file, mode=mode, encoding=encoding
             )
+            state.opened_files.append(fp)  # Track for cleanup
             aggregator = FileAggregator(fp, prefix=prefix, suffix=suffix, flush=flush)
         case _:
             assert False, "Unexpected aggregator"

src/pdl/pdl_interpreter_state.py

@@ -1,6 +1,6 @@
 from asyncio import AbstractEventLoop
 from pathlib import Path
-from typing import Any
+from typing import IO, Any
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -56,6 +56,8 @@ class InterpreterState(BaseModel):
     """Dictionary that associate runtime block ids with their values to be able to replay an execution."""
     llm_usage: PdlUsage = PdlUsage()
     """Data structure where to accumulate LLMs usage."""
+    opened_files: list[IO[Any]] = Field(default_factory=list)
+    """List of file handles opened during execution that need to be closed."""
 
     def with_yield_result(self: "InterpreterState", b: bool) -> "InterpreterState":
         return self.model_copy(update={"yield_result": b})