fix: delete coglet-managed output files after upload

crates/coglet-python/src/predictor.rs

@@ -166,9 +166,22 @@ 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): transfers the file to Coglet-managed storage.
 /// 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.
+fn send_path_output(
+    item: &Bound<'_, PyAny>,
+    slot_sender: &SlotSender,
+) -> Result<(), PredictionError> {
+    let path_str: String = item
+        .call_method0("__fspath__")
+        .and_then(|path| path.extract())
+        .map_err(|e| PredictionError::Failed(format!("Failed to get fspath: {}", e)))?;
+    slot_sender
+        .send_user_file_output(std::path::PathBuf::from(path_str), None)
+        .map_err(|e| PredictionError::Failed(format!("Failed to send file output: {}", e)))
+}
+
 fn send_output_item(
     py: Python<'_>,
     item: &Bound<'_, PyAny>,
@@ -189,14 +202,9 @@ 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)
-            .map_err(|e| PredictionError::Failed(format!("Failed to send file output: {}", e)))?;
+        // Returning a Path transfers ownership to Coglet, which moves the data
+        // into managed storage and deletes it after consumption.
+        send_path_output(item, slot_sender)?;
         return Ok(());
     }
 
@@ -949,15 +957,7 @@ impl PythonPredictor {
             .map_err(|e| PredictionError::Failed(format!("Failed to get io.IOBase: {}", e)))?;
 
         if result.is_instance(&pathlike).unwrap_or(false) {
-            let path_str: String = result
-                .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)
-                .map_err(|e| {
-                    PredictionError::Failed(format!("Failed to send file output: {}", e))
-                })?;
+            send_path_output(result, slot_sender)?;
             return Ok(PredictionOutput::Single(serde_json::Value::Null));
         }
 
@@ -1378,6 +1378,7 @@ mod tests {
 
     use std::path::PathBuf;
 
+    use coglet_core::bridge::protocol::SlotResponse;
     use pyo3::types::PyList;
 
     fn add_python_sdk_path(py: Python<'_>) {
@@ -1435,6 +1436,37 @@ sys.modules.setdefault('requests', requests)
         })
     }
 
+    #[test]
+    fn returned_path_transfers_ownership_to_slot_sender() {
+        pyo3::Python::initialize();
+        let source_dir = tempfile::tempdir().unwrap();
+        let source = source_dir.path().join("output.txt");
+        std::fs::write(&source, b"output").unwrap();
+        let output_dir = tempfile::tempdir().unwrap();
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        let sender = SlotSender::new(tx, output_dir.path().to_path_buf());
+
+        Python::attach(|py| {
+            let pathlib = py.import("pathlib").unwrap();
+            let path = pathlib
+                .getattr("Path")
+                .unwrap()
+                .call1((source.to_str().unwrap(),))
+                .unwrap();
+            send_path_output(&path, &sender).unwrap();
+        });
+
+        assert!(!source.exists());
+        match rx.try_recv().unwrap() {
+            SlotResponse::FileOutput {
+                filename,
+                managed: true,
+                ..
+            } => assert_eq!(std::fs::read(filename).unwrap(), b"output"),
+            response => panic!("expected managed FileOutput, got {response:?}"),
+        }
+    }
+
     #[test]
     fn class_with_run_loads() {
         let predictor = load_predictor_source(

crates/coglet/src/bridge/protocol.rs

@@ -327,6 +327,9 @@ pub enum SlotResponse {
         /// Explicit MIME type from the predictor. Falls back to mime_guess when None.
         #[serde(skip_serializing_if = "Option::is_none")]
         mime_type: Option<String>,
+        /// True if Coglet owns this file and may delete it after consumption.
+        #[serde(default)]
+        managed: bool,
     },
 
     /// Streaming output chunk for generator and iterator output.
@@ -592,6 +595,39 @@ mod tests {
         insta::assert_json_snapshot!(resp);
     }
 
+    #[test]
+    fn slot_file_output_managed_serializes() {
+        let resp = SlotResponse::FileOutput {
+            filename: "/tmp/coglet/predictions/pred_123/outputs/0.png".to_string(),
+            kind: FileOutputKind::FileType,
+            mime_type: Some("image/png".to_string()),
+            managed: true,
+        };
+        insta::assert_json_snapshot!(resp);
+    }
+
+    #[test]
+    fn slot_file_output_unmanaged_serializes() {
+        let resp = SlotResponse::FileOutput {
+            filename: "/home/user/model/output.wav".to_string(),
+            kind: FileOutputKind::FileType,
+            mime_type: None,
+            managed: false,
+        };
+        insta::assert_json_snapshot!(resp);
+    }
+
+    #[test]
+    fn slot_file_output_oversized_serializes() {
+        let resp = SlotResponse::FileOutput {
+            filename: "/tmp/coglet/predictions/pred_123/outputs/spill_abc.json".to_string(),
+            kind: FileOutputKind::Oversized,
+            mime_type: None,
+            managed: true,
+        };
+        insta::assert_json_snapshot!(resp);
+    }
+
     #[test]
     fn slot_metric_replace_serializes() {
         let resp = SlotResponse::Metric {

crates/coglet/src/bridge/snapshots/coglet__bridge__protocol__tests__slot_file_output_managed_serializes.snap

@@ -0,0 +1,13 @@
+---
+source: coglet/src/bridge/protocol.rs
+expression: resp
+---
+{
+  "type": "file_output",
+  "filename": "/tmp/coglet/predictions/pred_123/outputs/0.png",
+  "kind": {
+    "type": "file_type"
+  },
+  "mime_type": "image/png",
+  "managed": true
+}

crates/coglet/src/bridge/snapshots/coglet__bridge__protocol__tests__slot_file_output_oversized_serializes.snap

@@ -0,0 +1,12 @@
+---
+source: coglet/src/bridge/protocol.rs
+expression: resp
+---
+{
+  "type": "file_output",
+  "filename": "/tmp/coglet/predictions/pred_123/outputs/spill_abc.json",
+  "kind": {
+    "type": "oversized"
+  },
+  "managed": true
+}

crates/coglet/src/bridge/snapshots/coglet__bridge__protocol__tests__slot_file_output_unmanaged_serializes.snap

@@ -0,0 +1,12 @@
+---
+source: coglet/src/bridge/protocol.rs
+expression: resp
+---
+{
+  "type": "file_output",
+  "filename": "/home/user/model/output.wav",
+  "kind": {
+    "type": "file_type"
+  },
+  "managed": false
+}

crates/coglet/src/orchestrator.rs

@@ -84,6 +84,14 @@ fn ensure_trailing_slash(s: &str) -> String {
     }
 }
 
+fn read_file_output(filename: &str, managed: bool) -> std::io::Result<Vec<u8>> {
+    let bytes = std::fs::read(filename)?;
+    if managed && let Err(e) = std::fs::remove_file(filename) {
+        tracing::debug!(%filename, error = %e, "Failed to delete managed output file");
+    }
+    Ok(bytes)
+}
+
 /// Try to lock a prediction mutex.
 /// On poison: logs error, recovers to fail the prediction, returns None.
 /// Caller should remove the prediction from tracking if None is returned.
@@ -1072,9 +1080,9 @@ async fn run_event_loop(
                             predictions.remove(&slot_id);
                         }
                     }
-                    Ok(SlotResponse::FileOutput { filename, kind, mime_type }) => {
+                    Ok(SlotResponse::FileOutput { filename, kind, mime_type, managed }) => {
                         tracing::debug!(%slot_id, %filename, ?kind, "FileOutput received");
-                        let bytes = match std::fs::read(&filename) {
+                        let bytes = match read_file_output(&filename, managed) {
                             Ok(b) => b,
                             Err(e) => {
                                 tracing::error!(%slot_id, %filename, error = %e, "Failed to read FileOutput");
@@ -1309,6 +1317,30 @@ mod tests {
         assert!(!pending.contains("overflow"));
     }
 
+    #[test]
+    fn read_file_output_deletes_managed_file() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("managed.txt");
+        std::fs::write(&path, b"managed").unwrap();
+
+        let bytes = read_file_output(path.to_str().unwrap(), true).unwrap();
+
+        assert_eq!(bytes, b"managed");
+        assert!(!path.exists());
+    }
+
+    #[test]
+    fn read_file_output_preserves_unmanaged_file() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("unmanaged.txt");
+        std::fs::write(&path, b"unmanaged").unwrap();
+
+        let bytes = read_file_output(path.to_str().unwrap(), false).unwrap();
+
+        assert_eq!(bytes, b"unmanaged");
+        assert!(path.exists());
+    }
+
     #[test]
     fn wrap_outputs_schema_array_single_item() {
         // List[Path] with num_outputs=1 → ["url"] not "url"