diff --git a/apps/mobile/src/chat.tsx b/apps/mobile/src/chat.tsx index 683332b1..cc42712e 100644 --- a/apps/mobile/src/chat.tsx +++ b/apps/mobile/src/chat.tsx @@ -26,7 +26,7 @@ import { BackgroundUpdates } from "./background-updates"; import { BrowserRunContext, BrowserToolCard } from "./browser-tool-card"; import { BrowserThreadCard } from "./computer"; import { ConversationQueue, type QueuedMessage } from "./conversation-queue"; -import { runConversationTurn } from "./conversation-run"; +import { replayedRunError, runConversationTurn } from "./conversation-run"; import { MailToolCard } from "./mail-tool-card"; import { FileThreadCard, TaskThreadCard } from "./thread-artifacts"; import { type Selection, useMuseThread } from "./threads"; @@ -200,6 +200,8 @@ export function ChatScreen({ const [saveError, setSaveError] = useState(""); const [historyError, setHistoryError] = useState(""); const [historyAttempt, setHistoryAttempt] = useState(0); + // Loads still replaying history. A count, so an old load finishing does not end a newer one. + const replaying = useRef(0); useEffect(() => { if (!isReady) return; let active = true; @@ -213,12 +215,20 @@ export function ChatScreen({ async function hydrate() { try { if (richThreads) { - if (selection.existing) - await runConversationTurn( - agentId, - () => copilotkit.connectAgent({ agent }), - (onError) => copilotkit.subscribe({ onError }), - ); + if (selection.existing) { + // Replaying history re-emits past RUN_ERROR events; only connection failures block loading. + replaying.current += 1; + try { + await runConversationTurn( + agentId, + () => copilotkit.connectAgent({ agent }), + (onError) => copilotkit.subscribe({ onError }), + [replayedRunError], + ); + } finally { + replaying.current -= 1; + } + } } else { const { messages } = await api.request<{ messages: Message[] }>("/api/conversation"); if (active) agent.setMessages(messages); @@ -299,6 +309,8 @@ export function ChatScreen({ const subscription = copilotkit.subscribe({ onError: (event) => { if (event.context?.agentId && event.context.agentId !== agentId) return; + // A failed turn saved in history is already over; it is not a failure of this session. + if (replaying.current > 0 && event.code === replayedRunError) return; const failure = event.error instanceof Error ? event.error : new Error(String(event.error)); setError(failure.message); }, diff --git a/apps/mobile/src/conversation-run.ts b/apps/mobile/src/conversation-run.ts index 73e633ac..9eeae17e 100644 --- a/apps/mobile/src/conversation-run.ts +++ b/apps/mobile/src/conversation-run.ts @@ -1,14 +1,19 @@ -type RunError = { error: unknown; context?: { agentId?: string } }; +type RunError = { error: unknown; code?: string; context?: { agentId?: string } }; + +/** A RUN_ERROR event from the thread's saved history, not a failure of this call. */ +export const replayedRunError = "agent_run_error_event"; /** CopilotKit emits run failures through onError even when runAgent resolves. */ export async function runConversationTurn( agentId: string, execute: () => Promise, subscribe: (listener: (event: RunError) => void) => { unsubscribe: () => void }, + ignore: readonly string[] = [], ) { let failure: Error | undefined; const subscription = subscribe((event) => { if (event.context?.agentId && event.context.agentId !== agentId) return; + if (event.code && ignore.includes(event.code)) return; failure = event.error instanceof Error ? event.error : new Error(String(event.error)); }); try { diff --git a/tests/conversation-sdk.test.ts b/tests/conversation-sdk.test.ts index f266b302..8f92d817 100644 --- a/tests/conversation-sdk.test.ts +++ b/tests/conversation-sdk.test.ts @@ -1,10 +1,11 @@ import assert from "node:assert/strict"; import { test } from "node:test"; import { AbstractAgent } from "@ag-ui/client"; +import { EventType } from "@ag-ui/core"; import { CopilotKitCore } from "@copilotkit/core"; -import { throwError } from "rxjs"; +import { of, throwError } from "rxjs"; import { ConversationQueue } from "../apps/mobile/src/conversation-queue.ts"; -import { runConversationTurn } from "../apps/mobile/src/conversation-run.ts"; +import { replayedRunError, runConversationTurn } from "../apps/mobile/src/conversation-run.ts"; test("an emitted CopilotKit run error stops the queue even when runAgent resolves", async () => { let attempts = 0; @@ -36,3 +37,50 @@ test("an emitted CopilotKit run error stops the queue even when runAgent resolve ["second"], ); }); + +test("a failed turn saved in thread history does not block loading the conversation", async () => { + class ReplayingAgent extends AbstractAgent { + run() { + return throwError(() => new Error("not used")); + } + connect() { + return of( + { type: EventType.RUN_STARTED, threadId: "thread", runId: "old-run" }, + { type: EventType.RUN_ERROR, message: "Missing Authentication header" }, + ); + } + } + const agent = new ReplayingAgent({ agentId: "default", threadId: "thread" }); + const core = new CopilotKitCore({ agents__unsafe_dev_only: { default: agent } }); + const connect = (ignore?: string[]) => + runConversationTurn( + "default", + () => core.connectAgent({ agent }), + (onError) => core.subscribe({ onError }), + ignore, + ); + await assert.rejects(connect(), /Missing Authentication header/); + await connect([replayedRunError]); +}); + +test("a connection failure still blocks loading the conversation", async () => { + class UnreachableAgent extends AbstractAgent { + run() { + return throwError(() => new Error("not used")); + } + connect() { + return throwError(() => new Error("Thread service unavailable")); + } + } + const agent = new UnreachableAgent({ agentId: "default", threadId: "thread" }); + const core = new CopilotKitCore({ agents__unsafe_dev_only: { default: agent } }); + await assert.rejects( + runConversationTurn( + "default", + () => core.connectAgent({ agent }), + (onError) => core.subscribe({ onError }), + [replayedRunError], + ), + /Thread service unavailable/, + ); +});