refactor: lazy zen input bytes — Arc avoids clone on v2 non-zen path

add_input_vector: Arc::new(bytes) — zero-copy move, refcount bump into IoProxy
add_input_buffer: store &'static [u8] directly — no copy at registration
add_copied_input_buffer: one copy into Arc, shared with IoProxy (was two copies)

Vec<u8> for zenpipe built lazily in zen_execute_inner via ZenInput::to_vec(),
so v2 non-zen jobs pay zero clone cost even with zen-pipeline feature enabled.

New: IoBackend::ReadArc(Cursor<ArcBytes>), IoProxy::read_arc, IoProxy::as_input_slice

Author: lilith

imageflow_core/src/context.rs

@@ -31,6 +31,25 @@ use crate::graphics::bitmaps::{Bitmap, BitmapKey, BitmapWindowMut, BitmapsContai
 use imageflow_types::ImageInfo;
 use itertools::Itertools;
 
+/// Input bytes for the zen pipeline — either an Arc-wrapped owned Vec or a static slice.
+/// Stored instead of `Vec<u8>` so the v2 non-zen path pays zero clone cost at registration time.
+/// The `Vec<u8>` zenpipe needs is built lazily in `zen_execute_inner`.
+#[cfg(feature = "zen-pipeline")]
+enum ZenInput {
+    Owned(Arc<Vec<u8>>),
+    Static(&'static [u8]),
+}
+
+#[cfg(feature = "zen-pipeline")]
+impl ZenInput {
+    fn to_vec(&self) -> Vec<u8> {
+        match self {
+            ZenInput::Owned(a) => (**a).clone(),
+            ZenInput::Static(s) => s.to_vec(),
+        }
+    }
+}
+
 /// Something of a god object (which is necessary for a reasonable FFI interface).
 /// 1025 bytes including 5 heap allocations as of Oct 2025. If on the stack, 312 bytes are taken up
 pub struct Context {
@@ -60,8 +79,9 @@ pub struct Context {
     /// Input bytes stashed for the zen pipeline (feature-gated).
     /// Populated by `add_copied_input_buffer`, `add_input_vector`, etc.
     /// The zen pipeline reads from here instead of the codec containers.
+    /// Uses Arc to avoid cloning on the v2 non-zen path — Vec is built lazily at execute time.
     #[cfg(feature = "zen-pipeline")]
-    zen_input_bytes: std::collections::HashMap<i32, Vec<u8>>,
+    zen_input_bytes: std::collections::HashMap<i32, ZenInput>,
 
     /// Force a specific backend for testing. None = zen when zen-pipeline is compiled in.
     #[cfg(feature = "zen-pipeline")]
@@ -578,17 +598,31 @@ impl Context {
 
     pub fn add_copied_input_buffer(&mut self, io_id: i32, bytes: &[u8]) -> Result<()> {
         #[cfg(feature = "zen-pipeline")]
-        self.zen_input_bytes.insert(io_id, bytes.to_vec());
-
-        let io = IoProxy::copy_slice(self, io_id, bytes).map_err(|e| e.at(here!()))?;
-        self.add_io(io, io_id, IoDirection::In).map_err(|e| e.at(here!()))
+        {
+            let arc = Arc::new(bytes.to_vec());
+            self.zen_input_bytes.insert(io_id, ZenInput::Owned(arc.clone()));
+            let io = IoProxy::read_arc(self, io_id, arc).map_err(|e| e.at(here!()))?;
+            return self.add_io(io, io_id, IoDirection::In).map_err(|e| e.at(here!()));
+        }
+        #[cfg(not(feature = "zen-pipeline"))]
+        {
+            let io = IoProxy::copy_slice(self, io_id, bytes).map_err(|e| e.at(here!()))?;
+            self.add_io(io, io_id, IoDirection::In).map_err(|e| e.at(here!()))
+        }
     }
     pub fn add_input_vector(&mut self, io_id: i32, bytes: Vec<u8>) -> Result<()> {
         #[cfg(feature = "zen-pipeline")]
-        self.zen_input_bytes.insert(io_id, bytes.clone());
-
-        let io = IoProxy::read_vec(self, io_id, bytes).map_err(|e| e.at(here!()))?;
-        self.add_io(io, io_id, IoDirection::In).map_err(|e| e.at(here!()))
+        {
+            let arc = Arc::new(bytes);
+            self.zen_input_bytes.insert(io_id, ZenInput::Owned(arc.clone()));
+            let io = IoProxy::read_arc(self, io_id, arc).map_err(|e| e.at(here!()))?;
+            return self.add_io(io, io_id, IoDirection::In).map_err(|e| e.at(here!()));
+        }
+        #[cfg(not(feature = "zen-pipeline"))]
+        {
+            let io = IoProxy::read_vec(self, io_id, bytes).map_err(|e| e.at(here!()))?;
+            self.add_io(io, io_id, IoDirection::In).map_err(|e| e.at(here!()))
+        }
     }
 
     /// Zero-copy: borrows `bytes` without copying.
@@ -602,7 +636,7 @@ impl Context {
     /// In practice, the ABI layer (imageflow_abi) uses transmute to erase the real lifetime.
     pub fn add_input_buffer(&mut self, io_id: i32, bytes: &'static [u8]) -> Result<()> {
         #[cfg(feature = "zen-pipeline")]
-        self.zen_input_bytes.insert(io_id, bytes.to_vec());
+        self.zen_input_bytes.insert(io_id, ZenInput::Static(bytes));
 
         let io = IoProxy::read_slice(self, io_id, bytes).map_err(|e| e.at(here!()))?;
         self.add_io(io, io_id, IoDirection::In).map_err(|e| e.at(here!()))
@@ -875,8 +909,11 @@ impl Context {
         }
         let job_options = what.job_options.unwrap_or_default();
 
+        let io_bytes: std::collections::HashMap<i32, Vec<u8>> =
+            self.zen_input_bytes.iter().map(|(&id, z)| (id, z.to_vec())).collect();
+
         let output =
-            crate::zen::zen_execute(&what.framewise, &self.zen_input_bytes, &self.security, &job_options)
+            crate::zen::zen_execute(&what.framewise, &io_bytes, &self.security, &job_options)
                 .map_err(|e| e.at(here!()))?;
 
         // Store encoded outputs in Context's output buffer system.

imageflow_core/src/io.rs

@@ -6,11 +6,22 @@ use crate::{Context, ErrorKind, JsonResponse, Result};
 use imageflow_types::collections::AddRemoveSet;
 use imageflow_types::IoDirection;
 use std::rc::Rc;
+use std::sync::Arc;
 use uuid::Uuid;
 
+/// Newtype so `Cursor<ArcBytes>` implements `Read`/`BufRead`/`Seek`.
+/// `Arc<Vec<u8>>` doesn't implement `AsRef<[u8]>` directly; this bridges the gap.
+struct ArcBytes(Arc<Vec<u8>>);
+impl AsRef<[u8]> for ArcBytes {
+    fn as_ref(&self) -> &[u8] {
+        self.0.as_slice()
+    }
+}
+
 enum IoBackend {
     ReadSlice(Cursor<&'static [u8]>),
     ReadVec(Cursor<Vec<u8>>),
+    ReadArc(Cursor<ArcBytes>),
     WriteVec(Cursor<Vec<u8>>),
     ReadFile(BufReader<File>),
     WriteFile(BufWriter<File>),
@@ -27,6 +38,7 @@ impl IoBackend {
         match self {
             IoBackend::ReadSlice(w) => Some(w),
             IoBackend::ReadVec(w) => Some(w),
+            IoBackend::ReadArc(w) => Some(w),
             IoBackend::ReadFile(w) => Some(w),
             _ => None,
         }
@@ -35,6 +47,7 @@ impl IoBackend {
         match self {
             IoBackend::ReadSlice(w) => Some(w),
             IoBackend::ReadVec(w) => Some(w),
+            IoBackend::ReadArc(w) => Some(w),
             IoBackend::ReadFile(w) => Some(w),
             _ => None,
         }
@@ -43,6 +56,7 @@ impl IoBackend {
         match self {
             IoBackend::ReadSlice(w) => Some(w),
             IoBackend::ReadVec(w) => Some(w),
+            IoBackend::ReadArc(w) => Some(w),
             IoBackend::ReadFile(w) => Some(w),
             _ => None,
         }
@@ -103,6 +117,7 @@ impl IoProxy {
     pub fn try_get_length(&mut self) -> Option<u64> {
         match &self.backend {
             IoBackend::ReadVec(v) => Some(v.get_ref().len() as u64),
+            IoBackend::ReadArc(v) => Some(v.get_ref().0.len() as u64),
             IoBackend::ReadSlice(v) => Some(v.get_ref().len() as u64),
             IoBackend::ReadFile(v) => v.get_ref().metadata().map(|m| m.len()).ok(),
             _ => None,
@@ -112,12 +127,24 @@ impl IoProxy {
     pub fn try_get_position(&mut self) -> Option<u64> {
         match &self.backend {
             IoBackend::ReadVec(v) => Some(v.position()),
+            IoBackend::ReadArc(v) => Some(v.position()),
             IoBackend::ReadSlice(v) => Some(v.position()),
             IoBackend::ReadFile(v) => v.get_ref().stream_position().ok(),
             _ => None,
         }
     }
 
+    /// Return a slice view of the input bytes without copying.
+    /// Returns `None` for file-backed or write-backed proxies.
+    pub fn as_input_slice(&self) -> Option<&[u8]> {
+        match &self.backend {
+            IoBackend::ReadSlice(c) => Some(c.get_ref()),
+            IoBackend::ReadVec(c) => Some(c.get_ref()),
+            IoBackend::ReadArc(c) => Some(c.get_ref().0.as_slice()),
+            _ => None,
+        }
+    }
+
     pub fn read_exact(&mut self, buf: &mut [u8]) -> std::io::Result<()> {
         self.backend.get_read().expect("cannot read from writer").read_exact(buf)
     }
@@ -213,6 +240,12 @@ impl IoProxy {
         Ok(IoProxy { path: None, io_id, backend: IoBackend::ReadVec(Cursor::new(bytes)) })
     }
 
+    /// Wrap a shared `Arc<Vec<u8>>` as an input buffer — zero copy for the caller.
+    pub fn read_arc(context: &Context, io_id: i32, bytes: Arc<Vec<u8>>) -> Result<IoProxy> {
+        IoProxy::check_io_id(context, io_id)?;
+        Ok(IoProxy { path: None, io_id, backend: IoBackend::ReadArc(Cursor::new(ArcBytes(bytes))) })
+    }
+
     pub fn copy_slice(context: &Context, io_id: i32, bytes: &[u8]) -> Result<IoProxy> {
         IoProxy::check_io_id(context, io_id)?;