fix(coglet): clean up spill files and make output ownership explicit

architecture/04-container-runtime.md

@@ -179,7 +179,7 @@ Per-prediction data. Using separate sockets per slot avoids head-of-line blockin
 | ---------------------------------------------- | ---------------------------------------------------------------- |
 | `Log { source, data }`                         | Log line from run()                                              |
 | `Output { output }`                            | Yielded output value (for generators/streaming)                  |
-| `FileOutput { filename, kind, mime_type }`     | File produced by run() -- referenced by path, uploaded by parent |
+| `FileOutput { filename, kind, mime_type, upload_filename }` | Owned output snapshot referenced by path; optional original basename for uploads |
 | `Metric { name, value, mode }`                 | Custom metric (mode: `replace`, `increment`, or `append`)        |
 | `Done { id, output, predict_time, is_stream }` | Prediction completed successfully                                |
 | `Failed { id, error }`                         | Prediction failed                                                |
@@ -373,6 +373,8 @@ When a prediction input exceeds 6MiB, it's too large to send inline through the
 
 When run() produces file outputs (`cog.Path`), the worker sends a `FileOutput` message with the filename and MIME type. The parent handles uploading the file (or base64-encoding it for inline responses). The `output_dir` field in the `Predict` request tells the worker where to write output files.
 
+The worker snapshots a returned or yielded path into a unique temporary file in `output_dir` before resuming user code, so a later overwrite can't change an earlier streamed output. Async generators are collected before any item is sent, so their paths are snapshotted and deleted after the generator finishes. After the snapshot is sent, the worker deletes the source only if its directory resolves under the system temp dir; files elsewhere are left alone. Paths in a list, tuple, or nested output are deleted after the whole output is sent, so the same path can appear more than once. The parent deletes the snapshot right after reading it, before the upload starts; uploads use the bytes in memory. Cleanup removes files only, never directories.
+
 `FileOutputKind` distinguishes between normal file outputs (`FileType`) and oversized outputs (`Oversized`) that exceeded an inline size limit.
 
 ## Custom Metrics

crates/Cargo.toml

@@ -34,6 +34,9 @@ serde_json = "1"
 # Identifiers
 uuid = { version = "1", features = ["v4", "serde"] }
 
+# Filesystem
+tempfile = "3"
+
 # Observability
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }

crates/coglet-python/Cargo.toml

@@ -31,7 +31,7 @@ libc = "0.2"
 
 [dev-dependencies]
 pyo3 = { workspace = true, features = ["auto-initialize"] }
-tempfile = "3"
+tempfile.workspace = true
 
 [features]
 default = ["tracing", "tracing-grpc"]

crates/coglet-python/src/output.rs

@@ -12,27 +12,45 @@
 //!
 //! This replaces the Python modules cog.json and cog.files.
 
+use std::collections::HashSet;
+use std::path::PathBuf;
+
+use coglet_core::worker::release_output_path;
 use pyo3::prelude::*;
 use pyo3::types::{PyDict, PyFrozenSet, PyList, PySet, PyString, PyTuple};
 
 /// Process prediction output for JSON serialization.
 ///
 /// Calls make_encodeable() to normalize, then encode_files() to convert any
-/// remaining Path/IOBase objects to base64 data URLs.
+/// remaining Path/IOBase objects to base64 data URLs. Encoded paths are
+/// released once the whole output is encoded, so repeated paths work.
 pub fn process_output<'py>(
     py: Python<'py>,
     output: &Bound<'py, PyAny>,
 ) -> PyResult<Bound<'py, PyAny>> {
-    let encodeable = make_encodeable(py, output)?;
-    encode_files(py, &encodeable)
+    let mut paths = HashSet::new();
+    let encoded = process_output_item(py, output, &mut paths)?;
+    release_output_paths(paths);
+    Ok(encoded)
 }
 
-/// Process a single output item (for generator outputs).
+/// Process a single output item, collecting encoded paths instead of
+/// releasing them. Callers release `paths` after the last item that might
+/// reference them.
 pub fn process_output_item<'py>(
     py: Python<'py>,
     item: &Bound<'py, PyAny>,
+    paths: &mut HashSet<PathBuf>,
 ) -> PyResult<Bound<'py, PyAny>> {
-    process_output(py, item)
+    let encodeable = make_encodeable(py, item)?;
+    encode_files(py, &encodeable, paths)
+}
+
+/// Release every collected output path. See [`release_output_path`].
+pub fn release_output_paths(paths: HashSet<PathBuf>) {
+    for path in paths {
+        release_output_path(&path);
+    }
 }
 
 /// Normalize a Python object into a JSON-friendly form.
@@ -142,7 +160,11 @@ fn make_encodeable<'py>(py: Python<'py>, obj: &Bound<'py, PyAny>) -> PyResult<Bo
 }
 
 /// Recursively walk the output and encode any Path/IOBase objects to base64 data URLs.
-fn encode_files<'py>(py: Python<'py>, obj: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+fn encode_files<'py>(
+    py: Python<'py>,
+    obj: &Bound<'py, PyAny>,
+    paths: &mut HashSet<PathBuf>,
+) -> PyResult<Bound<'py, PyAny>> {
     // str -- return as-is (don't recurse into characters)
     if obj.is_instance_of::<PyString>() {
         return Ok(obj.clone());
@@ -152,7 +174,7 @@ fn encode_files<'py>(py: Python<'py>, obj: &Bound<'py, PyAny>) -> PyResult<Bound
     if let Ok(dict) = obj.cast_exact::<PyDict>() {
         let new_dict = PyDict::new(py);
         for (key, value) in dict.iter() {
-            new_dict.set_item(&key, encode_files(py, &value)?)?;
+            new_dict.set_item(&key, encode_files(py, &value, paths)?)?;
         }
         return Ok(new_dict.into_any());
     }
@@ -161,7 +183,7 @@ fn encode_files<'py>(py: Python<'py>, obj: &Bound<'py, PyAny>) -> PyResult<Bound
     if let Ok(list) = obj.cast_exact::<PyList>() {
         let items: Vec<Bound<'py, PyAny>> = list
             .iter()
-            .map(|item| encode_files(py, &item))
+            .map(|item| encode_files(py, &item, paths))
             .collect::<PyResult<_>>()?;
         let new_list = PyList::new(py, &items)?;
         return Ok(new_list.into_any());
@@ -171,11 +193,21 @@ fn encode_files<'py>(py: Python<'py>, obj: &Bound<'py, PyAny>) -> PyResult<Bound
     let os_mod = py.import("os")?;
     let pathlike_cls = os_mod.getattr("PathLike")?;
     if obj.is_instance(&pathlike_cls)? {
+        let path = obj.extract::<PathBuf>()?;
+        // Check before opening so a FIFO or device can't block the open.
+        if !std::fs::metadata(&path)?.is_file() {
+            return Err(pyo3::exceptions::PyValueError::new_err(format!(
+                "output path must be a regular file: {}",
+                path.display()
+            )));
+        }
         let builtins = py.import("builtins")?;
         let fh = builtins.getattr("open")?.call1((obj, "rb"))?;
         let result = file_to_base64(py, &fh);
         fh.call_method0("close")?;
-        return result;
+        let encoded = result?;
+        paths.insert(path);
+        return Ok(encoded);
     }
 
     // io.IOBase -> base64 encode
@@ -442,6 +474,120 @@ mod tests {
 
     // ── encode_files / file_to_base64 ────────────────────────────────
 
+    #[test]
+    fn nested_repeated_paths_are_encoded_then_consumed() {
+        let root = tempfile::tempdir().unwrap();
+        let file = root.path().join("result.txt");
+        std::fs::write(&file, b"nested").unwrap();
+        pyo3::Python::initialize();
+        Python::attach(|py| {
+            let path = py
+                .import("pathlib")
+                .unwrap()
+                .getattr("Path")
+                .unwrap()
+                .call1((&file,))
+                .unwrap();
+            let obj = PyDict::new(py);
+            obj.set_item("first", &path).unwrap();
+            obj.set_item("second", PyList::new(py, [&path]).unwrap())
+                .unwrap();
+            let encoded = process_output(py, obj.as_any()).unwrap();
+            assert_eq!(
+                encoded
+                    .get_item("first")
+                    .unwrap()
+                    .extract::<String>()
+                    .unwrap(),
+                "data:text/plain;base64,bmVzdGVk"
+            );
+            assert_eq!(
+                encoded
+                    .get_item("second")
+                    .unwrap()
+                    .get_item(0)
+                    .unwrap()
+                    .extract::<String>()
+                    .unwrap(),
+                "data:text/plain;base64,bmVzdGVk"
+            );
+            assert!(!file.exists());
+            assert!(
+                process_output(py, &path).is_err(),
+                "a stale path must not be reused"
+            );
+        });
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn nested_symlink_preserves_target() {
+        let root = tempfile::tempdir().unwrap();
+        let target = root.path().join("asset.txt");
+        let link = root.path().join("result.txt");
+        std::fs::write(&target, b"asset").unwrap();
+        std::os::unix::fs::symlink("asset.txt", &link).unwrap();
+        pyo3::Python::initialize();
+        Python::attach(|py| {
+            let path = py
+                .import("pathlib")
+                .unwrap()
+                .getattr("Path")
+                .unwrap()
+                .call1((&link,))
+                .unwrap();
+            process_output(py, &path).unwrap();
+        });
+        assert!(!link.exists());
+        assert_eq!(std::fs::read(target).unwrap(), b"asset");
+    }
+
+    #[test]
+    fn nested_path_outside_temp_dir_is_kept() {
+        let root = tempfile::Builder::new()
+            .tempdir_in(env!("CARGO_MANIFEST_DIR"))
+            .unwrap();
+        let file = root.path().join("asset.txt");
+        std::fs::write(&file, b"asset").unwrap();
+        pyo3::Python::initialize();
+        Python::attach(|py| {
+            let path = py
+                .import("pathlib")
+                .unwrap()
+                .getattr("Path")
+                .unwrap()
+                .call1((&file,))
+                .unwrap();
+            let obj = PyDict::new(py);
+            obj.set_item("file", &path).unwrap();
+            process_output(py, obj.as_any()).unwrap();
+        });
+        assert_eq!(std::fs::read(file).unwrap(), b"asset");
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn nested_fifo_is_rejected_without_blocking() {
+        let root = tempfile::tempdir().unwrap();
+        let fifo = root.path().join("pipe");
+        let c_path = std::ffi::CString::new(fifo.as_os_str().as_encoded_bytes()).unwrap();
+        assert_eq!(unsafe { libc::mkfifo(c_path.as_ptr(), 0o600) }, 0);
+        pyo3::Python::initialize();
+        Python::attach(|py| {
+            let path = py
+                .import("pathlib")
+                .unwrap()
+                .getattr("Path")
+                .unwrap()
+                .call1((&fifo,))
+                .unwrap();
+            let obj = PyDict::new(py);
+            obj.set_item("file", &path).unwrap();
+            assert!(process_output(py, obj.as_any()).is_err());
+        });
+        assert!(fifo.exists());
+    }
+
     #[test]
     fn encode_pathlike_to_base64() {
         let result = processed(

crates/coglet-python/src/predictor.rs

@@ -1,12 +1,14 @@
 //! Python predictor loading and invocation.
 
+use std::collections::HashSet;
+use std::path::PathBuf;
 use std::sync::{Arc, OnceLock};
 
 use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 use pyo3::types::{PyDict, PyTuple};
 
-use coglet_core::worker::SlotSender;
+use coglet_core::worker::{SlotSender, release_output_path};
 use coglet_core::{PredictionError, PredictionOutput, PredictionResult};
 
 use crate::cancel;
@@ -212,14 +214,18 @@ fn submit_async_coroutine(
 
 /// Send a single output item over IPC, routing file outputs to disk.
 ///
-/// For Path outputs (os.PathLike): sends the existing file path via send_file_output.
+/// For Path outputs (os.PathLike): snapshots the file via send_file_output.
 /// For IOBase outputs: reads bytes, writes to output_dir via write_file_output.
 /// For everything else: processes through make_encodeable + upload_files, then send_output.
+///
+/// Sent paths are added to `paths`; the caller releases them once no later
+/// item can reference them.
 fn send_output_item(
     py: Python<'_>,
     item: &Bound<'_, PyAny>,
     json_module: &Bound<'_, PyAny>,
     slot_sender: &SlotSender,
+    paths: &mut HashSet<PathBuf>,
 ) -> Result<(), PredictionError> {
     let os = py
         .import("os")
@@ -235,14 +241,15 @@ fn send_output_item(
         .map_err(|e| PredictionError::Failed(format!("Failed to get io.IOBase: {}", e)))?;
 
     if item.is_instance(&pathlike).unwrap_or(false) {
-        // Path output — file already on disk, send path reference
         let path_str: String = item
             .call_method0("__fspath__")
             .and_then(|p| p.extract())
             .map_err(|e| PredictionError::Failed(format!("Failed to get fspath: {}", e)))?;
-        slot_sender
-            .send_file_output(std::path::PathBuf::from(path_str), None)
+        let path = PathBuf::from(path_str);
+        // Copying a large file shouldn't block other slots on the GIL.
+        py.detach(|| slot_sender.send_file_output(path.clone(), None))
             .map_err(|e| PredictionError::Failed(format!("Failed to send file output: {}", e)))?;
+        paths.insert(path);
         return Ok(());
     }
 
@@ -281,7 +288,7 @@ fn send_output_item(
     }
 
     // Non-file output - process normally
-    let processed = output::process_output_item(py, item)
+    let processed = output::process_output_item(py, item, paths)
         .map_err(|e| PredictionError::Failed(format!("Failed to process output item: {}", e)))?;
 
     let item_str: String = json_module
@@ -981,7 +988,10 @@ impl PythonPredictor {
                 PredictionError::Failed(format!("Generator iteration error: {}", e))
             })?;
 
-            send_output_item(py, &item, json_module, slot_sender)?;
+            // Release before the generator resumes so it can reuse the path.
+            let mut paths = HashSet::new();
+            send_output_item(py, &item, json_module, slot_sender, &mut paths)?;
+            output::release_output_paths(paths);
         }
 
         // Outputs already streamed over IPC — return empty stream
@@ -1018,11 +1028,12 @@ impl PythonPredictor {
                 .call_method0("__fspath__")
                 .and_then(|p| p.extract())
                 .map_err(|e| PredictionError::Failed(format!("Failed to get fspath: {}", e)))?;
-            slot_sender
-                .send_file_output(std::path::PathBuf::from(path_str), None)
+            let path = PathBuf::from(path_str);
+            py.detach(|| slot_sender.send_file_output(path.clone(), None))
                 .map_err(|e| {
                     PredictionError::Failed(format!("Failed to send file output: {}", e))
                 })?;
+            release_output_path(&path);
             return Ok(PredictionOutput::Single(serde_json::Value::Null));
         }
 
@@ -1060,16 +1071,21 @@ impl PythonPredictor {
         // List/tuple output — iterate items so file outputs (Path, IOBase)
         // go through the FileOutput IPC path for upload instead of being
         // base64-encoded inline by process_output.
+        // Paths are released after the last item so `[p, p]` works.
         if let Ok(list) = result.cast::<pyo3::types::PyList>() {
+            let mut paths = HashSet::new();
             for item in list.iter() {
-                send_output_item(py, &item, json_module, slot_sender)?;
+                send_output_item(py, &item, json_module, slot_sender, &mut paths)?;
             }
+            output::release_output_paths(paths);
             return Ok(PredictionOutput::Stream(vec![]));
         }
         if let Ok(tuple) = result.cast::<pyo3::types::PyTuple>() {
+            let mut paths = HashSet::new();
             for item in tuple.iter() {
-                send_output_item(py, &item, json_module, slot_sender)?;
+                send_output_item(py, &item, json_module, slot_sender, &mut paths)?;
             }
+            output::release_output_paths(paths);
             return Ok(PredictionOutput::Stream(vec![]));
         }
 
@@ -1198,9 +1214,11 @@ impl PythonPredictor {
         let output = if is_async_gen {
             // Result is a pre-collected list — stream each item over IPC
             if let Ok(list) = result.extract::<Vec<Bound<'_, PyAny>>>() {
+                let mut paths = HashSet::new();
                 for item in list {
-                    send_output_item(py, &item, &json_module, slot_sender)?;
+                    send_output_item(py, &item, &json_module, slot_sender, &mut paths)?;
                 }
+                output::release_output_paths(paths);
             }
             PredictionOutput::Stream(vec![])
         } else {