fix(backends): bound temporary scratch files (#11941)

Backend processes shared the host temporary directory, so crashes could leave request images and audio behind until the filesystem filled. Give each process a locked LocalAI-owned runtime, remove scratch on exit, and sweep only marked abandoned runtimes at the next start.

Also close known request error-path leaks in the Python media backends, CrispASR, LongCat Video, and stable-diffusion.cpp.

Assisted-by: Codex:gpt-5

Co-authored-by: Ettore Di Giacinto <mudler@localai.io>

backend/python/common/temp_utils.py

@@ -0,0 +1,36 @@
+import base64
+import contextlib
+import os
+import tempfile
+
+
+@contextlib.contextmanager
+def materialize_base64(data, suffix=""):
+    """Materialize base64 data for a path-only library and always remove it."""
+    descriptor, path = tempfile.mkstemp(prefix="localai-media-", suffix=suffix)
+    try:
+        with os.fdopen(descriptor, "wb") as output:
+            descriptor = None
+            output.write(base64.b64decode(data))
+        yield path
+    finally:
+        if descriptor is not None:
+            os.close(descriptor)
+        try:
+            os.remove(path)
+        except OSError:
+            pass
+
+
+@contextlib.contextmanager
+def cleanup_paths():
+    """Collect temporary paths and remove them on success or failure."""
+    paths = []
+    try:
+        yield paths
+    finally:
+        for path in paths:
+            try:
+                os.remove(path)
+            except OSError:
+                pass

backend/python/common/temp_utils_test.py

@@ -0,0 +1,41 @@
+import os
+import tempfile
+import unittest
+from unittest import mock
+
+from temp_utils import cleanup_paths, materialize_base64
+
+
+class MaterializeBase64Test(unittest.TestCase):
+    def test_removes_materialized_file_after_success(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with mock.patch.object(tempfile, "tempdir", directory):
+                with materialize_base64("aGVsbG8=", suffix=".data") as path:
+                    with open(path, "rb") as materialized:
+                        self.assertEqual(materialized.read(), b"hello")
+                self.assertFalse(os.path.exists(path))
+
+    def test_removes_materialized_file_when_consumer_fails(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with mock.patch.object(tempfile, "tempdir", directory):
+                with self.assertRaisesRegex(RuntimeError, "decode failed"):
+                    with materialize_base64("aGVsbG8="):
+                        raise RuntimeError("decode failed")
+                self.assertEqual(os.listdir(directory), [])
+
+
+class CleanupPathsTest(unittest.TestCase):
+    def test_removes_every_registered_path_after_failure(self):
+        with tempfile.TemporaryDirectory() as directory:
+            paths = [os.path.join(directory, name) for name in ("one.wav", "two.wav")]
+            with self.assertRaisesRegex(RuntimeError, "merge failed"):
+                with cleanup_paths() as registered:
+                    for path in paths:
+                        open(path, "wb").close()
+                        registered.append(path)
+                    raise RuntimeError("merge failed")
+            self.assertEqual(os.listdir(directory), [])
+
+
+if __name__ == "__main__":
+    unittest.main()