diff --git a/CHANGELOG.md b/CHANGELOG.md index acefca2c2..2c656ba9b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/crates/zeph-core/src/agent/persistence/store.rs b/crates/zeph-core/src/agent/persistence/store.rs index ba1fee63c..0da4e3c3f 100644 --- a/crates/zeph-core/src/agent/persistence/store.rs +++ b/crates/zeph-core/src/agent/persistence/store.rs @@ -20,6 +20,13 @@ impl Agent { /// `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 Agent { ); } + // 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 = 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(); diff --git a/crates/zeph-core/src/agent/persistence/tests.rs b/crates/zeph-core/src/agent/persistence/tests.rs index 52eb60b57..0c2dc2d46 100644 --- a/crates/zeph-core/src/agent/persistence/tests.rs +++ b/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" + ); + } +}