apunkt/rowboat
1
0
Fork 0
mirror of https://github.com/rowboatlabs/rowboat.git synced 2026-07-12 21:02:17 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(x): preserve block-level provider metadata in the turns bridge

Browse source 
The bridge dropped providerMetadata from content-block stream events,
so reasoning parts persisted as bare {type, text}. Anthropic thinking
signatures (which arrive on a reasoning-delta with an empty text delta),
Gemini thoughtSignatures (on text-end / reasoning-end / tool-call), and
OpenAI encrypted reasoning were all lost — breaking multi-step tool
turns whenever extended thinking is enabled, since providers require
signed blocks to be echoed back verbatim within the tool loop.

Now every -start event opens a new part (distinct blocks keep distinct
signatures) and metadata from each event of a block is merged opaquely
onto that part's providerOptions, which convertFromMessages already
echoes verbatim. The bridge never interprets the contents; each
provider reads back only its own keys. Empty blocks with no metadata
are dropped, matching the previous lazy-creation behavior. finish-step
metadata stays top-level as before.

Schemas already allowed providerOptions on all part types, so persisted
files remain readable in both directions.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Ramnique Singh 2026-07-10 09:36:53 +05:30
parent 7a0e729703
commit 091f3e3acd
2 changed files with 251 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

160
apps/x/packages/core/src/turns/bridges/real-model-registry.test.ts
View file

@ -138,6 +138,166 @@ describe("RealModelRegistry", () => {
]);
});
it("captures block-level provider metadata opaquely onto parts (Anthropic-style signatures)", async () => {
// Anthropic delivers the thinking signature on a reasoning-delta with
// an EMPTY text delta, and redacted thinking as metadata on
// reasoning-start. Both must land on the right reasoning part, and
// distinct blocks must stay distinct parts.
const registry = makeRegistry(
[
{ type: "reasoning-start" },
{ type: "reasoning-delta", text: "let me think" },
{ type: "reasoning-delta", text: "", providerMetadata: { anthropic: { signature: "sig-1" } } },
{ type: "reasoning-end" },
{
type: "reasoning-start",
providerMetadata: { anthropic: { redactedData: "opaque-blob" } },
},
{ type: "reasoning-end" },
{ type: "text-start" },
{ type: "text-delta", text: "answer" },
{ type: "text-end" },
{ type: "finish-step", finishReason: "stop", usage: {} },
],
[],
);
const events = await collect(registry, request());
const completed = events[events.length - 1];
expect(
completed.type === "completed" ? completed.message.content : undefined,
).toEqual([
{
type: "reasoning",
text: "let me think",
providerOptions: { anthropic: { signature: "sig-1" } },
},
{
type: "reasoning",
text: "",
providerOptions: { anthropic: { redactedData: "opaque-blob" } },
},
{ type: "text", text: "answer" },
]);
});
it("captures metadata on text and tool-call parts (Gemini-style thoughtSignatures)", async () => {
const registry = makeRegistry(
[
{ type: "text-start" },
{ type: "text-delta", text: "calling a tool" },
{ type: "text-end", providerMetadata: { google: { thoughtSignature: "ts-text" } } },
{
type: "tool-call",
toolCallId: "tc1",
toolName: "echo",
input: { x: 1 },
providerMetadata: { google: { thoughtSignature: "ts-call" } },
},
{ type: "finish-step", finishReason: "tool-calls", usage: {} },
],
[],
);
const events = await collect(registry, request());
const completed = events[events.length - 1];
expect(
completed.type === "completed" ? completed.message.content : undefined,
).toEqual([
{
type: "text",
text: "calling a tool",
providerOptions: { google: { thoughtSignature: "ts-text" } },
},
{
type: "tool-call",
toolCallId: "tc1",
toolName: "echo",
arguments: { x: 1 },
providerOptions: { google: { thoughtSignature: "ts-call" } },
},
]);
});
it("merges metadata from multiple events of one block, later fields winning", async () => {
const registry = makeRegistry(
[
{ type: "reasoning-start", providerMetadata: { openai: { itemId: "r-1" } } },
{ type: "reasoning-delta", text: "…" },
{
type: "reasoning-end",
providerMetadata: { openai: { itemId: "r-1", reasoningEncryptedContent: "enc" } },
},
{ type: "finish-step", finishReason: "stop", usage: {} },
],
[],
);
const events = await collect(registry, request());
const completed = events[events.length - 1];
expect(
completed.type === "completed" ? completed.message.content : undefined,
).toEqual([
{
type: "reasoning",
text: "…",
providerOptions: { openai: { itemId: "r-1", reasoningEncryptedContent: "enc" } },
},
]);
});
it("drops empty blocks that carry no metadata", async () => {
const registry = makeRegistry(
[
{ type: "reasoning-start" },
{ type: "reasoning-end" },
{ type: "text-start" },
{ type: "text-delta", text: "hi" },
{ type: "text-end" },
{ type: "finish-step", finishReason: "stop", usage: {} },
],
[],
);
const events = await collect(registry, request());
const completed = events[events.length - 1];
expect(
completed.type === "completed" ? completed.message.content : undefined,
).toEqual([{ type: "text", text: "hi" }]);
});
it("round-trips part-level providerOptions through encodeMessages", async () => {
const registry = makeRegistry([], []);
const model = await registry.resolve({ provider: "openai", model: "gpt-test" });
const encoded = model.encodeMessages([
{
role: "assistant",
content: [
{
type: "reasoning",
text: "thought",
providerOptions: { anthropic: { signature: "sig-1" } },
},
{
type: "tool-call",
toolCallId: "tc1",
toolName: "echo",
arguments: { x: 1 },
providerOptions: { google: { thoughtSignature: "ts-call" } },
},
],
},
] as never) as Array<{ role: string; content: Array<Record<string, unknown>> }>;
expect(encoded[0].content[0]).toMatchObject({
type: "reasoning",
text: "thought",
providerOptions: { anthropic: { signature: "sig-1" } },
});
expect(encoded[0].content[1]).toMatchObject({
type: "tool-call",
toolCallId: "tc1",
input: { x: 1 },
providerOptions: { google: { thoughtSignature: "ts-call" } },
});
});
it("throws on provider error parts (a model failure, not a completion)", async () => {
const registry = makeRegistry(
[{ type: "error", error: new Error("rate limited") }],

104
apps/x/packages/core/src/turns/bridges/real-model-registry.ts
View file

@ -121,12 +121,33 @@ export class RealModelRegistry implements IModelRegistry {
};
const parts: Array<z.infer<typeof AssistantContentPart>> = [];
// Per-block accumulation: providers attach round-trip metadata to
// individual content blocks (Anthropic thinking signatures arrive on
// reasoning-delta, redacted thinking on reasoning-start; Gemini
// thoughtSignatures on text-end / reasoning-end / tool-call; OpenAI
// encrypted reasoning on reasoning events). Each -start opens a new
// part so distinct blocks keep distinct signatures, and metadata from
// every event of a block is merged onto that part's providerOptions —
// which is exactly what the AI SDK providers read back when the
// message is echoed in later steps.
let currentText: TextContentPart | null = null;
let currentReasoning: ReasoningContentPart | null = null;
let textBuffer = "";
let reasoningBuffer = "";
let finishReason = "unknown";
let usage: z.infer<typeof TurnUsage> = {};
let providerMetadata: JsonValue | undefined;
const tagPart = (
part: TextContentPart | ReasoningContentPart,
metadata: unknown,
) => {
const merged = mergeProviderOptions(part.providerOptions, metadata);
if (merged !== undefined) {
part.providerOptions = merged;
}
};
// Anthropic-family models get cache_control breakpoints; everything
// else passes through byte-identical (transport-only, not persisted).
const prompt = applyPromptCaching(flavor, modelId, {
@ -157,56 +178,74 @@ export class RealModelRegistry implements IModelRegistry {
error?: unknown;
};
switch (event.type) {
case "text-start":
case "text-start": {
textBuffer = "";
currentText = { type: "text", text: "" };
tagPart(currentText, event.providerMetadata);
parts.push(currentText);
yield { type: "step_event", event: { type: "text_start" } };
break;
}
case "text-delta": {
const delta = event.text ?? "";
textBuffer += delta;
const last = parts[parts.length - 1];
if (last?.type === "text") {
last.text += delta;
} else {
parts.push({ type: "text", text: delta });
if (currentText === null) {
currentText = { type: "text", text: "" };
parts.push(currentText);
}
currentText.text += delta;
tagPart(currentText, event.providerMetadata);
yield { type: "text_delta", delta };
break;
}
case "text-end":
if (currentText !== null) {
tagPart(currentText, event.providerMetadata);
currentText = null;
}
yield {
type: "step_event",
event: { type: "text_end", text: textBuffer },
};
break;
case "reasoning-start":
case "reasoning-start": {
reasoningBuffer = "";
currentReasoning = { type: "reasoning", text: "" };
tagPart(currentReasoning, event.providerMetadata);
parts.push(currentReasoning);
yield { type: "step_event", event: { type: "reasoning_start" } };
break;
}
case "reasoning-delta": {
const delta = event.text ?? "";
reasoningBuffer += delta;
const last = parts[parts.length - 1];
if (last?.type === "reasoning") {
last.text += delta;
} else {
parts.push({ type: "reasoning", text: delta });
if (currentReasoning === null) {
currentReasoning = { type: "reasoning", text: "" };
parts.push(currentReasoning);
}
currentReasoning.text += delta;
tagPart(currentReasoning, event.providerMetadata);
yield { type: "reasoning_delta", delta };
break;
}
case "reasoning-end":
if (currentReasoning !== null) {
tagPart(currentReasoning, event.providerMetadata);
currentReasoning = null;
}
yield {
type: "step_event",
event: { type: "reasoning_end", text: reasoningBuffer },
};
break;
case "tool-call": {
const partOptions = mergeProviderOptions(undefined, event.providerMetadata);
const toolCall = {
type: "tool-call" as const,
toolCallId: String(event.toolCallId),
toolName: String(event.toolName),
arguments: event.input,
...(partOptions === undefined ? {} : { providerOptions: partOptions }),
};
parts.push(toolCall);
yield { type: "step_event", event: { type: "tool_call", toolCall } };
@ -238,11 +277,21 @@ export class RealModelRegistry implements IModelRegistry {
}
}
// Blocks that ended up with no text and no round-trip metadata carry
// no information; dropping them matches the pre-block-tracking
// behavior (parts used to be created lazily on first delta).
const content = parts.filter(
(part) =>
part.type === "tool-call"
|| part.text !== ""
|| part.providerOptions !== undefined,
);
yield {
type: "completed",
message: {
role: "assistant",
content: parts.length > 0 ? parts : "",
content: content.length > 0 ? content : "",
},
finishReason,
usage,
@ -251,6 +300,35 @@ export class RealModelRegistry implements IModelRegistry {
}
}
type TextContentPart = Extract<z.infer<typeof AssistantContentPart>, { type: "text" }>;
type ReasoningContentPart = Extract<z.infer<typeof AssistantContentPart>, { type: "reasoning" }>;
type PartProviderOptions = NonNullable<TextContentPart["providerOptions"]>;
// Round-trip metadata is relayed opaquely: whatever JSON-safe, per-provider
// record a stream event carries is merged onto the part (later events win
// per field within a provider's namespace). The bridge never interprets the
// contents — Anthropic signatures, Gemini thoughtSignatures, and OpenAI
// encrypted reasoning all ride the same channel, and each provider reads
// back only its own keys.
function mergeProviderOptions(
base: PartProviderOptions | undefined,
incoming: unknown,
): PartProviderOptions | undefined {
const sanitized = toJsonValue(incoming);
if (sanitized === undefined || sanitized === null || typeof sanitized !== "object" || Array.isArray(sanitized)) {
return base;
}
let merged: PartProviderOptions | undefined;
for (const [provider, fields] of Object.entries(sanitized)) {
if (fields === null || typeof fields !== "object" || Array.isArray(fields)) {
continue;
}
merged = merged ?? { ...(base ?? {}) };
merged[provider] = { ...(merged[provider] ?? {}), ...fields };
}
return merged ?? base;
}
function mapUsage(
usage: Record<string, number | undefined> | undefined,
): z.infer<typeof TurnUsage> {