Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- **fix(sse):** Anthropic requests that reject a completed historical thinking signature now retry once with only that historical thinking omitted, while valid requests and active tool-use cycles remain unchanged. ([#7906](https://github.com/diegosouzapw/OmniRoute/pull/7906))
42 changes: 41 additions & 1 deletion open-sse/handlers/chatCore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ import {
redactPassthroughThinkingSignatures,
isClaudeCodeSemanticPassthroughRequest,
} from "./chatCore/passthroughHelpers.ts";
import { recoverAnthropicThinkingSignature } from "./chatCore/thinkingSignatureRecovery.ts";
import {
buildStreamingResponseHeaders,
materializeDeduplicatedExecutionResult,
Expand Down Expand Up @@ -3391,7 +3392,7 @@ export async function handleChatCore({
await persistCodexQuotaState(normalizeHeaders(providerResponse.headers), providerResponse.status);

// Check provider response - return error info for fallback handling
if (!providerResponse.ok) {
providerFailure: if (!providerResponse.ok) {
trackPendingRequest(model, provider, connectionId, false);

let statusCode = providerResponse.status;
Expand All @@ -3414,6 +3415,45 @@ export async function handleChatCore({
upstreamErrorType = details.errorType as string | undefined;
}

const signatureRecovery = await recoverAnthropicThinkingSignature({
provider,
statusCode,
message,
body: translatedBody,
execute: async (recoveryBody) => {
translatedBody = recoveryBody as typeof translatedBody;
return executeProviderRequest(currentModel, false);
},
parseError: (response) => parseUpstreamError(response, provider),
});
if (signatureRecovery.attempted && signatureRecovery.execution) {
providerResponse = signatureRecovery.execution.response;
if (signatureRecovery.succeeded) {
providerUrl = signatureRecovery.execution.url;
providerHeaders = signatureRecovery.execution.headers;
finalBody = providerRequestCapture.body(signatureRecovery.execution.transformedBody);
reqLogger.logTargetRequest(providerUrl, providerHeaders, finalBody);
updatePendingScope(pendingScope, {
providerRequest: finalBody,
providerUrl,
stage: "provider_response_started",
});
log?.info?.(
"THINKING_SIGNATURE",
`Recovered ${provider}/${currentModel} after one historical-thinking retry`
);
} else if (signatureRecovery.error) {
statusCode = signatureRecovery.error.statusCode;
message = signatureRecovery.error.message;
retryAfterMs = signatureRecovery.error.retryAfterMs;
upstreamErrorBody = signatureRecovery.error.responseBody;
upstreamErrorCode = signatureRecovery.error.errorCode as string | undefined;
upstreamErrorType = signatureRecovery.error.errorType as string | undefined;
}
}

if (signatureRecovery.succeeded) break providerFailure;

// T06/T10/T36: classify provider errors and persist terminal account states.
let errorType = classifyProviderError(statusCode, message, provider);
if (statusCode === 429 && isModelScope()) {
Expand Down
156 changes: 156 additions & 0 deletions open-sse/handlers/chatCore/passthroughHelpers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,162 @@ export function redactPassthroughThinkingSignatures(
return messages;
}

type MessageLike = {
role?: unknown;
content?: unknown;
};

type ThinkingSignatureError = {
provider?: string | null;
status?: number | null;
message?: string | null;
};

function isThinkingBlock(block: unknown): boolean {
if (!block || typeof block !== "object") return false;
const type = (block as { type?: unknown }).type;
return type === "thinking" || type === "redacted_thinking";
}

function hasBlock(message: MessageLike | null | undefined, type: string): boolean {
return (
!!message &&
Array.isArray(message.content) &&
message.content.some(
(block) => !!block && typeof block === "object" && (block as { type?: unknown }).type === type
)
);
}

/**
* Match only the Anthropic validation failure this recovery path understands.
* Generic 400s and the separate "latest assistant message cannot be modified"
* validation error must continue through the normal error path unchanged.
*/
export function isAnthropicThinkingSignatureError({
provider,
status,
message,
}: ThinkingSignatureError): boolean {
const isAnthropicTarget =
provider === "claude" ||
(typeof provider === "string" && provider.startsWith("anthropic-compatible-"));
if (!isAnthropicTarget || status !== 400 || typeof message !== "string") return false;

return /invalid\s+[`'\"]?signature[`'\"]?\s+in\s+[`'\"]?thinking[`'\"]?\s+block/i.test(message);
}

/**
* Build a one-shot recovery body after Anthropic has explicitly rejected a
* thinking signature. Historical thinking blocks are omitted, but the complete
* active tool-use cycle is preserved verbatim: when the request ends in one or
* more `user[tool_result]` turns, every paired assistant `tool_use` turn in that
* still-open cycle keeps its thinking blocks. A trailing unresolved assistant
* `tool_use` turn is protected as well.
*
* This helper is intentionally NOT used eagerly. Normal same-model requests must
* retain their thinking history, cache shape, and current-model semantics.
* Returns the original body reference when no safe recovery change is possible.
*/
export function stripHistoricalThinkingForSignatureRecovery<T>(body: T): T {
if (!body || typeof body !== "object" || Array.isArray(body)) return body;

const record = body as Record<string, unknown>;
if (!Array.isArray(record.messages)) return body;

const messages = record.messages as MessageLike[];
const protectedAssistantIndexes = new Set<number>();
let cursor = messages.length - 1;

// Some internal callers can resume from an unresolved assistant tool_use.
if (
cursor >= 0 &&
messages[cursor]?.role === "assistant" &&
hasBlock(messages[cursor], "tool_use")
) {
protectedAssistantIndexes.add(cursor);
cursor -= 1;
}

// Walk the complete trailing tool-result chain. Interleaved thinking can span
// several assistant/tool_result pairs, so protecting only the latest assistant
// message is insufficient.
while (
cursor >= 0 &&
messages[cursor]?.role === "user" &&
hasBlock(messages[cursor], "tool_result")
) {
cursor -= 1;
while (cursor >= 0 && messages[cursor]?.role !== "assistant") cursor -= 1;
if (cursor < 0 || !hasBlock(messages[cursor], "tool_use")) break;
protectedAssistantIndexes.add(cursor);
cursor -= 1;
}

let changed = false;
const recoveredMessages = messages.map((message, index) => {
if (
!message ||
message.role !== "assistant" ||
!Array.isArray(message.content) ||
protectedAssistantIndexes.has(index)
) {
return message;
}

const content = message.content.filter((block) => !isThinkingBlock(block));
if (content.length === message.content.length) return message;
changed = true;
return { ...message, content };
});

if (!changed) return body;
return { ...record, messages: recoveredMessages } as T;
}

type SignatureRecoveryExecution<T> = {
result: T;
retried: boolean;
recoveryBody: unknown | null;
};

/** Execute the normal body once, then perform at most one exact-error recovery. */
export async function executeWithAnthropicThinkingSignatureRecovery<T>(args: {
provider?: string | null;
body: unknown;
execute: (body: unknown) => Promise<T>;
getError: (
result: T
) =>
| { status?: number | null; message?: string | null }
| null
| Promise<{ status?: number | null; message?: string | null } | null>;
}): Promise<SignatureRecoveryExecution<T>> {
const first = await args.execute(args.body);
const failure = await args.getError(first);
if (
!failure ||
!isAnthropicThinkingSignatureError({
provider: args.provider,
status: failure.status,
message: failure.message,
})
) {
return { result: first, retried: false, recoveryBody: null };
}

const recoveryBody = stripHistoricalThinkingForSignatureRecovery(args.body);
if (recoveryBody === args.body) {
return { result: first, retried: false, recoveryBody: null };
}

return {
result: await args.execute(recoveryBody),
retried: true,
recoveryBody,
};
}

export function isClaudeCodeSemanticPassthroughRequest({
provider,
sourceFormat,
Expand Down
97 changes: 97 additions & 0 deletions open-sse/handlers/chatCore/thinkingSignatureRecovery.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
import {
executeWithAnthropicThinkingSignatureRecovery,
isAnthropicThinkingSignatureError,
} from "./passthroughHelpers.ts";

type ProviderExecution = {
response: Response;
url?: string;
headers?: Headers | Record<string, string>;
transformedBody?: unknown;
};

type ParsedError = {
statusCode: number;
message: string;
retryAfterMs: number | null;
responseBody: unknown;
errorCode?: unknown;
errorType?: unknown;
};

export type ThinkingSignatureRecoveryResult = {
attempted: boolean;
succeeded: boolean;
execution: ProviderExecution | null;
error: ParsedError | null;
recoveryBody: unknown | null;
};

/**
* Retry exactly once after Anthropic's explicit thinking-signature validation
* error. The caller keeps all connection/provider resilience state untouched
* until this request-scoped recovery has finished.
*/
export async function recoverAnthropicThinkingSignature(args: {
provider?: string | null;
statusCode: number;
message: string;
body: unknown;
execute: (body: unknown) => Promise<ProviderExecution>;
parseError: (response: Response) => Promise<ParsedError>;
}): Promise<ThinkingSignatureRecoveryResult> {
if (
!isAnthropicThinkingSignatureError({
provider: args.provider,
status: args.statusCode,
message: args.message,
})
) {
return {
attempted: false,
succeeded: false,
execution: null,
error: null,
recoveryBody: null,
};
}

const firstFailure = {
response: new Response(null, { status: args.statusCode }),
status: args.statusCode,
message: args.message,
};
const recovery = await executeWithAnthropicThinkingSignatureRecovery({
provider: args.provider,
body: args.body,
execute: async (requestBody) => {
if (requestBody === args.body) return firstFailure;
return args.execute(requestBody);
},
getError: async (result) => {
if (result === firstFailure) return { status: result.status, message: result.message };
if (result.response.ok) return null;
const details = await args.parseError(result.response.clone());
return { status: details.statusCode, message: details.message };
},
});

if (!recovery.retried) {
return {
attempted: false,
succeeded: false,
execution: null,
error: null,
recoveryBody: null,
};
}

const execution = recovery.result as ProviderExecution;
return {
attempted: true,
succeeded: execution.response.ok,
execution,
error: execution.response.ok ? null : await args.parseError(execution.response),
recoveryBody: recovery.recoveryBody,
};
}
1 change: 1 addition & 0 deletions stryker.conf.json
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
"tests/unit/account-fallback-route-restriction-403.test.ts",
"tests/unit/account-fallback-service.test.ts",
"tests/unit/accountfallback-ratelimit-400-4976.test.ts",
"tests/unit/anthropic-thinking-signature-recovery.test.ts",
"tests/unit/antigravity-429-quota-tdd.test.ts",
"tests/unit/api-key-rotator-health.test.ts",
"tests/unit/appearance-widget-settings-schema.test.ts",
Expand Down
Loading
Loading