fix(core): strip ephemeral Image parts before persistence writes

CHANGELOG.md

@@ -120,6 +120,16 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   per-task (`SchedulerAction::Verify`) and whole-plan (`run_whole_plan_verify`) scopes emit an
   independent `Note: ...` message; previously this case was silently logged at `debug`/`warn`
   only and never surfaced to the user (#6265).
+- **zeph-core**: `Agent::persist_message` now strips every `MessagePart::Image` from `parts`
+  before either persistence writer sees it — SQLite `parts_json`, the Qdrant embed path, and the
+  durable JSONL session log all receive an Image-free slice (spec-072 §4 C1, #6239, P1 of 4).
+  The strip applies to all `Image` parts, not only future MCP-sourced ones, closing pre-existing
+  persistence waste on the existing user-upload image path (base64 bytes were written to SQLite
+  then silently dropped on rehydrate, since `hydrate.rs` never reconstructs `Image` parts). The
+  in-memory `Message` pushed via `push_message` is unaffected — `Image` parts remain available
+  for the current turn's provider request; only the two persistence writers receive the stripped
+  copy. No runtime-visible behavior change beyond no longer persisting image bytes that were
+  already dead weight.
 
 ### Changed
 

crates/zeph-core/src/agent/persistence/store.rs

@@ -20,6 +20,13 @@ impl<C: Channel> Agent<C> {
     /// `has_injection_flags` controls whether Qdrant embedding is skipped for this message.
     /// When `true` and `guard_memory_writes` is enabled, only `SQLite` is written — the message
     /// is saved for conversation continuity but will not pollute semantic search (M2, D2).
+    ///
+    /// `MessagePart::Image` parts are deliberately stripped before either persistence writer
+    /// below sees `parts` — they are ephemeral, current-turn-only content (spec-072 §4, C1) and
+    /// must never reach `SQLite` `parts_json`, the Qdrant embed path, or the durable JSONL session
+    /// log. This is a single, explicit strip point above both writers, not an omission; the
+    /// `parts` slice passed in by the caller is untouched, so the in-memory `Message` already
+    /// pushed via `push_message` keeps its `Image` parts for the current turn's provider request.
     #[tracing::instrument(name = "core.persist.persist_message", skip_all, level = "debug")]
     pub(crate) async fn persist_message(
         &mut self,
@@ -48,19 +55,33 @@ impl<C: Channel> Agent<C> {
             );
         }
 
+        // C1 (spec-072 §4): strip Image parts once, above both persistence writers below.
+        // Neither `sink.record_message` nor `PersistMessageRequest::from_borrowed`/
+        // `svc.persist_message` may see an unstripped `parts` slice — see the doc comment above.
+        let persisted_parts: Vec<MessagePart> = parts
+            .iter()
+            .filter(|p| !matches!(p, MessagePart::Image(_)))
+            .cloned()
+            .collect();
+
         // INV-SP-1 (spec-068 §13): the durable event log must be appended and flushed before the
         // SQLite `messages` projection is written — the projection must never lead the log. A
         // failed session-log write is logged and the turn proceeds; a crash between the two
         // leaves the log ahead of the projection, which INV-SP-3 reconciles on next open.
         if let Some(sink) = self.services.session.session_sink.clone() {
             tracing::debug!("persist_message: session_sink.record_message start");
-            if let Err(e) = sink.record_message(role, content, parts).await {
+            if let Err(e) = sink.record_message(role, content, &persisted_parts).await {
                 tracing::warn!(error = %e, "failed to append session event log entry");
             }
             tracing::debug!("persist_message: session_sink.record_message done");
         }
 
-        let req = PersistMessageRequest::from_borrowed(role, content, parts, has_injection_flags);
+        let req = PersistMessageRequest::from_borrowed(
+            role,
+            content,
+            &persisted_parts,
+            has_injection_flags,
+        );
 
         let mut unsummarized = self.services.memory.persistence.unsummarized_count;
         let memory_arc = self.services.memory.persistence.memory.clone();

crates/zeph-core/src/agent/persistence/tests.rs

@@ -2797,3 +2797,234 @@ async fn regression_3168_corrupt_parts_row_skipped_on_load() {
         "orphaned ToolResult must not survive load_history; loaded={loaded}"
     );
 }
+
+// --- issue #6239 / spec-072 §4 C1: ephemeral Image persistence strip ---
+
+mod image_persistence_strip {
+    use super::*;
+    use zeph_llm::provider::{ImageData, Message, MessageMetadata, MessagePart};
+
+    fn png_image_part() -> MessagePart {
+        MessagePart::Image(Box::new(ImageData {
+            data: vec![0x89, 0x50, 0x4E, 0x47, 1, 2, 3, 4],
+            mime_type: "image/png".to_owned(),
+        }))
+    }
+
+    #[tokio::test]
+    async fn test_persist_message_strips_image_before_sqlite() {
+        // A ToolResult sibling survives (mirrors the future MCP sibling-Image emission shape),
+        // but the Image part itself must never reach the SQLite `parts_json` column.
+        let provider = mock_provider(vec![]);
+        let channel = MockChannel::new(vec![]);
+        let registry = create_test_registry();
+        let executor = MockToolExecutor::no_tools();
+
+        let memory = test_memory(&AnyProvider::Mock(zeph_llm::mock::MockProvider::default())).await;
+        let cid = memory.sqlite().create_conversation().await.unwrap();
+
+        let mut agent = Agent::new(provider, channel, registry, None, 5, executor).with_memory(
+            std::sync::Arc::new(memory),
+            cid,
+            50,
+            5,
+            100,
+        );
+
+        let parts = vec![
+            MessagePart::ToolResult {
+                tool_use_id: "call_img_1".to_owned(),
+                content: "see attached image".to_owned(),
+                is_error: false,
+            },
+            png_image_part(),
+        ];
+        agent
+            .persist_message(Role::User, "[tool_result: call_img_1]", &parts, false)
+            .await;
+
+        let history = agent
+            .services
+            .memory
+            .persistence
+            .memory
+            .as_ref()
+            .unwrap()
+            .sqlite()
+            .load_history(cid, 50)
+            .await
+            .unwrap();
+
+        assert_eq!(history.len(), 1);
+        assert!(
+            !history[0]
+                .parts
+                .iter()
+                .any(|p| matches!(p, MessagePart::Image(_))),
+            "SQLite parts_json must not contain an Image part"
+        );
+        assert!(
+            history[0]
+                .parts
+                .iter()
+                .any(|p| matches!(p, MessagePart::ToolResult { tool_use_id, .. } if tool_use_id == "call_img_1")),
+            "the non-Image sibling must survive the strip"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_persist_message_strips_image_before_embed() {
+        // Role::User always takes the Qdrant embed path (should_embed_message). An
+        // Image-only `parts` slice must still persist (as an empty parts array) without
+        // ever routing image bytes into the embed text or the persisted parts_json.
+        let provider = mock_provider(vec![]);
+        let channel = MockChannel::new(vec![]);
+        let registry = create_test_registry();
+        let executor = MockToolExecutor::no_tools();
+
+        let (tx, rx) = tokio::sync::watch::channel(MetricsSnapshot::default());
+        let memory = test_memory(&AnyProvider::Mock(zeph_llm::mock::MockProvider::default())).await;
+        let cid = memory.sqlite().create_conversation().await.unwrap();
+
+        let mut agent = Agent::new(provider, channel, registry, None, 5, executor)
+            .with_metrics(tx)
+            .with_memory(std::sync::Arc::new(memory), cid, 50, 5, 100);
+
+        let parts = vec![png_image_part()];
+        agent
+            .persist_message(Role::User, "hello with an image", &parts, false)
+            .await;
+
+        // sqlite_message_count increments regardless of Qdrant availability in the test
+        // harness — proves the message (sans Image) still reached the persist path.
+        assert_eq!(rx.borrow().sqlite_message_count, 1);
+
+        let history = agent
+            .services
+            .memory
+            .persistence
+            .memory
+            .as_ref()
+            .unwrap()
+            .sqlite()
+            .load_history(cid, 50)
+            .await
+            .unwrap();
+        assert_eq!(history.len(), 1);
+        assert_eq!(history[0].content, "hello with an image");
+        assert!(
+            history[0].parts.is_empty(),
+            "an Image-only parts slice must persist as empty, not carry the image through"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_persist_message_strips_image_before_session_log() {
+        use std::sync::Arc;
+        use zeph_agent_persistence::SessionSink;
+        use zeph_session::{SessionEvent, SessionEventLog, SessionStore};
+
+        let provider = mock_provider(vec![]);
+        let channel = MockChannel::new(vec![]);
+        let registry = create_test_registry();
+        let executor = MockToolExecutor::no_tools();
+
+        let dir = tempfile::tempdir().unwrap();
+        let log = Arc::new(SessionEventLog::open(dir.path()).await.unwrap());
+        let db_config = zeph_db::DbConfig {
+            url: ":memory:".to_owned(),
+            ..Default::default()
+        };
+        let pool = db_config.connect().await.unwrap();
+        zeph_db::run_migrations(&pool).await.unwrap();
+        let store = SessionStore::new(pool);
+        store.create("s-6239").await.unwrap();
+        let sink = SessionSink::new(log.clone(), store, zeph_common::SessionId::new("s-6239"));
+
+        let mut agent = Agent::new(provider, channel, registry, None, 5, executor)
+            .with_session_sink(Some(Arc::new(sink)));
+
+        let parts = vec![
+            MessagePart::Text {
+                text: "here is the result".to_owned(),
+            },
+            png_image_part(),
+        ];
+        agent
+            .persist_message(Role::Assistant, "here is the result", &parts, false)
+            .await;
+
+        let events = log.read_all().await.unwrap();
+        assert_eq!(events.len(), 1);
+        let SessionEvent::AssistantMessage {
+            parts: logged_parts,
+        } = &events[0].kind
+        else {
+            panic!("expected AssistantMessage event");
+        };
+        assert!(
+            !logged_parts
+                .iter()
+                .any(|p| matches!(p, MessagePart::Image(_))),
+            "durable JSONL session log must not contain an Image part"
+        );
+        assert!(
+            logged_parts
+                .iter()
+                .any(|p| matches!(p, MessagePart::Text { text } if text == "here is the result")),
+            "the non-Image sibling must survive the strip"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_persist_message_inmemory_message_keeps_image() {
+        // The strip is persistence-only: the in-memory Message object (pushed separately by
+        // the caller via push_message, not by persist_message itself) must retain its Image
+        // part after persist_message runs on the same underlying parts.
+        let provider = mock_provider(vec![]);
+        let channel = MockChannel::new(vec![]);
+        let registry = create_test_registry();
+        let executor = MockToolExecutor::no_tools();
+
+        let memory = test_memory(&AnyProvider::Mock(zeph_llm::mock::MockProvider::default())).await;
+        let cid = memory.sqlite().create_conversation().await.unwrap();
+
+        let mut agent = Agent::new(provider, channel, registry, None, 5, executor).with_memory(
+            std::sync::Arc::new(memory),
+            cid,
+            50,
+            5,
+            100,
+        );
+
+        let parts = vec![
+            MessagePart::Text {
+                text: "here is the result".to_owned(),
+            },
+            png_image_part(),
+        ];
+
+        // Simulate the real call order: the in-memory Message is pushed first (tier_loop.rs),
+        // then persist_message is invoked with a borrow of the same parts.
+        agent.msg.messages.push(Message {
+            role: Role::Assistant,
+            content: "here is the result".to_owned(),
+            parts: parts.clone(),
+            metadata: MessageMetadata::default(),
+        });
+
+        agent
+            .persist_message(Role::Assistant, "here is the result", &parts, false)
+            .await;
+
+        let in_memory = agent.msg.messages.last().unwrap();
+        assert_eq!(in_memory.parts.len(), 2);
+        assert!(
+            in_memory
+                .parts
+                .iter()
+                .any(|p| matches!(p, MessagePart::Image(_))),
+            "in-memory Message must keep its Image part — the strip is persistence-only"
+        );
+    }
+}