fix(session-continuation): recover image history and settled turns

src-tauri/crates/agent-core/src/core/session/persistence/mod.rs

@@ -13,6 +13,8 @@
 mod crud;
 pub(crate) mod linked_work_item;
 mod messages;
+mod native_transfer;
+pub use native_transfer::load_messages_for_native_transfer;
 mod sidebar;
 
 // Re-exports kept at the `session::persistence::` surface — these are

src-tauri/crates/agent-core/src/core/session/persistence/native_transfer.rs

@@ -0,0 +1,65 @@
+//! Exact native-transfer projection of durable Agent rows. Display history keeps
+//! disk refs; this read embeds attachments and restores the producing event's
+//! accepted intent by message id, never by prompt text or position.
+
+use std::collections::HashMap;
+
+use database::db::get_connection;
+use rusqlite::params;
+use serde_json::Value;
+
+use crate::persistence::db_helpers as shared;
+
+pub fn load_messages_for_native_transfer(session_id: &str) -> rusqlite::Result<Vec<Value>> {
+    let messages = shared::load_native_message_rows("agent", session_id)?;
+    let conn = get_connection()?;
+    // One scoped query. Project only identity fields so historical base64 image
+    // payloads do not cross this boundary a second time.
+    let mut stmt = conn.prepare(
+        "SELECT json_extract(result_json, '$.messageId'),
+                json_extract(result_json, '$.turnIntentId')
+         FROM events WHERE session_id = ?1 AND function_name = 'user_message'
+         AND json_valid(result_json)",
+    )?;
+    let rows = stmt.query_map(params![session_id], |row| {
+        Ok((
+            row.get::<_, Option<String>>(0)?,
+            row.get::<_, Option<String>>(1)?,
+        ))
+    })?;
+    let mut intents = HashMap::new();
+    for row in rows {
+        let (message_id, intent) = row?;
+        if let (Some(message_id), Some(intent)) = (message_id, intent) {
+            if intent.is_empty() {
+                continue;
+            }
+            if let Some(previous) = intents.insert(message_id.clone(), intent.clone()) {
+                if previous != intent {
+                    return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
+                        std::io::Error::other(format!(
+                            "conflicting native turn identity for message {message_id}"
+                        )),
+                    )));
+                }
+            }
+        }
+    }
+    messages
+        .into_iter()
+        .map(|message| {
+            let intent = (message.role == shared::message_role::USER)
+                .then(|| intents.get(&message.id))
+                .flatten();
+            let mut value = shared::to_json_value(&message)?;
+            if let Some(intent) = intent {
+                value["turnIntentId"] = Value::String(intent.clone());
+            }
+            Ok(value)
+        })
+        .collect()
+}
+
+#[cfg(test)]
+#[path = "native_transfer_tests.rs"]
+mod tests;

src-tauri/crates/agent-core/src/core/session/persistence/native_transfer_tests.rs

@@ -0,0 +1,136 @@
+use super::*;
+
+fn seed(session_id: &str) {
+    let conn = get_connection().unwrap();
+    crate::persistence::test_schema::ensure_agent_sessions_schema(&conn);
+    conn.execute_batch(
+        "CREATE TABLE IF NOT EXISTS agent_messages (
+        id TEXT PRIMARY KEY, session_id TEXT NOT NULL, role TEXT NOT NULL,
+        content TEXT NOT NULL, tool_name TEXT, tool_call_id TEXT, tool_input TEXT,
+        tool_output TEXT, model TEXT, sequence INTEGER NOT NULL, created_at TEXT NOT NULL,
+        images TEXT, compact_from_sequence INTEGER, compact_tokens_before INTEGER,
+        compact_tokens_after INTEGER, tool_is_error INTEGER NOT NULL DEFAULT 0
+    );
+    CREATE TABLE IF NOT EXISTS events (
+        id TEXT PRIMARY KEY, session_id TEXT NOT NULL, event_type TEXT NOT NULL,
+        function_name TEXT, result_json TEXT NOT NULL, created_at TEXT NOT NULL
+    );",
+    )
+    .unwrap();
+    conn.execute(
+        "INSERT INTO agent_sessions
+        (session_id,name,status,created_at,updated_at)
+        VALUES (?1,'Transfer fixture','completed','now','now')",
+        [session_id],
+    )
+    .unwrap();
+}
+
+fn bind(session_id: &str, message_id: &str, intent: &str) {
+    let conn = get_connection().unwrap();
+    conn.execute(
+        "INSERT INTO events
+        (id,session_id,event_type,function_name,result_json,created_at)
+        VALUES (?1,?2,'user_message','user_message',?3,'now')",
+        params![
+            format!("user-message-{session_id}-{message_id}"),
+            session_id,
+            serde_json::json!({"messageId":message_id,"turnIntentId":intent}).to_string()
+        ],
+    )
+    .unwrap();
+}
+
+#[tokio::test]
+async fn native_ipc_embeds_disk_images_without_cli_refs_and_preserves_display_rows() {
+    let _sandbox = test_helpers::test_env::sandbox();
+    let sid = "sdeagent-transfer-images";
+    seed(sid);
+    let url = "data:image/png;base64,QUJD".to_string();
+    let id = super::super::save_user_msg(sid, "inspect", Some(&[url.clone()])).unwrap();
+    bind(sid, &id, "image-intent");
+    let stored = super::super::load_messages(sid).unwrap();
+    let paths: Vec<String> = serde_json::from_str(stored[0].images.as_ref().unwrap()).unwrap();
+    assert!(!paths[0].starts_with("data:"));
+    // No code_session_image_refs or native-transcript-id table is needed by Agent history.
+    for _ in 0..2 {
+        let native = crate::state::commands::agent_load_messages(sid.into(), Some(true))
+            .await
+            .unwrap();
+        assert_eq!(native[0]["turnIntentId"], "image-intent");
+        assert_eq!(
+            native[0]["images"],
+            serde_json::to_string(&[url.clone()]).unwrap()
+        );
+    }
+    let display = crate::state::commands::agent_load_messages(sid.into(), None)
+        .await
+        .unwrap();
+    assert_eq!(display[0]["images"], stored[0].images.clone().unwrap());
+    assert!(display[0].get("turnIntentId").is_none());
+    assert_eq!(
+        super::super::load_messages(sid).unwrap()[0].images,
+        stored[0].images
+    );
+}
+
+#[test]
+fn native_identity_uses_durable_message_ids_for_repeated_expanded_prompts() {
+    let _sandbox = test_helpers::test_env::sandbox();
+    let sid = "sdeagent-transfer-force-send";
+    seed(sid);
+    let first = super::super::save_user_msg(sid, "expanded provider prompt", None).unwrap();
+    let second = super::super::save_user_msg(sid, "expanded provider prompt", None).unwrap();
+    bind(sid, &first, "old-turn");
+    bind(sid, &second, "force-send-turn");
+    seed("sdeagent-other");
+    bind("sdeagent-other", &second, "unrelated-turn");
+    let rows = load_messages_for_native_transfer(sid).unwrap();
+    assert_eq!(rows[0]["turnIntentId"], "old-turn");
+    assert_eq!(rows[1]["turnIntentId"], "force-send-turn");
+    assert_eq!(rows[1]["content"], "expanded provider prompt");
+}
+
+#[test]
+fn exact_transfer_keeps_missing_attachments_as_visible_errors() {
+    let _sandbox = test_helpers::test_env::sandbox();
+    let sid = "sdeagent-transfer-missing";
+    seed(sid);
+    let id =
+        super::super::save_user_msg(sid, "inspect", Some(&["data:image/png;base64,QUJD".into()]))
+            .unwrap();
+    let stored = super::super::load_messages(sid).unwrap();
+    let paths: Vec<String> = serde_json::from_str(stored[0].images.as_ref().unwrap()).unwrap();
+    std::fs::remove_file(&paths[0]).unwrap();
+    assert!(load_messages_for_native_transfer(sid)
+        .unwrap_err()
+        .to_string()
+        .contains("historical image is unavailable"));
+    assert_eq!(super::super::load_messages(sid).unwrap()[0].id, id);
+}
+
+#[test]
+fn compacted_away_images_do_not_block_the_effective_native_frame() {
+    let _sandbox = test_helpers::test_env::sandbox();
+    let sid = "sdeagent-transfer-compacted";
+    seed(sid);
+    super::super::save_user_msg(sid, "superseded image", None).unwrap();
+    let conn = get_connection().unwrap();
+    conn.execute(
+        "UPDATE agent_messages SET images='[\"missing.png\"]' WHERE session_id=?1",
+        [sid],
+    )
+    .unwrap();
+    super::super::save_user_msg(sid, "retained", None).unwrap();
+    conn.execute(
+        "INSERT INTO agent_messages
+        (id,session_id,role,content,sequence,created_at,compact_from_sequence)
+        VALUES ('summary',?1,'system','summary',2,'now',1)",
+        [sid],
+    )
+    .unwrap();
+    let rows = load_messages_for_native_transfer(sid).unwrap();
+    assert_eq!(rows.len(), 2);
+    assert_eq!(rows[0]["content"], "summary");
+    assert_eq!(rows[1]["content"], "retained");
+}

src-tauri/crates/agent-core/src/foundation/persistence/db_helpers/messages/load_llm.rs

@@ -73,8 +73,23 @@ pub fn load_native_history(
     prefix: &str,
     session_id: &str,
 ) -> rusqlite::Result<Vec<serde_json::Value>> {
+    let messages = load_native_message_rows(prefix, session_id)?;
+    Ok(reconstruct_with_projection(&messages, true))
+}
+
+/// Effective message frame shared by native verification and the IPC reader.
+pub fn load_native_message_rows(
+    prefix: &str,
+    session_id: &str,
+) -> rusqlite::Result<Vec<AgentMessageRow>> {
     let mut messages = visible_rows(&load_messages(prefix, session_id)?);
-    for message in &mut messages {
+    embed_native_message_images(&mut messages)?;
+    Ok(messages)
+}
+
+/// Exact-transfer projection; disk references remain valid persisted data.
+fn embed_native_message_images(messages: &mut [AgentMessageRow]) -> rusqlite::Result<()> {
+    for message in messages {
         if message.role != message_role::USER {
             continue;
         }
@@ -96,7 +111,7 @@ pub fn load_native_history(
                 Some(serde_json::to_string(&embedded).expect("string array serialization"));
         }
     }
-    Ok(reconstruct_with_projection(&messages, true))
+    Ok(())
 }
 
 /// Turns elapsed since the given tool was last called in this session,

src-tauri/crates/agent-core/src/foundation/persistence/db_helpers/messages/mod.rs

@@ -31,5 +31,5 @@ pub use cleanup::{clear_messages, truncate_messages_from_sequence};
 pub use load_llm::{
     compact_cutoff_sequence, load_llm_history, load_llm_history_start_sequences,
     load_llm_history_text_only, load_llm_history_text_only_bounded, load_native_history,
-    turns_since_last_tool_call,
+    load_native_message_rows, turns_since_last_tool_call,
 };

src-tauri/crates/agent-core/src/state/commands/session/persistence.rs

@@ -20,8 +20,14 @@ use super::common::review_session_ids;
 
 /// Load conversation messages for a session.
 #[tauri::command]
-pub async fn agent_load_messages(session_id: String) -> Result<Vec<serde_json::Value>, String> {
+pub async fn agent_load_messages(
+    session_id: String,
+    for_native_transfer: Option<bool>,
+) -> Result<Vec<serde_json::Value>, String> {
     shared::spawn_blocking_cmd(move || {
+        if for_native_transfer.unwrap_or(false) {
+            return session_persistence::load_messages_for_native_transfer(&session_id);
+        }
         let messages = session_persistence::load_messages(&session_id)?;
         messages.into_iter().map(shared::to_json_value).collect()
     })

src/api/tauri/agent/session.ts

@@ -86,9 +86,10 @@ export async function cancelSession(
 }
 
 export async function loadMessages(
-  sessionId: string
+  sessionId: string,
+  forNativeTransfer?: boolean
 ): Promise<SessionMessage[]> {
-  return rpc.agentSession.loadMessages({ sessionId });
+  return rpc.agentSession.loadMessages({ sessionId, forNativeTransfer });
 }
 
 export async function getSession(

src/api/tauri/rpc/procedures/agentSession.ts

@@ -48,7 +48,11 @@ export const agentSession = {
     .output(z.boolean())
     .build(),
   loadMessages: defineProcedure("agent_load_messages")
-    .input(schemas.agentSession.SessionIdInput)
+    .input(
+      schemas.agentSession.SessionIdInput.extend({
+        forNativeTransfer: z.boolean().optional(),
+      })
+    )
     .output(z.array(schemas.agentSession.SessionMessageSchema))
     .build(),
   getSession: defineProcedure("agent_get_session")

src/engines/SessionCore/conversations/localConversationContinuation.test.ts

@@ -795,7 +795,11 @@ describe("local native conversation continuation", () => {
     const result = await recoverLocalConversationTurn({
       root,
       title: "Shared",
-      timeline: [...history, currentUser],
+      timeline: [
+        ...history,
+        currentUser,
+        event("canonical-answer", "assistant", "recovered answer"),
+      ],
       displayText: "continue",
       target,
       turnIntentId: "turn-recover-adopted",
@@ -828,6 +832,46 @@ describe("local native conversation continuation", () => {
     expect(mocks.sendMessage).not.toHaveBeenCalled();
   });
 
+  it("retires an irreparable completed anchor without reserving the adopted turn forever", async () => {
+    const history = [event("prior", "user", "prior request")];
+    childEvents = [
+      ...history,
+      event("orphan-answer", "assistant", "orphan output"),
+    ];
+    mocks.invokeTauri.mockResolvedValue([
+      { sessionId: "agentsession-child", updatedAt: "now" },
+    ]);
+    mocks.getAgentSession.mockResolvedValue({
+      status: "completed",
+      workspacePath: "/repo",
+      accountId: "account-1",
+      model: "model-1",
+      agentDefinitionId: "builtin:sde",
+    });
+    mocks.turnIntentStatus.mockResolvedValue({
+      status: "completed",
+      turnIntentId: "orphan-intent",
+    });
+    await expect(
+      recoverLocalConversationTurn({
+        root,
+        title: "Shared",
+        timeline: history,
+        displayText: "accepted request",
+        target,
+        turnIntentId: "orphan-intent",
+        runnerSessionId: "agentsession-child",
+      })
+    ).rejects.toBeInstanceOf(QueuedConversationRecoveryBlockedError);
+    expect(mocks.markTerminal).toHaveBeenCalledWith(
+      "agentsession-child",
+      "completed",
+      { generation: 3 }
+    );
+    expect(mocks.recoverNativeAfterMismatch).toHaveBeenCalledOnce();
+    expect(mocks.sendMessage).not.toHaveBeenCalled();
+  });
+
   it("restores provider acceptance before validating a restarted runner", async () => {
     mocks.turnIntentStatus.mockResolvedValue({
       sessionId: "agentsession-missing",

src/engines/SessionCore/conversations/localConversationContinuation.ts

@@ -281,9 +281,20 @@ export async function recoverLocalConversationTurn(
     );
   }
 
-  const timeline = params.timeline.filter(
-    (event) => conversationTurnIdOf(event) !== params.turnIntentId
+  // A restarted local root already contains the accepted user and its output.
+  // Filtering just that user's id leaves its answer in the supposed pre-turn
+  // prefix and makes every authoritative reread fail deterministically.
+  const currentUserIndex = params.timeline.findIndex(
+    (event) =>
+      event.source === "user" &&
+      conversationTurnIdOf(event) === params.turnIntentId
   );
+  const timeline =
+    currentUserIndex >= 0
+      ? params.timeline.slice(0, currentUserIndex)
+      : params.timeline.filter(
+          (event) => conversationTurnIdOf(event) !== params.turnIntentId
+        );
   const { events } = await loadAuthoritativeSessionEvents(
     params.runnerSessionId
   );
@@ -330,7 +341,12 @@ export async function recoverLocalConversationTurn(
       agentTail: finished.agentTail,
     };
   } catch (error) {
-    if (error instanceof QueuedConversationRecoveryPendingError) throw error;
+    if (
+      error instanceof QueuedConversationRecoveryPendingError ||
+      error instanceof QueuedConversationRecoveryBlockedError
+    ) {
+      throw error;
+    }
     throw new QueuedConversationRecoveryPendingError(
       error instanceof Error ? error.message : String(error)
     );