fix(app): settle submissions in either acknowledgement order

Complete submission transactions when RPC and provider acknowledgement arrive in either order. Cache only transaction-owned local rows as transient data, preserve canonical ID-less prompts, and invalidate ambiguous legacy display caches instead of inventing provider identity.
2026-11-06 16:34:36 +00:00 · 2026-07-27 17:49:42 +02:00
parent 292cea4bcc
commit 5334008e44
7 changed files with 107 additions and 16 deletions
@@ -340,6 +340,37 @@ async function expectCompletedSubmissionClearsAfterMissedRunningTransition(
  }
 }

+async function expectProviderAcknowledgementBeforeRpcAcceptanceSettlesSubmission(
+  page: Page,
+  testInfo: { workerIndex: number },
+): Promise<void> {
+  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) => {
@@ -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<string>;
@@ -125,6 +136,7 @@ export async function installDaemonWebSocketGate(page: Page) {
  };
  const clientRequestCounts = new Map<string, number>();
  const serverMessageCounts = new Map<string, number>();
+  const agentStreamItemCounts = new Map<string, number>();
  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<void>((resolve) => serverMessageWaiters.add(resolve));
      }
    },
+    async waitForAgentStreamItem(type: string, count = 1): Promise<void> {
+      while ((agentStreamItemCounts.get(type) ?? 0) < count) {
+        await new Promise<void>((resolve) => serverMessageWaiters.add(resolve));
+      }
+    },
  };
 }
@@ -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({
@@ -87,6 +87,28 @@ async function waitForCurrentSubmissionExcludedFromCache(
    .toBe(true);
 }

+async function waitForCachedMessageWithoutProviderId(page: Page, prompt: string): Promise<void> {
+  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<Record<string, unknown>> } | 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<void> {
  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();
@@ -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[];
@@ -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