Classify RPC errors and recover quietly from Durable Object resets

packages/workshop-backend/__integration__/open-gadget-rpc.test.ts

@@ -1,3 +1,4 @@
+import { abortAllDurableObjects } from "cloudflare:test";
 import { exports } from "cloudflare:workers";
 import { newWebSocketRpcSession, type RpcStub } from "capnweb";
 import {
@@ -133,3 +134,46 @@ describe.skip("openGadget errors across native RPC and Cap'n Web", () => {
     expectRpcCode(browserError, OPEN_GADGET_ERROR_CODES.workspaceAccessDenied);
   });
 });
+
+// The DO-reset recovery contract: a user-DO reset must not poison the API session. E-order is
+// per-stub, so the authenticated API shares one user-DO stub per session while it's healthy; a
+// reset permanently breaks that stub, so a stub-breaking rejection drops the cache, and a
+// rejection that proves the call was never sent (the flagless dead-capability error — the
+// flagged one only reaches calls in flight at reset time) is transparently re-issued once on a
+// fresh stub. abortAllDurableObjects() is the non-graceful teardown, the local stand-in for the
+// storage-timeout/overload resets observed in production. (Deliberately not
+// evictDurableObject(): eviction is graceful — it drains in-flight work and never breaks a
+// stub — so it cannot reproduce this failure.)
+describe("user-DO reset recovery", () => {
+  it("recovers on the same session after the user DO is reset", async () => {
+    using publicApi = await connect();
+    const account = await createAccount(publicApi, "reset");
+    using authenticated = await publicApi.authenticate(account.token);
+
+    expect(await authenticated.listModels()).toBeInstanceOf(Array);
+
+    await abortAllDurableObjects();
+
+    // Same socket, same AuthenticatedApiImpl. The cached stub is dead, so this call rejects
+    // locally without reaching the DO and is re-issued once on a fresh stub — it must succeed
+    // with no client-visible failure. This doubles as the canary for workerd's dead-capability
+    // message: if that string drifts, the never-sent retry stops firing and this fails loudly.
+    expect(await authenticated.listModels()).toBeInstanceOf(Array);
+  });
+
+  it("re-arms the cached stub after recovery instead of churning or staying poisoned", async () => {
+    using publicApi = await connect();
+    const account = await createAccount(publicApi, "rearm");
+    using authenticated = await publicApi.authenticate(account.token);
+
+    expect(await authenticated.listModels()).toBeInstanceOf(Array);
+
+    await abortAllDurableObjects();
+
+    // First call recovers via the never-sent retry; the calls after it must ride the re-armed
+    // cached stub (a poisoned or thrashing cache would reject here).
+    expect(await authenticated.listModels()).toBeInstanceOf(Array);
+    expect(await authenticated.isOnboardingCompleted()).toBeTypeOf("boolean");
+    expect(await authenticated.whoami()).toBeTruthy();
+  });
+});

packages/workshop-backend/src/server.ts

@@ -1,7 +1,7 @@
 import { RpcStub, RpcTarget, newWorkersRpcResponse } from "capnweb";
 import { validateRpc } from "capnweb-validate";
 import type { JWTPayload } from "jose";
-import { PublicApi, AuthenticatedApi, Overseer, GadgetMetadataWithTimestamps, AiChatAuthorInfo, AiModelConfig, AiGatewayInfo, AiModelProvider, ConnectedAccountsSubscriber, ConnectedAccountsFilter, GatekeeperVendorFilter, ObserverConfigCallback, BlueprintLibrarySummary, BlueprintPublicInfo, BlueprintUserSummary, BlueprintBindingAssignment, AgentSpawnerConfig, WorkpieceId, BLUEPRINT_SCREENSHOT_PATH_PREFIX, BLUEPRINT_SCREENSHOT_R2_PREFIX, blueprintScreenshotUrl, ServerConfig, CloudflareUsageInfo, CloudflareAccountOption, LoginAttempt, GatekeeperAppInfo, AdminApi, GatekeeperVendorInfo, OutputFormatOffer, ListOutputsResult, createOpenGadgetError, getOpenGadgetErrorCode, OPEN_GADGET_ERROR_CODES } from '@gadgets/workshop-shared/api';
+import { PublicApi, AuthenticatedApi, Overseer, GadgetMetadataWithTimestamps, AiChatAuthorInfo, AiModelConfig, AiGatewayInfo, AiModelProvider, ConnectedAccountsSubscriber, ConnectedAccountsFilter, GatekeeperVendorFilter, ObserverConfigCallback, BlueprintLibrarySummary, BlueprintPublicInfo, BlueprintUserSummary, BlueprintBindingAssignment, AgentSpawnerConfig, WorkpieceId, BLUEPRINT_SCREENSHOT_PATH_PREFIX, BLUEPRINT_SCREENSHOT_R2_PREFIX, blueprintScreenshotUrl, ServerConfig, CloudflareUsageInfo, CloudflareAccountOption, LoginAttempt, GatekeeperAppInfo, AdminApi, GatekeeperVendorInfo, OutputFormatOffer, ListOutputsResult, createOpenGadgetError, getOpenGadgetErrorCode, OPEN_GADGET_ERROR_CODES, AUTH_ERROR_CODES, createAuthError, WORKERD_DEAD_CAPABILITY_MESSAGE } from '@gadgets/workshop-shared/api';
 import type { UiFeatureFlags } from "@gadgets/workshop-shared/feature-flags";
 import { getServerConfig } from "./deployment-config.js";
 import { isPasswordAuthEnabled, getAuthGatekeeperAllowlist } from "./auth/config.js";
@@ -74,7 +74,7 @@ type Env = Cloudflare.Env & {
 @validateRpc()
 class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi {
   constructor(private ctx: ExecutionContext, private env: Env,
-      private user: DurableObjectStub<UserDurableObject>,
+      private userId: DurableObjectId,
       private abortSession: (reason: Error) => void) {
     super();
 
@@ -87,6 +87,71 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi {
   private adminSettings: DurableObjectNamespace<AdminSettings>;
   private users: DurableObjectNamespace<UserDurableObject>;
 
+  // One stub per session, because e-order (in-order delivery to the DO) is guaranteed per stub.
+  // A stub is permanently broken once its incarnation of the object resets, so the wrapper below
+  // drops both fields on a stub-breaking rejection and the next call re-resolves fresh ones.
+  // `rawUserStub` is the Proxy's target, kept so the never-sent retry can re-issue a call
+  // without re-entering the wrapper.
+  private userStub?: DurableObjectStub<UserDurableObject>;
+  private rawUserStub?: DurableObjectStub<UserDurableObject>;
+
+  private get user(): DurableObjectStub<UserDurableObject> {
+    return this.userStub ??=
+        this.#wrapUserStub(this.rawUserStub = this.users.get(this.userId));
+  }
+
+  // Calls on a stub whose DO already reset while idle reject flagless with the brokenness
+  // reason; the flagged (`durableObjectReset`) error only reaches calls in flight at reset time.
+  // The distinction matters: a flagless rejection matching one of these proves the call never
+  // reached the DO, so re-issuing it on a fresh stub cannot double-execute — safe even for writes.
+  static readonly #DEAD_CAPABILITY_MESSAGES = [
+    // What production resets leave as the brokenness reason; the frontend canary test pins it.
+    WORKERD_DEAD_CAPABILITY_MESSAGE,
+    // What vitest-pool-workers' abortAllDurableObjects() leaves — never occurs in production;
+    // listed so integration tests exercise the real never-sent retry path.
+    "Application called abortAllDurableObjects().",
+  ];
+
+  // Intercepts every method call on the user-DO stub. On a stub-breaking rejection, drops the
+  // cached stub so the next call re-resolves; if the rejection proves the call was never sent
+  // (see #DEAD_CAPABILITY_MESSAGES), re-issues it once on the fresh stub. Flagged errors are
+  // rethrown untouched — the call may have executed, and the frontend owns that recovery.
+  #wrapUserStub(stub: DurableObjectStub<UserDurableObject>): DurableObjectStub<UserDurableObject> {
+    return new Proxy(stub, {
+      get: (target, prop) => {
+        const value = Reflect.get(target, prop);
+        if (typeof value !== "function") return value;
+        return (...args: unknown[]) => {
+          // Reflect.apply, not value.apply(): on a JSRPC method proxy, `.apply` is an RPC path
+          // segment (it would invoke a remote method named "apply"), not Function.prototype.apply.
+          const result = Reflect.apply(value, target, args);
+          if (typeof (result as PromiseLike<unknown> | null)?.then !== "function") return result;
+          // JsRpcPromise.then validates its first parameter as a Function — no `undefined` slot.
+          return (result as Promise<unknown>).then((v: unknown) => v, (err: unknown) => {
+            const flags = err as { durableObjectReset?: unknown, retryable?: unknown } | null;
+            const flagged = flags?.durableObjectReset === true || flags?.retryable === true;
+            const neverSent = !flagged && err instanceof Error &&
+                AuthenticatedApiImpl.#DEAD_CAPABILITY_MESSAGES.some(m => err.message.includes(m));
+            if (flagged || neverSent) {
+              // Guard against thrashing: concurrent failures on the same dead stub must not
+              // each discard the replacement the first one already resolved.
+              if (this.rawUserStub === target) this.userStub = this.rawUserStub = undefined;
+              if (neverSent) {
+                // Retry exactly once, on the raw target of the re-armed cache — retrying through
+                // the proxy would retry unboundedly under repeated resets.
+                void this.user;
+                const fresh = this.rawUserStub!;
+                return Reflect.apply(Reflect.get(fresh, prop) as (...a: unknown[]) => unknown,
+                    fresh, args);
+              }
+            }
+            throw err;
+          });
+        };
+      },
+    });
+  }
+
   #isAdmin(): boolean {
     let name = this.user.id.name;
     let admins = this.env.ADMINS;
@@ -664,7 +729,7 @@ class PublicApiImpl extends RpcTarget implements PublicApi {
   async authenticate(token: string): Promise<AuthenticatedApi> {
     let split = token.split(':');
     if (split.length !== 2) {
-      throw new Error("Invalid session token.");
+      throw createAuthError(AUTH_ERROR_CODES.invalidSessionToken);
     }
 
     let userId = this.users.idFromName(split[0]);
@@ -675,12 +740,12 @@ class PublicApiImpl extends RpcTarget implements PublicApi {
       user_id: userId.toString(),
       source: "session_token",
     });
-    return new AuthenticatedApiImpl(this.ctx, this.env, stub, this.abortSession);
+    return new AuthenticatedApiImpl(this.ctx, this.env, userId, this.abortSession);
   }
 
   async authenticateFromCfAccess(): Promise<AuthenticatedApi> {
     if (!this.accessPayload) {
-      throw new Error("Not authenticated with Access.");
+      throw createAuthError(AUTH_ERROR_CODES.notAuthenticatedWithAccess);
     }
 
     let email = this.accessPayload.email as string;
@@ -700,7 +765,7 @@ class PublicApiImpl extends RpcTarget implements PublicApi {
       user_id: userId.toString(),
       source: "cf_access",
     });
-    return new AuthenticatedApiImpl(this.ctx, this.env, stub, this.abortSession);
+    return new AuthenticatedApiImpl(this.ctx, this.env, userId, this.abortSession);
   }
 
   async login(username: string, passwordHash: Uint8Array): Promise<string | null> {

packages/workshop-backend/src/user.ts

@@ -1,5 +1,5 @@
 import { RpcStub } from "capnweb";
-import { GadgetMetadataWithTimestamps, AiChatAuthorInfo, AiModelConfig, SUGGESTED_MODELS, CollaboratorRole, ConnectedAccountsSubscriber, ConnectedAccountsFilter, GatekeeperVendorFilter, GadgetMetadata, BlueprintMetadata, BlueprintLibrarySummary, BlueprintSource, BlueprintUserSummary, BLUEPRINT_SCREENSHOT_R2_PREFIX, GatekeeperVendorInfo, BlueprintOutput, OutputSummary, WorkpieceId, ListOutputsResult } from '@gadgets/workshop-shared/api';
+import { GadgetMetadataWithTimestamps, AiChatAuthorInfo, AiModelConfig, SUGGESTED_MODELS, CollaboratorRole, ConnectedAccountsSubscriber, ConnectedAccountsFilter, GatekeeperVendorFilter, GadgetMetadata, BlueprintMetadata, BlueprintLibrarySummary, BlueprintSource, BlueprintUserSummary, BLUEPRINT_SCREENSHOT_R2_PREFIX, GatekeeperVendorInfo, BlueprintOutput, OutputSummary, WorkpieceId, ListOutputsResult, AUTH_ERROR_CODES, createAuthError } from '@gadgets/workshop-shared/api';
 import { Gatekeeper, GatekeeperUser, GatekeeperUserVerifier, GatekeeperVendor, AccountDescription, VendorDescription, GatekeeperConnectCallback, SupportedResource, ResourceConfiguratorFrame, AppUiContext, GatekeeperUiFrame } from "@gadgets/workshop-shared/gatekeeper";
 import { shouldAutoProvisionAccount, ambientGatekeeperMode } from "./provisioning-policy.js";
 import { CloudflareGatekeeperUser } from "@gadgets/workshop-shared/cloudflare-gatekeeper";
@@ -296,12 +296,19 @@ export class UserDurableObject extends DurableObject<Cloudflare.Env> {
   }
 
   async authenticate(token: string): Promise<void> {
-    let tokenBytes = Uint8Array.fromBase64(token);
+    let tokenBytes: Uint8Array;
+    try {
+      tokenBytes = Uint8Array.fromBase64(token);
+    } catch {
+      // A corrupt (non-Base64) token must classify as an auth failure like any other bad token,
+      // not surface as the decoder's SyntaxError.
+      throw createAuthError(AUTH_ERROR_CODES.invalidSessionToken);
+    }
     let hash = await crypto.subtle.digest('SHA-256', tokenBytes);
     let tokenId = new Uint8Array(hash).toHex();
     let session = this.storage.sessions.get(tokenId);
     if (!session) {
-      throw new Error("invalid session token");
+      throw createAuthError(AUTH_ERROR_CODES.invalidSessionToken);
     }
   }
 

packages/workshop-frontend/src/BlueprintLandingPage.tsx

@@ -1,3 +1,4 @@
+import { logRpcFailure } from './rpcErrors'
 import { useState, useEffect, useCallback, useMemo, useRef, type ReactNode } from 'react'
 import { useNavigate, useParams, useRouter } from '@tanstack/react-router'
 import { RpcStub, RpcTarget } from 'capnweb'
@@ -120,7 +121,9 @@ export default function BlueprintLandingPage({ rpcStub }: Props) {
   // When authenticated, fetch models for binding assignment.
   useEffect(() => {
     if (isAuthenticated && authenticatedApi) {
-      authenticatedApi.listModels().then(setModels).catch(console.error)
+      authenticatedApi.listModels()
+        .then(setModels)
+        .catch(err => logRpcFailure('Failed to load models:', err))
     } else {
       setModels([])
     }
@@ -190,7 +193,7 @@ export default function BlueprintLandingPage({ rpcStub }: Props) {
         }
       })
       .catch(err => {
-        console.error('Failed to subscribe to connected accounts:', err)
+        logRpcFailure('Failed to subscribe to connected accounts:', err)
       })
 
     return () => {

packages/workshop-frontend/src/ChatInterface.tsx

@@ -1,3 +1,4 @@
+import { isTransientRpcError, logRpcFailure } from "./rpcErrors";
 import {
   Fragment,
   memo,
@@ -1780,6 +1781,7 @@ export const ChatInput = ({
   attachLabel,
   draftUpdateBanner,
   blockedReason,
+  chatKey,
   onStop,
   showThinkingTraces = true,
   onToggleThinkingTraces,
@@ -1825,6 +1827,8 @@ export const ChatInput = ({
   /** When set, the composer is disabled and shows this message — the user must resolve something
    * (e.g. accept/deny a pending connection request) before they can type or send. */
   blockedReason?: string;
+  /** Identity of the chat the composer is bound to; a change clears chat-scoped hints. */
+  chatKey?: number | null;
   onStop?: () => void;
   showThinkingTraces?: boolean;
   onToggleThinkingTraces?: () => void;
@@ -1838,6 +1842,10 @@ export const ChatInput = ({
   const [capsules, setCapsules] = useState<InputCapsule[]>([]);
   const [pendingAttachments, setPendingAttachments] = useState<PendingAttachment[]>([]);
   const [isSending, setIsSending] = useState(false);
+  // The chat the "may not have been sent" hint belongs to; the render condition scopes it, and
+  // leaving the chat dismisses it.
+  const [sendHiccup, setSendHiccup] = useState<{ chatKey?: number | null } | null>(null);
+  useEffect(() => setSendHiccup(null), [chatKey]);
   const [isAttachmentDragActive, setIsAttachmentDragActive] = useState(false);
   const [selectedSlashCommand, setSelectedSlashCommand] = useState<SelectedSlashCommand | null>(null);
   // The caret the slash command picker parses at. Deliberately updated only when it moves to a
@@ -2276,6 +2284,7 @@ export const ChatInput = ({
 
   const handleSend = async () => {
     if (sendInFlightRef.current || isSending || isBlocked) return;
+    setSendHiccup(null);
     const attachmentsSnapshot = pendingAttachments;
     const readyAttachments = attachmentsSnapshot
       .filter((attachment) => attachment.uploadState === "ready" && attachment.ref)
@@ -2454,8 +2463,10 @@ export const ChatInput = ({
   };
 
   const submitMessage = () => {
+    const submittedChatKey = chatKey;
     void handleSend().catch((err) => {
-      console.error("Failed to send chat message:", err);
+      // The onSend handlers already log; the composer only needs the hint state.
+      if (isTransientRpcError(err)) setSendHiccup({ chatKey: submittedChatKey });
     });
   };
 
@@ -3050,6 +3061,11 @@ export const ChatInput = ({
           </div>
         )}
         {draftUpdateBanner}
+        {sendHiccup && sendHiccup.chatKey === chatKey && (
+          <div className="px-4 pt-2 text-xs text-kumo-warning">
+            Connection hiccup — your message may not have been sent. Check the thread, then try again.
+          </div>
+        )}
         {/* Textarea */}
         <div className="relative px-4 pb-1 pt-3">
           {slashCommandPicker.popup}
@@ -5213,9 +5229,10 @@ function ChatInterface({
           forceUpdate();
         }
       } catch (err) {
-        console.error("Failed to subscribe to chats:", err);
-        reportIssue('chat.subscription-load', err)
-        toasts.add({ title: "Unable to load conversations", variant: "error" });
+        if (!logRpcFailure("Failed to subscribe to chats:", err)) {
+          reportIssue('chat.subscription-load', err)
+          toasts.add({ title: "Unable to load conversations", variant: "error" });
+        }
       }
     };
 
@@ -5366,8 +5383,9 @@ function ChatInterface({
         );
       }
     } catch (err) {
-      console.error("Failed to send message:", err);
-      toasts.add({ title: "Failed to send message", variant: "error" });
+      if (!logRpcFailure("Failed to send message:", err, { reportSite: "chat.send" })) {
+        toasts.add({ title: "Failed to send message", variant: "error" });
+      }
       throw err;
     }
   };
@@ -5388,8 +5406,9 @@ function ChatInterface({
           message, model, capsules, attachments, formats);
       onNavigateToChatRef.current(newChatId);
     } catch (err) {
-      console.error("Failed to create new chat:", err);
-      toasts.add({ title: "Failed to start conversation", variant: "error" });
+      if (!logRpcFailure("Failed to create new chat:", err, { reportSite: "chat.new" })) {
+        toasts.add({ title: "Failed to start conversation", variant: "error" });
+      }
       throw err;
     }
   };
@@ -7576,6 +7595,7 @@ function ChatInterface({
               <div className={`flex-shrink-0 bg-kumo-base ${sidebarMode ? "" : "border-t border-kumo-line"}`}>
                 <div className={useConstrainedChatWidth ? "mx-auto w-full max-w-[920px]" : ""}>
                   <ChatInput
+                    chatKey={selectedChatId}
                     createCapsuleGatekeeper={(accountId, url) =>
                       overseer.newGatekeeper(accountId, url)
                     }

packages/workshop-frontend/src/Connections.tsx

@@ -70,6 +70,8 @@ export default function Connections({ overseer, gadget, chatId, authenticatedApi
       setHooks(hookList.filter((hook) => hook.gadgetId === id))
       onHasGatekeepersChange?.(bindingList.length > 0)
     } catch (err) {
+      // Loud on purpose: this panel has no retry path, so a quieted transient failure would
+      // silently render "no connected resources".
       console.error('Failed to load gatekeepers:', err)
       reportIssue('connections.load', err)
       toasts.add({ title: 'Failed to load connections', variant: 'error' })

packages/workshop-frontend/src/GatekeeperModal.tsx

@@ -1,3 +1,4 @@
+import { logRpcFailure } from './rpcErrors'
 import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
 import { Dialog, useKumoToastManager, type PortalContainer } from '@cloudflare/kumo'
 import {
@@ -440,7 +441,7 @@ export default function GatekeeperModal({
         }
       })
       .catch(error => {
-        console.error('Failed to subscribe to connected accounts:', error)
+        logRpcFailure('Failed to subscribe to connected accounts:', error)
       })
 
     return () => {