mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
Anthropic message threads (#11545)
Anthropic can keep a conversation on the server (message threads beta). A request can then continue from the previous reply and send only the new messages, instead of the whole history every turn. pi uses this when a session id is set and caching is on. It still builds the full request, compares it with the previous one, and only sends the new part if nothing earlier changed. - Default: on only for the anthropic provider on api.anthropic.com. Gateways that drop unknown fields would lose the conversation. Override with compat.supportsMessageThreads. - If the server rejects a continuation, pi resends the full request as a new thread. Other errors turn threads off for the session. - New option StreamOptions.disableContinuation. The cache warmer sets it so warm requests don't replace the session's thread. - cleanupSessionResources(sessionId) frees a session's thread state.
This commit is contained in:
David Brailovsky
GitHub
parent
3551ec07eb
commit
a758f3ae4c
14 changed files with 885 additions and 22 deletions
|
|
@ -52,6 +52,7 @@ type ProxySerializableStreamOptions = Pick<
|
|||
| "reasoning"
|
||||
| "cacheRetention"
|
||||
| "sessionId"
|
||||
| "disableContinuation"
|
||||
| "headers"
|
||||
| "metadata"
|
||||
| "transport"
|
||||
|
|
@ -95,6 +96,7 @@ function buildProxyRequestOptions(options: ProxyStreamOptions): ProxySerializabl
|
|||
reasoning: options.reasoning,
|
||||
cacheRetention: options.cacheRetention,
|
||||
sessionId: options.sessionId,
|
||||
disableContinuation: options.disableContinuation,
|
||||
headers: options.headers,
|
||||
metadata: options.metadata,
|
||||
transport: options.transport,
|
||||
|
|
|
|||
|
|
@ -2,6 +2,11 @@
|
|||
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
|
||||
- Anthropic requests with a `sessionId` (and `cacheRetention` not `"none"`) now use message threads (`message-threads-2026-08-12` beta): the server continues from the previous reply and pi sends only the new messages instead of the whole conversation. When anything earlier in the request changed, or the server can't continue the thread, the full request is sent. Token counts and prompt caching are unchanged. On by default only for the `anthropic` provider on `api.anthropic.com`; set `compat.supportsMessageThreads` to change that, for example for gateways that forward the `thread` field. Call `cleanupSessionResources(sessionId)` when a session ends to free the per-session state ([#11545](https://github.com/earendil-works/pi/pull/11545))
|
||||
- Added `StreamOptions.disableContinuation`, which keeps a request from using or replacing server-side continuation state such as Anthropic message threads. Set it on requests that repeat an earlier one, such as cache warming ([#11545](https://github.com/earendil-works/pi/pull/11545))
|
||||
|
||||
### Fixed
|
||||
|
||||
- Fixed OpenCode Go Claude Haiku 5.5 failing with a 400 (`Invalid signature in thinking block ... bound to a different conversation`) after the tool list, system prompt, or earlier history changed, which left the session failing on every retry. OpenCode Go and Zen Claude 5.5 models, Opus 5, and Fable 5.1 now use per-message effort and send `thinking.block_binding: drop_block`, like the Anthropic provider ([#10954](https://github.com/earendil-works/pi/issues/10954))
|
||||
|
|
|
|||
|
|
@ -1438,6 +1438,8 @@ Built-in API implementations live under `./api/<api-id>`:
|
|||
|
||||
Importing an implementation module loads its SDK. The `./api/<id>.lazy` wrappers (used by the provider factories) defer that load to the first request when the runtime or bundler supports dynamic import chunking. Legacy raw API subpaths from older releases (`./anthropic`, `./google`, `./mistral`, `./openai-completions`, ...) were removed; use `@earendil-works/pi-ai/api/<api-id>`.
|
||||
|
||||
**Anthropic message threads:** with `sessionId` set and `cacheRetention` not `"none"`, Anthropic requests continue from the previous reply on the server (`message-threads-2026-08-12` beta) and send only the new messages. When anything earlier in the request changed, or the server can't continue, the full request is sent. Token counts and prompt caching are unchanged. On by default only for the `anthropic` provider on `api.anthropic.com`; set `compat.supportsMessageThreads` to change that, for example for gateways that forward the `thread` field. Set `disableContinuation: true` on requests that repeat an earlier one, such as cache warming. Call `cleanupSessionResources(sessionId)` when a session ends.
|
||||
|
||||
### OpenAI Compatibility Settings
|
||||
|
||||
The `openai-completions` API is implemented by many providers with minor differences. By default, the library auto-detects compatibility settings based on `baseUrl` for a small set of known OpenAI-compatible providers (Cerebras, xAI, Chutes, DeepSeek, NVIDIA NIM, Together AI, zAi, OpenCode, Cloudflare Workers AI, etc.). For custom proxies or unknown endpoints, you can override these settings via the `compat` field. For `openai-responses` models, the compat field supports Responses-specific flags.
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import {
|
|||
ANTHROPIC_WORKSPACE_ID_ENV,
|
||||
} from "../env-api-keys.ts";
|
||||
import { calculateCost } from "../models.ts";
|
||||
import { registerSessionResourceCleanup } from "../session-resources.ts";
|
||||
import type {
|
||||
Api,
|
||||
AssistantMessage,
|
||||
|
|
@ -37,6 +38,7 @@ import type {
|
|||
} from "../types.ts";
|
||||
import { appendAssistantMessageDiagnostic } from "../utils/diagnostics.ts";
|
||||
import { AssistantMessageEventStream } from "../utils/event-stream.ts";
|
||||
import { shortHash } from "../utils/hash.ts";
|
||||
import { headersToRecord } from "../utils/headers.ts";
|
||||
import { type ClientIdentity, getIdentityHeaders } from "../utils/identity.ts";
|
||||
import { parseJsonWithRepair } from "../utils/json-parse.ts";
|
||||
|
|
@ -203,6 +205,7 @@ const SERVER_SIDE_FALLBACK_BETA = "server-side-fallback-2026-07-01";
|
|||
const MID_CONVERSATION_OUTPUT_CONFIG_BETA = "mid-conversation-output-config-2026-07-01";
|
||||
const THINKING_BINDING_CONTROLS_BETA = "thinking-binding-controls-2026-08-01";
|
||||
const INLINE_TOOLS_BETA = "inline-tools-2026-09-15";
|
||||
const MESSAGE_THREADS_BETA = "message-threads-2026-08-12";
|
||||
|
||||
/**
|
||||
* Stable deferred tool declared whenever native tool changes are in use. Anthropic adds
|
||||
|
|
@ -235,9 +238,234 @@ function getAnthropicCompat(model: Model<"anthropic-messages">) {
|
|||
supportsStrictTools: model.compat?.supportsStrictTools ?? false,
|
||||
supportsMidConvoSystemMessages: model.compat?.supportsMidConvoSystemMessages ?? false,
|
||||
supportsMidConvoToolChanges: model.compat?.supportsMidConvoToolChanges ?? false,
|
||||
// Gateways that drop unknown request fields would silently turn a `continue` into a request
|
||||
// without the earlier conversation, so the default is limited to Anthropic's own endpoint.
|
||||
// Widen or remove this restriction once proxies generally forward the `thread` field.
|
||||
supportsMessageThreads:
|
||||
model.compat?.supportsMessageThreads ??
|
||||
(model.provider === "anthropic" && model.baseUrl.includes("api.anthropic.com")),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Message threads: the server keeps the conversation of a thread, so a request can continue
|
||||
* from the previous reply and send only the new messages. pi still builds the full request;
|
||||
* `prepareThreadRequest()` compares it with the previous request of the session and sends
|
||||
* only the new part when nothing earlier changed.
|
||||
*/
|
||||
type ThreadParam = { type: "create" } | { type: "continue"; previous_message_id: string };
|
||||
|
||||
/** The pinned SDK types do not include the `thread` request field yet. */
|
||||
type ThreadedParams = MessageCreateParamsStreaming & { thread?: ThreadParam };
|
||||
|
||||
interface ThreadEntry {
|
||||
/** Number of messages in the last full request as sent plus the converted reply. */
|
||||
prefixCount: number;
|
||||
/** `hashThreadPrefix()` of the request's settings and those messages. */
|
||||
prefixHash: string;
|
||||
/** Id of the reply to continue from. */
|
||||
replyId: string;
|
||||
}
|
||||
|
||||
/** Per session: where the next request can continue, or "off" after the endpoint rejected threads. */
|
||||
const threadState = new Map<string, ThreadEntry | "off">();
|
||||
const MAX_THREAD_STATE_ENTRIES = 10_000;
|
||||
|
||||
/** Rejections of a `continue` that a new thread fixes: expired, raced, or a fixed setting changed. */
|
||||
const NEW_THREAD_ERRORS: ReadonlySet<string> = new Set([
|
||||
"thread_not_found",
|
||||
"thread_already_continued",
|
||||
"thread_continue_in_flight",
|
||||
"thread_fingerprint_mismatch",
|
||||
]);
|
||||
|
||||
/**
|
||||
* Sets a session's thread state, keeping at most `MAX_THREAD_STATE_ENTRIES` sessions.
|
||||
*
|
||||
* State is normally freed by `cleanupSessionResources(sessionId)` when a session ends, as pi's
|
||||
* coding agent does. The cap bounds memory for library users that never call it, for example a
|
||||
* server passing a new `sessionId` per conversation. Dropping the least recently updated session
|
||||
* is cheap: its thread has most likely expired on the server already (after a few minutes idle),
|
||||
* and otherwise its next request just starts a new thread.
|
||||
*/
|
||||
function setThreadState(sessionId: string, value: ThreadEntry | "off"): void {
|
||||
// Re-insert so Map order stays oldest-first for the cap.
|
||||
threadState.delete(sessionId);
|
||||
threadState.set(sessionId, value);
|
||||
if (threadState.size > MAX_THREAD_STATE_ENTRIES) {
|
||||
const oldest = threadState.keys().next().value;
|
||||
if (oldest !== undefined) threadState.delete(oldest);
|
||||
}
|
||||
}
|
||||
|
||||
registerSessionResourceCleanup((sessionId) => {
|
||||
if (sessionId) threadState.delete(sessionId);
|
||||
else threadState.clear();
|
||||
});
|
||||
|
||||
/**
|
||||
* Hash of everything a `continue` builds on: the settings that must stay the same within a thread
|
||||
* and the messages the server already holds.
|
||||
*
|
||||
* - Settings, as sent: the endpoint (a thread exists on one endpoint only), and the fixed request
|
||||
* fields. A `continue` omits `system` and `tools` and the server keeps the thread's own, so only
|
||||
* this check can notice a change to them, including their cache markers. The server rejects
|
||||
* changes to the other fields with `thread_fingerprint_mismatch`; checking them here saves that
|
||||
* rejected request. A fixed field missing here costs only that rejection, which also starts a
|
||||
* new thread.
|
||||
* - Messages, normalized so messages that mean the same to the API hash the same.
|
||||
*/
|
||||
function hashThreadPrefix(
|
||||
model: Model<"anthropic-messages">,
|
||||
params: MessageCreateParamsStreaming,
|
||||
messages: readonly unknown[],
|
||||
): string {
|
||||
const { system, tools, thinking, output_config, tool_choice, betas, fallbacks } = params;
|
||||
const settings = JSON.stringify([
|
||||
model.provider,
|
||||
model.baseUrl,
|
||||
params.model,
|
||||
system,
|
||||
tools,
|
||||
thinking,
|
||||
output_config,
|
||||
tool_choice,
|
||||
betas,
|
||||
fallbacks,
|
||||
]);
|
||||
const conversation = JSON.stringify(messages, (key, value: unknown) => {
|
||||
// Cache markers don't change what a message means, but they move: pi puts one on the newest
|
||||
// message, so the message that had it in the last request has none in this one. Extensions
|
||||
// may add markers anywhere too.
|
||||
if (key === "cache_control") return undefined;
|
||||
// To attach that marker, the newest user message's string content is sent as one text block;
|
||||
// the next request sends the same message as a string: "hi" vs [{ type: "text", text: "hi" }].
|
||||
if (key === "content" && typeof value === "string") return [{ type: "text", text: value }];
|
||||
return value;
|
||||
});
|
||||
return shortHash(settings + conversation);
|
||||
}
|
||||
|
||||
interface PreparedThreadRequest {
|
||||
sessionId: string;
|
||||
body: ThreadedParams;
|
||||
}
|
||||
|
||||
function createThreadRequest(sessionId: string, params: MessageCreateParamsStreaming): PreparedThreadRequest {
|
||||
return { sessionId, body: { ...params, thread: { type: "create" } } };
|
||||
}
|
||||
|
||||
function prepareThreadRequest(
|
||||
model: Model<"anthropic-messages">,
|
||||
sessionId: string,
|
||||
params: MessageCreateParamsStreaming,
|
||||
): PreparedThreadRequest | undefined {
|
||||
const state = threadState.get(sessionId);
|
||||
if (state === "off") return undefined;
|
||||
if (
|
||||
state !== undefined &&
|
||||
params.messages.length > state.prefixCount &&
|
||||
hashThreadPrefix(model, params, params.messages.slice(0, state.prefixCount)) === state.prefixHash
|
||||
) {
|
||||
// Each reply can be continued once; deleting also keeps a concurrent request from trying.
|
||||
threadState.delete(sessionId);
|
||||
const { system: _system, tools: _tools, ...rest } = params;
|
||||
return {
|
||||
body: {
|
||||
...rest,
|
||||
messages: params.messages.slice(state.prefixCount),
|
||||
thread: { type: "continue", previous_message_id: state.replyId },
|
||||
},
|
||||
sessionId,
|
||||
};
|
||||
}
|
||||
return createThreadRequest(sessionId, params);
|
||||
}
|
||||
|
||||
/** The `error_code` Anthropic puts in `error.details` of thread rejections. */
|
||||
function getThreadErrorCode(error: InstanceType<typeof Anthropic.APIError>): string | undefined {
|
||||
const body = error.error as { error?: { details?: { error_code?: unknown } } } | undefined;
|
||||
const code = body?.error?.details?.error_code;
|
||||
return typeof code === "string" ? code : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a thread request. A rejection about the thread (never a started stream) is resent once in
|
||||
* full: as a new thread when the old one can't be continued, or without `thread` when the endpoint
|
||||
* can't use threads, which also turns threads off for the session. Other errors are passed on.
|
||||
* Returns the response and the thread request that produced it, if any.
|
||||
*/
|
||||
async function sendThreadRequest(
|
||||
thread: PreparedThreadRequest,
|
||||
params: MessageCreateParamsStreaming,
|
||||
send: (body: ThreadedParams) => Promise<Response>,
|
||||
output: AssistantMessage,
|
||||
signal: AbortSignal | undefined,
|
||||
): Promise<{ response: Response; thread: PreparedThreadRequest | undefined }> {
|
||||
try {
|
||||
return { response: await send(thread.body), thread };
|
||||
} catch (error) {
|
||||
if (signal?.aborted || !(error instanceof Anthropic.APIError)) throw error;
|
||||
const errorCode = getThreadErrorCode(error);
|
||||
const startNewThread = errorCode !== undefined && NEW_THREAD_ERRORS.has(errorCode);
|
||||
// The endpoint can't use threads: it can't thread this request, or it doesn't know the
|
||||
// `thread` field and rejects it without a code ("thread: Extra inputs are not permitted"),
|
||||
// e.g. with an `anthropic-beta` header that lacks the threads beta.
|
||||
const threadsUnsupported =
|
||||
errorCode === "thread_unsupported_request" ||
|
||||
(errorCode === undefined && error.status === 400 && error.message.includes("thread"));
|
||||
// Any other error (prompt too long, unknown model, strict tool limits, ...) is not about the
|
||||
// thread, and resending would only fail again.
|
||||
if (!startNewThread && !threadsUnsupported) throw error;
|
||||
// Recorded before the resend, so a failed resend keeps it.
|
||||
appendAssistantMessageDiagnostic(output, {
|
||||
type: "anthropic_thread_fallback",
|
||||
timestamp: Date.now(),
|
||||
error: { message: error.message },
|
||||
details: { reason: errorCode ?? String(error.status) },
|
||||
});
|
||||
if (startNewThread) {
|
||||
const created = createThreadRequest(thread.sessionId, params);
|
||||
return { response: await send(created.body), thread: created };
|
||||
}
|
||||
// With the threads beta, a request without `thread` is a normal request.
|
||||
const response = await send(params);
|
||||
setThreadState(thread.sessionId, "off");
|
||||
return { response, thread: undefined };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Records where the session's next request can continue: after `output`, the clean reply to the
|
||||
* thread request `thread`, whose full request was `params`.
|
||||
*/
|
||||
function recordThreadReply(
|
||||
model: Model<"anthropic-messages">,
|
||||
thread: PreparedThreadRequest,
|
||||
params: MessageCreateParamsStreaming,
|
||||
output: AssistantMessage,
|
||||
isOAuth: boolean,
|
||||
allowEmptySignature: boolean,
|
||||
): void {
|
||||
// `responseModel` is set only when a fallback model answered; the server does not store those
|
||||
// replies, so they cannot be continued.
|
||||
if (output.responseId === undefined || output.responseModel !== undefined) return;
|
||||
// Convert the reply like the next request will, keeping only the assistant message
|
||||
// (transformMessages adds placeholder results for trailing tool calls).
|
||||
const reply = convertMessages(
|
||||
transformMessages([output], model, normalizeToolCallId),
|
||||
isOAuth,
|
||||
undefined,
|
||||
allowEmptySignature,
|
||||
).messages.filter((message) => message.role === "assistant");
|
||||
const prefix = [...params.messages, ...reply];
|
||||
setThreadState(thread.sessionId, {
|
||||
prefixCount: prefix.length,
|
||||
prefixHash: hashThreadPrefix(model, params, prefix),
|
||||
replyId: output.responseId,
|
||||
});
|
||||
}
|
||||
|
||||
export interface AnthropicOptions extends StreamOptions {
|
||||
/**
|
||||
* Enable extended thinking.
|
||||
|
|
@ -580,7 +808,8 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
|
|||
options?: AnthropicOptions,
|
||||
): AssistantMessageEventStream => {
|
||||
const stream = new AssistantMessageEventStream();
|
||||
const normalizedContext = resolveTranscript(context, getAnthropicCompat(model).supportsMidConvoSystemMessages);
|
||||
const compat = getAnthropicCompat(model);
|
||||
const normalizedContext = resolveTranscript(context, compat.supportsMidConvoSystemMessages);
|
||||
const currentTools = getCurrentTools(normalizedContext.messages);
|
||||
|
||||
(async () => {
|
||||
|
|
@ -607,6 +836,12 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
|
|||
let isOAuth: boolean;
|
||||
let usageModel = model;
|
||||
let inputTransformations: BetaInputTransformation[] | undefined;
|
||||
const cacheRetention = resolveCacheRetention(options?.cacheRetention, options?.env);
|
||||
const cacheSessionId = cacheRetention === "none" ? undefined : options?.sessionId;
|
||||
// Session whose requests may use threads; undefined when threads don't apply. Sessions where
|
||||
// the endpoint rejected threads ("off") still send the threads beta, which the resend that
|
||||
// turned them off was accepted with; `prepareThreadRequest()` just sends no `thread`.
|
||||
const threadSessionId = compat.supportsMessageThreads ? cacheSessionId : undefined;
|
||||
|
||||
if (options?.client) {
|
||||
client = options.client;
|
||||
|
|
@ -625,9 +860,6 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
|
|||
});
|
||||
}
|
||||
|
||||
const cacheRetention = resolveCacheRetention(options?.cacheRetention, options?.env);
|
||||
const cacheSessionId = cacheRetention === "none" ? undefined : options?.sessionId;
|
||||
|
||||
const created = createClient(
|
||||
model,
|
||||
apiKey,
|
||||
|
|
@ -646,26 +878,39 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
|
|||
...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}),
|
||||
maxRetries: 0,
|
||||
};
|
||||
const send = (body: ThreadedParams) =>
|
||||
retryProviderRequest(() => client.beta.messages.create(body, requestOptions).asResponse(), {
|
||||
maxRetries: options?.maxRetries,
|
||||
maxRetryDelayMs: options?.maxRetryDelayMs,
|
||||
signal: options?.signal,
|
||||
});
|
||||
let disablePreferredStrictTools = false;
|
||||
let response: Response;
|
||||
let params: MessageCreateParamsStreaming;
|
||||
// The thread request that produced the response; undefined when no `thread` was sent.
|
||||
let thread: PreparedThreadRequest | undefined;
|
||||
for (;;) {
|
||||
options?.signal?.throwIfAborted();
|
||||
const built = buildParams(model, normalizedContext, isOAuth, options, disablePreferredStrictTools);
|
||||
const built = buildParams(model, normalizedContext, isOAuth, options, {
|
||||
disablePreferredStrictTools,
|
||||
messageThreads: threadSessionId !== undefined,
|
||||
});
|
||||
if (built.providerThinkingLevel) output.providerThinkingLevel = built.providerThinkingLevel;
|
||||
let params = built.params;
|
||||
params = built.params;
|
||||
const nextParams = await options?.onPayload?.(params, model);
|
||||
if (nextParams !== undefined) {
|
||||
params = { ...(nextParams as MessageCreateParamsStreaming), stream: true };
|
||||
}
|
||||
const prepared =
|
||||
threadSessionId !== undefined && options?.disableContinuation !== true
|
||||
? prepareThreadRequest(model, threadSessionId, params)
|
||||
: undefined;
|
||||
try {
|
||||
response = await retryProviderRequest(
|
||||
() => client.beta.messages.create(params, requestOptions).asResponse(),
|
||||
{
|
||||
maxRetries: options?.maxRetries,
|
||||
maxRetryDelayMs: options?.maxRetryDelayMs,
|
||||
signal: options?.signal,
|
||||
},
|
||||
);
|
||||
const sent = prepared
|
||||
? await sendThreadRequest(prepared, params, send, output, options?.signal)
|
||||
: { response: await send(params), thread: undefined };
|
||||
response = sent.response;
|
||||
thread = sent.thread;
|
||||
break;
|
||||
} catch (error) {
|
||||
// Only downgrade a rejected request, never an already-started stream.
|
||||
|
|
@ -899,6 +1144,7 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
|
|||
if (output.stopReason === "aborted" || output.stopReason === "error") {
|
||||
throw new Error(output.errorMessage || "An unknown error occurred");
|
||||
}
|
||||
if (thread) recordThreadReply(model, thread, params, output, isOAuth, compat.allowEmptySignature);
|
||||
if (inputTransformations && inputTransformations.length > 0) {
|
||||
appendAssistantMessageDiagnostic(output, {
|
||||
type: "anthropic_input_transformations",
|
||||
|
|
@ -1106,12 +1352,19 @@ function createClient(
|
|||
return { client, isOAuthToken: false };
|
||||
}
|
||||
|
||||
interface BetaFeatureFlags {
|
||||
isOAuthToken: boolean;
|
||||
/** Tool changes are sent as native `tool_addition` / `tool_removal` blocks. */
|
||||
nativeToolChanges: boolean;
|
||||
/** The session uses message threads. */
|
||||
messageThreads: boolean;
|
||||
}
|
||||
|
||||
function getBetaFeatures(
|
||||
model: Model<"anthropic-messages">,
|
||||
context: TranscriptContext,
|
||||
isOAuthToken: boolean,
|
||||
nativeToolChanges: boolean,
|
||||
options?: AnthropicOptions,
|
||||
options: AnthropicOptions | undefined,
|
||||
{ isOAuthToken, nativeToolChanges, messageThreads }: BetaFeatureFlags,
|
||||
): NonNullable<MessageCreateParamsStreaming["betas"]> {
|
||||
let configuredFeatures: string | null | undefined;
|
||||
for (const headers of [model.headers, options?.headers]) {
|
||||
|
|
@ -1147,15 +1400,23 @@ function getBetaFeatures(
|
|||
features.push(MID_CONVERSATION_OUTPUT_CONFIG_BETA, THINKING_BINDING_CONTROLS_BETA);
|
||||
}
|
||||
if (nativeToolChanges) features.push(INLINE_TOOLS_BETA);
|
||||
if (messageThreads) features.push(MESSAGE_THREADS_BETA);
|
||||
return [...new Set(features)];
|
||||
}
|
||||
|
||||
interface BuildParamsFlags {
|
||||
/** Send `strict: "prefer"` tools without strict schemas, after Anthropic rejected them. */
|
||||
disablePreferredStrictTools: boolean;
|
||||
/** The session uses message threads, so the threads beta is sent. */
|
||||
messageThreads: boolean;
|
||||
}
|
||||
|
||||
function buildParams(
|
||||
model: Model<"anthropic-messages">,
|
||||
context: TranscriptContext,
|
||||
isOAuthToken: boolean,
|
||||
options?: AnthropicOptions,
|
||||
disablePreferredStrictTools = false,
|
||||
options: AnthropicOptions | undefined,
|
||||
flags: BuildParamsFlags,
|
||||
): {
|
||||
params: MessageCreateParamsStreaming;
|
||||
hasPreferredStrictTools: boolean;
|
||||
|
|
@ -1180,7 +1441,7 @@ function buildParams(
|
|||
compat.supportsEagerToolInputStreaming,
|
||||
compat.supportsStrictTools,
|
||||
toolCacheControl,
|
||||
disablePreferredStrictTools,
|
||||
flags.disablePreferredStrictTools,
|
||||
);
|
||||
if (
|
||||
definitions.some(
|
||||
|
|
@ -1207,7 +1468,11 @@ function buildParams(
|
|||
nativeToolChanges ? convertToolDefinitions : undefined,
|
||||
);
|
||||
const activeEffort = options?.effort ?? "high";
|
||||
const betaFeatures = getBetaFeatures(model, context, isOAuthToken, nativeToolChanges, options);
|
||||
const betaFeatures = getBetaFeatures(model, context, options, {
|
||||
isOAuthToken,
|
||||
nativeToolChanges,
|
||||
messageThreads: flags.messageThreads,
|
||||
});
|
||||
const params: MessageCreateParamsStreaming = {
|
||||
model: model.id,
|
||||
messages: effortMessages ? insertThinkingLevelMessages(converted, activeEffort) : converted.messages,
|
||||
|
|
|
|||
|
|
@ -64,6 +64,7 @@ export function buildBaseOptions(
|
|||
transport: options?.transport,
|
||||
cacheRetention: options?.cacheRetention,
|
||||
sessionId: options?.sessionId,
|
||||
disableContinuation: options?.disableContinuation,
|
||||
headers: options?.headers,
|
||||
identity: options?.identity,
|
||||
onPayload: options?.onPayload,
|
||||
|
|
|
|||
|
|
@ -492,6 +492,10 @@ const AnthropicMessagesCompatProperties = {
|
|||
default: false,
|
||||
}),
|
||||
),
|
||||
supportsMessageThreads: optionalCompatBoolean({
|
||||
description:
|
||||
"Whether the endpoint supports Anthropic message threads, which let a request with a session id send only the new messages and continue from the previous reply. Default: true for provider anthropic with a baseUrl on api.anthropic.com, false otherwise. Set true for gateways that forward the thread request field.",
|
||||
}),
|
||||
allowedFallbackModels: opt(
|
||||
arr(AnthropicAllowedFallbackModelSchema, {
|
||||
maxItems: 3,
|
||||
|
|
|
|||
|
|
@ -255,6 +255,13 @@ export interface StreamOptions extends ProviderRequestOptions<Model<Api>> {
|
|||
* session-aware features. Ignored by providers that don't support it.
|
||||
*/
|
||||
sessionId?: string;
|
||||
/**
|
||||
* Do not use or record server-side continuation state for this request (for example
|
||||
* Anthropic message threads). Set this for replays such as cache warming, which must not
|
||||
* replace the state of the real conversation. Providers that do not keep continuation
|
||||
* state ignore it.
|
||||
*/
|
||||
disableContinuation?: boolean;
|
||||
/**
|
||||
* WebSocket connect timeout in milliseconds for providers that support
|
||||
* WebSocket transports. This covers the connection/open handshake only;
|
||||
|
|
|
|||
560
packages/ai/test/anthropic-message-threads.test.ts
Normal file
560
packages/ai/test/anthropic-message-threads.test.ts
Normal file
|
|
@ -0,0 +1,560 @@
|
|||
import { Type } from "typebox";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { type AnthropicOptions, stream } from "../src/api/anthropic-messages.ts";
|
||||
import { cleanupSessionResources } from "../src/session-resources.ts";
|
||||
import type { AssistantMessage, Message, Model, Tool, ToolResultMessage } from "../src/types.ts";
|
||||
import { normalizeContext } from "../src/utils/transcript.ts";
|
||||
|
||||
const THREADS_BETA = "message-threads-2026-08-12";
|
||||
const grammarError =
|
||||
"The compiled grammar is too large, which would cause performance issues. Simplify your tool schemas or reduce the number of strict tools.";
|
||||
|
||||
interface WireMessage {
|
||||
role: string;
|
||||
content: unknown;
|
||||
output_config?: { effort?: string };
|
||||
}
|
||||
|
||||
interface WireBody {
|
||||
model: string;
|
||||
messages: WireMessage[];
|
||||
system?: unknown;
|
||||
tools?: unknown[];
|
||||
max_tokens: number;
|
||||
thinking?: unknown;
|
||||
output_config?: unknown;
|
||||
tool_choice?: unknown;
|
||||
fallbacks?: unknown;
|
||||
thread?: { type: string; previous_message_id?: string };
|
||||
}
|
||||
|
||||
interface CapturedRequest {
|
||||
body: WireBody;
|
||||
betas: string[];
|
||||
}
|
||||
|
||||
interface ReplySpec {
|
||||
id?: string;
|
||||
model?: string;
|
||||
text?: string;
|
||||
thinking?: string;
|
||||
toolCalls?: Array<{ id: string; name: string; input: Record<string, unknown> }>;
|
||||
stopReason?: string;
|
||||
/** Event to emit instead of finishing normally. */
|
||||
end?: "error_event" | "no_message_stop";
|
||||
}
|
||||
|
||||
function sse(events: Array<{ type: string } & Record<string, unknown>>): Response {
|
||||
return new Response(events.map((event) => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join(""), {
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
});
|
||||
}
|
||||
|
||||
function reply(request: CapturedRequest, spec: ReplySpec = {}): Response {
|
||||
const events: Array<{ type: string } & Record<string, unknown>> = [
|
||||
{
|
||||
type: "message_start",
|
||||
message: {
|
||||
id: spec.id ?? "msg_x",
|
||||
model: spec.model ?? request.body.model,
|
||||
usage: { input_tokens: 10, output_tokens: 0 },
|
||||
},
|
||||
},
|
||||
];
|
||||
let index = 0;
|
||||
if (spec.thinking !== undefined) {
|
||||
events.push(
|
||||
{ type: "content_block_start", index, content_block: { type: "thinking", thinking: "", signature: "" } },
|
||||
{ type: "content_block_delta", index, delta: { type: "thinking_delta", thinking: spec.thinking } },
|
||||
{ type: "content_block_delta", index, delta: { type: "signature_delta", signature: "sig" } },
|
||||
{ type: "content_block_stop", index },
|
||||
);
|
||||
index++;
|
||||
}
|
||||
events.push(
|
||||
{ type: "content_block_start", index, content_block: { type: "text", text: "" } },
|
||||
{ type: "content_block_delta", index, delta: { type: "text_delta", text: spec.text ?? "answer" } },
|
||||
{ type: "content_block_stop", index },
|
||||
);
|
||||
index++;
|
||||
for (const call of spec.toolCalls ?? []) {
|
||||
events.push(
|
||||
{
|
||||
type: "content_block_start",
|
||||
index,
|
||||
content_block: { type: "tool_use", id: call.id, name: call.name, input: {} },
|
||||
},
|
||||
{
|
||||
type: "content_block_delta",
|
||||
index,
|
||||
delta: { type: "input_json_delta", partial_json: JSON.stringify(call.input) },
|
||||
},
|
||||
{ type: "content_block_stop", index },
|
||||
);
|
||||
index++;
|
||||
}
|
||||
if (spec.end === "error_event") {
|
||||
return new Response(
|
||||
`${events.map((event) => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join("")}event: error\ndata: ${JSON.stringify({ type: "error", error: { type: "overloaded_error", message: "Overloaded" } })}\n\n`,
|
||||
{ headers: { "content-type": "text/event-stream" } },
|
||||
);
|
||||
}
|
||||
const stopReason = spec.stopReason ?? (spec.toolCalls ? "tool_use" : "end_turn");
|
||||
events.push({ type: "message_delta", delta: { stop_reason: stopReason }, usage: { output_tokens: 1 } });
|
||||
if (spec.end !== "no_message_stop") events.push({ type: "message_stop" });
|
||||
return sse(events);
|
||||
}
|
||||
|
||||
function errorResponse(
|
||||
status: number,
|
||||
message: string,
|
||||
errorCode?: string,
|
||||
headers?: Record<string, string>,
|
||||
): Response {
|
||||
const type = status === 404 ? "not_found_error" : status === 429 ? "rate_limit_error" : "invalid_request_error";
|
||||
return Response.json(
|
||||
{ type: "error", error: { type, message, ...(errorCode ? { details: { error_code: errorCode } } : {}) } },
|
||||
{ status, headers },
|
||||
);
|
||||
}
|
||||
|
||||
type Responder = (request: CapturedRequest) => Response;
|
||||
|
||||
let sessionCounter = 0;
|
||||
|
||||
/** The fields Anthropic requires to stay the same within a thread (observed live). */
|
||||
function fingerprint({ body, betas }: CapturedRequest): string {
|
||||
const { model, thinking, output_config, tool_choice, fallbacks } = body;
|
||||
return JSON.stringify([model, thinking, output_config, tool_choice, fallbacks !== undefined, betas]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Fake Anthropic endpoint: records request bodies and beta lists, and answers from a queue or with a
|
||||
* plain reply. Like the real server, it rejects a `continue` whose fixed fields differ from the thread's.
|
||||
*/
|
||||
function harness() {
|
||||
const requests: CapturedRequest[] = [];
|
||||
const queue: Responder[] = [];
|
||||
const threadFingerprints = new Map<string, string>();
|
||||
const sessionId = `thread-test-${++sessionCounter}`;
|
||||
const fetch: NonNullable<AnthropicOptions["fetch"]> = async (_input, init) => {
|
||||
const body = JSON.parse(init?.body as string) as WireBody;
|
||||
const beta = new Headers(init?.headers).get("anthropic-beta");
|
||||
const request = { body, betas: beta ? beta.split(",") : [] };
|
||||
requests.push(request);
|
||||
const next = queue.shift();
|
||||
if (next) return next(request);
|
||||
const id = `msg_${requests.length}`;
|
||||
const previous = body.thread?.previous_message_id;
|
||||
const expected = previous === undefined ? fingerprint(request) : threadFingerprints.get(previous);
|
||||
if (expected !== fingerprint(request)) {
|
||||
return errorResponse(400, "fixed field changed", "thread_fingerprint_mismatch");
|
||||
}
|
||||
if (body.thread) threadFingerprints.set(id, expected);
|
||||
return reply(request, { id });
|
||||
};
|
||||
const options: AnthropicOptions = { apiKey: "test-key", fetch, sessionId, maxRetries: 0 };
|
||||
return { requests, queue, fetch, sessionId, options };
|
||||
}
|
||||
|
||||
const baseModel: Model<"anthropic-messages"> = {
|
||||
id: "claude-test",
|
||||
name: "Test Claude",
|
||||
api: "anthropic-messages",
|
||||
provider: "anthropic",
|
||||
baseUrl: "https://api.anthropic.com",
|
||||
reasoning: false,
|
||||
input: ["text"],
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
|
||||
contextWindow: 200000,
|
||||
maxTokens: 100,
|
||||
};
|
||||
|
||||
function tool(name: string): Tool {
|
||||
return { name, description: `Tool ${name}`, parameters: Type.Object({ value: Type.String() }) };
|
||||
}
|
||||
|
||||
function user(text: string): Message {
|
||||
return { role: "user", content: text, timestamp: 1 };
|
||||
}
|
||||
|
||||
function toolResult(id: string, name: string): ToolResultMessage {
|
||||
return {
|
||||
role: "toolResult",
|
||||
toolCallId: id,
|
||||
toolName: name,
|
||||
content: [{ type: "text", text: "result" }],
|
||||
isError: false,
|
||||
timestamp: 1,
|
||||
};
|
||||
}
|
||||
|
||||
interface TurnInput {
|
||||
model?: Model<"anthropic-messages">;
|
||||
systemPrompt?: string;
|
||||
tools?: Tool[];
|
||||
options?: AnthropicOptions;
|
||||
}
|
||||
|
||||
async function turn(
|
||||
h: ReturnType<typeof harness>,
|
||||
messages: Message[],
|
||||
input: TurnInput = {},
|
||||
): Promise<AssistantMessage> {
|
||||
const context = normalizeContext({
|
||||
systemPrompt: input.systemPrompt ?? "You are a test assistant.",
|
||||
tools: input.tools ?? [tool("lookup")],
|
||||
messages,
|
||||
});
|
||||
return stream(input.model ?? baseModel, context, { ...h.options, ...input.options }).result();
|
||||
}
|
||||
|
||||
/** Runs one turn and appends the reply to the conversation. */
|
||||
async function say(
|
||||
h: ReturnType<typeof harness>,
|
||||
messages: Message[],
|
||||
input: TurnInput = {},
|
||||
): Promise<AssistantMessage> {
|
||||
const result = await turn(h, messages, input);
|
||||
messages.push(result);
|
||||
return result;
|
||||
}
|
||||
|
||||
function threadOf(request: CapturedRequest | undefined) {
|
||||
return request?.body.thread;
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
cleanupSessionResources();
|
||||
});
|
||||
|
||||
describe("Anthropic message threads", () => {
|
||||
it("creates a thread, then continues it with only the new messages", async () => {
|
||||
const h = harness();
|
||||
const messages = [user("first")];
|
||||
await say(h, messages);
|
||||
messages.push(user("second"));
|
||||
const result = await say(h, messages);
|
||||
|
||||
expect(result.stopReason).toBe("stop");
|
||||
const [create, next] = h.requests;
|
||||
expect(create.body.thread).toEqual({ type: "create" });
|
||||
expect(create.body.system).toBeDefined();
|
||||
expect(create.body.tools).toBeDefined();
|
||||
expect(next.body.thread).toEqual({ type: "continue", previous_message_id: "msg_1" });
|
||||
expect(next.body).not.toHaveProperty("system");
|
||||
expect(next.body).not.toHaveProperty("tools");
|
||||
// The first user message had string content turned into a cached block on request 1,
|
||||
// and plain string content on request 2; the cache marker moved to the new message.
|
||||
expect(create.body.messages[0].content).toEqual([
|
||||
{ type: "text", text: "first", cache_control: { type: "ephemeral" } },
|
||||
]);
|
||||
expect(next.body.messages).toEqual([
|
||||
{ role: "user", content: [{ type: "text", text: "second", cache_control: { type: "ephemeral" } }] },
|
||||
]);
|
||||
expect(create.betas).toContain(THREADS_BETA);
|
||||
expect(next.betas).toEqual(create.betas);
|
||||
});
|
||||
|
||||
it("continues after a reply that ends in tool calls, without placeholder results in the prefix", async () => {
|
||||
const h = harness();
|
||||
h.queue.push((request) =>
|
||||
reply(request, { id: "msg_tool", toolCalls: [{ id: "toolu_1", name: "lookup", input: { value: "x" } }] }),
|
||||
);
|
||||
const messages = [user("look it up")];
|
||||
const first = await say(h, messages);
|
||||
expect(first.stopReason).toBe("toolUse");
|
||||
messages.push(toolResult("toolu_1", "lookup"));
|
||||
await say(h, messages);
|
||||
|
||||
const next = h.requests[1];
|
||||
expect(next.body.thread).toEqual({ type: "continue", previous_message_id: "msg_tool" });
|
||||
expect(next.body.messages).toHaveLength(1);
|
||||
expect(next.body.messages[0]).toMatchObject({
|
||||
role: "user",
|
||||
content: [{ type: "tool_result", tool_use_id: "toolu_1" }],
|
||||
});
|
||||
});
|
||||
|
||||
it("continues when the effort level changes between turns", async () => {
|
||||
const model: Model<"anthropic-messages"> = {
|
||||
...baseModel,
|
||||
reasoning: true,
|
||||
compat: { forceAdaptiveThinking: true, supportsMidConvoEffort: true },
|
||||
};
|
||||
const h = harness();
|
||||
h.queue.push((request) => reply(request, { id: "msg_a", thinking: "thought" }));
|
||||
const messages = [user("first")];
|
||||
await say(h, messages, { model, options: { thinkingEnabled: true, effort: "high" } });
|
||||
messages.push(user("second"));
|
||||
await say(h, messages, { model, options: { thinkingEnabled: true, effort: "low" } });
|
||||
|
||||
expect(h.requests[0].body.messages.map((message) => message.output_config?.effort ?? message.role)).toEqual([
|
||||
"user",
|
||||
"high",
|
||||
]);
|
||||
const next = h.requests[1];
|
||||
expect(next.body.thread).toEqual({ type: "continue", previous_message_id: "msg_a" });
|
||||
expect(next.body.messages.map((message) => message.output_config?.effort ?? message.role)).toEqual([
|
||||
"user",
|
||||
"low",
|
||||
]);
|
||||
});
|
||||
|
||||
it("continues when a system message with a native tool change follows the reply", async () => {
|
||||
const model: Model<"anthropic-messages"> = {
|
||||
...baseModel,
|
||||
compat: { supportsMidConvoSystemMessages: true, supportsMidConvoToolChanges: true },
|
||||
};
|
||||
const h = harness();
|
||||
const messages: Message[] = [user("first")];
|
||||
await say(h, messages, { model });
|
||||
messages.push({ role: "system", content: "Update", toolsAdded: [tool("other")], timestamp: 2 }, user("second"));
|
||||
await say(h, messages, { model });
|
||||
|
||||
const next = h.requests[1];
|
||||
expect(next.body.thread?.type).toBe("continue");
|
||||
expect(next.body.messages.map((message) => message.role)).toEqual(["user", "system"]);
|
||||
expect(JSON.stringify(next.body.messages[1].content)).toContain("tool_addition");
|
||||
});
|
||||
|
||||
it.each<[string, TurnInput, TurnInput]>([
|
||||
["system prompt", {}, { systemPrompt: "Different prompt." }],
|
||||
["tools", {}, { tools: [tool("lookup"), tool("other")] }],
|
||||
["model", {}, { model: { ...baseModel, id: "claude-other" } }],
|
||||
["tool choice", {}, { options: { toolChoice: "any" } }],
|
||||
["cache retention", {}, { options: { cacheRetention: "long" } }],
|
||||
[
|
||||
"thinking level",
|
||||
{ model: { ...baseModel, reasoning: true }, options: { thinkingEnabled: true, thinkingBudgetTokens: 1024 } },
|
||||
{ model: { ...baseModel, reasoning: true }, options: { thinkingEnabled: true, thinkingBudgetTokens: 2048 } },
|
||||
],
|
||||
])("creates a new thread when the %s changes", async (_name, before, after) => {
|
||||
const h = harness();
|
||||
const messages = [user("first")];
|
||||
await say(h, messages, before);
|
||||
messages.push(user("second"));
|
||||
await say(h, messages, after);
|
||||
expect(threadOf(h.requests[1])).toEqual({ type: "create" });
|
||||
expect(h.requests[1].body.messages).toHaveLength(3);
|
||||
});
|
||||
|
||||
it("creates a new thread when the request has no messages beyond the stored prefix", async () => {
|
||||
const h = harness();
|
||||
const messages = [user("first")];
|
||||
await say(h, messages);
|
||||
await turn(h, messages);
|
||||
expect(threadOf(h.requests[1])).toEqual({ type: "create" });
|
||||
});
|
||||
|
||||
it.each([
|
||||
["thread_not_found", 404],
|
||||
["thread_already_continued", 400],
|
||||
["thread_continue_in_flight", 400],
|
||||
["thread_fingerprint_mismatch", 400],
|
||||
])("starts a new thread with the full request after %s", async (errorCode, status) => {
|
||||
const h = harness();
|
||||
const messages = [user("first")];
|
||||
await say(h, messages);
|
||||
messages.push(user("second"));
|
||||
h.queue.push(() => errorResponse(status, "thread rejected", errorCode));
|
||||
const onPayload = vi.fn();
|
||||
const result = await say(h, messages, { options: { onPayload } });
|
||||
|
||||
expect(result.stopReason).toBe("stop");
|
||||
expect(onPayload).toHaveBeenCalledTimes(1);
|
||||
expect(h.requests).toHaveLength(3);
|
||||
const [, rejected, resend] = h.requests;
|
||||
expect(rejected.body.thread?.type).toBe("continue");
|
||||
expect(resend.body.thread).toEqual({ type: "create" });
|
||||
expect(resend.body.messages).toHaveLength(3);
|
||||
expect(resend.body.system).toBeDefined();
|
||||
expect(resend.betas).toEqual(rejected.betas);
|
||||
expect(result.diagnostics).toHaveLength(1);
|
||||
expect(result.diagnostics![0]).toMatchObject({
|
||||
type: "anthropic_thread_fallback",
|
||||
details: { reason: errorCode },
|
||||
});
|
||||
|
||||
messages.push(user("third"));
|
||||
await say(h, messages);
|
||||
expect(threadOf(h.requests[3])).toEqual({ type: "continue", previous_message_id: "msg_3" });
|
||||
});
|
||||
|
||||
it.each([
|
||||
["thread_unsupported_request", "thread_unsupported_request", "rejected"],
|
||||
["a 400 without error code about the thread field", undefined, "thread: Extra inputs are not permitted"],
|
||||
])("turns threads off for the session after %s on continue", async (_name, errorCode, message) => {
|
||||
const h = harness();
|
||||
const messages = [user("first")];
|
||||
await say(h, messages);
|
||||
messages.push(user("second"));
|
||||
h.queue.push(() => errorResponse(400, message, errorCode));
|
||||
const result = await say(h, messages);
|
||||
|
||||
expect(result.stopReason).toBe("stop");
|
||||
expect(result.diagnostics?.[0]).toMatchObject({
|
||||
type: "anthropic_thread_fallback",
|
||||
details: { reason: errorCode ?? "400" },
|
||||
});
|
||||
messages.push(user("third"));
|
||||
await say(h, messages);
|
||||
const later = h.requests[3];
|
||||
expect(later.body).not.toHaveProperty("thread");
|
||||
// The resend was accepted with the threads beta, so the beta list stays the same.
|
||||
expect(later.betas).toEqual(h.requests[0].betas);
|
||||
});
|
||||
|
||||
it("shows the error when the resend fails too and keeps threads on", async () => {
|
||||
const h = harness();
|
||||
const messages = [user("first")];
|
||||
await say(h, messages);
|
||||
messages.push(user("second"));
|
||||
h.queue.push(
|
||||
() => errorResponse(400, "thread: Extra inputs are not permitted"),
|
||||
() => errorResponse(400, "Real problem"),
|
||||
);
|
||||
const failed = await turn(h, messages);
|
||||
expect(failed.stopReason).toBe("error");
|
||||
expect(failed.errorMessage).toContain("Real problem");
|
||||
expect(h.requests).toHaveLength(3);
|
||||
|
||||
await turn(h, messages);
|
||||
expect(threadOf(h.requests[3])).toEqual({ type: "create" });
|
||||
});
|
||||
|
||||
it.each([
|
||||
[400, "prompt is too long: 250000 tokens > 200000 maximum"],
|
||||
[404, "model: claude-test"],
|
||||
])("passes a %s that is not about the thread on without resending", async (status, message) => {
|
||||
const h = harness();
|
||||
const messages = [user("first")];
|
||||
await say(h, messages);
|
||||
messages.push(user("second"));
|
||||
h.queue.push(() => errorResponse(status, message));
|
||||
const failed = await turn(h, messages);
|
||||
expect(failed.stopReason).toBe("error");
|
||||
expect(failed.errorMessage).toContain(message);
|
||||
expect(h.requests).toHaveLength(2);
|
||||
expect(failed.diagnostics).toBeUndefined();
|
||||
|
||||
// Threads stay on: after a fix such as compaction, the next request starts a new thread.
|
||||
await turn(h, messages);
|
||||
expect(threadOf(h.requests[2])).toEqual({ type: "create" });
|
||||
});
|
||||
|
||||
it("leaves strict tool rejections to the strict tool fallback", async () => {
|
||||
const model: Model<"anthropic-messages"> = { ...baseModel, compat: { supportsStrictTools: true } };
|
||||
const strictTool: Tool = { ...tool("lookup"), constrainedSampling: { type: "json_schema", strict: "prefer" } };
|
||||
const h = harness();
|
||||
h.queue.push(() => errorResponse(400, grammarError));
|
||||
const result = await turn(h, [user("first")], { model, tools: [strictTool] });
|
||||
expect(result.stopReason).toBe("stop");
|
||||
expect(h.requests).toHaveLength(2);
|
||||
expect(h.requests.map((request) => threadOf(request)?.type)).toEqual(["create", "create"]);
|
||||
expect(result.diagnostics?.map((diagnostic) => diagnostic.type)).toEqual(["anthropic_strict_tools_fallback"]);
|
||||
});
|
||||
|
||||
it("starts a new thread when a retried continue finds the thread in flight", async () => {
|
||||
const h = harness();
|
||||
const messages = [user("first")];
|
||||
await say(h, messages);
|
||||
messages.push(user("second"));
|
||||
h.queue.push(
|
||||
() => errorResponse(503, "overloaded", undefined, { "retry-after-ms": "0" }),
|
||||
() => errorResponse(400, "in flight", "thread_continue_in_flight"),
|
||||
);
|
||||
const result = await say(h, messages, { options: { maxRetries: 1 } });
|
||||
expect(result.stopReason).toBe("stop");
|
||||
expect(h.requests.map((request) => threadOf(request)?.type)).toEqual([
|
||||
"create",
|
||||
"continue",
|
||||
"continue",
|
||||
"create",
|
||||
]);
|
||||
});
|
||||
|
||||
it.each<[string, string | null]>([
|
||||
["without the threads beta", "fine-grained-tool-streaming-2025-05-14"],
|
||||
["set to null", null],
|
||||
])("fails once, then turns threads off, with an explicit anthropic-beta header %s", async (_name, header) => {
|
||||
const h = harness();
|
||||
const respond: Responder = (request) =>
|
||||
request.body.thread && !request.betas.includes(THREADS_BETA)
|
||||
? errorResponse(400, "thread: Extra inputs are not permitted")
|
||||
: reply(request, { id: `msg_${h.requests.length}` });
|
||||
h.queue.push(respond, respond, respond);
|
||||
const options: AnthropicOptions = { headers: { "anthropic-beta": header } };
|
||||
const messages = [user("first")];
|
||||
const result = await say(h, messages, { options });
|
||||
expect(result.stopReason).toBe("stop");
|
||||
messages.push(user("second"));
|
||||
await say(h, messages, { options });
|
||||
expect(h.requests.map((request) => threadOf(request)?.type)).toEqual(["create", undefined, undefined]);
|
||||
});
|
||||
|
||||
it.each<[string, ReplySpec | "abort"]>([
|
||||
["an aborted reply", "abort"],
|
||||
["a stream error", { end: "error_event" }],
|
||||
["a stream that ends early", { end: "no_message_stop" }],
|
||||
["a refusal", { stopReason: "refusal" }],
|
||||
["a fallback model reply", { model: "claude-fallback" }],
|
||||
])("does not record %s", async (_name, spec) => {
|
||||
const h = harness();
|
||||
const controller = new AbortController();
|
||||
if (spec !== "abort") h.queue.push((request) => reply(request, { id: "msg_bad", ...spec }));
|
||||
const messages = [user("first")];
|
||||
const first = await turn(h, messages, {
|
||||
options:
|
||||
spec === "abort"
|
||||
? {
|
||||
signal: controller.signal,
|
||||
onProviderStreamEvent: (event) => {
|
||||
if ((event as { type: string }).type === "message_start") controller.abort();
|
||||
},
|
||||
}
|
||||
: {},
|
||||
});
|
||||
messages.push(first, user("second"));
|
||||
await turn(h, messages);
|
||||
expect(threadOf(h.requests[1])).toEqual({ type: "create" });
|
||||
});
|
||||
|
||||
it("does not use threads with cacheRetention none", async () => {
|
||||
const h = harness();
|
||||
const messages = [user("first")];
|
||||
await say(h, messages, { options: { cacheRetention: "none" } });
|
||||
expect(h.requests[0].body).not.toHaveProperty("thread");
|
||||
expect(h.requests[0].betas).not.toContain(THREADS_BETA);
|
||||
});
|
||||
|
||||
it("matches Claude Code tool names across turns with OAuth", async () => {
|
||||
const h = harness();
|
||||
h.queue.push((request) =>
|
||||
reply(request, { id: "msg_oauth", toolCalls: [{ id: "toolu_1", name: "Read", input: { value: "x" } }] }),
|
||||
);
|
||||
const options: AnthropicOptions = { apiKey: "sk-ant-oat-test" };
|
||||
const messages = [user("read it")];
|
||||
const first = await say(h, messages, { tools: [tool("read")], options });
|
||||
expect(first.content.find((block) => block.type === "toolCall")).toMatchObject({ name: "read" });
|
||||
messages.push(toolResult("toolu_1", "read"));
|
||||
await say(h, messages, { tools: [tool("read")], options });
|
||||
expect(threadOf(h.requests[1])).toEqual({ type: "continue", previous_message_id: "msg_oauth" });
|
||||
});
|
||||
|
||||
it("sends cache warming replays without a thread and keeps the conversation's thread", async () => {
|
||||
const h = harness();
|
||||
const messages = [user("first")];
|
||||
await say(h, messages);
|
||||
await turn(h, messages.slice(0, 1), { options: { maxTokens: 1, disableContinuation: true } });
|
||||
messages.push(user("second"));
|
||||
await say(h, messages);
|
||||
|
||||
const [real, warm, next] = h.requests;
|
||||
expect(warm.body).not.toHaveProperty("thread");
|
||||
expect(warm.body.max_tokens).toBe(1);
|
||||
expect(warm.body.messages).toEqual(real.body.messages);
|
||||
expect(warm.betas).toEqual(real.betas);
|
||||
expect(next.body.thread).toEqual({ type: "continue", previous_message_id: "msg_1" });
|
||||
});
|
||||
});
|
||||
|
|
@ -5,6 +5,7 @@
|
|||
### Added
|
||||
|
||||
- `pi update` now lists each updated package with its npm version or git commit change, for example `npm:foo 1.2.0 -> 1.3.0`, and prints `Packages are already up to date` when nothing changed. When nothing changed but an update printed a warning, for example because npm left the old version installed after resolving stale registry metadata, it prints `No packages were updated, see the warnings above` instead; in offline mode it prints `Skipped package updates (offline mode)`. `PackageManager.update()` reports each changed package as a new `result` progress event whose `message` holds that line ([#11707](https://github.com/earendil-works/pi/pull/11707), [#11711](https://github.com/earendil-works/pi/pull/11711))
|
||||
- Anthropic sessions now use message threads: each request sends only the new messages and the server continues from the previous reply, falling back to the full request when the conversation changed (for example after compaction or a tool list change). On by default for the `anthropic` provider on `api.anthropic.com`; set `compat.supportsMessageThreads` in `models.json` to enable it for gateways that forward the `thread` field, or to turn it off. Cache warming does not touch the session's thread ([#11545](https://github.com/earendil-works/pi/pull/11545))
|
||||
|
||||
## [1.5.0] - 2026-11-02
|
||||
|
||||
|
|
|
|||
|
|
@ -878,6 +878,8 @@ pi.on("before_provider_request", (event, ctx) => {
|
|||
|
||||
This is mainly useful for debugging provider serialization and cache behavior.
|
||||
|
||||
The payload is the full request. Providers that continue from a previous response (OpenAI Codex over WebSocket, Anthropic message threads) may send only the part that is new since the previous request.
|
||||
|
||||
#### after_provider_response
|
||||
|
||||
Fired after an HTTP response is received and before its stream body is consumed. Handlers run in extension load order.
|
||||
|
|
|
|||
|
|
@ -172,6 +172,7 @@ Also applies to `azure-openai-responses` and `openai-codex-responses`.
|
|||
| `supportsMidConvoEffort` | boolean | `false` | Whether the exact model transport supports effort-only system messages and thinking binding controls. |
|
||||
| `supportsMidConvoSystemMessages` | boolean | `false` | Whether the exact model accepts system-role messages inside the conversation. When false, later system messages are folded into the top-level system prompt. |
|
||||
| `supportsMidConvoToolChanges` | boolean | `false` | Whether the exact model accepts mid-conversation tool_addition and tool_removal blocks. Requires supportsMidConvoSystemMessages. |
|
||||
| `supportsMessageThreads` | boolean | | Whether the endpoint supports Anthropic message threads, which let a request with a session id send only the new messages and continue from the previous reply. Default: true for provider anthropic with a baseUrl on api.anthropic.com, false otherwise. Set true for gateways that forward the thread request field. |
|
||||
| `allowedFallbackModels` | `object[]` | | Models Anthropic accepts for server-side refusal fallback, with local pricing metadata for returned fallback responses. When absent or empty, callers must omit fallbacks; Anthropic rejects the field for models with no permitted fallback targets. |
|
||||
|
||||
### `bedrock-converse-stream`
|
||||
|
|
|
|||
|
|
@ -1081,6 +1081,10 @@
|
|||
"description": "Whether the exact model accepts mid-conversation tool_addition and tool_removal blocks. Requires supportsMidConvoSystemMessages.",
|
||||
"default": false
|
||||
},
|
||||
"supportsMessageThreads": {
|
||||
"type": "boolean",
|
||||
"description": "Whether the endpoint supports Anthropic message threads, which let a request with a session id send only the new messages and continue from the previous reply. Default: true for provider anthropic with a baseUrl on api.anthropic.com, false otherwise. Set true for gateways that forward the thread request field."
|
||||
},
|
||||
"allowedFallbackModels": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
|
|
|
|||
|
|
@ -333,6 +333,8 @@ export class CacheWarmer {
|
|||
.streamSimple(run.model, run.context, {
|
||||
...run.options,
|
||||
maxTokens: 1,
|
||||
// The warm request repeats the last request; it must not replace the conversation's thread state.
|
||||
disableContinuation: true,
|
||||
maxRetries: 0,
|
||||
signal: run.controller.signal,
|
||||
})
|
||||
|
|
|
|||
|
|
@ -155,7 +155,14 @@ describe("cache warming", () => {
|
|||
|
||||
expect(calls[0]).toMatchObject({
|
||||
model: adaptiveModel,
|
||||
options: { reasoning: "high", sessionId: "s", transformHeaders, maxTokens: 1, maxRetries: 0 },
|
||||
options: {
|
||||
reasoning: "high",
|
||||
sessionId: "s",
|
||||
transformHeaders,
|
||||
maxTokens: 1,
|
||||
maxRetries: 0,
|
||||
disableContinuation: true,
|
||||
},
|
||||
});
|
||||
expect(calls[0].options?.signal).not.toBe(signal);
|
||||
expect(events[0]).toMatchObject({
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue