fix(threads): stop turn/steer fallback to turn/start

Author: Dimillian

src/features/threads/hooks/useThreadMessaging.test.tsx

@@ -204,6 +204,7 @@ describe("useThreadMessaging telemetry", () => {
   });
 
   it("uses turn/steer when steer mode is enabled and an active turn is present", async () => {
+    const dispatch = vi.fn();
     const { result } = renderHook(() =>
       useThreadMessaging({
         activeWorkspace: workspace,
@@ -229,7 +230,7 @@ describe("useThreadMessaging telemetry", () => {
         },
         rateLimitsByWorkspace: {},
         pendingInterruptsRef: { current: new Set<string>() },
-        dispatch: vi.fn(),
+        dispatch,
         getCustomName: vi.fn(() => undefined),
         markProcessing: vi.fn(),
         markReviewing: vi.fn(),
@@ -263,14 +264,15 @@ describe("useThreadMessaging telemetry", () => {
       [],
     );
     expect(sendUserMessageService).not.toHaveBeenCalled();
+    expect(dispatch).not.toHaveBeenCalledWith(
+      expect.objectContaining({ type: "upsertItem" }),
+    );
   });
 
-  it("falls back to turn/start when turn/steer is unsupported and remembers fallback", async () => {
+  it("does not fall back to turn/start when turn/steer fails", async () => {
+    const pushThreadErrorMessage = vi.fn();
     vi.mocked(steerTurnService).mockResolvedValueOnce({
-      error: {
-        message:
-          "Invalid request: unknown variant `turn/steer`, expected one of `turn/start`, `turn/interrupt`",
-      },
+      error: { message: "no active turn to steer" },
     } as unknown as Awaited<ReturnType<typeof steerTurnService>>);
 
     const { result } = renderHook(() =>
@@ -306,89 +308,7 @@ describe("useThreadMessaging telemetry", () => {
         recordThreadActivity: vi.fn(),
         safeMessageActivity: vi.fn(),
         onDebug: vi.fn(),
-        pushThreadErrorMessage: vi.fn(),
-        ensureThreadForActiveWorkspace: vi.fn(async () => "thread-1"),
-        ensureThreadForWorkspace: vi.fn(async () => "thread-1"),
-        refreshThread: vi.fn(async () => null),
-        forkThreadForWorkspace: vi.fn(async () => null),
-        updateThreadParent: vi.fn(),
-      }),
-    );
-
-    await act(async () => {
-      await result.current.sendUserMessageToThread(
-        workspace,
-        "thread-1",
-        "fallback once",
-        [],
-      );
-    });
-    await act(async () => {
-      await result.current.sendUserMessageToThread(
-        workspace,
-        "thread-1",
-        "fallback twice",
-        [],
-      );
-    });
-
-    expect(steerTurnService).toHaveBeenCalledTimes(1);
-    expect(sendUserMessageService).toHaveBeenCalledTimes(2);
-    expect(sendUserMessageService).toHaveBeenNthCalledWith(
-      1,
-      "ws-1",
-      "thread-1",
-      "fallback once",
-      expect.any(Object),
-    );
-    expect(sendUserMessageService).toHaveBeenNthCalledWith(
-      2,
-      "ws-1",
-      "thread-1",
-      "fallback twice",
-      expect.any(Object),
-    );
-  });
-
-  it("falls back to turn/start when remote reports unknown method turn_steer", async () => {
-    vi.mocked(steerTurnService).mockRejectedValueOnce(
-      new Error("unknown method: turn_steer"),
-    );
-
-    const { result } = renderHook(() =>
-      useThreadMessaging({
-        activeWorkspace: workspace,
-        activeThreadId: "thread-1",
-        accessMode: "current",
-        model: null,
-        effort: null,
-        collaborationMode: null,
-        reviewDeliveryMode: "inline",
-        steerEnabled: true,
-        customPrompts: [],
-        threadStatusById: {
-          "thread-1": {
-            isProcessing: true,
-            isReviewing: false,
-            hasUnread: false,
-            processingStartedAt: 0,
-            lastDurationMs: null,
-          },
-        },
-        activeTurnIdByThread: {
-          "thread-1": "turn-1",
-        },
-        rateLimitsByWorkspace: {},
-        pendingInterruptsRef: { current: new Set<string>() },
-        dispatch: vi.fn(),
-        getCustomName: vi.fn(() => undefined),
-        markProcessing: vi.fn(),
-        markReviewing: vi.fn(),
-        setActiveTurnId: vi.fn(),
-        recordThreadActivity: vi.fn(),
-        safeMessageActivity: vi.fn(),
-        onDebug: vi.fn(),
-        pushThreadErrorMessage: vi.fn(),
+        pushThreadErrorMessage,
         ensureThreadForActiveWorkspace: vi.fn(async () => "thread-1"),
         ensureThreadForWorkspace: vi.fn(async () => "thread-1"),
         refreshThread: vi.fn(async () => null),
@@ -401,18 +321,16 @@ describe("useThreadMessaging telemetry", () => {
       await result.current.sendUserMessageToThread(
         workspace,
         "thread-1",
-        "fallback remote method",
+        "steer should fail",
         [],
       );
     });
 
     expect(steerTurnService).toHaveBeenCalledTimes(1);
-    expect(sendUserMessageService).toHaveBeenCalledTimes(1);
-    expect(sendUserMessageService).toHaveBeenCalledWith(
-      "ws-1",
+    expect(sendUserMessageService).not.toHaveBeenCalled();
+    expect(pushThreadErrorMessage).toHaveBeenCalledWith(
       "thread-1",
-      "fallback remote method",
-      expect.any(Object),
+      "Turn steer failed: no active turn to steer",
     );
   });
 });

src/features/threads/hooks/useThreadMessaging.ts

@@ -1,4 +1,4 @@
-import { useCallback, useRef } from "react";
+import { useCallback } from "react";
 import type { Dispatch, MutableRefObject } from "react";
 import * as Sentry from "@sentry/react";
 import type {
@@ -26,7 +26,6 @@ import {
   extractRpcErrorMessage,
   parseReviewTarget,
 } from "@threads/utils/threadNormalize";
-import { isUnsupportedTurnSteerError } from "@threads/utils/threadRpc";
 import type { ThreadAction, ThreadState } from "./useThreadsReducer";
 import { useReviewPrompt } from "./useReviewPrompt";
 import { formatRelativeTime } from "@utils/time";
@@ -113,8 +112,6 @@ export function useThreadMessaging({
   updateThreadParent,
   registerDetachedReviewChild,
 }: UseThreadMessagingOptions) {
-  const steerSupportedByWorkspaceRef = useRef<Record<string, boolean>>({});
-
   const sendMessageToThread = useCallback(
     async (
       workspace: WorkspaceInfo,
@@ -157,29 +154,8 @@ export function useThreadMessaging({
 
       const isProcessing = threadStatusById[threadId]?.isProcessing ?? false;
       const activeTurnId = activeTurnIdByThread[threadId] ?? null;
-      const steerSupported = steerSupportedByWorkspaceRef.current[workspace.id] !== false;
       const shouldSteer =
-        isProcessing && steerEnabled && Boolean(activeTurnId) && steerSupported;
-      if (isProcessing && steerEnabled) {
-        const optimisticText = finalText;
-        if (optimisticText || images.length > 0) {
-          dispatch({
-            type: "upsertItem",
-            workspaceId: workspace.id,
-            threadId,
-            item: {
-              id: `optimistic-user-${Date.now()}-${Math.random()
-                .toString(36)
-                .slice(2, 8)}`,
-              kind: "message",
-              role: "user",
-              text: optimisticText,
-              images: images.length > 0 ? images : undefined,
-            },
-            hasCustomName: Boolean(getCustomName(workspace.id, threadId)),
-          });
-        }
-      }
+        isProcessing && steerEnabled && Boolean(activeTurnId);
       Sentry.metrics.count("prompt_sent", 1, {
         attributes: {
           workspace_id: workspace.id,
@@ -217,7 +193,7 @@ export function useThreadMessaging({
           collaborationMode: sanitizedCollaborationMode,
         },
       });
-      let requestMode: "start" | "steer" = shouldSteer ? "steer" : "start";
+      const requestMode: "start" | "steer" = shouldSteer ? "steer" : "start";
       try {
         const startTurn = () => {
           const payload: {
@@ -245,49 +221,26 @@ export function useThreadMessaging({
           );
         };
 
-        let response: Record<string, unknown>;
-        if (shouldSteer) {
-          try {
-            response = (await (appMentions.length > 0
-              ? steerTurnService(
-                workspace.id,
-                threadId,
-                activeTurnId ?? "",
-                finalText,
-                images,
-                appMentions,
-              )
-              : steerTurnService(
-                workspace.id,
-                threadId,
-                activeTurnId ?? "",
-                finalText,
-                images,
-              ))) as Record<string, unknown>;
-          } catch (error) {
-            const message = error instanceof Error ? error.message : String(error);
-            if (!isUnsupportedTurnSteerError(message)) {
-              throw error;
-            }
-            steerSupportedByWorkspaceRef.current[workspace.id] = false;
-            requestMode = "start";
-            response = (await startTurn()) as Record<string, unknown>;
-          }
-        } else {
-          response = (await startTurn()) as Record<string, unknown>;
-        }
+        const response: Record<string, unknown> = shouldSteer
+          ? (await (appMentions.length > 0
+            ? steerTurnService(
+              workspace.id,
+              threadId,
+              activeTurnId ?? "",
+              finalText,
+              images,
+              appMentions,
+            )
+            : steerTurnService(
+              workspace.id,
+              threadId,
+              activeTurnId ?? "",
+              finalText,
+              images,
+            ))) as Record<string, unknown>
+          : (await startTurn()) as Record<string, unknown>;
 
-        let rpcError = extractRpcErrorMessage(response);
-        if (
-          rpcError
-          && requestMode === "steer"
-          && isUnsupportedTurnSteerError(rpcError)
-        ) {
-          steerSupportedByWorkspaceRef.current[workspace.id] = false;
-          requestMode = "start";
-          response = (await startTurn()) as Record<string, unknown>;
-          rpcError = extractRpcErrorMessage(response);
-        }
+        const rpcError = extractRpcErrorMessage(response);
 
         onDebug?.({
           id: `${Date.now()}-${requestMode === "steer" ? "server-turn-steer" : "server-turn-start"}`,

src/features/threads/utils/threadRpc.ts

@@ -61,13 +61,3 @@ export function getResumedActiveTurnId(thread: Record<string, unknown>): string
   }
   return null;
 }
-
-export function isUnsupportedTurnSteerError(message: string): boolean {
-  const normalized = message.toLowerCase();
-  const mentionsSteerMethod =
-    normalized.includes("turn/steer") || normalized.includes("turn_steer");
-  return normalized.includes("unknown variant `turn/steer`")
-    || normalized.includes("unknown variant \"turn/steer\"")
-    || (normalized.includes("unknown request") && mentionsSteerMethod)
-    || (normalized.includes("unknown method") && mentionsSteerMethod);
-}