diff --git a/src-tauri/crates/agent-core/src/core/session/persistence/mod.rs b/src-tauri/crates/agent-core/src/core/session/persistence/mod.rs index 96f8c47d31..9e73cdf9f5 100644 --- a/src-tauri/crates/agent-core/src/core/session/persistence/mod.rs +++ b/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 diff --git a/src-tauri/crates/agent-core/src/core/session/persistence/native_transfer.rs b/src-tauri/crates/agent-core/src/core/session/persistence/native_transfer.rs new file mode 100644 index 0000000000..962a0dec45 --- /dev/null +++ b/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> { + 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>(0)?, + row.get::<_, Option>(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; diff --git a/src-tauri/crates/agent-core/src/core/session/persistence/native_transfer_tests.rs b/src-tauri/crates/agent-core/src/core/session/persistence/native_transfer_tests.rs new file mode 100644 index 0000000000..a6ac27015e --- /dev/null +++ b/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 = 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 = 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"); +} diff --git a/src-tauri/crates/agent-core/src/foundation/persistence/db_helpers/messages/load_llm.rs b/src-tauri/crates/agent-core/src/foundation/persistence/db_helpers/messages/load_llm.rs index 392f6cd45e..694c8f6072 100644 --- a/src-tauri/crates/agent-core/src/foundation/persistence/db_helpers/messages/load_llm.rs +++ b/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> { + 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> { 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, diff --git a/src-tauri/crates/agent-core/src/foundation/persistence/db_helpers/messages/mod.rs b/src-tauri/crates/agent-core/src/foundation/persistence/db_helpers/messages/mod.rs index 95422ed13d..71c085a4f4 100644 --- a/src-tauri/crates/agent-core/src/foundation/persistence/db_helpers/messages/mod.rs +++ b/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, }; diff --git a/src-tauri/crates/agent-core/src/state/commands/session/persistence.rs b/src-tauri/crates/agent-core/src/state/commands/session/persistence.rs index 951f10d23b..500a930a07 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/persistence.rs +++ b/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, String> { +pub async fn agent_load_messages( + session_id: String, + for_native_transfer: Option, +) -> Result, 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() }) diff --git a/src/api/tauri/agent/session.ts b/src/api/tauri/agent/session.ts index 137eab42e9..e52a151b69 100644 --- a/src/api/tauri/agent/session.ts +++ b/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 { - return rpc.agentSession.loadMessages({ sessionId }); + return rpc.agentSession.loadMessages({ sessionId, forNativeTransfer }); } export async function getSession( diff --git a/src/api/tauri/rpc/procedures/agentSession.ts b/src/api/tauri/rpc/procedures/agentSession.ts index 6729462198..94b257f397 100644 --- a/src/api/tauri/rpc/procedures/agentSession.ts +++ b/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") diff --git a/src/engines/SessionCore/conversations/localConversationContinuation.test.ts b/src/engines/SessionCore/conversations/localConversationContinuation.test.ts index 8f4d6fd408..f2399669ae 100644 --- a/src/engines/SessionCore/conversations/localConversationContinuation.test.ts +++ b/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", diff --git a/src/engines/SessionCore/conversations/localConversationContinuation.ts b/src/engines/SessionCore/conversations/localConversationContinuation.ts index 6c043d7e0f..f2b220d90f 100644 --- a/src/engines/SessionCore/conversations/localConversationContinuation.ts +++ b/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) ); diff --git a/src/engines/SessionCore/conversations/localConversationSettledTail.test.ts b/src/engines/SessionCore/conversations/localConversationSettledTail.test.ts index e3c0c89a5e..8d49d52b0b 100644 --- a/src/engines/SessionCore/conversations/localConversationSettledTail.test.ts +++ b/src/engines/SessionCore/conversations/localConversationSettledTail.test.ts @@ -8,7 +8,10 @@ import { nativeSourceEventId, projectNativeConversationItems, } from "./nativeConversationMaterializer"; -import { QueuedConversationRecoveryPendingError } from "./queuedConversationContract"; +import { + QueuedConversationRecoveryBlockedError, + QueuedConversationRecoveryPendingError, +} from "./queuedConversationContract"; import { recordEmptyFailedAttempt, retryLineageForMessage, @@ -450,6 +453,26 @@ describe("native settled tail lifecycle boundary", () => { ).rejects.toBeInstanceOf(QueuedConversationRecoveryPendingError); }); + it("blocks a completed turn after bounded rereads when no durable user anchor exists", async () => { + const raw = [ + ...before, + event("orphan-answer", "assistant", "orphan output", "assistant"), + ]; + mocks.reconcile.mockResolvedValueOnce(raw); + mocks.recover.mockResolvedValueOnce(raw); + await expect( + loadSettledTail( + "runner", + before, + "force-send-intent", + { text: "accepted request", images: [] }, + "completed" + ) + ).rejects.toBeInstanceOf(QueuedConversationRecoveryBlockedError); + expect(mocks.reconcile).toHaveBeenCalledOnce(); + expect(mocks.recover).toHaveBeenCalledOnce(); + }); + it("does not certify a failed prompt when a later concurrent user exists", async () => { const raw = [...native, event("other-user", "user", "another turn")]; mocks.reconcile.mockResolvedValueOnce(raw); diff --git a/src/engines/SessionCore/conversations/localConversationSettledTail.ts b/src/engines/SessionCore/conversations/localConversationSettledTail.ts index 31c8de1952..684ce206b8 100644 --- a/src/engines/SessionCore/conversations/localConversationSettledTail.ts +++ b/src/engines/SessionCore/conversations/localConversationSettledTail.ts @@ -31,7 +31,10 @@ import { isNativeTerminalDiagnosticOrEcho, nativeTerminalDiagnosticSources, } from "./nativeTerminalDiagnostic"; -import { QueuedConversationRecoveryPendingError } from "./queuedConversationContract"; +import { + QueuedConversationRecoveryBlockedError, + QueuedConversationRecoveryPendingError, +} from "./queuedConversationContract"; const log = createLogger("localConversationContinuation"); @@ -316,7 +319,10 @@ export async function loadSettledTail( true ); if (agentTail) return { agentTail, events }; - throw new Error( + // Durable completed output plus bounded authoritative rereads cannot be + // repaired by indefinitely retrying the same accepted intent. Retire its + // execution owner through the queue's visible failure boundary. + throw new QueuedConversationRecoveryBlockedError( `conversation turn ${turnIntentId} is missing its native transcript anchor` ); } diff --git a/src/engines/SessionCore/conversations/localConversationTurnDispatch.ts b/src/engines/SessionCore/conversations/localConversationTurnDispatch.ts index 8983a5dff7..519ac28b35 100644 --- a/src/engines/SessionCore/conversations/localConversationTurnDispatch.ts +++ b/src/engines/SessionCore/conversations/localConversationTurnDispatch.ts @@ -43,7 +43,10 @@ import { supportsNativeConversationTarget, } from "./nativeConversationMaterializer"; import { nativeTurnFailureDiagnostic } from "./nativeTerminalDiagnostic"; -import { QueuedConversationRecoveryPendingError } from "./queuedConversationContract"; +import { + QueuedConversationRecoveryBlockedError, + QueuedConversationRecoveryPendingError, +} from "./queuedConversationContract"; const TURN_WAIT_WINDOW_MS = 60_000; const log = createLogger("localConversationContinuation"); @@ -163,7 +166,18 @@ export async function finishConversationTurn(params: { params.turnIntentId, params.providerRequest, terminalStatus - ); + ).catch((error: unknown) => { + if ( + params.settleAdoptedLifecycle && + error instanceof QueuedConversationRecoveryBlockedError + ) { + // This renderer adopted a terminal turn after restart. Its irreparable + // transcript mismatch retires the queue owner, so release only this + // adopted generation too; transient recovery keeps it reserved. + settleUserIntentLifecycle(params, terminalStatus); + } + throw error; + }); // Fresh sends are closed only by the CLI/Agent lifecycle coordinator. // Crash recovery created a synthetic frontend lifecycle after the original // terminal event, so it alone closes that adopted generation here. diff --git a/src/engines/SessionCore/hooks/session/__tests__/useQueueDispatch.intervention.test.ts b/src/engines/SessionCore/hooks/session/__tests__/useQueueDispatch.intervention.test.ts index bfb459a362..a9d261ea0e 100644 --- a/src/engines/SessionCore/hooks/session/__tests__/useQueueDispatch.intervention.test.ts +++ b/src/engines/SessionCore/hooks/session/__tests__/useQueueDispatch.intervention.test.ts @@ -1046,6 +1046,33 @@ describe("useQueueDispatch Agent Org intervention", () => { ); }); + it("dispatches the next queued turn after retiring a completed anchor mismatch", async () => { + const dispatched: string[] = []; + mocks.dispatchCanonicalConversation.mockImplementation( + async (_store, message, callbacks) => { + dispatched.push(message.id); + await callbacks.onAccepted(`runner-${message.id}`); + if (message.id === "missing-anchor") { + throw new QueuedConversationRecoveryBlockedError( + "missing native anchor" + ); + } + } + ); + await mountWithMessages([ + makeCanonicalMessage("missing-anchor"), + makeCanonicalMessage("next-request"), + ]); + await vi.waitFor(() => + expect(dispatched).toEqual(["missing-anchor", "next-request"]) + ); + await vi.waitFor(() => + expect(store.get(activeMessageDeliveriesAtom)).toEqual([]) + ); + expect(store.get(messageQueueAtom)).toEqual([]); + expect(mocks.dispatchCanonicalConversation).toHaveBeenCalledTimes(2); + }); + it.each(["recovery-blocked", "turn-closed", "accepted-turn-closed"])( "restores a reconciled-away failed row before retiring its %s owner", async (verdict) => { diff --git a/src/engines/SessionCore/ingestion/agentMessageAdapters.ts b/src/engines/SessionCore/ingestion/agentMessageAdapters.ts index bc606992ef..5f2793dcd1 100644 --- a/src/engines/SessionCore/ingestion/agentMessageAdapters.ts +++ b/src/engines/SessionCore/ingestion/agentMessageAdapters.ts @@ -54,6 +54,8 @@ export interface PersistedMessage { sequence: number; createdAt: string; images: string | null; + /** Producing user event identity, restored for exact native transfer. */ + turnIntentId?: string; /** * Set on compact-boundary rows: sequence number the preserved tail starts * at. Serialized by Rust's `AgentMessageRow` (serde camelCase); non-null @@ -290,6 +292,7 @@ export function persistedMessageToSessionEvent( result = { type: "user", message: { content: msg.content, role: "user" }, + ...(msg.turnIntentId ? { turnIntentId: msg.turnIntentId } : {}), ...(displayImages && displayImages.length > 0 ? { images: displayImages } : {}), diff --git a/src/engines/SessionCore/sync/adapters/__tests__/createRustAgentAdapter.authoritative.test.ts b/src/engines/SessionCore/sync/adapters/__tests__/createRustAgentAdapter.authoritative.test.ts index b442ea6c3d..f7593c0650 100644 --- a/src/engines/SessionCore/sync/adapters/__tests__/createRustAgentAdapter.authoritative.test.ts +++ b/src/engines/SessionCore/sync/adapters/__tests__/createRustAgentAdapter.authoritative.test.ts @@ -97,6 +97,40 @@ describe("Rust Agent authoritative history", () => { }); }); + it("requests embedded native rows and preserves durable force-send identity", async () => { + const loadMessages = vi.fn(async (_sessionId: string, native?: boolean) => [ + row("persisted-image-user", "user", { + content: "expanded provider payload", + images: JSON.stringify([ + native + ? "data:image/png;base64,QUJD" + : "C:\\Users\\fixture\\image.png", + ]), + ...(native ? { turnIntentId: "force-send-intent" } : {}), + }), + ]); + const adapter = createRustAgentAdapter({ + category: "agent", + features: {}, + loadMessages, + cancel: async () => {}, + }); + const events = await adapter.loadAuthoritativeHistory!( + "sdeagent-parent", + new AbortController().signal + ); + expect(loadMessages).toHaveBeenCalledWith("sdeagent-parent", true); + expect(projectNativeConversationItems(events)).toMatchObject([ + { + kind: "message", + role: "user", + text: "expanded provider payload", + images: ["data:image/png;base64,QUJD"], + turnId: "force-send-intent", + }, + ]); + }); + it("keeps display backfills out of canonical tool arguments across repeated reads", async () => { const args = { description: "Fast echo reply", diff --git a/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts b/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts index 8cf145ec85..9c68dd9617 100644 --- a/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts +++ b/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts @@ -65,7 +65,10 @@ export interface RustAgentConfig { category: string; /** Async function to load persisted messages */ - loadMessages: (sessionId: string) => Promise; + loadMessages: ( + sessionId: string, + forNativeTransfer?: boolean + ) => Promise; /** Async function to cancel/stop the session */ cancel: (sessionId: string, reason: CancelReason) => Promise; @@ -116,7 +119,7 @@ export function createRustAgentAdapter( sessionId: string, signal: AbortSignal ): Promise => { - const messages = await loadMessages(sessionId); + const messages = await loadMessages(sessionId, true); if (signal.aborted || !messages?.length) return []; // Match Rust load_native_history's complete effective frame. // The boundary is appended after its retained tail, so simply clearing @@ -233,8 +236,10 @@ export function createRustAgentAdapter( /** Unified agent configuration — handles all Rust-native agents (OS, SDE, custom). */ export const AGENT_CONFIG: RustAgentConfig = { category: "agent", - loadMessages: (sessionId) => - loadMessages(sessionId) as Promise as Promise, + loadMessages: (sessionId, forNativeTransfer) => + loadMessages(sessionId, forNativeTransfer) as Promise as Promise< + PersistedMessage[] + >, cancel: (sessionId, reason) => cancelSession(sessionId, reason) as unknown as Promise, tokenUsageCommand: "get_session_token_usage_records",