diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 0463b92005..5d9c1571ae 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -24,7 +24,11 @@ import { resolve } from 'node:path'; import { describe, test } from 'node:test'; import type { ModelMessage, ModelStreamResult } from '../model-protocol.js'; import { MockLanguageModelV4, simulateReadableStream } from 'ai/test'; -import { APICallError, type LanguageModelV4StreamPart } from '@ai-sdk/provider'; +import { + APICallError, + type LanguageModelV4StreamPart, + type LanguageModelV4Usage, +} from '@ai-sdk/provider'; import type { AgentRunHeader } from '@maka/core/agent-run'; import type { AttachmentByteReader } from '@maka/core/attachments'; import type { BackendSendInput } from '@maka/core/backend-types'; @@ -7775,25 +7779,20 @@ describe('AiSdkBackend usage telemetry', () => { ); }); - test('does not record fabricated zero telemetry when provider usage is unavailable', async () => { + test('does not record fabricated zero telemetry when completed response usage is unavailable', async () => { const events: SessionEvent[] = []; const model = new MockLanguageModelV4({ doStream: { stream: simulateReadableStream({ chunks: [ { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-1' }, + { type: 'text-delta', id: 'text-1', delta: 'Done.' }, + { type: 'text-end', id: 'text-1' }, { type: 'finish', finishReason: { unified: 'stop', raw: 'stop' }, - usage: { - inputTokens: { - total: undefined, - noCache: undefined, - cacheRead: undefined, - cacheWrite: undefined, - }, - outputTokens: { total: undefined, text: undefined, reasoning: undefined }, - } as never, + usage: unavailableUsage(), }, ], initialDelayInMs: null, @@ -9861,7 +9860,11 @@ describe('AiSdkBackend RunTrace', () => { return { events: (async function* () { input.onStreamActivity(); - yield { kind: 'finish' as const, finishReason: 'stop' }; + yield { + kind: 'finish' as const, + finishReason: 'stop', + disposition: 'authoritative' as const, + }; finishConsumed.release(); await new Promise((_resolve, reject) => { const abort = () => reject(input.abortSignal.reason ?? new Error('aborted')); @@ -9874,6 +9877,7 @@ describe('AiSdkBackend RunTrace', () => { finishReason: 'stop', request: { messages: [] }, continuation: 'none', + hasResponseEvidence: false, }), }; }; @@ -12565,6 +12569,7 @@ describe('AiSdkBackend thinking persistence', () => { }, request: { messages: [] }, continuation: 'none', + hasResponseEvidence: true, }), }); @@ -14039,6 +14044,18 @@ function emptyUsage() { }; } +function unavailableUsage(): LanguageModelV4Usage { + return { + inputTokens: { + total: undefined, + noCache: undefined, + cacheRead: undefined, + cacheWrite: undefined, + }, + outputTokens: { total: undefined, text: undefined, reasoning: undefined }, + }; +} + function imageReplayBackend( model: MockLanguageModelV4, options: { supportsVision: boolean; readAttachmentBytes: AttachmentByteReader }, @@ -14556,3 +14573,692 @@ function runtimeExecute( }) ).result; } + +describe('AiSdkBackend provider finish recovery', () => { + test('retries an empty stop after a settled tool without replaying the tool or spending its step budget', async () => { + const durable = durableTurnHarness('turn-empty-stop-after-tool', 'read notes.md'); + const appended: StoredMessage[] = []; + const usageCheckpoints: Array<{ inputTokens: number; outputTokens: number }> = []; + let calls = 0; + let executions = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + const chunks: LanguageModelV4StreamPart[] = + calls === 1 + ? [ + { type: 'stream-start', warnings: [] }, + { + type: 'tool-call', + toolCallId: 'tool-1', + toolName: 'Read', + input: JSON.stringify({ path: 'notes.md' }), + }, + { + type: 'finish', + finishReason: { unified: 'tool-calls', raw: 'tool_calls' }, + usage: { + inputTokens: { total: 3, noCache: 3, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 1, text: 0, reasoning: 1 }, + }, + }, + ] + : calls === 2 + ? [ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: unavailableUsage(), + }, + ] + : [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-final' }, + { type: 'text-delta', id: 'text-final', delta: 'Recovered' }, + { type: 'text-end', id: 'text-final' }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: { + inputTokens: { total: 5, noCache: 5, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 2, text: 2, reasoning: 0 }, + }, + }, + ]; + return { + stream: simulateReadableStream({ + chunks, + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async (message) => { + appended.push(message); + }, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [ + { + ...testTool('Read', z.object({ path: z.string() })), + impl: async () => { + executions += 1; + return { ok: true }; + }, + }, + ], + maxSteps: 2, + loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents, + recordUsageCheckpoint: async (usage) => { + usageCheckpoints.push(usage); + }, + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events = await drainDurably(backend.send(durable.input()), durable); + + assert.equal(calls, 3); + assert.equal(executions, 1); + assert.deepEqual( + appended + .filter((message): message is AssistantMessage => message.type === 'assistant') + .map((message) => message.text), + ['Recovered'], + ); + assert.deepEqual(model.doStreamCalls[2]?.prompt, model.doStreamCalls[1]?.prompt); + assert.match(JSON.stringify(model.doStreamCalls[2]?.prompt), /"ok":true/); + assert.deepEqual( + events + .filter((event) => event.type === 'provider_retry') + .map(({ phase, attempt, maxAttempts, reason }) => ({ + phase, + attempt, + maxAttempts, + reason, + })), + [ + { phase: 'scheduled', attempt: 2, maxAttempts: 2, reason: 'provider_unavailable' }, + { phase: 'started', attempt: 2, maxAttempts: 2, reason: 'provider_unavailable' }, + ], + ); + assert.deepEqual( + usageCheckpoints.map(({ inputTokens, outputTokens }) => ({ inputTokens, outputTokens })), + [ + { inputTokens: 3, outputTokens: 1 }, + { inputTokens: 8, outputTokens: 3 }, + ], + ); + assert.equal( + events.some((event) => event.type === 'error'), + false, + ); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'end_turn'); + }); + + test('fails metering closed when an after-step stop overtakes an empty-stop retry', async () => { + const durable = durableTurnHarness('turn-empty-stop-after-step-stop', 'read notes.md'); + let backend!: AiSdkBackend; + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + if (calls === 2) await backend.stop('user_stop', 'after_step'); + const chunks: LanguageModelV4StreamPart[] = + calls === 1 + ? [ + { type: 'stream-start', warnings: [] }, + { + type: 'tool-call', + toolCallId: 'tool-1', + toolName: 'Read', + input: JSON.stringify({ path: 'notes.md' }), + }, + { + type: 'finish', + finishReason: { unified: 'tool-calls', raw: 'tool_calls' }, + usage: { + inputTokens: { total: 3, noCache: 3, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 1, text: 0, reasoning: 1 }, + }, + }, + ] + : [ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: unavailableUsage(), + }, + ]; + return { + stream: simulateReadableStream({ + chunks, + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [testTool('Read', z.object({ path: z.string() }))], + maxSteps: 2, + loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents, + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events = await drainDurably(backend.send(durable.input()), durable); + + assert.equal(calls, 2); + assert.equal( + events.some((event) => event.type === 'provider_retry'), + false, + ); + assert.equal( + events.some((event) => event.type === 'token_usage'), + false, + 'the metered tool step is only a partial total after the unmetered empty stop', + ); + assert.equal(events.find((event) => event.type === 'error')?.reason, 'provider_unavailable'); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'error'); + }); + + test('classifies exhausted empty stop retries as provider unavailable', async () => { + const durable = durableTurnHarness('turn-empty-stop-exhausted', 'continue'); + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + return { + stream: simulateReadableStream({ + chunks: [ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: unavailableUsage(), + }, + ], + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + maxSteps: 1, + loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents, + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events = await drainDurably(backend.send(durable.input()), durable); + const error = events.find( + (event): event is Extract => event.type === 'error', + ); + + assert.equal(calls, 2); + assert.equal(error?.reason, 'provider_unavailable'); + assert.equal(error?.message, 'Provider returned an empty stop without output or usable usage'); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'error'); + }); + + test('routes an output-free network_error finish through the network retry path', async () => { + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + const chunks: LanguageModelV4StreamPart[] = + calls === 1 + ? [ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'other', raw: 'network_error' }, + usage: unavailableUsage(), + }, + ] + : [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-1' }, + { type: 'text-delta', id: 'text-1', delta: 'Recovered' }, + { type: 'text-end', id: 'text-1' }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: emptyUsage(), + }, + ]; + return { + stream: simulateReadableStream({ + chunks, + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + maxSteps: 1, + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events: SessionEvent[] = []; + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + } + + assert.equal(calls, 2); + assert.deepEqual( + events + .filter((event) => event.type === 'provider_retry') + .map(({ phase, attempt, maxAttempts, reason }) => ({ + phase, + attempt, + maxAttempts, + reason, + })), + [ + { phase: 'scheduled', attempt: 2, maxAttempts: 10, reason: 'network' }, + { phase: 'started', attempt: 2, maxAttempts: 10, reason: 'network' }, + ], + ); + assert.equal( + events.some((event) => event.type === 'error'), + false, + ); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'end_turn'); + }); + + test('preserves valid usage from a retried network_error finish', async () => { + const usageCheckpoints: Array<{ inputTokens: number; outputTokens: number }> = []; + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + const chunks: LanguageModelV4StreamPart[] = + calls === 1 + ? [ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'other', raw: 'network_error' }, + usage: { + inputTokens: { total: 4, noCache: 4, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 1, text: 1, reasoning: 0 }, + }, + }, + ] + : [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-1' }, + { type: 'text-delta', id: 'text-1', delta: 'Recovered' }, + { type: 'text-end', id: 'text-1' }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: { + inputTokens: { total: 5, noCache: 5, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 2, text: 2, reasoning: 0 }, + }, + }, + ]; + return { + stream: simulateReadableStream({ + chunks, + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + maxSteps: 1, + recordUsageCheckpoint: async (usage) => { + usageCheckpoints.push(usage); + }, + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events: SessionEvent[] = []; + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + } + const usage = events.find( + (event): event is Extract => + event.type === 'token_usage', + ); + + assert.equal(calls, 2); + assert.deepEqual( + usageCheckpoints.map(({ inputTokens, outputTokens }) => ({ inputTokens, outputTokens })), + [ + { inputTokens: 4, outputTokens: 1 }, + { inputTokens: 9, outputTokens: 3 }, + ], + ); + assert.equal(usage?.input, 9); + assert.equal(usage?.output, 3); + assert.equal(usage?.runtimeSteps, 1); + assert.equal( + events.some((event) => event.type === 'error'), + false, + ); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'end_turn'); + }); + + test('does not retry a network_error finish after partial output', async () => { + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + return { + stream: simulateReadableStream({ + chunks: [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-1' }, + { type: 'text-delta', id: 'text-1', delta: 'Partial' }, + { + type: 'finish', + finishReason: { unified: 'other', raw: 'network_error' }, + usage: unavailableUsage(), + }, + ], + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events: SessionEvent[] = []; + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + } + + assert.equal(calls, 1); + assert.equal( + events.some((event) => event.type === 'provider_retry'), + false, + ); + assert.equal(events.find((event) => event.type === 'error')?.reason, 'network'); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'error'); + }); + + test('fails metering closed when a partial network_error follows a metered tool step', async () => { + const durable = durableTurnHarness('turn-partial-network-after-tool', 'read notes.md'); + const usageCheckpoints: Array<{ inputTokens: number; outputTokens: number }> = []; + let calls = 0; + let executions = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + const chunks: LanguageModelV4StreamPart[] = + calls === 1 + ? [ + { type: 'stream-start', warnings: [] }, + { + type: 'tool-call', + toolCallId: 'tool-1', + toolName: 'Read', + input: JSON.stringify({ path: 'notes.md' }), + }, + { + type: 'finish', + finishReason: { unified: 'tool-calls', raw: 'tool_calls' }, + usage: { + inputTokens: { total: 3, noCache: 3, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 1, text: 0, reasoning: 1 }, + }, + }, + ] + : [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-1' }, + { type: 'text-delta', id: 'text-1', delta: 'Partial' }, + { + type: 'finish', + finishReason: { unified: 'other', raw: 'network_error' }, + usage: unavailableUsage(), + }, + ]; + return { + stream: simulateReadableStream({ + chunks, + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [ + { + ...testTool('Read', z.object({ path: z.string() })), + impl: async () => { + executions += 1; + return { ok: true }; + }, + }, + ], + maxSteps: 2, + loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents, + recordUsageCheckpoint: async (usage) => { + usageCheckpoints.push(usage); + }, + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events = await drainDurably(backend.send(durable.input()), durable); + + assert.equal(calls, 2); + assert.equal(executions, 1); + assert.deepEqual( + usageCheckpoints.map(({ inputTokens, outputTokens }) => ({ inputTokens, outputTokens })), + [{ inputTokens: 3, outputTokens: 1 }], + ); + assert.equal( + events.some((event) => event.type === 'provider_retry'), + false, + ); + assert.equal( + events.some((event) => event.type === 'token_usage'), + false, + ); + assert.equal(events.find((event) => event.type === 'error')?.reason, 'network'); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'error'); + }); + + test('fails metering closed when the exhausted network_error retry loses usage', async () => { + const usageCheckpoints: Array<{ inputTokens: number; outputTokens: number }> = []; + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + return { + stream: simulateReadableStream({ + chunks: [ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'other', raw: 'network_error' }, + usage: + calls < 10 + ? { + inputTokens: { total: 1, noCache: 1, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 1, text: 1, reasoning: 0 }, + } + : unavailableUsage(), + }, + ], + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + maxSteps: 1, + recordUsageCheckpoint: async (usage) => { + usageCheckpoints.push(usage); + }, + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events: SessionEvent[] = []; + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + } + + assert.equal(calls, 10); + assert.equal(usageCheckpoints.length, 9); + assert.deepEqual( + usageCheckpoints.at(-1) + ? { + inputTokens: usageCheckpoints.at(-1)!.inputTokens, + outputTokens: usageCheckpoints.at(-1)!.outputTokens, + } + : undefined, + { inputTokens: 9, outputTokens: 9 }, + ); + assert.equal( + events.some((event) => event.type === 'token_usage'), + false, + ); + assert.equal(events.find((event) => event.type === 'error')?.reason, 'network'); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'error'); + }); + + test('does not retry a network_error finish after metadata-only response evidence', async () => { + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + return { + stream: simulateReadableStream({ + chunks: [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-1' }, + { + type: 'text-end', + id: 'text-1', + providerMetadata: { openai: { itemId: 'message-1' } }, + }, + { + type: 'finish', + finishReason: { unified: 'other', raw: 'network_error' }, + usage: unavailableUsage(), + }, + ], + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events: SessionEvent[] = []; + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + } + + assert.equal(calls, 1); + assert.equal( + events.some((event) => event.type === 'provider_retry'), + false, + ); + assert.equal(events.find((event) => event.type === 'error')?.reason, 'network'); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'error'); + }); +}); diff --git a/packages/runtime/src/__tests__/model-adapter-onerror.test.ts b/packages/runtime/src/__tests__/model-adapter-onerror.test.ts index 0b2dd3de7c..155720a66a 100644 --- a/packages/runtime/src/__tests__/model-adapter-onerror.test.ts +++ b/packages/runtime/src/__tests__/model-adapter-onerror.test.ts @@ -27,12 +27,23 @@ import { } from '@ai-sdk/provider'; import { ModelAdapter, settleModelStepOutcome } from '../model-adapter.js'; +import type { ModelStepOutcome, ModelStreamEvent, ModelStreamResult } from '../model-protocol.js'; const ZERO_USAGE: LanguageModelV4Usage = { inputTokens: { total: 0, noCache: 0, cacheRead: 0, cacheWrite: 0 }, outputTokens: { total: 0, text: 0, reasoning: 0 }, }; +const UNAVAILABLE_USAGE: LanguageModelV4Usage = { + inputTokens: { + total: undefined, + noCache: undefined, + cacheRead: undefined, + cacheWrite: undefined, + }, + outputTokens: { total: undefined, text: undefined, reasoning: undefined }, +}; + function newAdapter(): ModelAdapter { return new ModelAdapter({ connection: { providerType: 'openai' } as never, @@ -58,13 +69,30 @@ describe('settleModelStepOutcome', () => { failure, sawFinish: true, finishReason: 'content-filter', + finishDisposition: 'authoritative', request: {}, + hasResponseEvidence: false, }); assert.equal(outcome.kind, 'retryable-failure'); if (outcome.kind !== 'retryable-failure') return; assert.equal(outcome.failure, failure); }); + + test('preserves the missing terminal finish diagnostic after authoritative step evidence', () => { + const outcome = settleModelStepOutcome({ + aborted: false, + sawFinish: false, + finishReason: 'stop', + finishDisposition: 'authoritative', + request: {}, + hasResponseEvidence: true, + }); + + assert.equal(outcome.kind, 'truncated'); + if (outcome.kind !== 'truncated') return; + assert.equal(outcome.failure.message, 'Provider stream ended without finishing (stop)'); + }); }); describe('ModelAdapter.startStream onError', () => { @@ -110,6 +138,7 @@ describe('ModelAdapter.startStream onError', () => { failure: failures[0], request: { messages: [{ role: 'user', content: 'hi' }] }, continuation: 'none', + hasResponseEvidence: false, }); }); @@ -192,6 +221,7 @@ describe('ModelAdapter.startStream onError', () => { }, request: { messages: [{ role: 'user', content: 'hi' }] }, continuation: 'none', + hasResponseEvidence: false, }); }); @@ -208,6 +238,150 @@ describe('ModelAdapter.startStream onError', () => { assert.equal(outcome.continuation, 'none'); }); + test('settles an empty stop without usable usage as truncated', async () => { + const { events, outcome } = await observe([ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: UNAVAILABLE_USAGE, + }, + ]); + + assert.equal(outcome.kind, 'truncated'); + if (outcome.kind !== 'truncated') return; + assert.deepEqual(outcome.failure, { + type: 'model_failure', + kind: 'provider_unavailable', + message: 'Provider returned an empty stop without output or usable usage', + retryable: false, + }); + assert.equal(outcome.continuation, 'none'); + assert.equal(outcome.hasResponseEvidence, false); + assert.deepEqual(boundaryDispositions(events), [ + { kind: 'step-finish', disposition: 'incomplete' }, + { kind: 'finish', disposition: 'incomplete' }, + ]); + }); + + test('keeps an empty stop with authoritative zero usage completed', async () => { + const { events, outcome } = await observe([ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: ZERO_USAGE, + }, + ]); + + assert.equal(outcome.kind, 'completed'); + if (outcome.kind !== 'completed') return; + assert.equal(outcome.finishReason, 'stop'); + assert.equal(outcome.usage?.totalTokens, 0); + assert.equal(outcome.hasResponseEvidence, false); + assert.deepEqual(boundaryDispositions(events), [ + { kind: 'step-finish', disposition: 'authoritative' }, + { kind: 'finish', disposition: 'authoritative' }, + ]); + }); + + test('keeps nonempty text completed when provider usage is unavailable', async () => { + const { events, outcome } = await observe([ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-1' }, + { type: 'text-delta', id: 'text-1', delta: 'Done.' }, + { type: 'text-end', id: 'text-1' }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: UNAVAILABLE_USAGE, + }, + ]); + + assert.equal(outcome.kind, 'completed'); + if (outcome.kind !== 'completed') return; + assert.equal(outcome.finishReason, 'stop'); + assert.equal(outcome.usage, undefined); + assert.equal(outcome.hasResponseEvidence, true); + assert.deepEqual(boundaryDispositions(events), [ + { kind: 'step-finish', disposition: 'authoritative' }, + { kind: 'finish', disposition: 'authoritative' }, + ]); + }); + + test('uses terminal total usage when no step-finish usage fact exists', async () => { + const { events, outcome } = await observeAdapterChunks( + [ + { type: 'start-step' }, + { + type: 'finish', + finishReason: 'stop', + rawFinishReason: 'stop', + totalUsage: ZERO_USAGE, + }, + ], + ZERO_USAGE, + ); + + assert.deepEqual(boundaryDispositions(events), [ + { kind: 'finish', disposition: 'authoritative' }, + ]); + assert.equal(outcome.kind, 'completed'); + assert.equal(outcome.usage?.totalTokens, 0); + }); + + test('resets step evidence without losing request-level retry safety', async () => { + const { events, outcome } = await observeAdapterChunks([ + { type: 'start-step' }, + { type: 'text-delta', text: 'first step' }, + { type: 'finish-step', finishReason: 'stop', usage: ZERO_USAGE }, + { type: 'start-step' }, + { type: 'finish-step', finishReason: 'stop', usage: UNAVAILABLE_USAGE }, + { + type: 'finish', + finishReason: 'stop', + rawFinishReason: 'stop', + totalUsage: ZERO_USAGE, + }, + ]); + + assert.deepEqual(boundaryDispositions(events), [ + { kind: 'step-finish', disposition: 'authoritative' }, + { kind: 'step-finish', disposition: 'incomplete' }, + { kind: 'finish', disposition: 'incomplete' }, + ]); + assert.equal(outcome.kind, 'truncated'); + assert.equal(outcome.hasResponseEvidence, true); + }); + + test('settles an explicit provider network_error finish as retryable', async () => { + const { events, outcome } = await observe([ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'other', raw: 'network_error' }, + usage: UNAVAILABLE_USAGE, + }, + ]); + + assert.equal(outcome.kind, 'retryable-failure'); + if (outcome.kind !== 'retryable-failure') return; + assert.deepEqual(outcome.failure, { + type: 'model_failure', + kind: 'network', + code: 'network_error', + message: 'Network error', + retryable: true, + }); + assert.equal(outcome.continuation, 'none'); + assert.equal(outcome.hasResponseEvidence, false); + assert.equal(events.find((event) => event.kind === 'finish')?.finishReason, 'network-error'); + assert.deepEqual(boundaryDispositions(events), [ + { kind: 'step-finish', disposition: 'retryable-network-failure' }, + { kind: 'finish', disposition: 'retryable-network-failure' }, + ]); + }); + test('preserves a provider reason hidden by the SDK other bucket', async () => { const outcome = await settle([ { type: 'stream-start', warnings: [] }, @@ -240,6 +414,40 @@ describe('ModelAdapter.startStream onError', () => { assert.equal(outcome.failure.message, 'Provider stopped the stream on a content filter'); }); + test('normalizes a provider sensitive finish to the content-filter policy', async () => { + const { events, outcome } = await observe([ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'other', raw: 'sensitive' }, + usage: ZERO_USAGE, + }, + ]); + + assert.equal(events.find((event) => event.kind === 'finish')?.finishReason, 'content-filter'); + assert.equal(outcome.kind, 'terminal-failure'); + if (outcome.kind !== 'terminal-failure') return; + assert.equal(outcome.failure.kind, 'unknown'); + assert.equal(outcome.failure.message, 'Provider stopped the stream on a content filter'); + }); + + test('keeps a provider error finish on the existing terminal error policy', async () => { + const { events, outcome } = await observe([ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'other', raw: 'error' }, + usage: ZERO_USAGE, + }, + ]); + + assert.equal(events.find((event) => event.kind === 'finish')?.finishReason, 'error'); + assert.equal(outcome.kind, 'terminal-failure'); + if (outcome.kind !== 'terminal-failure') return; + assert.equal(outcome.failure.kind, 'provider_unavailable'); + assert.equal(outcome.failure.message, 'Provider stopped the stream with an error'); + }); + test('settles an aborted request as aborted', async () => { const controller = new AbortController(); controller.abort(); @@ -317,6 +525,13 @@ async function settle( chunks: LanguageModelV4StreamPart[], abortSignal = new AbortController().signal, ) { + return (await observe(chunks, abortSignal)).outcome; +} + +async function observe( + chunks: LanguageModelV4StreamPart[], + abortSignal = new AbortController().signal, +): Promise<{ events: ModelStreamEvent[]; outcome: ModelStepOutcome }> { const model = new MockLanguageModelV4({ doStream: async () => ({ stream: convertArrayToReadableStream(chunks) }), }); @@ -329,6 +544,47 @@ async function settle( abortSignal, repairToolCall: async () => null, }); - for await (const _event of result.events) void _event; - return await result.outcome; + const events: ModelStreamEvent[] = []; + for await (const event of result.events) events.push(event); + return { events, outcome: await result.outcome }; +} + +function boundaryDispositions(events: readonly ModelStreamEvent[]) { + return events + .filter((event) => event.kind === 'step-finish' || event.kind === 'finish') + .map(({ kind, disposition }) => ({ kind, disposition })); +} + +async function observeAdapterChunks( + chunks: readonly Record[], + sdkUsage?: LanguageModelV4Usage, +): Promise<{ events: ModelStreamEvent[]; outcome: ModelStepOutcome }> { + const adapter = newAdapter() as unknown as { + toModelStreamResult( + sdk: unknown, + onStreamActivity: () => void, + continuation: { + requestMessages: Array<{ role: 'user'; content: string }>; + abortSignal: AbortSignal; + }, + ): ModelStreamResult; + }; + const result = adapter.toModelStreamResult( + { + stream: (async function* () { + for (const chunk of chunks) yield chunk; + })(), + usage: Promise.resolve(sdkUsage), + finishReason: Promise.resolve('stop'), + response: Promise.resolve({ id: 'response-1' }), + }, + () => {}, + { + requestMessages: [{ role: 'user', content: 'hi' }], + abortSignal: new AbortController().signal, + }, + ); + const events: ModelStreamEvent[] = []; + for await (const event of result.events) events.push(event); + return { events, outcome: await result.outcome }; } diff --git a/packages/runtime/src/__tests__/model-adapter.test.ts b/packages/runtime/src/__tests__/model-adapter.test.ts index 99b08f8991..dd5856756d 100644 --- a/packages/runtime/src/__tests__/model-adapter.test.ts +++ b/packages/runtime/src/__tests__/model-adapter.test.ts @@ -184,7 +184,7 @@ describe('ModelAdapter stream and error normalization', () => { { type: 'unknown-provider-chunk' }, ]; - const events: ModelStreamEvent[] = chunks.flatMap((chunk) => adapter.translateChunk(chunk)); + const events = chunks.flatMap((chunk) => adapter.translateChunk(chunk)); // Tool results and unknown chunks are inert; returned tool calls and errors // cross the adapter as Maka-owned events. @@ -482,7 +482,7 @@ describe('ModelAdapter stream and error normalization', () => { { type: 'text-delta', text: 'two' }, { type: 'finish-step', finishReason: { unified: 'stop', raw: 'stop' } }, ]; - const events: ModelStreamEvent[] = chunks.flatMap((chunk) => adapter.translateChunk(chunk)); + const events = chunks.flatMap((chunk) => adapter.translateChunk(chunk)); assert.deepEqual( events.map((event) => event.kind), @@ -494,9 +494,7 @@ describe('ModelAdapter stream and error normalization', () => { .map((event) => (event as { text: string }).text), ['one', 'two'], ); - const stepFinishes = events.filter((event) => event.kind === 'step-finish') as Array< - Extract - >; + const stepFinishes = events.filter((event) => event.kind === 'step-finish'); assert.deepEqual( stepFinishes.map((event) => event.finishReason), ['tool_calls', 'stop'], @@ -522,7 +520,7 @@ describe('ModelAdapter stream and error normalization', () => { }, { type: 'reasoning-end' }, ]; - const events: ModelStreamEvent[] = chunks.flatMap((chunk) => adapter.translateChunk(chunk)); + const events = chunks.flatMap((chunk) => adapter.translateChunk(chunk)); assert.deepEqual( events.map((event) => event.kind), diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 6f1a31eb36..118dbe379c 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -756,7 +756,7 @@ export interface AiSdkBackendInput extends AiSdkCompactionCapabilities { recordToolInvocation?: ToolTelemetryRecorder; /** Optional Phase 2 SQLite T1/T2 boundary for real tool execution. */ runtimeCommitSink?: RuntimeCommitSink; - /** Durable session-lifetime cumulative usage checkpoint after each completed provider step. */ + /** Durable cumulative usage after completed steps and explicitly metered failed requests. */ recordUsageCheckpoint?: ( usage: NormalizedAiSdkUsage & { costUsd?: number }, ) => void | Promise; @@ -938,10 +938,6 @@ function providerRetryReason(kind: ModelFailureKind): ProviderRetryReason { } } -function isIncompleteProviderFinishReason(reason: ModelFinishReason | undefined): boolean { - return reason === undefined || reason === 'other' || reason === 'unknown'; -} - function sleepForProviderRetry(delayMs: number, signal: AbortSignal): Promise { if (signal.aborted) { return Promise.reject(Object.assign(new Error('aborted'), { name: 'AbortError' })); @@ -1582,15 +1578,26 @@ export class AiSdkBackend implements AgentBackend { }; let tokenUsage: NormalizedAiSdkUsage | undefined; let tokenUsageCostUsd: number | undefined; - // Per-send sum of every COMPLETED step's usage, merged at each finish-step - // boundary. When the send aborts (mid-turn exhaust, user stop, stream - // error) the SDK's cumulative `usage` promise may not resolve, but this sum is - // real provider-reported evidence for the steps that did finish — IF every - // completed step produced a usable sample. One unusable sample makes the - // sum a partial cost, and LlmCallRecord has no partial marker, so the flag - // fails the whole fallback closed (#972: incomplete usage is no usage). + // Per-send sum of every completed step's usage plus any explicitly + // accounted failed request whose usage is still authoritative. When the + // send aborts (mid-turn exhaust, user stop, stream error), the SDK's cumulative `usage` + // promise may not resolve, but this sum remains real provider-reported + // evidence. One unusable sample makes the sum a partial cost, and + // LlmCallRecord has no partial marker, so the flag fails the whole fallback + // closed (#972: incomplete usage is no usage). let completedStepUsage: NormalizedAiSdkUsage | undefined; let sawUnusableStepUsage = false; + const recordAccountedStepUsage = async (stepUsage: NormalizedAiSdkUsage): Promise => { + completedStepUsage = mergeNormalizedUsage(completedStepUsage, stepUsage); + this.cumulativeUsageCheckpoint = mergeNormalizedUsage( + this.cumulativeUsageCheckpoint, + stepUsage, + ); + await this.input.recordUsageCheckpoint?.({ + ...this.cumulativeUsageCheckpoint, + costUsd: this.computeTokenUsageCostUsd(this.cumulativeUsageCheckpoint), + }); + }; // Input tokens from the last completed step — the actual prompt token count // of the final API request. Used to compute contextRemaining for the TUI // statusline ctx segment (#1067): contextRemaining = contextWindow - this. @@ -2239,16 +2246,10 @@ export class AiSdkBackend implements AgentBackend { // are cleared after flushStep(), so they cannot decide whether a // later stream failure is safe to retry. let attemptSawText = false; - let attemptSawThinking = false; let attemptSawToolActivity = false; let attemptSawContinuationMetadata = false; let attemptReachedStepBoundary = false; - const attemptHasNoObservableOutput = () => - !attemptSawText && - !attemptSawThinking && - !attemptSawToolActivity && - !attemptSawContinuationMetadata && - !attemptReachedStepBoundary; + let attemptRecordedSettledUsage = false; const attemptCanRecoverFromIdleTimeout = () => !attemptSawText && !attemptSawToolActivity && @@ -2318,10 +2319,12 @@ export class AiSdkBackend implements AgentBackend { // trailer and consume the one authoritative outcome below. break; } - const incompleteFinish = - (event.kind === 'finish' || event.kind === 'step-finish') && - isIncompleteProviderFinishReason(event.finishReason); - if ((event.kind === 'finish' || event.kind === 'step-finish') && !incompleteFinish) { + const finishDisposition = + event.kind === 'finish' || event.kind === 'step-finish' + ? event.disposition + : undefined; + const authoritativeFinish = finishDisposition === 'authoritative'; + if (authoritativeFinish) { attemptReachedStepBoundary = true; } if (event.kind === 'step-finish') { @@ -2329,7 +2332,7 @@ export class AiSdkBackend implements AgentBackend { // stream reaches EOF without a terminal frame. That is not a // completed model step and must not consume the step budget or // checkpoint imaginary usage before the safe retry below. - if (!incompleteFinish) { + if (authoritativeFinish) { // Step boundary: AI SDK 7 delimits steps with `finish-step` // (and `step-finish` for legacy replay fixtures); the adapter // reduces both to this event. A duplicate boundary is harmless: @@ -2343,16 +2346,20 @@ export class AiSdkBackend implements AgentBackend { // step's usage does not leave a stale value from an earlier step. lastStepInputTokens = stepUsage?.inputTokens; if (stepUsage) { - completedStepUsage = mergeNormalizedUsage(completedStepUsage, stepUsage); - this.cumulativeUsageCheckpoint = mergeNormalizedUsage( - this.cumulativeUsageCheckpoint, - stepUsage, - ); - await this.input.recordUsageCheckpoint?.({ - ...this.cumulativeUsageCheckpoint, - costUsd: this.computeTokenUsageCostUsd(this.cumulativeUsageCheckpoint), - }); + await recordAccountedStepUsage(stepUsage); + attemptRecordedSettledUsage = true; } + } else if ( + finishDisposition === 'retryable-network-failure' && + event.usage && + !attemptRecordedSettledUsage + ) { + // A provider may report exact usage while declaring the + // request failed. Preserve that billed request without + // promoting it to a logical step or flushing an Assistant + // message; the retry owns the eventual step completion. + await recordAccountedStepUsage(event.usage); + attemptRecordedSettledUsage = true; } } if (event.kind === 'text-start') { @@ -2380,7 +2387,6 @@ export class AiSdkBackend implements AgentBackend { } else if (event.kind === 'thinking') { sawStepThinking = true; stepThinking += event.text; - if (event.text.length > 0) attemptSawThinking = true; if (event.providerOptions !== undefined) { if (event.providerOptionsOrigin !== 'maka_transport') { attemptSawContinuationMetadata = true; @@ -2476,7 +2482,7 @@ export class AiSdkBackend implements AgentBackend { ), } satisfies ToolResultEvent); providerToolInputs.delete(event.toolCallId); - } else if (event.kind === 'step-finish' && !incompleteFinish) { + } else if (event.kind === 'step-finish' && authoritativeFinish) { // The step's text/thinking deltas are all in (the stream is // drained in order), so flush this step's AssistantMessage and // rotate to a fresh id for the next step. Tool settlement @@ -2500,13 +2506,33 @@ export class AiSdkBackend implements AgentBackend { // must not be reported as the already-handled watchdog timeout. const settledWatchdogTimeout = consumeWatchdogTimeout(); providerOutcome = await result.outcome; + const settleFailureUsageBeforePolicy = + providerOutcome.kind === 'truncated' || + (providerOutcome.kind === 'retryable-failure' && + providerOutcome.failure.kind === 'network' && + providerOutcome.failure.code === 'network_error'); + if (settleFailureUsageBeforePolicy) { + // These provider terminal frames are billable even though they + // cannot complete a logical model step. The stream normally + // exposes their exact usage at `step-finish`; retain the outcome + // fallback for providers that settle it only through the SDK's + // terminal promise. If neither seam has usable usage, fail the + // send-level aggregate closed before stop, exhaustion, or retry + // policy can bypass the retry branch below. Context-overflow + // recovery intentionally keeps its established completed-step + // diagnostic semantics: the rejected request has no finish + // frame and is not one of these terminal-frame classifications. + if (!attemptRecordedSettledUsage && providerOutcome.usage) { + await recordAccountedStepUsage(providerOutcome.usage); + attemptRecordedSettledUsage = true; + } + if (!attemptRecordedSettledUsage) sawUnusableStepUsage = true; + } const incompleteStreamTerminal = providerOutcome.kind === 'truncated'; const incompleteStreamHasNoObservableOutput = - incompleteStreamTerminal && - !attemptSawText && - !attemptSawThinking && - !attemptSawToolActivity && - !attemptSawContinuationMetadata; + incompleteStreamTerminal && !providerOutcome.hasResponseEvidence; + const attemptHasNoObservableOutput = + !providerOutcome.hasResponseEvidence && !attemptReachedStepBoundary; const attemptFailure = settledWatchdogTimeout?.error ?? (providerOutcome.kind === 'completed' ? undefined : providerOutcome.failure); @@ -2525,7 +2551,7 @@ export class AiSdkBackend implements AgentBackend { // nothing left to grant it, so the error is terminal. const stepBudgetRemains = maxSteps === undefined || runtimeSteps < maxSteps; const recovered = - stepBudgetRemains && attemptHasNoObservableOutput() + stepBudgetRemains && attemptHasNoObservableOutput ? await this.compaction.recoverFromOverflowError({ error: attemptFailure, retryAlreadyUsed: overflowRetryUsed, @@ -2582,7 +2608,7 @@ export class AiSdkBackend implements AgentBackend { failure.kind !== 'context_overflow' && providerAttempt < MAX_PROVIDER_ATTEMPTS_PER_STEP && stepBudgetRemains && - (attemptHasNoObservableOutput() || idleWatchdogRecovery || incompleteStreamRecovery) + (attemptHasNoObservableOutput || idleWatchdogRecovery || incompleteStreamRecovery) ) { if (idleWatchdogRecovery) { idleWatchdogRetryCount += 1; @@ -2592,9 +2618,11 @@ export class AiSdkBackend implements AgentBackend { } } if (incompleteStreamRecovery) incompleteStreamRetryCount += 1; - // The failed request did not return authoritative usage. Keep - // effectiveness recoverable, but fail final metering closed. - sawUnusableStepUsage = true; + // Other retryable failures reach this branch only when a new + // physical request will actually be dispatched. Preserve the + // established fail-closed rule without changing overflow + // recovery, which continues above after rebuilding context. + if (!attemptRecordedSettledUsage) sawUnusableStepUsage = true; const delayMs = providerRetryDelayMs(providerAttempt, failure.retryAfterMs); const nextAttempt = providerAttempt + 1; const maxAttempts = diff --git a/packages/runtime/src/model-adapter.ts b/packages/runtime/src/model-adapter.ts index 9c99c2527a..52c2d4c30a 100644 --- a/packages/runtime/src/model-adapter.ts +++ b/packages/runtime/src/model-adapter.ts @@ -35,6 +35,7 @@ import type { ModelStreamEvent, ModelStreamResult, ModelStepOutcome, + ModelFinishDisposition, ModelFinishReason, ModelFailure, ModelFailureKind, @@ -48,6 +49,7 @@ export type { ModelStreamEvent, ModelStreamResult, ModelStepOutcome, + ModelFinishDisposition, ModelFinishReason, ModelFailure, ModelFailureKind, @@ -55,6 +57,12 @@ export type { ModelToolSet, } from './model-protocol.js'; +type ModelFinishBoundaryEvent = Extract; +type UnclassifiedModelStreamEvent = + | Exclude + | Omit, 'disposition'> + | Omit, 'disposition'>; + import { resolveModelRuntime, type ResolvedModelRuntime } from './model-runtime.js'; import { classifyError, @@ -336,16 +344,48 @@ export class ModelAdapter { async *[Symbol.asyncIterator]() { let failure: ModelFailure | undefined; let sawFinish = false; + let requestHasResponseEvidence = false; + let stepHasResponseEvidence = false; + let latestStepFinishHadUsableUsage: boolean | undefined; let streamedFinishReason: string | undefined; + let streamedFinishDisposition: ModelFinishDisposition | undefined; try { for await (const chunk of sdk.stream as AsyncIterable) { onStreamActivity(); - for (const event of translateChunk(chunk, openAiChatReasoningTransportState)) { - if (event.kind === 'error') failure = event.failure; + // AI SDK emits start-step before each provider step. Keep the + // request-level fact monotonic for retry safety, but do not let a + // previous step's evidence or usage authorize a later empty stop. + if (chunk.type === 'start-step') { + stepHasResponseEvidence = false; + latestStepFinishHadUsableUsage = undefined; + } + for (const translated of translateChunk(chunk, openAiChatReasoningTransportState)) { + if (isModelResponseEvidence(translated)) { + requestHasResponseEvidence = true; + stepHasResponseEvidence = true; + } + if (translated.kind === 'error') failure = translated.failure; + if (translated.kind !== 'finish' && translated.kind !== 'step-finish') { + yield translated; + continue; + } + + const hasUsableStepUsage = + translated.kind === 'step-finish' + ? translated.usage !== undefined + : (latestStepFinishHadUsableUsage ?? translated.usage !== undefined); + const disposition = classifyModelFinishBoundary({ + finishReason: translated.finishReason, + hasResponseEvidence: stepHasResponseEvidence, + hasUsableStepUsage, + }); + const event: ModelFinishBoundaryEvent = { ...translated, disposition }; if (event.kind === 'finish') sawFinish = true; - if (event.kind === 'finish' || event.kind === 'step-finish') { - streamedFinishReason = event.finishReason ?? streamedFinishReason; + if (event.kind === 'step-finish') { + latestStepFinishHadUsableUsage = event.usage !== undefined; } + streamedFinishReason = event.finishReason ?? streamedFinishReason; + streamedFinishDisposition = disposition; yield event; } } @@ -365,8 +405,10 @@ export class ModelAdapter { failure, sawFinish, finishReason, + finishDisposition: streamedFinishDisposition ?? 'incomplete', usage, request, + hasResponseEvidence: requestHasResponseEvidence, }); try { @@ -416,12 +458,11 @@ export class ModelAdapter { /** * Translate one raw AI SDK stream chunk into zero or more Maka-owned - * `ModelStreamEvent`s. This is the sole place that parses SDK chunk names - * (`text-delta` / `reasoning-delta` / `finish-step` / `finish` / `error` / …); - * the backend never sees them. Pure and side-effect-free so it is directly - * testable through the Maka-owned event contract. + * event candidates. Finish-boundary disposition depends on stream-level + * evidence and is stamped by `toModelStreamResult`; this pure hook only + * exposes raw-chunk lowering for focused adapter tests. */ - translateChunk(chunk: AiSdkStreamChunk): ModelStreamEvent[] { + translateChunk(chunk: AiSdkStreamChunk): UnclassifiedModelStreamEvent[] { return translateChunk( chunk, this.runtime.reasoningReplay.kind === 'openai-chat-plaintext' @@ -473,18 +514,69 @@ interface ModelStepSettlementEvidence { failure?: ModelFailure; sawFinish: boolean; finishReason: ModelFinishReason; + finishDisposition: ModelFinishDisposition; usage?: NormalizedUsage; request: ModelRequestMetadata; + hasResponseEvidence: boolean; +} + +function isModelResponseEvidence(event: UnclassifiedModelStreamEvent): boolean { + switch (event.kind) { + case 'text': + return event.text.length > 0; + case 'text-metadata': + case 'thinking-signature': + case 'provider-tool-input': + case 'tool-call': + case 'provider-tool-result': + return true; + case 'thinking': + return ( + event.text.length > 0 || + (event.providerOptions !== undefined && event.providerOptionsOrigin !== 'maka_transport') + ); + default: + return false; + } +} + +function classifyModelFinishBoundary(input: { + finishReason: ModelFinishReason | undefined; + hasResponseEvidence: boolean; + hasUsableStepUsage: boolean; +}): 'authoritative' | 'incomplete' | 'retryable-network-failure' { + if (input.finishReason === NETWORK_ERROR_FINISH_REASON) return 'retryable-network-failure'; + if ( + input.finishReason === undefined || + input.finishReason === 'other' || + input.finishReason === 'unknown' + ) { + return 'incomplete'; + } + if (input.finishReason === 'stop' && !input.hasResponseEvidence && !input.hasUsableStepUsage) { + return 'incomplete'; + } + return 'authoritative'; } export function settleModelStepOutcome(evidence: ModelStepSettlementEvidence): ModelStepOutcome { - const { aborted, failure, sawFinish, finishReason, usage, request } = evidence; + const { + aborted, + failure, + sawFinish, + finishReason, + finishDisposition, + usage, + request, + hasResponseEvidence, + } = evidence; if (aborted || failure?.kind === 'abort') { return failedStepOutcome( 'aborted', failure ?? normalizeModelFailure(Object.assign(new Error('aborted'), { name: 'AbortError' })), request, usage, + hasResponseEvidence, ); } if (failure) { @@ -493,17 +585,36 @@ export function settleModelStepOutcome(evidence: ModelStepSettlementEvidence): M failure, request, usage, + hasResponseEvidence, + ); + } + if (finishDisposition === 'retryable-network-failure') { + return failedStepOutcome( + 'retryable-failure', + { + type: 'model_failure', + kind: 'network', + code: 'network_error', + message: 'Network error', + retryable: true, + }, + request, + usage, + hasResponseEvidence, ); } - if (!sawFinish || finishReason === 'other' || finishReason === 'unknown') { + if (!sawFinish || finishDisposition === 'incomplete') { return failedStepOutcome( 'truncated', modelStepFailure( 'provider_unavailable', - `Provider stream ended without finishing (${finishReason})`, + sawFinish && finishReason === 'stop' + ? 'Provider returned an empty stop without output or usable usage' + : `Provider stream ended without finishing (${finishReason})`, ), request, usage, + hasResponseEvidence, ); } if (finishReason === 'content-filter' || finishReason === 'error') { @@ -517,6 +628,7 @@ export function settleModelStepOutcome(evidence: ModelStepSettlementEvidence): M ), request, usage, + hasResponseEvidence, ); } return { @@ -525,6 +637,7 @@ export function settleModelStepOutcome(evidence: ModelStepSettlementEvidence): M ...(usage ? { usage } : {}), request, continuation: 'none', + hasResponseEvidence, }; } @@ -536,7 +649,8 @@ function failedStepOutcome( kind: Exclude, failure: ModelFailure, request: ModelRequestMetadata, - usage?: NormalizedUsage, + usage: NormalizedUsage | undefined, + hasResponseEvidence: boolean, ): Exclude { return { kind, @@ -544,6 +658,7 @@ function failedStepOutcome( ...(usage ? { usage } : {}), request, continuation: 'none', + hasResponseEvidence, }; } @@ -613,6 +728,7 @@ interface AiSdkStreamChunk { output?: unknown; isError?: boolean; usage?: AiSdkUsageLike; + totalUsage?: AiSdkUsageLike; finishReason?: unknown; /** What the provider itself called it, before the SDK bucketed it. */ rawFinishReason?: unknown; @@ -642,15 +758,24 @@ interface SdkStreamResult { * own spelling. Unified is the right thing to forward — `RuntimeKernel` and * the backend compare against `'tool-calls'`, which is a name only the SDK * uses. Except when unified is `other`, which is not a reason but the SDK - * declining to name one; there it hides the only distinction that matters - * downstream. `other` with a provider spelling is a model that stopped for a - * reason we have no case for — an ordinary finished turn. `other` with nothing - * behind it is a stream that died without anyone saying so. + * declining to name one. Known provider spellings with established semantics + * are mapped to Maka's canonical vocabulary here; unknown raw spellings still + * pass through for tolerant forward compatibility. `other` with nothing + * behind it remains an unnamed, incomplete stream. */ +const NETWORK_ERROR_FINISH_REASON = 'network-error'; +const RAW_FINISH_REASON_ALIASES = new Map([ + ['network_error', NETWORK_ERROR_FINISH_REASON], + ['sensitive', 'content-filter'], +]); + function chunkFinishReason(chunk: AiSdkStreamChunk): string | undefined { const unified = rawFinishReasonString(chunk.finishReason); - if (unified !== 'other' && unified !== 'unknown') return unified; - return rawFinishReasonString(chunk.rawFinishReason) ?? unified; + if (unified !== 'other' && unified !== 'unknown') { + return unified === undefined ? undefined : (RAW_FINISH_REASON_ALIASES.get(unified) ?? unified); + } + const raw = rawFinishReasonString(chunk.rawFinishReason); + return raw === undefined ? unified : (RAW_FINISH_REASON_ALIASES.get(raw) ?? raw); } /** @@ -697,7 +822,7 @@ function openAiResponsesReasoningProviderOptionsFromChunk( function translateChunk( chunk: AiSdkStreamChunk, openAiChatReasoningTransportState?: OpenAiChatReasoningTransportState, -): ModelStreamEvent[] { +): UnclassifiedModelStreamEvent[] { switch (chunk.type) { case 'text-start': return [{ kind: 'text-start' }]; @@ -784,7 +909,16 @@ function translateChunk( } case 'finish': { const finishReason = chunkFinishReason(chunk); - return [{ kind: 'finish', ...(finishReason ? { finishReason } : {}) }]; + const usage = normalizeAiSdkUsage(chunk.totalUsage ?? chunk.usage, { + rawFinishReason: finishReason, + }); + return [ + { + kind: 'finish', + ...(usage ? { usage } : {}), + ...(finishReason ? { finishReason } : {}), + }, + ]; } case 'reasoning-start': case 'start-step': diff --git a/packages/runtime/src/model-protocol.ts b/packages/runtime/src/model-protocol.ts index fe1a1db769..b6090a31ca 100644 --- a/packages/runtime/src/model-protocol.ts +++ b/packages/runtime/src/model-protocol.ts @@ -319,6 +319,9 @@ export function rawFinishReasonString(reason: unknown): string | undefined { */ export type ModelFinishReason = string; +/** Adapter-owned classification of a provider finish boundary. */ +export type ModelFinishDisposition = 'authoritative' | 'incomplete' | 'retryable-network-failure'; + // --------------------------------------------------------------------------- // Failure contract // --------------------------------------------------------------------------- @@ -377,10 +380,11 @@ export interface ModelRequestMetadata { * delivered out-of-band from the thinking text. * - `step-finish`: a provider step boundary. Carries the step's normalized * usage (already reduced to `NormalizedUsage`) and normalized finish - * reason. The backend owns step counting, the per-step `AssistantMessage` - * flush, and the messageId rotation. - * - `finish`: the terminal stream boundary, carrying the normalized finish - * reason. + * reason plus the adapter's authoritative disposition. The backend owns + * step counting, the per-step `AssistantMessage` flush, and messageId + * rotation, but does not reclassify the boundary. + * - `finish`: the terminal stream boundary, carrying normalized total usage, + * finish reason, and the same adapter-owned disposition. * - `error`: a request-level provider failure, already classified and scrubbed * by the adapter. The backend uses its stable kind for overflow/transport * recovery and terminal error emission. @@ -407,10 +411,25 @@ export type ModelStreamEvent = output: unknown; isError?: boolean; } - | { kind: 'step-finish'; usage?: NormalizedUsage; finishReason?: ModelFinishReason } - | { kind: 'finish'; finishReason?: ModelFinishReason } + | { + kind: 'step-finish'; + usage?: NormalizedUsage; + finishReason?: ModelFinishReason; + disposition: ModelFinishDisposition; + } + | { + kind: 'finish'; + usage?: NormalizedUsage; + finishReason?: ModelFinishReason; + disposition: ModelFinishDisposition; + } | { kind: 'error'; failure: ModelFailure }; +/** + * Authoritative settlement for one physical provider request. The monotonic + * response-evidence fact lets Runtime retry policy avoid replaying observable + * model or provider activity without parsing the stream a second time. + */ export type ModelStepOutcome = | { kind: 'completed'; @@ -418,6 +437,7 @@ export type ModelStepOutcome = usage?: NormalizedUsage; request: ModelRequestMetadata; continuation: 'none' | 'pending'; + hasResponseEvidence: boolean; } | { kind: 'truncated' | 'retryable-failure' | 'terminal-failure' | 'aborted'; @@ -425,6 +445,7 @@ export type ModelStepOutcome = usage?: NormalizedUsage; request: ModelRequestMetadata; continuation: 'none'; + hasResponseEvidence: boolean; }; /** One physical provider request: live output plus one authoritative settlement. */