diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index e9ba7e842e..b852e340a8 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -104,7 +104,7 @@ The daemon's accepted response already waits for the correlated run start, but i directory update reach client state separately. An accepted transaction remains active until the directory observes that run or canonical ingestion acknowledges the prompt, bridging those ordered authorities without inspecting timeline snapshots. Either signal clears only an RPC-accepted -transaction; it cannot settle a fresh send. +transaction, regardless of which arrived first; it cannot settle a fresh send. Overlapping sends settle independently rather than collapsing to one newest pending message. Canonical submitted user rows carry the provider's `messageId` and Paseo's optional diff --git a/packages/app/e2e/agent-message-submission.spec.ts b/packages/app/e2e/agent-message-submission.spec.ts index 36d800ff15..d67332e689 100644 --- a/packages/app/e2e/agent-message-submission.spec.ts +++ b/packages/app/e2e/agent-message-submission.spec.ts @@ -340,6 +340,37 @@ async function expectCompletedSubmissionClearsAfterMissedRunningTransition( } } +async function expectProviderAcknowledgementBeforeRpcAcceptanceSettlesSubmission( + page: Page, + testInfo: { workerIndex: number }, +): Promise { + const gate = await installDaemonWebSocketGate(page); + const agent = await seedMockAgentWorkspace({ + repoPrefix: `submission-ack-before-rpc-${testInfo.workerIndex}-`, + title: "Submission acknowledgement before RPC", + model: "ten-second-stream", + }); + const prompt = "Settle this provider-acknowledged submission."; + try { + await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId }); + await expectComposerVisible(page); + await expectAgentIdle(page); + gate.setServerMessageSuppressed("agent_status", true); + gate.setServerMessageSuppressed("agent_update", true); + gate.holdNextServerMessage("send_agent_message_response"); + const userMessage = await submitMessageWithImage(page, prompt); + await gate.waitForHeldServerMessage(); + await gate.waitForAgentStreamItem("user_message"); + gate.releaseHeldServerMessage(); + await gate.drop(); + await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); + await expect(userMessage).toHaveAttribute("aria-busy", "false"); + } finally { + gate.restore(); + await agent.cleanup(); + } +} + async function expectLegacyAssistantStartsAfterInterruptedPrompt( page: Page, testInfo: { workerIndex: number }, @@ -567,6 +598,13 @@ test.describe("Agent message submission", () => { await expectCompletedSubmissionClearsAfterMissedRunningTransition(page, testInfo); }); + test("clears a provider acknowledgement that arrives before RPC acceptance", async ({ + page, + }, testInfo) => { + test.setTimeout(90_000); + await expectProviderAcknowledgementBeforeRpcAcceptanceSettlesSubmission(page, testInfo); + }); + test("keeps an old-daemon replacement answer after its interrupted prompt", async ({ page, }, testInfo) => { diff --git a/packages/app/e2e/helpers/daemon-websocket-gate.ts b/packages/app/e2e/helpers/daemon-websocket-gate.ts index d376c84916..379717c668 100644 --- a/packages/app/e2e/helpers/daemon-websocket-gate.ts +++ b/packages/app/e2e/helpers/daemon-websocket-gate.ts @@ -89,6 +89,17 @@ function readAgentStreamEventType(message: ClientRequest | null): string | null return typeof event?.type === "string" ? event.type : null; } +function readAgentStreamItemType(message: ClientRequest | null): string | null { + if (message?.type !== "agent_stream" || !message.payload || typeof message.payload !== "object") { + return null; + } + const event = (message.payload as { event?: { type?: unknown; item?: { type?: unknown } } }) + .event; + return event?.type === "timeline" && typeof event.item?.type === "string" + ? event.item.type + : null; +} + function shouldSuppressServerMessage(input: { message: ClientRequest | null; messageTypes: ReadonlySet; @@ -125,6 +136,7 @@ export async function installDaemonWebSocketGate(page: Page) { }; const clientRequestCounts = new Map(); const serverMessageCounts = new Map(); + const agentStreamItemCounts = new Map(); const serverMessageWaiters = new Set<() => void>(); await page.routeWebSocket(daemonWsRoutePattern(), (ws) => { @@ -192,6 +204,15 @@ export async function installDaemonWebSocketGate(page: Page) { for (const resolve of serverMessageWaiters) resolve(); serverMessageWaiters.clear(); } + const agentStreamItemType = readAgentStreamItemType(serverMessage); + if (agentStreamItemType) { + agentStreamItemCounts.set( + agentStreamItemType, + (agentStreamItemCounts.get(agentStreamItemType) ?? 0) + 1, + ); + for (const resolve of serverMessageWaiters) resolve(); + serverMessageWaiters.clear(); + } if (serverMessage?.type === heldServerMessageType) { heldServerMessage = { browser: ws, message: outboundMessage }; resolveHeldServerMessage?.(); @@ -360,5 +381,10 @@ export async function installDaemonWebSocketGate(page: Page) { await new Promise((resolve) => serverMessageWaiters.add(resolve)); } }, + async waitForAgentStreamItem(type: string, count = 1): Promise { + while ((agentStreamItemCounts.get(type) ?? 0) < count) { + await new Promise((resolve) => serverMessageWaiters.add(resolve)); + } + }, }; } diff --git a/packages/app/e2e/new-workspace.spec.ts b/packages/app/e2e/new-workspace.spec.ts index cd5cd86082..06512a12d6 100644 --- a/packages/app/e2e/new-workspace.spec.ts +++ b/packages/app/e2e/new-workspace.spec.ts @@ -600,6 +600,7 @@ test.describe("New workspace flow", () => { assertHeader: false, }); createdWorktreeDirectories.add(createdWorkspace.workspaceDirectory); + await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); const rowTestId = `sidebar-workspace-row-${serverId}:${createdWorkspace.workspaceId}`; await expectWorkspaceStatusGroupEvents({ diff --git a/packages/app/e2e/rewind-menu.ui-contract.spec.ts b/packages/app/e2e/rewind-menu.ui-contract.spec.ts index 6e3f494487..7c12f3f5b0 100644 --- a/packages/app/e2e/rewind-menu.ui-contract.spec.ts +++ b/packages/app/e2e/rewind-menu.ui-contract.spec.ts @@ -87,6 +87,28 @@ async function waitForCurrentSubmissionExcludedFromCache( .toBe(true); } +async function waitForCachedMessageWithoutProviderId(page: Page, prompt: string): Promise { + await expect + .poll(() => + page.evaluate((messageText) => { + const raw = localStorage.getItem("@paseo:replica-cache"); + if (!raw) return false; + const cache = JSON.parse(raw) as { + hosts?: Array<{ timeline?: { items?: Array> } | null }>; + }; + return cache.hosts + ?.flatMap((host) => host.timeline?.items ?? []) + .some( + (item) => + item.kind === "user_message" && + item.text === messageText && + item.messageId === undefined, + ); + }, prompt), + ) + .toBe(true); +} + async function expectPendingSubmissionNotRestoredAfterReload(page: Page): Promise { const prompt = "Keep this cached submission pending."; const gate = await installDaemonWebSocketGate(page); @@ -117,7 +139,7 @@ test.describe("Rewind sheet", () => { await expectPendingSubmissionNotRestoredAfterReload(page); }); - test("keeps rewind available for a message restored from the legacy cache", async ({ page }) => { + test("does not invent rewind identity for an ID-less cached message", async ({ page }) => { const prompt = "Restore this rewind identity from the legacy cache."; const gate = await installDaemonWebSocketGate(page); const session = await seedMockAgentWorkspace({ @@ -139,7 +161,8 @@ test.describe("Rewind sheet", () => { const restoredMessage = userMessage(page, prompt); await expect(restoredMessage).toBeVisible(); await restoredMessage.hover(); - await expect(restoredMessage.getByTestId("rewind-menu-trigger")).toBeVisible(); + await expect(restoredMessage.getByTestId("rewind-menu-trigger")).toHaveCount(0); + await waitForCachedMessageWithoutProviderId(page, prompt); } finally { if (heldTimelineRequest) gate.releaseHeldClientRequest(); gate.restore(); diff --git a/packages/app/src/composer/submission/model.ts b/packages/app/src/composer/submission/model.ts index a9e8fc154b..2a46280571 100644 --- a/packages/app/src/composer/submission/model.ts +++ b/packages/app/src/composer/submission/model.ts @@ -50,7 +50,7 @@ export function acceptMessageSubmission( (submission) => submission.clientMessageId === clientMessageId, ); if (index < 0) return submissions as MessageSubmissionRecord[]; - if (isAgentRunning) { + if (isAgentRunning || submissions[index].providerAcknowledged) { return submissions.filter((_, submissionIndex) => submissionIndex !== index); } if (submissions[index].rpcAccepted) return submissions as MessageSubmissionRecord[]; diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index f2d9eea418..b9bfab5c0c 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -19,9 +19,10 @@ import { } from "@/stores/session-store"; import type { StreamItem } from "@/types/stream"; import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; +import { getSendingClientMessageIds } from "@/composer/submission/model"; const STORAGE_KEY = "@paseo:replica-cache"; -const CACHE_VERSION = 1; +const CACHE_VERSION = 2; const PERSIST_DELAY_MS = 750; const MAX_TIMELINE_ITEMS = 50; const MAX_CACHE_BYTES = 1024 * 1024; @@ -129,15 +130,6 @@ function decodeDates(value: unknown): unknown { return Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, decodeDates(entry)])); } -function restoreCachedStreamItem(item: StreamItem): StreamItem | null { - if (item.kind !== "user_message" || item.messageId) return item; - const legacyItem = item as StreamItem & { optimistic?: true }; - if (legacyItem.optimistic || item.clientMessageId) return null; - // COMPAT(replicaCacheUserMessageId): before v0.2.3, cache v1 stored the provider ID - // only in `id`. Remove after 2027-01-27, once cache v1 predating v0.2.3 is unsupported. - return { ...item, messageId: item.id }; -} - function deserializeTimeline(stored: StoredHost["timeline"]): SessionReplica["timeline"] { if (!stored) { return null; @@ -148,7 +140,7 @@ function deserializeTimeline(stored: StoredHost["timeline"]): SessionReplica["ti } return { agentId: stored.agentId, - items: decoded.flatMap((item) => restoreCachedStreamItem(item) ?? []), + items: decoded, cursor: stored.cursor, hasOlder: stored.hasOlder, }; @@ -379,10 +371,21 @@ export class ReplicaCache { (workspace) => workspace.workspaceDirectory === focusedAgent.cwd, )) : undefined; + const localSubmissionIds = new Set( + getSendingClientMessageIds( + focusedAgentId ? session.messageSubmissions.get(focusedAgentId) : undefined, + ), + ); const items = focusedAgentId ? session.agentStreamTail .get(focusedAgentId) - ?.filter((item) => item.kind !== "user_message" || item.messageId !== undefined) + ?.filter( + (item) => + item.kind !== "user_message" || + item.messageId !== undefined || + !item.clientMessageId || + !localSubmissionIds.has(item.clientMessageId), + ) : undefined; const timeline = focusedAgent && items