fix(webapp,sdk): keep chat.agent snapshots on one object store (#3679)

(`OBJECT_STORE_BASE_URL`) and a named protocol provider
(`OBJECT_STORE_DEFAULT_PROTOCOL=s3`), chat.agent session snapshot writes
landed in the named provider but reads fell through to the default — so
the recovery boot couldn't find the snapshot it had just written.

After a mid-stream cancel, the missing snapshot triggered a fallback
replay path that dropped the user's follow-up message, leaving the chat
stuck in `submitted` indefinitely.

Fix:
- New `/api/v1/sessions/:id/snapshot-url` route handles PUT + GET
  symmetrically — both prefix unprefixed keys with
  `OBJECT_STORE_DEFAULT_PROTOCOL` so they always round-trip through the
  same store.
- `Session.chatSnapshotStoragePath` persists the resolved URI on first
  write so future protocol changes don't strand existing snapshots.
  Reads prefer the stored URI and fall back to the computed default for
  pre-column sessions.
- SDK calls `createChatSnapshotUploadUrl` / `getChatSnapshotUrl`; the
  generic v1/v2 packets endpoints are unchanged.

## Test plan
- [x] Configure local with two providers (R2 default + MinIO `s3` named)
      and `OBJECT_STORE_DEFAULT_PROTOCOL=s3`.
- [x] Reproduce hang: send a message, cancel mid-stream, send another —
      without the fix it hangs in `submitted`; with the fix it streams.
- [x] Snapshot lands in the `s3`-protocol bucket and
      `Session.chatSnapshotStoragePath` is set after first write.
- [x] SDK unit tests pass; webapp typecheck passes.

Author: ericallam

.server-changes/chat-snapshot-default-protocol.md

@@ -0,0 +1,8 @@
+---
+area: webapp
+type: fix
+---
+
+Pin chat.agent session snapshots to a single object store so writes and reads
+always round-trip through the same provider when `OBJECT_STORE_DEFAULT_PROTOCOL`
+is set.

apps/webapp/app/routes/api.v1.sessions.$sessionId.snapshot-url.ts

@@ -0,0 +1,69 @@
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { chatSnapshotStoragePathForSession } from "~/services/realtime/chatSnapshot.server";
+import { resolveSessionByIdOrExternalId } from "~/services/realtime/sessions.server";
+import {
+  createActionApiRoute,
+  createLoaderApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
+import { generatePresignedUrl } from "~/v3/objectStore.server";
+
+const ParamsSchema = z.object({
+  sessionId: z.string(),
+});
+
+// `chatSnapshotStoragePath` is stamped on every new Session at row creation
+// (see api.v1.sessions.ts). The fallback handles sessions created before
+// the column existed — read against the currently-configured default
+// protocol and compute the same path the SDK uploaded under.
+function snapshotKey(session: { friendlyId: string; chatSnapshotStoragePath: string | null }) {
+  return session.chatSnapshotStoragePath ?? chatSnapshotStoragePathForSession(session.friendlyId);
+}
+
+const routeConfig = {
+  params: ParamsSchema,
+  allowJWT: true,
+  corsStrategy: "all" as const,
+  findResource: async (params: z.infer<typeof ParamsSchema>, auth: { environment: { id: string } }) =>
+    resolveSessionByIdOrExternalId($replica, auth.environment.id, params.sessionId),
+};
+
+export const { action } = createActionApiRoute(
+  { ...routeConfig, method: "PUT" },
+  async ({ authentication, resource: session }) => {
+    if (!session) {
+      return json({ error: "Session not found" }, { status: 404 });
+    }
+
+    const signed = await generatePresignedUrl(
+      authentication.environment.project.externalRef,
+      authentication.environment.slug,
+      snapshotKey(session),
+      "PUT"
+    );
+    if (!signed.success) {
+      return json({ error: `Failed to generate presigned URL: ${signed.error}` }, { status: 500 });
+    }
+
+    return json({ presignedUrl: signed.url });
+  }
+);
+
+export const loader = createLoaderApiRoute(routeConfig, async ({ authentication, resource: session }) => {
+  if (!session) {
+    return json({ error: "Session not found" }, { status: 404 });
+  }
+
+  const signed = await generatePresignedUrl(
+    authentication.environment.project.externalRef,
+    authentication.environment.slug,
+    snapshotKey(session),
+    "GET"
+  );
+  if (!signed.success) {
+    return json({ error: `Failed to generate presigned URL: ${signed.error}` }, { status: 500 });
+  }
+
+  return json({ presignedUrl: signed.url });
+});

apps/webapp/app/routes/api.v1.sessions.ts

@@ -17,6 +17,7 @@ import {
   ensureRunForSession,
   type SessionTriggerConfig,
 } from "~/services/realtime/sessionRunManager.server";
+import { chatSnapshotStoragePathForSession } from "~/services/realtime/chatSnapshot.server";
 import { serializeSession } from "~/services/realtime/sessions.server";
 import { SessionsRepository } from "~/services/sessionsRepository/sessionsRepository.server";
 import {
@@ -181,6 +182,7 @@ const { action } = createActionApiRoute(
             environmentType: authentication.environment.type,
             organizationId: authentication.environment.organizationId,
             streamBasinName: authentication.environment.organization.streamBasinName,
+            chatSnapshotStoragePath: chatSnapshotStoragePathForSession(friendlyId),
           },
           update: { triggerConfig: triggerConfigJson },
         });
@@ -201,6 +203,7 @@ const { action } = createActionApiRoute(
             environmentType: authentication.environment.type,
             organizationId: authentication.environment.organizationId,
             streamBasinName: authentication.environment.organization.streamBasinName,
+            chatSnapshotStoragePath: chatSnapshotStoragePathForSession(friendlyId),
           },
         });
       }

apps/webapp/app/services/apiRateLimit.server.ts

@@ -70,6 +70,7 @@ export const apiRateLimiter = authorizationRateLimitMiddleware({
     // customer-facing surface so customer rate limits shouldn't apply.
     /^\/api\/v1\/packets\//,
     /^\/api\/v2\/packets\//,
+    /^\/api\/v1\/sessions\/[^\/]+\/snapshot-url$/,
   ],
   log: {
     rejections: env.API_RATE_LIMIT_REJECTION_LOGS_ENABLED === "1",

apps/webapp/app/services/realtime/chatSnapshot.server.ts

@@ -0,0 +1,13 @@
+import { env } from "~/env.server";
+
+/**
+ * Canonical storage URI for a session's chat.agent snapshot. Stamped on
+ * `Session.chatSnapshotStoragePath` at row creation so PUT/GET presigns
+ * resolve to the same store even if `OBJECT_STORE_DEFAULT_PROTOCOL`
+ * changes later.
+ */
+export function chatSnapshotStoragePathForSession(friendlyId: string): string {
+  const path = `sessions/${friendlyId}/snapshot.json`;
+  const protocol = env.OBJECT_STORE_DEFAULT_PROTOCOL;
+  return protocol ? `${protocol}://${path}` : path;
+}

apps/webapp/test/chat-snapshot-integration.test.ts

@@ -26,6 +26,7 @@ import {
 import type { UIMessage } from "ai";
 import { afterEach, describe, expect, vi } from "vitest";
 import { env } from "~/env.server";
+import { chatSnapshotStoragePathForSession } from "~/services/realtime/chatSnapshot.server";
 import { generatePresignedUrl } from "~/v3/objectStore.server";
 
 vi.setConfig({ testTimeout: 60_000 });
@@ -54,22 +55,21 @@ function makeSnapshot(opts: { messages?: UIMessage[]; lastOutEventId?: string }
 
 /**
  * Stub `apiClientManager.clientOrThrow()` so the SDK helpers see a fake
- * api client whose `getPayloadUrl` / `createUploadPayloadUrl` return
- * presigned URLs minted by the webapp's real `generatePresignedUrl`
- * (which signs against MinIO).
- *
- * The SDK helpers internally do `fetch(presignedUrl, ...)` to read/write
- * the blob, so MinIO ends up holding the actual bytes.
+ * api client. Mirrors the snapshot-url route: derive the canonical
+ * `sessions/{id}/snapshot.json` key (with optional default-protocol
+ * prefix) and sign it via `generatePresignedUrl` against MinIO.
  */
 function stubApiClient(opts: { projectRef: string; envSlug: string }) {
   vi.spyOn(apiClientManager, "clientOrThrow").mockReturnValue({
-    async getPayloadUrl(filename: string) {
-      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, filename, "GET");
+    async getChatSnapshotUrl(sessionId: string) {
+      const key = chatSnapshotStoragePathForSession(sessionId);
+      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, key, "GET");
       if (!result.success) throw new Error(result.error);
       return { presignedUrl: result.url };
     },
-    async createUploadPayloadUrl(filename: string) {
-      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, filename, "PUT");
+    async createChatSnapshotUploadUrl(sessionId: string) {
+      const key = chatSnapshotStoragePathForSession(sessionId);
+      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, key, "PUT");
       if (!result.success) throw new Error(result.error);
       return { presignedUrl: result.url };
     },

apps/webapp/test/replay-after-crash.test.ts

@@ -33,6 +33,7 @@ import {
 import type { UIMessageChunk } from "ai";
 import { afterEach, describe, expect, vi } from "vitest";
 import { env } from "~/env.server";
+import { chatSnapshotStoragePathForSession } from "~/services/realtime/chatSnapshot.server";
 import { generatePresignedUrl } from "~/v3/objectStore.server";
 
 vi.setConfig({ testTimeout: 60_000 });
@@ -77,13 +78,15 @@ function stubApiClient(opts: {
     })
   );
   vi.spyOn(apiClientManager, "clientOrThrow").mockReturnValue({
-    async getPayloadUrl(filename: string) {
-      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, filename, "GET");
+    async getChatSnapshotUrl(sessionId: string) {
+      const key = chatSnapshotStoragePathForSession(sessionId);
+      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, key, "GET");
       if (!result.success) throw new Error(result.error);
       return { presignedUrl: result.url };
     },
-    async createUploadPayloadUrl(filename: string) {
-      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, filename, "PUT");
+    async createChatSnapshotUploadUrl(sessionId: string) {
+      const key = chatSnapshotStoragePathForSession(sessionId);
+      const result = await generatePresignedUrl(opts.projectRef, opts.envSlug, key, "PUT");
       if (!result.success) throw new Error(result.error);
       return { presignedUrl: result.url };
     },

internal-packages/database/prisma/migrations/20260520170000_add_session_chat_snapshot_storage_path/migration.sql

@@ -0,0 +1 @@
+ALTER TABLE "Session" ADD COLUMN IF NOT EXISTS "chatSnapshotStoragePath" TEXT;

internal-packages/database/prisma/schema.prisma

@@ -830,6 +830,11 @@ model Session {
   /// (OSS, or pre-backfill); reads fall back to the global basin.
   streamBasinName String?
 
+  /// Storage URI (with protocol prefix) for this session's chat.agent
+  /// snapshot blob. Set on first snapshot write. Null = pre-column session,
+  /// fall back to computed default path.
+  chatSnapshotStoragePath String?
+
   runs SessionRun[]
 
   /// Idempotency: `(env, externalId)` uniquely identifies a session.

packages/core/src/v3/apiClient/index.ts

@@ -602,6 +602,32 @@ export class ApiClient {
     );
   }
 
+  /** Presigned PUT URL for a `chat.agent` session snapshot. */
+  createChatSnapshotUploadUrl(sessionId: string, requestOptions?: ZodFetchOptions) {
+    return zodfetch(
+      CreateUploadPayloadUrlResponseBody,
+      `${this.baseUrl}/api/v1/sessions/${encodeURIComponent(sessionId)}/snapshot-url`,
+      {
+        method: "PUT",
+        headers: this.#getHeaders(false),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
+  /** Presigned GET URL for a `chat.agent` session snapshot. */
+  getChatSnapshotUrl(sessionId: string, requestOptions?: ZodFetchOptions) {
+    return zodfetch(
+      CreateUploadPayloadUrlResponseBody,
+      `${this.baseUrl}/api/v1/sessions/${encodeURIComponent(sessionId)}/snapshot-url`,
+      {
+        method: "GET",
+        headers: this.#getHeaders(false),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
   retrieveRun(runId: string, requestOptions?: ZodFetchOptions) {
     return zodfetch(
       RetrieveRunResponse,