diff --git a/apps/desktop/src/main/__tests__/artifact-visibility.test.ts b/apps/desktop/src/main/__tests__/artifact-visibility.test.ts index 042a9af61e..2563a36f2d 100644 --- a/apps/desktop/src/main/__tests__/artifact-visibility.test.ts +++ b/apps/desktop/src/main/__tests__/artifact-visibility.test.ts @@ -25,6 +25,7 @@ describe('generated artifact visibility', () => { 'synthesis_cache_block', 'history_compact_block', 'history_compact_source', + 'provider_request_capture', 'user_upload', ]; diff --git a/apps/desktop/src/main/session-stream.ts b/apps/desktop/src/main/session-stream.ts index e64c6ac0d9..c9a72eadbf 100644 --- a/apps/desktop/src/main/session-stream.ts +++ b/apps/desktop/src/main/session-stream.ts @@ -11,6 +11,7 @@ import { buildLlmHistorySummarizer, buildMcpTools, buildProviderOptions, + createProviderRequestCaptureRecorder, getAIModel, loadHistoryCompactBlocksFromArtifacts, loadSynthesisCacheBlocksFromArtifacts, @@ -38,6 +39,7 @@ import { createAttachmentByteReader, createTelemetryRepo, openRuntimeEventPersistence, + persistProviderRequestCaptureArtifact, } from '@maka/storage'; import { WEB_SEARCH_TOOL_NAME } from './web-search/agent-tool.js'; import { @@ -252,6 +254,25 @@ export function createAiSdkBackendFactory(deps: AiSdkBackendFactoryDeps): Backen }, }), recordRunTrace: ctx.recordRunTrace, + ...(ctx.recordProviderRequestCapture + ? { + recordProviderRequestCapture: createProviderRequestCaptureRecorder({ + persistArtifact: async (capture) => { + const artifact = await persistProviderRequestCaptureArtifact(artifactStore, { + sessionId: ctx.sessionId, + turnId: capture.turnId, + captureId: capture.captureId, + step: capture.step, + serializedRequest: capture.serializedRequest, + now: Date.now(), + }); + return { artifactId: artifact.id }; + }, + recordLedger: ctx.recordProviderRequestCapture, + }), + recordProviderRequestAttempt: ctx.recordProviderRequestAttempt, + } + : {}), recordHistoryCompactCheckpoint: ctx.recordHistoryCompactCheckpoint, loadTurnRuntimeEvents: ctx.loadTurnRuntimeEvents, recordActiveFullCompactBlock: ctx.recordActiveFullCompactBlock, diff --git a/apps/desktop/src/renderer/artifact-visibility.ts b/apps/desktop/src/renderer/artifact-visibility.ts index dbcc96afef..4b233a1ab3 100644 --- a/apps/desktop/src/renderer/artifact-visibility.ts +++ b/apps/desktop/src/renderer/artifact-visibility.ts @@ -6,6 +6,7 @@ const USER_VISIBLE_ARTIFACT_SOURCES = { synthesis_cache_block: false, history_compact_block: false, history_compact_source: false, + provider_request_capture: false, user_upload: false, export: true, snapshot: true, diff --git a/packages/cli/src/runtime-bootstrap.ts b/packages/cli/src/runtime-bootstrap.ts index ef0ccdecc3..9552d88228 100644 --- a/packages/cli/src/runtime-bootstrap.ts +++ b/packages/cli/src/runtime-bootstrap.ts @@ -19,6 +19,7 @@ import { buildRuntimeEventModelReplayPlan, buildChildAgentTools, createBuiltinSandboxManager, + createProviderRequestCaptureRecorder, createFilesystemWorkerLaunchSpecProvider, createLocalContinuationSafetyInspector, FilesystemWorkerClient, @@ -63,6 +64,7 @@ import { createSettingsStore, createShellRunStore, type ForeignSessionStore, + persistProviderRequestCaptureArtifact, } from '@maka/storage'; import type { ToolPermissionRule } from '@maka/core/permission'; import { fetchProviderModels } from '@maka/runtime'; @@ -631,6 +633,25 @@ export async function createMakaCliRuntimeContext( turnTailPrompt: ({ cwd }) => buildCliTurnTailPrompt({ cwd, sessionId: ctx.sessionId, automationManager, goalManager }), shellRunContextSummary: ctx.shellRunContextSummary, + ...(ctx.recordProviderRequestCapture + ? { + recordProviderRequestCapture: createProviderRequestCaptureRecorder({ + persistArtifact: async (capture) => { + const artifact = await persistProviderRequestCaptureArtifact(artifactStore, { + sessionId: ctx.sessionId, + turnId: capture.turnId, + captureId: capture.captureId, + step: capture.step, + serializedRequest: capture.serializedRequest, + now: Date.now(), + }); + return { artifactId: artifact.id }; + }, + recordLedger: ctx.recordProviderRequestCapture, + }), + recordProviderRequestAttempt: ctx.recordProviderRequestAttempt, + } + : {}), newId: randomUUID, now: Date.now, ...(input.maxSteps !== undefined ? { maxSteps: input.maxSteps } : {}), diff --git a/packages/core/src/__tests__/agent-run-provider-request.test.ts b/packages/core/src/__tests__/agent-run-provider-request.test.ts new file mode 100644 index 0000000000..bfab43f0b0 --- /dev/null +++ b/packages/core/src/__tests__/agent-run-provider-request.test.ts @@ -0,0 +1,19 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; + +import { decodeAgentRunEvent } from '../agent-run.js'; + +test('AgentRun accepts provider request capture and attempt trace rows', () => { + for (const type of ['provider_request_captured', 'provider_request_attempt_recorded']) { + const decoded = decodeAgentRunEvent({ + type, + id: `${type}-1`, + runId: 'run-1', + sessionId: 'session-1', + turnId: 'turn-1', + ts: 1, + data: { traceId: 'provider-trace-1' }, + }); + assert.equal(decoded.type, type); + } +}); diff --git a/packages/core/src/__tests__/runtime-event.test.ts b/packages/core/src/__tests__/runtime-event.test.ts index d09fc95184..73395c2c76 100644 --- a/packages/core/src/__tests__/runtime-event.test.ts +++ b/packages/core/src/__tests__/runtime-event.test.ts @@ -1,4 +1,5 @@ import { describe, test } from 'node:test'; +import assert from 'node:assert/strict'; import { expect } from '../test-helpers.js'; import { RUNTIME_EVENT_AUTHORS, @@ -7,6 +8,7 @@ import { RUNTIME_EVENT_STATUSES, TERMINAL_RUNTIME_EVENT_STATUSES, createRuntimeEventId, + decodeRuntimeEvent, isRuntimeEventAuthor, isRuntimeEventRole, isRuntimeEventStatus, @@ -312,6 +314,17 @@ describe('createRuntimeEventId', () => { }); describe('RuntimeEvent shape compile-time contract', () => { + test('accepts a provider-request trace reference and rejects a non-string reference', () => { + const event = baseEvent({ refs: { providerRequestTraceId: 'provider-trace-1' } }); + expect(decodeRuntimeEvent(event).refs?.providerRequestTraceId).toBe('provider-trace-1'); + assert.throws(() => + decodeRuntimeEvent({ + ...event, + refs: { providerRequestTraceId: 123 }, + }), + ); + }); + test('a full user event satisfies the type', () => { const event: RuntimeEvent = { id: 'evt-u1', diff --git a/packages/core/src/agent-run.ts b/packages/core/src/agent-run.ts index dd159cd2a5..1dd892dbeb 100644 --- a/packages/core/src/agent-run.ts +++ b/packages/core/src/agent-run.ts @@ -98,6 +98,8 @@ export const AGENT_RUN_EVENT_TYPES = [ 'sandbox_escalation_failed', 'sandbox_denial_detected', 'usage_recorded', + 'provider_request_captured', + 'provider_request_attempt_recorded', 'history_compact_checkpoint_recorded', 'active_full_compact_block_recorded', 'semantic_compact_block_recorded', diff --git a/packages/core/src/artifacts.ts b/packages/core/src/artifacts.ts index ca34331240..e4625ffecd 100644 --- a/packages/core/src/artifacts.ts +++ b/packages/core/src/artifacts.ts @@ -6,6 +6,7 @@ export type ArtifactSource = | 'synthesis_cache_block' | 'history_compact_block' | 'history_compact_source' + | 'provider_request_capture' | 'user_upload' | 'export' | 'snapshot' diff --git a/packages/core/src/events.ts b/packages/core/src/events.ts index be7b2314c7..f576a94c42 100644 --- a/packages/core/src/events.ts +++ b/packages/core/src/events.ts @@ -518,6 +518,8 @@ export interface TokenUsageEvent extends BaseEvent { requestShapeChangeReason?: PrefixChangeReason; promptSegments?: PromptSegmentEstimate[]; contextBudget?: ContextBudgetDiagnostic; + /** Links this aggregate to per-physical-request AgentRun trace rows. */ + providerRequestTraceId?: string; } /** diff --git a/packages/core/src/runtime-event.ts b/packages/core/src/runtime-event.ts index 5f8b0d9b89..73f6912ae8 100644 --- a/packages/core/src/runtime-event.ts +++ b/packages/core/src/runtime-event.ts @@ -298,6 +298,8 @@ export interface RuntimeEventRefs { traceEventId?: string; toolCallId?: string; providerEventId?: string; + /** Trace-group id linking aggregate usage to physical provider attempts. */ + providerRequestTraceId?: string; artifactId?: string; /** Runtime-owned durable identity for one tool side-effect boundary. */ operationId?: string; @@ -453,6 +455,7 @@ const RUNTIME_REFS_SHAPE = defineObjectShape()( 'traceEventId', 'toolCallId', 'providerEventId', + 'providerRequestTraceId', 'artifactId', 'operationId', 'stepId', @@ -594,6 +597,7 @@ function isRuntimeEventRefs(value: unknown): value is RuntimeEventRefs { value.traceEventId, value.toolCallId, value.providerEventId, + value.providerRequestTraceId, value.artifactId, value.operationId, value.stepId, diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 0dccdfe9d2..47a9dd68fe 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -329,6 +329,7 @@ export interface TokenUsageMessage { requestShapeChangeReason?: PrefixChangeReason; promptSegments?: PromptSegmentEstimate[]; contextBudget?: ContextBudgetDiagnostic; + providerRequestTraceId?: string; } export interface TurnStateMessage { @@ -424,6 +425,7 @@ const TOKEN_USAGE_MESSAGE_SHAPE = defineObjectShape()( 'requestShapeChangeReason', 'promptSegments', 'contextBudget', + 'providerRequestTraceId', ], ); const TURN_STATE_MESSAGE_SHAPE = defineObjectShape()( @@ -540,7 +542,8 @@ function decodeStoredMessage( if ( hasExactShape(message, TOKEN_USAGE_MESSAGE_SHAPE) && hasMessageEnvelope(message, true) && - isTokenUsageFields(message) + isTokenUsageFields(message) && + isOptionalString(message.providerRequestTraceId) ) return message as unknown as TokenUsageMessage; break; diff --git a/packages/headless/src/__tests__/harbor-cell.test.ts b/packages/headless/src/__tests__/harbor-cell.test.ts index ee2bcedd06..e307ba6d29 100644 --- a/packages/headless/src/__tests__/harbor-cell.test.ts +++ b/packages/headless/src/__tests__/harbor-cell.test.ts @@ -15,9 +15,12 @@ import { PermissionEngine, PiAgentBackend, type AgentBackend, + type AiSdkBackendInput, type BackendFactoryContext, type PiAgentTransport, + type ProviderRequestCaptureRecord, type SessionStore, + type SynthesisCacheWriteInput, type ToolResultArchiveReader, type ToolResultArchiveRecorder, } from '@maka/runtime'; @@ -1961,6 +1964,7 @@ describe('runHarborCell', () => { systemPrompt: DEFAULT_HEADLESS_SYSTEM_PROMPT, }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, @@ -1992,6 +1996,133 @@ describe('runHarborCell', () => { }); }); + test('Harbor persists provider captures and synthesis blocks under the run storage root', async () => { + await withDirs(async ({ workspaceDir, outputDir, storageRoot }) => { + const registry = new BackendRegistry(); + const toolExecutor = fakeToolExecutor(); + const register = buildAiSdkCellBackendRegistration({ + provider: 'openai', + model: 'gpt-4o-mini', + env: { + OPENAI_API_KEY: 'test-key', + MAKA_STORAGE_ROOT: outputDir, + MAKA_CONTEXT_SYNTHESIS_CACHE: 'on', + MAKA_CONTEXT_SYNTHESIS_CACHE_MODE: 'read_write', + }, + now: () => 123, + newId: testIdFactory(), + }); + const context: HeadlessBackendContext = { + config: { + id: 'harbor-ai-sdk', + backend: 'ai-sdk', + llmConnectionSlug: 'openai', + model: 'gpt-4o-mini', + }, + task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + workspaceDir, + storageRoot, + realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, + toolExecutor, + }; + await register(registry, context); + + const backend = await registry.build('ai-sdk', { + ...backendContext(workspaceDir), + recordProviderRequestCapture: async () => {}, + }); + const backendInput = ( + backend as unknown as { + input: Pick< + AiSdkBackendInput, + 'loadSynthesisCache' | 'writeSynthesisCache' | 'recordProviderRequestCapture' + >; + } + ).input; + assert.ok(backendInput.loadSynthesisCache); + assert.ok(backendInput.writeSynthesisCache); + assert.ok(backendInput.recordProviderRequestCapture); + + await backendInput.loadSynthesisCache({ sessionId: 'session-1' }); + const capture: ProviderRequestCaptureRecord = { + schemaVersion: 1, + traceId: 'trace-1', + captureId: 'capture-1', + turnId: 'turn-1', + step: 0, + providerId: 'openai', + modelId: 'gpt-4o-mini', + requestHash: 'sha256:request', + requestBytes: 18, + segments: [], + serializedRequest: '{"prompt":"hello"}', + }; + await backendInput.recordProviderRequestCapture(capture); + + const sourceResult = { key: 'key-alpha', sentinel: 'SYNTH_SENTINEL' }; + const serializedResult = JSON.stringify(sourceResult); + const sourceBodySha256 = sha256(serializedResult); + const synthesisWrite: SynthesisCacheWriteInput = { + sessionId: 'session-1', + turnId: 'turn-1', + source: { + createdFrom: 'gated_archive_retrieval', + query: 'Recover key-alpha sentinel', + hydratedRuntimeEvents: [ + { + id: 'runtime-result-1', + sessionId: 'session-1', + runId: 'run-1', + turnId: 'turn-1', + invocationId: 'invocation-1', + ts: 122, + partial: false, + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: 'tool-call-1', + name: 'Read', + result: sourceResult, + }, + }, + ], + retrievedArchiveRefs: [ + { + kind: 'archived_tool_result', + sessionId: 'session-1', + turnId: 'turn-1', + runtimeEventId: 'runtime-result-1', + toolCallId: 'tool-call-1', + toolName: 'Read', + artifactId: 'archive-artifact-1', + bodySha256: sourceBodySha256, + originalEstimatedTokens: 12, + originalBytes: Buffer.byteLength(serializedResult, 'utf8'), + placeholderReason: 'stale_tool_result_pruned_before_compact', + }, + ], + archiveRetrievalMode: 'history_search_gated', + }, + limits: { + maxBlocks: 1, + maxBlockEstimatedTokens: 1_024, + maxEstimatedTokens: 2_048, + charsPerToken: 4, + }, + }; + const synthesisResult = await backendInput.writeSynthesisCache(synthesisWrite); + assert.equal(synthesisResult?.blocks.length, 1); + + const records = await createArtifactStore(storageRoot).list('session-1'); + assert.deepEqual(records.map((record) => record.source).sort(), [ + 'provider_request_capture', + 'synthesis_cache_block', + ]); + assert.deepEqual(await createArtifactStore(outputDir).list('session-1'), []); + }); + }); + test('Harbor ai-sdk backend uses the discovered GitHub Copilot wire', async () => { await withDirs(async ({ workspaceDir }) => { const registry = new BackendRegistry(); @@ -2014,6 +2145,7 @@ describe('runHarborCell', () => { model: 'gpt-5.4', }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, @@ -2087,6 +2219,7 @@ describe('runHarborCell', () => { model: 'gpt-4o-mini', }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, @@ -2119,6 +2252,7 @@ describe('runHarborCell', () => { model: 'gpt-4o-mini', }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, @@ -2198,6 +2332,7 @@ describe('runHarborCell', () => { systemPrompt: candidatePrompt, }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, @@ -2237,6 +2372,7 @@ describe('runHarborCell', () => { model: 'deepseek-v4-flash', }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, @@ -2295,6 +2431,7 @@ describe('runHarborCell', () => { model: 'gpt-4o-mini', }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, @@ -2394,6 +2531,7 @@ describe('runHarborCell', () => { model: 'gpt-4o-mini', }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, @@ -2441,6 +2579,7 @@ describe('runHarborCell', () => { model: 'gpt-4o-mini', }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, @@ -2485,6 +2624,7 @@ describe('runHarborCell', () => { model: 'gpt-4o-mini', }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', @@ -2562,6 +2702,7 @@ describe('runHarborCell', () => { model: 'gpt-4o-mini', }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, @@ -2596,6 +2737,7 @@ describe('runHarborCell', () => { model: 'gpt-4o-mini', }, task: { id: 'harbor-cell', instruction: 'solve', workspaceDir }, + storageRoot: workspaceDir, workspaceDir, realBackendIsolation: { kind: 'external', label: 'Harbor task container', toolExecutor }, toolExecutor, diff --git a/packages/headless/src/__tests__/provider-request-trace.test.ts b/packages/headless/src/__tests__/provider-request-trace.test.ts new file mode 100644 index 0000000000..511cbe2b9f --- /dev/null +++ b/packages/headless/src/__tests__/provider-request-trace.test.ts @@ -0,0 +1,106 @@ +import { mkdtemp, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import assert from 'node:assert/strict'; + +import * as traceAnalysis from '../provider-request-trace.js'; + +test('derives the first changed cacheable segment from the existing AgentRun trace', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const base = { + type: 'provider_request_captured', + runId: 'run-1', + sessionId: 'session-1', + turnId: 'turn-1', + ts: 1, + }; + const capture = (id: string, step: number, messageHashes: string[]): Record => ({ + ...base, + id, + ts: step + 1, + data: { + schemaVersion: 1, + traceId: 'provider-trace-1', + captureId: id, + artifactId: `artifact-${id}`, + step, + providerId: 'openai', + modelId: 'gpt-test', + requestHash: `sha256:request-${step}`, + requestBytes: 100 + step, + segments: [ + { + kind: 'system_prompt', + index: 0, + cacheable: true, + hash: 'sha256:system', + bytes: 10, + }, + ...messageHashes.map((hash, index) => ({ + kind: 'message', + index, + role: index === 0 ? 'user' : 'assistant', + cacheable: true, + hash, + bytes: 10, + })), + ], + }, + }); + await writeFile( + traceEventsPath, + `${[ + capture('capture-1', 0, ['sha256:user']), + capture('capture-2', 1, ['sha256:user', 'sha256:assistant']), + ] + .map((event) => JSON.stringify(event)) + .join('\n')}\n`, + ); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.equal(result.traceId, 'provider-trace-1'); + assert.equal(result.captures.length, 2); + assert.deepEqual(result.captures[1]?.firstChangedCacheableSegment, { + kind: 'message', + index: 1, + role: 'assistant', + }); +}); + +test('keeps complete provider captures when the AgentRun trace ends with a torn record', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + await writeFile( + traceEventsPath, + `${JSON.stringify({ + type: 'provider_request_captured', + id: 'capture-1', + runId: 'run-1', + sessionId: 'session-1', + turnId: 'turn-1', + ts: 1, + data: { + schemaVersion: 1, + traceId: 'provider-trace-1', + captureId: 'capture-1', + artifactId: 'artifact-capture-1', + step: 0, + providerId: 'openai', + modelId: 'gpt-test', + requestHash: 'sha256:request-1', + requestBytes: 100, + segments: [], + }, + })}\n{"type":"provider_request_captured"`, + ); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual( + result.captures.map((capture) => capture.captureId), + ['capture-1'], + ); +}); diff --git a/packages/headless/src/harbor-cell.ts b/packages/headless/src/harbor-cell.ts index 73e84dd585..ee26610a7f 100644 --- a/packages/headless/src/harbor-cell.ts +++ b/packages/headless/src/harbor-cell.ts @@ -12,6 +12,7 @@ import { buildChildAgentTools, buildProviderOptions, buildSubscriptionModelFetch, + createProviderRequestCaptureRecorder, getAIModel, getBuiltinPricing, loadSynthesisCacheBlocksFromArtifacts, @@ -25,6 +26,7 @@ import { createArtifactStore, createRuntimeEventStore, createSessionStore, + persistProviderRequestCaptureArtifact, } from '@maka/storage'; import { registerFakeBackend } from './backends.js'; import { @@ -259,6 +261,7 @@ export async function runHarborCell(input: RunHarborCellInput): Promise { const subscriptionFetch = buildSubscriptionModelFetch({ connection, @@ -793,6 +795,25 @@ export function buildAiSdkCellBackendRegistration(input: { newId: input.newId, now: input.now, recordRunTrace: ctx.recordRunTrace, + ...(ctx.recordProviderRequestCapture + ? { + recordProviderRequestCapture: createProviderRequestCaptureRecorder({ + persistArtifact: async (capture) => { + const artifact = await persistProviderRequestCaptureArtifact(artifactStore, { + sessionId: ctx.sessionId, + turnId: capture.turnId, + captureId: capture.captureId, + step: capture.step, + serializedRequest: capture.serializedRequest, + now: input.now(), + }); + return { artifactId: artifact.id }; + }, + recordLedger: ctx.recordProviderRequestCapture, + }), + recordProviderRequestAttempt: ctx.recordProviderRequestAttempt, + } + : {}), recordActiveFullCompactBlock: ctx.recordActiveFullCompactBlock, recordSemanticCompactBlock: ctx.recordSemanticCompactBlock, ...(input.recordUsageCheckpoint @@ -839,13 +860,10 @@ async function writeHarborCellArtifact(path: string, contents: string): Promise< } function buildHarborCellSynthesisCacheCallbacks( - env: RunHarborCellEnv, + artifactStore: ReturnType, enabled: boolean, ): { loadSynthesisCache?: SynthesisCacheLoader; writeSynthesisCache?: SynthesisCacheWriter } { if (!enabled) return {}; - const outputDir = env.MAKA_OUTPUT_DIR ?? '/logs/agent'; - const storageRoot = env.MAKA_STORAGE_ROOT ?? join(outputDir, 'maka-storage'); - const artifactStore = createArtifactStore(storageRoot); return { loadSynthesisCache: (event) => loadSynthesisCacheBlocksFromArtifacts(artifactStore, event), writeSynthesisCache: (event) => persistSynthesisCacheBlocksToArtifacts(artifactStore, event), diff --git a/packages/headless/src/index.ts b/packages/headless/src/index.ts index bf7d279836..878db2dc45 100644 --- a/packages/headless/src/index.ts +++ b/packages/headless/src/index.ts @@ -4,6 +4,11 @@ // package-local entrypoints, not the root API. Minimal usage is // `runExperiment(config, task, { storageRoot })`. export { runPromptOptimizationRun } from './prompt-optimization-run.js'; +export { readProviderRequestTrace } from './provider-request-trace.js'; +export type { + ProviderRequestTraceAnalysis, + ProviderRequestTraceCaptureAnalysis, +} from './provider-request-trace.js'; export type { MakaChangeAuditRecord } from './change-audit.js'; export type { PromptOptimizationRunInput, diff --git a/packages/headless/src/isolation.ts b/packages/headless/src/isolation.ts index f5462bf125..b3ae901ec8 100644 --- a/packages/headless/src/isolation.ts +++ b/packages/headless/src/isolation.ts @@ -186,6 +186,8 @@ export type RealBackendIsolation = ExternalRealBackendIsolation; export interface HeadlessBackendContext extends Partial { config: Config; task: Task; + /** Authoritative persistence root for this run. */ + storageRoot: string; /** Absolute throwaway workspace path for this run. */ workspaceDir: string; /** diff --git a/packages/headless/src/provider-request-trace.ts b/packages/headless/src/provider-request-trace.ts new file mode 100644 index 0000000000..0ef1589b54 --- /dev/null +++ b/packages/headless/src/provider-request-trace.ts @@ -0,0 +1,116 @@ +import { readFile } from 'node:fs/promises'; +import { decodeAgentRunEvent } from '@maka/core'; +import { + findFirstChangedCacheableSegment, + type PreparedRequestSegment, + type PreparedRequestSegmentRef, +} from '@maka/runtime'; + +export interface ProviderRequestTraceCaptureAnalysis { + traceId: string; + captureId: string; + artifactId: string; + turnId: string; + step: number; + providerId: string; + modelId: string; + requestHash: string; + requestBytes: number; + segments: PreparedRequestSegment[]; + firstChangedCacheableSegment?: PreparedRequestSegmentRef; +} + +export interface ProviderRequestTraceAnalysis { + traceId?: string; + captures: ProviderRequestTraceCaptureAnalysis[]; +} + +/** Read Harbor's existing AgentRun events.jsonl; no provider-proxy sidecar is required. */ +export async function readProviderRequestTrace( + traceEventsPath: string, +): Promise { + const text = await readFile(traceEventsPath, 'utf8'); + const captures: ProviderRequestTraceCaptureAnalysis[] = []; + for (const line of text.split('\n')) { + if (!line.trim()) continue; + let event: ReturnType; + try { + event = decodeAgentRunEvent(JSON.parse(line)); + } catch { + continue; + } + if (event.type !== 'provider_request_captured') continue; + const capture = captureFromEvent(event.turnId, event.data); + if (!capture) continue; + const prior = captures.at(-1); + captures.push({ + ...capture, + ...(prior + ? { + firstChangedCacheableSegment: findFirstChangedCacheableSegment(capture, prior), + } + : {}), + }); + } + return { + ...(captures[0] ? { traceId: captures[0].traceId } : {}), + captures, + }; +} + +function captureFromEvent( + turnId: string, + data: Record | undefined, +): ProviderRequestTraceCaptureAnalysis | undefined { + if (!data) return undefined; + const segments = Array.isArray(data.segments) + ? data.segments.map(segmentFromValue).filter((value) => value !== undefined) + : []; + if ( + typeof data.traceId !== 'string' || + typeof data.captureId !== 'string' || + typeof data.artifactId !== 'string' || + !isNonNegativeInteger(data.step) || + typeof data.providerId !== 'string' || + typeof data.modelId !== 'string' || + typeof data.requestHash !== 'string' || + !isNonNegativeInteger(data.requestBytes) || + segments.length !== (Array.isArray(data.segments) ? data.segments.length : 0) + ) { + return undefined; + } + return { + traceId: data.traceId, + captureId: data.captureId, + artifactId: data.artifactId, + turnId, + step: data.step, + providerId: data.providerId, + modelId: data.modelId, + requestHash: data.requestHash, + requestBytes: data.requestBytes, + segments, + }; +} + +function segmentFromValue(value: unknown): PreparedRequestSegment | undefined { + if (!value || typeof value !== 'object') return undefined; + const segment = value as Record; + if ( + !['tool_schema', 'system_prompt', 'message', 'provider_options'].includes( + String(segment.kind), + ) || + !isNonNegativeInteger(segment.index) || + typeof segment.cacheable !== 'boolean' || + typeof segment.hash !== 'string' || + !isNonNegativeInteger(segment.bytes) || + (segment.role !== undefined && typeof segment.role !== 'string') + ) { + return undefined; + } + return segment as unknown as PreparedRequestSegment; +} + +function isNonNegativeInteger(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; +} diff --git a/packages/headless/src/runner.ts b/packages/headless/src/runner.ts index f593bc0f30..c6d25a1d03 100644 --- a/packages/headless/src/runner.ts +++ b/packages/headless/src/runner.ts @@ -113,6 +113,7 @@ export async function runExperiment( await registerBackends(backends, { config: effectiveConfig, task, + storageRoot: deps.storageRoot, workspaceDir: agentWorkspaceDir, ...sessionCapabilities.capabilities, ...(backendNeedsIsolation(config.backend) diff --git a/packages/headless/src/task-agent-controller.ts b/packages/headless/src/task-agent-controller.ts index e22a26e91a..6732052222 100644 --- a/packages/headless/src/task-agent-controller.ts +++ b/packages/headless/src/task-agent-controller.ts @@ -285,6 +285,7 @@ export async function runTaskOnce( await registerBackends(backends, { config: effectiveConfig, task, + storageRoot: deps.storageRoot, workspaceDir: agentWorkspaceDir, heavyTaskMode, ...(heavyTaskProgress ? { heavyTaskProgress } : {}), @@ -1172,6 +1173,14 @@ function createSingleRunActiveSession( header, store, recordRunTrace: (event) => boundRun?.recordRunTrace(event), + recordProviderRequestCapture: (capture) => { + if (!boundRun) { + return Promise.reject(new Error('No active AgentRun for provider request capture')); + } + return boundRun.recordProviderRequestCapture(capture); + }, + recordProviderRequestAttempt: (attempt) => + boundRun?.recordProviderRequestAttempt(attempt), recordActiveFullCompactBlock: (block) => boundRun?.recordActiveFullCompactBlock(block), recordSemanticCompactBlock: (block) => boundRun?.recordSemanticCompactBlock(block), }); diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 0e63d9a459..ed1408592d 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -4,7 +4,7 @@ import { createHash } from 'node:crypto'; import { describe, test } from 'node:test'; import type { ModelMessage } from 'ai'; import { MockLanguageModelV4, simulateReadableStream } from 'ai/test'; -import type { LanguageModelV4StreamPart } from '@ai-sdk/provider'; +import { APICallError, type LanguageModelV4StreamPart } from '@ai-sdk/provider'; import type { AgentRunHeader, AttachmentByteReader, @@ -35,7 +35,11 @@ import { import type { MakaTool } from '../tool-runtime.js'; import { LOAD_TOOLS_NAME } from '../tool-availability.js'; import { PermissionEngine } from '../permission-engine.js'; -import { canonicalizeToolSet, computeRequestShapeDiagnostic } from '../request-shape.js'; +import { + canonicalizeToolSet, + computeRequestShapeDiagnostic, + findFirstChangedCacheableSegment, +} from '../request-shape.js'; import { ARCHIVED_TOOL_RESULT_PLACEHOLDER_KIND, ARCHIVED_TOOL_RESULT_REWRITE_VERSION, @@ -58,6 +62,10 @@ import { buildRuntimeEventModelReplayPlan, buildSteeringEnvelope } from '../mode import type { ActiveFullCompactBlock } from '../active-full-compact.js'; import type { SemanticCompactBlock } from '../semantic-compact.js'; import { HistoryCompactSummarizerError } from '../history-compact-summarizer.js'; +import type { + ProviderRequestAttemptRecord, + ProviderRequestCaptureRecord, +} from '../provider-request-telemetry.js'; describe('AiSdkBackend model history', () => { test('omits an empty system prompt from the provider request', async () => { @@ -7920,6 +7928,313 @@ describe('AiSdkBackend context budget and prompt attribution', () => { }); describe('AiSdkBackend RunTrace', () => { + for (const protocol of ['openai-compatible', 'anthropic-compatible'] as const) { + test(`records ${protocol} multi-step requests and reconciles complete attempt usage`, async () => { + const captures: ProviderRequestCaptureRecord[] = []; + const attempts: ProviderRequestAttemptRecord[] = []; + let calls = 0; + const usageFor = (step: number) => { + if (protocol === 'openai-compatible') { + const input = step === 0 ? 10 : 20; + const cached = step === 0 ? 4 : 5; + const output = step === 0 ? 2 : 3; + return { + inputTokens: { + total: input, + noCache: input - cached, + cacheRead: cached, + cacheWrite: undefined, + }, + outputTokens: { + total: output, + text: output - (step === 0 ? 0 : 1), + reasoning: step === 0 ? 0 : 1, + }, + raw: { + prompt_tokens: input, + completion_tokens: output, + prompt_tokens_details: { cached_tokens: cached }, + completion_tokens_details: { reasoning_tokens: step === 0 ? 0 : 1 }, + }, + }; + } + const noCache = step === 0 ? 6 : 12; + const cacheRead = step === 0 ? 3 : 6; + const cacheWrite = step === 0 ? 1 : 2; + const output = step === 0 ? 2 : 3; + return { + inputTokens: { + total: noCache + cacheRead + cacheWrite, + noCache, + cacheRead, + cacheWrite, + }, + outputTokens: { total: output, text: undefined, reasoning: undefined }, + raw: { + input_tokens: noCache, + output_tokens: output, + cache_read_input_tokens: cacheRead, + cache_creation_input_tokens: cacheWrite, + }, + }; + }; + const model = new MockLanguageModelV4({ + doStream: async () => { + const step = calls++; + const chunks: LanguageModelV4StreamPart[] = + step === 0 + ? [ + { type: 'stream-start', warnings: [] }, + { + type: 'tool-call', + toolCallId: 'read-1', + toolName: 'Read', + input: JSON.stringify({ path: 'notes.md' }), + }, + { + type: 'finish', + finishReason: { unified: 'tool-calls', raw: 'tool_calls' }, + usage: usageFor(step), + }, + ] + : [ + { 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: usageFor(step), + }, + ]; + return { + stream: simulateReadableStream({ + chunks, + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => model, + tools: [testTool('Read', z.object({ path: z.string() }))], + newId: idGenerator(), + now: monotonicClock(), + recordProviderRequestCapture: async (capture) => { + captures.push(capture); + return { artifactId: `artifact-${captures.length}` }; + }, + recordProviderRequestAttempt: (attempt) => { + attempts.push(attempt); + }, + }); + + const events: SessionEvent[] = []; + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + } + + assert.equal(captures.length, 2); + assert.deepEqual( + attempts.map(({ step, attempt, status }) => ({ step, attempt, status })), + [ + { step: 0, attempt: 1, status: 'completed' }, + { step: 1, attempt: 1, status: 'completed' }, + ], + ); + assert.equal(findFirstChangedCacheableSegment(captures[1]!, captures[0]!)?.kind, 'message'); + const aggregate = events.find( + (event): event is Extract => + event.type === 'token_usage', + ); + assert.ok(aggregate); + const sum = (field: keyof ProviderRequestAttemptRecord) => + attempts.reduce( + (total, attempt) => total + ((attempt[field] as number | undefined) ?? 0), + 0, + ); + assert.equal(sum('inputTokens'), aggregate.input); + assert.equal(sum('outputTokens'), aggregate.output); + assert.equal(sum('cacheReadInputTokens'), aggregate.cacheHitInput); + assert.equal(sum('cacheMissInputTokens'), aggregate.cacheMissInput); + assert.equal(sum('cacheWriteInputTokens'), aggregate.cacheWriteInput); + }); + } + + test('captures the prepared request before the provider call and records its physical attempt', async () => { + const captures: ProviderRequestCaptureRecord[] = []; + const attempts: ProviderRequestAttemptRecord[] = []; + const model = new MockLanguageModelV4({ + doStream: async () => { + assert.equal(captures.length, 1, 'capture must be durable before provider dispatch'); + return { + stream: simulateReadableStream({ + chunks: [ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: { + inputTokens: { total: 4, noCache: 4, cacheRead: 0, cacheWrite: undefined }, + outputTokens: { total: 2, text: 2, reasoning: 0 }, + raw: { + prompt_tokens: 4, + completion_tokens: 2, + prompt_tokens_details: { cached_tokens: 0 }, + }, + }, + }, + ], + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + recordProviderRequestCapture: async (capture) => { + captures.push(capture); + return { artifactId: `artifact-${captures.length}` }; + }, + recordProviderRequestAttempt: async (attempt) => { + attempts.push(attempt); + }, + }); + + const events: SessionEvent[] = []; + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + } + + assert.equal(captures.length, 1); + assert.equal(attempts.length, 1); + assert.equal(attempts[0]?.step, 0); + assert.equal(attempts[0]?.attempt, 1); + assert.equal(attempts[0]?.status, 'completed'); + assert.equal(attempts[0]?.captureId, captures[0]?.captureId); + assert.equal(attempts[0]?.cacheMissInputSource, 'derived'); + assert.equal( + events.find((event) => event.type === 'token_usage')?.providerRequestTraceId, + captures[0]?.traceId, + ); + }); + + test('does not call the provider when prepared-request persistence fails', async () => { + const model = completionModel(); + const backend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + recordProviderRequestCapture: async () => { + throw new Error('capture unavailable'); + }, + recordProviderRequestAttempt: () => {}, + }); + + const events: SessionEvent[] = []; + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + } + + assert.equal(model.doStreamCalls.length, 0); + assert.equal(events.at(-1)?.type, 'complete'); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'error'); + }); + + test('records each AI SDK internal retry as a physical attempt against one capture', async () => { + const captures: ProviderRequestCaptureRecord[] = []; + const attempts: ProviderRequestAttemptRecord[] = []; + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + if (calls === 1) { + throw new APICallError({ + message: 'retry me', + url: 'https://provider.invalid/v1/messages', + requestBodyValues: {}, + statusCode: 503, + responseHeaders: { 'retry-after-ms': '0' }, + }); + } + return { + stream: simulateReadableStream({ + chunks: [ + { type: 'stream-start', warnings: [] }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: emptyUsage(), + }, + ], + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = new AiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + permissionEngine: new PermissionEngine({ newId: () => 'permission-id', now: () => 1 }), + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + recordProviderRequestCapture: async (capture) => { + captures.push(capture); + return { artifactId: 'artifact-1' }; + }, + recordProviderRequestAttempt: (attempt) => { + attempts.push(attempt); + }, + }); + + await drain(backend.send({ turnId: 'turn-1', text: 'hi', context: [] })); + + assert.equal(calls, 2); + assert.equal(captures.length, 1); + assert.deepEqual( + attempts.map(({ attempt, status }) => ({ attempt, status })), + [ + { attempt: 1, status: 'failed' }, + { attempt: 2, status: 'completed' }, + ], + ); + }); + test('records the continuation replay gate and blocking diagnostics on stream failure', async () => { const trace: RunTraceEvent[] = []; const model = new MockLanguageModelV4({ diff --git a/packages/runtime/src/__tests__/ai-sdk-flow.test.ts b/packages/runtime/src/__tests__/ai-sdk-flow.test.ts index c1f48ca7ce..f8caafa7bf 100644 --- a/packages/runtime/src/__tests__/ai-sdk-flow.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-flow.test.ts @@ -143,6 +143,7 @@ describe('AiSdkFlow seam', () => { output: 5, costUsd: 0.001, systemPromptHash: 'sys-hash', + providerRequestTraceId: 'provider-trace-1', }), ev({ type: 'complete', stopReason: 'end_turn' }), ], @@ -174,6 +175,7 @@ describe('AiSdkFlow seam', () => { costUsd: 0.001, systemPromptHash: 'sys-hash', }); + assert.deepEqual(out[3].refs, { providerRequestTraceId: 'provider-trace-1' }); // Stream closes with a terminal event. assert.equal(isTerminalRuntimeEvent(out[out.length - 1]), true); assert.equal(out[out.length - 1].status, 'completed'); diff --git a/packages/runtime/src/__tests__/provider-request-telemetry.test.ts b/packages/runtime/src/__tests__/provider-request-telemetry.test.ts new file mode 100644 index 0000000000..ad57cca754 --- /dev/null +++ b/packages/runtime/src/__tests__/provider-request-telemetry.test.ts @@ -0,0 +1,599 @@ +import { describe, test } from 'node:test'; +import assert from 'node:assert/strict'; + +import * as telemetry from '../provider-request-telemetry.js'; + +describe('strict provider-request usage', () => { + test('preserves Anthropic cache fields as provider-reported values', () => { + const usage = telemetry.strictProviderRequestUsage({ + inputTokens: { total: 100, noCache: 40, cacheRead: 50, cacheWrite: 10 }, + outputTokens: { total: 12, text: undefined, reasoning: undefined }, + raw: { + input_tokens: 40, + output_tokens: 12, + cache_creation_input_tokens: 10, + cache_read_input_tokens: 50, + }, + }); + + assert.deepEqual(usage, { + inputTokens: 100, + cacheReadInputTokens: 50, + cacheReadInputSource: 'provider', + cacheMissInputTokens: 40, + cacheMissInputSource: 'provider', + cacheWriteInputTokens: 10, + cacheWriteInputSource: 'provider', + outputTokens: 12, + }); + }); + + test('reconciles Anthropic compaction iterations with normalized input usage', () => { + const usage = telemetry.strictProviderRequestUsage({ + inputTokens: { total: 115, noCache: 105, cacheRead: 10, cacheWrite: 0 }, + outputTokens: { total: 9, text: 9, reasoning: undefined }, + raw: { + input_tokens: 5, + output_tokens: 9, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 10, + iterations: [ + { type: 'message', input_tokens: 5, output_tokens: 4 }, + { type: 'compaction', input_tokens: 100, output_tokens: 5 }, + ], + }, + }); + + assert.deepEqual(usage, { + inputTokens: 115, + cacheReadInputTokens: 10, + cacheReadInputSource: 'provider', + cacheMissInputTokens: 105, + cacheMissInputSource: 'provider', + cacheWriteInputTokens: 0, + cacheWriteInputSource: 'provider', + outputTokens: 9, + }); + }); + + test('marks OpenAI cache miss as derived and leaves unsupported cache-write missing', () => { + const usage = telemetry.strictProviderRequestUsage({ + inputTokens: { total: 100, noCache: 30, cacheRead: 70, cacheWrite: undefined }, + outputTokens: { total: 20, text: 15, reasoning: 5 }, + raw: { + prompt_tokens: 100, + completion_tokens: 20, + prompt_tokens_details: { cached_tokens: 70 }, + completion_tokens_details: { reasoning_tokens: 5 }, + }, + }); + + assert.deepEqual(usage, { + inputTokens: 100, + cacheReadInputTokens: 70, + cacheReadInputSource: 'provider', + cacheMissInputTokens: 30, + cacheMissInputSource: 'derived', + outputTokens: 20, + reasoningTokens: 5, + }); + }); + + test('preserves OpenAI Chat cache-write and derives cache miss from the raw total', () => { + const usage = telemetry.strictProviderRequestUsage({ + inputTokens: { total: 100, noCache: 50, cacheRead: 20, cacheWrite: 30 }, + outputTokens: { total: 20, text: 20, reasoning: 0 }, + raw: { + prompt_tokens: 100, + completion_tokens: 20, + prompt_tokens_details: { cached_tokens: 20, cache_write_tokens: 30 }, + }, + }); + + assert.deepEqual(usage, { + inputTokens: 100, + cacheReadInputTokens: 20, + cacheReadInputSource: 'provider', + cacheMissInputTokens: 50, + cacheMissInputSource: 'derived', + cacheWriteInputTokens: 30, + cacheWriteInputSource: 'provider', + outputTokens: 20, + }); + }); + + test('preserves Google usage and derives cache miss from raw usage metadata', () => { + const usage = telemetry.strictProviderRequestUsage({ + inputTokens: { total: 100, noCache: 60, cacheRead: 40, cacheWrite: undefined }, + outputTokens: { total: 20, text: 15, reasoning: 5 }, + raw: { + promptTokenCount: 100, + candidatesTokenCount: 15, + cachedContentTokenCount: 40, + thoughtsTokenCount: 5, + }, + }); + + assert.deepEqual(usage, { + inputTokens: 100, + cacheReadInputTokens: 40, + cacheReadInputSource: 'provider', + cacheMissInputTokens: 60, + cacheMissInputSource: 'derived', + outputTokens: 20, + reasoningTokens: 5, + }); + }); + + test('does not inherit Google adapter zeroes for omitted raw cache and reasoning fields', () => { + const usage = telemetry.strictProviderRequestUsage({ + inputTokens: { total: 100, noCache: 100, cacheRead: 0, cacheWrite: undefined }, + outputTokens: { total: 20, text: 20, reasoning: 0 }, + raw: { + promptTokenCount: 100, + candidatesTokenCount: 20, + }, + }); + + assert.deepEqual(usage, { inputTokens: 100, outputTokens: 20 }); + }); + + test('does not turn omitted provider cache details into zero-valued evidence', () => { + const usage = telemetry.strictProviderRequestUsage({ + inputTokens: { total: 100, noCache: 100, cacheRead: 0, cacheWrite: undefined }, + outputTokens: { total: 20, text: 20, reasoning: 0 }, + raw: { prompt_tokens: 100, completion_tokens: 20 }, + }); + + assert.deepEqual(usage, { inputTokens: 100, outputTokens: 20 }); + }); + + test('does not inherit normalized zero totals when the raw provider fields are missing', () => { + const usage = telemetry.strictProviderRequestUsage({ + inputTokens: { total: 0, noCache: 0, cacheRead: 10, cacheWrite: undefined }, + outputTokens: { total: 0, text: 0, reasoning: 0 }, + raw: { prompt_tokens_details: { cached_tokens: 10 } }, + }); + + assert.deepEqual(usage, { + cacheReadInputTokens: 10, + cacheReadInputSource: 'provider', + }); + }); + + test('keeps normalized totals when no raw provider payload is available', () => { + const usage = telemetry.strictProviderRequestUsage({ + inputTokens: { total: 8, noCache: 8, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: 3, text: 3, reasoning: undefined }, + }); + + assert.deepEqual(usage, { inputTokens: 8, outputTokens: 3 }); + }); + + test('does not derive cache miss from inconsistent provider components', () => { + const usage = telemetry.strictProviderRequestUsage({ + inputTokens: { total: 10, noCache: 0, cacheRead: 20, cacheWrite: undefined }, + outputTokens: { total: 0, text: 0, reasoning: 0 }, + raw: { + prompt_tokens: 10, + prompt_tokens_details: { cached_tokens: 20 }, + }, + }); + + assert.deepEqual(usage, { + inputTokens: 10, + cacheReadInputTokens: 20, + cacheReadInputSource: 'provider', + }); + }); +}); + +describe('provider request capture commit', () => { + test('links body-free metadata and returns the committed artifact reference', async () => { + const ledgerCaptures: Array> = []; + const recordCapture = telemetry.createProviderRequestCaptureRecorder({ + persistArtifact: async () => ({ artifactId: 'artifact-capture-1' }), + recordLedger: async (capture) => { + ledgerCaptures.push(capture as unknown as Record); + }, + }); + + const result = await recordCapture({ + schemaVersion: 1, + traceId: 'trace-1', + captureId: 'capture-1', + turnId: 'turn-1', + step: 0, + providerId: 'openai', + modelId: 'gpt-test', + requestHash: 'sha256:request', + requestBytes: 2, + segments: [], + serializedRequest: '{}', + }); + + assert.deepEqual(result, { artifactId: 'artifact-capture-1' }); + assert.equal(ledgerCaptures.length, 1); + assert.equal(ledgerCaptures[0]?.artifactId, 'artifact-capture-1'); + assert.equal(Object.hasOwn(ledgerCaptures[0]!, 'serializedRequest'), false); + }); + + test('retains the request artifact when a failed ledger append may have landed', async () => { + const ledgerError = new Error('capture ledger append failed'); + const ledgerCaptures: Array> = []; + const persistedArtifactIds = new Set(); + const createRecorder = Reflect.get( + telemetry, + 'createProviderRequestCaptureRecorder', + ) as unknown as + | ((input: Record) => (capture: Record) => Promise) + | undefined; + assert.equal(typeof createRecorder, 'function'); + const recordCapture = createRecorder!({ + persistArtifact: async () => { + persistedArtifactIds.add('artifact-capture-1'); + return { artifactId: 'artifact-capture-1' }; + }, + recordLedger: async (capture: Record) => { + ledgerCaptures.push(capture); + throw ledgerError; + }, + }); + + await assert.rejects( + recordCapture({ + schemaVersion: 1, + traceId: 'trace-1', + captureId: 'capture-1', + turnId: 'turn-1', + step: 0, + providerId: 'openai', + modelId: 'gpt-test', + requestHash: 'sha256:request', + requestBytes: 2, + segments: [], + serializedRequest: '{}', + }), + (error) => error === ledgerError, + ); + assert.equal(ledgerCaptures.length, 1); + assert.deepEqual([...persistedArtifactIds], ['artifact-capture-1']); + }); +}); + +describe('provider request tracker', () => { + test('persists a logical capture before each physical attempt and reuses it for retries', async () => { + const captures: Array<{ + captureId: string; + requestHash: string; + serializedRequest: string; + }> = []; + const attempts: Array<{ step: number; attempt: number; status: string; captureId: string }> = + []; + const Tracker = Reflect.get(telemetry, 'ProviderRequestTracker') as unknown as + | (new ( + input: Record, + ) => { + setStep(step: number): void; + trackStream(input: Record): Promise<{ stream: ReadableStream }>; + }) + | undefined; + assert.equal(typeof Tracker, 'function'); + let id = 0; + const tracker = new Tracker!({ + traceId: 'trace-1', + turnId: 'turn-1', + now: () => Date.now(), + newId: () => `id-${++id}`, + persistCapture: async (capture: { + captureId: string; + requestHash: string; + serializedRequest: string; + }) => { + captures.push(capture); + return { artifactId: `artifact-${captures.length}` }; + }, + recordAttempt: async (attempt: { + step: number; + attempt: number; + status: string; + captureId: string; + }) => attempts.push(attempt), + }); + tracker.setStep(2); + const params = preparedParams('hello'); + + await assert.rejects( + tracker.trackStream({ + providerId: 'openai', + modelId: 'gpt-test', + params, + abortSignal: new AbortController().signal, + doStream: async () => { + throw new Error('network'); + }, + }), + /network/, + ); + const result = await tracker.trackStream({ + providerId: 'openai', + modelId: 'gpt-test', + params, + abortSignal: new AbortController().signal, + doStream: async () => ({ + stream: streamOf([ + { type: 'text-delta', id: 'text-1', delta: 'ok' }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: { + inputTokens: { total: 10, noCache: 6, cacheRead: 4, cacheWrite: undefined }, + outputTokens: { total: 2, text: 2, reasoning: 0 }, + raw: { + prompt_tokens: 10, + completion_tokens: 2, + prompt_tokens_details: { cached_tokens: 4 }, + }, + }, + }, + ]), + }), + }); + await drain(result.stream); + + assert.equal(captures.length, 1); + assert.deepEqual(JSON.parse(captures[0]!.serializedRequest), params); + assert.deepEqual( + attempts.map(({ step, attempt, status, captureId }) => ({ + step, + attempt, + status, + captureId, + })), + [ + { step: 2, attempt: 1, status: 'failed', captureId: captures[0]!.captureId }, + { step: 2, attempt: 2, status: 'completed', captureId: captures[0]!.captureId }, + ], + ); + assert.equal((attempts[1] as Record).cacheReadInputSource, 'provider'); + assert.equal((attempts[1] as Record).cacheMissInputSource, 'derived'); + }); + + test('captures a changed logical body separately and blocks provider calls on capture failure', async () => { + const captures: string[] = []; + const Tracker = Reflect.get(telemetry, 'ProviderRequestTracker') as unknown as new ( + input: Record, + ) => { + setStep(step: number): void; + trackStream(input: Record): Promise<{ stream: ReadableStream }>; + }; + let providerCalls = 0; + const tracker = new Tracker({ + traceId: 'trace-2', + turnId: 'turn-2', + now: () => Date.now(), + newId: () => `capture-${captures.length + 1}`, + persistCapture: async (capture: { requestHash: string }) => { + captures.push(capture.requestHash); + if (captures.length === 2) throw new Error('capture unavailable'); + return { artifactId: 'artifact-1' }; + }, + recordAttempt: () => {}, + }); + tracker.setStep(0); + const completed = await tracker.trackStream({ + providerId: 'anthropic', + modelId: 'claude-test', + params: preparedParams('before'), + abortSignal: new AbortController().signal, + doStream: async () => { + providerCalls += 1; + return { stream: streamOf([finishPart()]) }; + }, + }); + await drain(completed.stream); + + await assert.rejects( + tracker.trackStream({ + providerId: 'anthropic', + modelId: 'claude-test', + params: preparedParams('after'), + abortSignal: new AbortController().signal, + doStream: async () => { + providerCalls += 1; + return { stream: streamOf([finishPart()]) }; + }, + }), + /capture unavailable/, + ); + assert.equal(providerCalls, 1); + assert.equal(captures.length, 2); + assert.notEqual(captures[0], captures[1]); + }); + + test('records an errored stream after output as interrupted', async () => { + const attempts: Array<{ status: string }> = []; + const Tracker = Reflect.get(telemetry, 'ProviderRequestTracker') as unknown as new ( + input: Record, + ) => { + setStep(step: number): void; + trackStream(input: Record): Promise<{ stream: ReadableStream }>; + }; + const tracker = new Tracker({ + traceId: 'trace-3', + turnId: 'turn-3', + now: () => Date.now(), + newId: () => 'id', + persistCapture: async () => ({ artifactId: 'artifact' }), + recordAttempt: async (attempt: { status: string }) => attempts.push(attempt), + }); + tracker.setStep(0); + const result = await tracker.trackStream({ + providerId: 'openai', + modelId: 'gpt-test', + params: preparedParams('hello'), + abortSignal: new AbortController().signal, + doStream: async () => ({ stream: interruptedStream() }), + }); + await assert.rejects(drain(result.stream), /stream broke/); + assert.equal(attempts[0]?.status, 'interrupted'); + }); + + test('records an in-flight attempt as aborted when its signal is cancelled', async () => { + const attempts: Array<{ status: string }> = []; + const abort = new AbortController(); + const tracker = new telemetry.ProviderRequestTracker({ + traceId: 'trace-4', + turnId: 'turn-4', + now: () => Date.now(), + newId: () => 'id', + persistCapture: async () => ({ artifactId: 'artifact' }), + recordAttempt: async (attempt) => { + attempts.push(attempt); + }, + }); + tracker.setStep(0); + await tracker.trackStream({ + providerId: 'openai', + modelId: 'gpt-test', + params: preparedParams('hello'), + abortSignal: abort.signal, + doStream: async () => ({ stream: new ReadableStream() }), + }); + + abort.abort(); + await Promise.resolve(); + + assert.equal(attempts[0]?.status, 'aborted'); + }); + + test('does not capture or record an attempt when cancellation predates dispatch', async () => { + let captures = 0; + let attempts = 0; + let providerCalls = 0; + const abort = new AbortController(); + abort.abort(); + const tracker = new telemetry.ProviderRequestTracker({ + traceId: 'trace-5', + turnId: 'turn-5', + now: () => Date.now(), + newId: () => 'id', + persistCapture: async () => { + captures += 1; + return { artifactId: 'artifact' }; + }, + recordAttempt: async () => { + attempts += 1; + }, + }); + + await assert.rejects( + tracker.trackStream({ + providerId: 'openai', + modelId: 'gpt-test', + params: preparedParams('hello'), + abortSignal: abort.signal, + doStream: async () => { + providerCalls += 1; + return { stream: streamOf([finishPart()]) }; + }, + }), + { name: 'AbortError' }, + ); + + assert.equal(captures, 0); + assert.equal(attempts, 0); + assert.equal(providerCalls, 0); + }); + + test('does not dispatch or record an attempt when cancellation happens during capture', async () => { + let captures = 0; + let attempts = 0; + let providerCalls = 0; + const abort = new AbortController(); + const tracker = new telemetry.ProviderRequestTracker({ + traceId: 'trace-6', + turnId: 'turn-6', + now: () => Date.now(), + newId: () => 'id', + persistCapture: async () => { + captures += 1; + abort.abort(); + return { artifactId: 'artifact' }; + }, + recordAttempt: async () => { + attempts += 1; + }, + }); + + await assert.rejects( + tracker.trackStream({ + providerId: 'openai', + modelId: 'gpt-test', + params: preparedParams('hello'), + abortSignal: abort.signal, + doStream: async () => { + providerCalls += 1; + return { stream: streamOf([finishPart()]) }; + }, + }), + { name: 'AbortError' }, + ); + + assert.equal(captures, 1); + assert.equal(attempts, 0); + assert.equal(providerCalls, 0); + }); +}); + +function preparedParams(text: string): Record { + return { + prompt: [ + { role: 'system', content: 'system' }, + { role: 'user', content: [{ type: 'text', text }] }, + ], + tools: [{ type: 'function', name: 'Read', inputSchema: { type: 'object' } }], + providerOptions: { test: { cacheControl: true } }, + }; +} + +function finishPart(): Record { + return { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: { + inputTokens: { total: 1, noCache: 1, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: 1, text: 1, reasoning: undefined }, + raw: { input_tokens: 1, output_tokens: 1 }, + }, + }; +} + +function streamOf(parts: unknown[]): ReadableStream { + return new ReadableStream({ + start(controller) { + for (const part of parts) controller.enqueue(part); + controller.close(); + }, + }); +} + +function interruptedStream(): ReadableStream { + let pulls = 0; + return new ReadableStream({ + pull(controller) { + pulls += 1; + if (pulls === 1) { + controller.enqueue({ type: 'text-delta', id: 'text', delta: 'partial' }); + } else { + controller.error(new Error('stream broke')); + } + }, + }); +} + +async function drain(stream: ReadableStream): Promise { + for await (const _part of stream) { + // Drain to trigger terminal telemetry. + } +} diff --git a/packages/runtime/src/__tests__/request-shape.test.ts b/packages/runtime/src/__tests__/request-shape.test.ts index 11c36b102a..8a6a0d7b0b 100644 --- a/packages/runtime/src/__tests__/request-shape.test.ts +++ b/packages/runtime/src/__tests__/request-shape.test.ts @@ -6,6 +6,7 @@ import { toolSchemaCharsForDiagnostics, computeRequestShapeDiagnostic, } from '../request-shape.js'; +import * as requestShape from '../request-shape.js'; import type { MakaTool } from '../tool-runtime.js'; function tool(name: string): MakaTool { @@ -113,3 +114,119 @@ describe('diagnostics measure the provider-visible (active) tool subset', () => assert.equal(after.prefixChangeReason, 'tool_schema_changed'); }); }); + +describe('prepared provider request capture', () => { + test('records cacheable request segments in provider-prefix order', () => { + const capture = Reflect.get(requestShape, 'capturePreparedProviderRequest') as + | ((input: { + providerId: string; + modelId: string; + instructions: string; + messages: Array<{ role: string; content: string }>; + tools: Array>; + providerOptions: Record; + }) => { + requestHash: string; + requestBytes: number; + serializedRequest: string; + segments: Array<{ + kind: string; + index: number; + cacheable: boolean; + hash: string; + bytes: number; + role?: string; + }>; + }) + | undefined; + + assert.equal(typeof capture, 'function'); + const result = capture!({ + providerId: 'anthropic', + modelId: 'claude-test', + instructions: 'system', + messages: [{ role: 'user', content: 'hello' }], + tools: [{ name: 'Bash', description: 'Run a command', inputSchema: { type: 'object' } }], + providerOptions: { anthropic: { thinking: { type: 'enabled', budgetTokens: 1_024 } } }, + }); + + assert.deepEqual( + result.segments.map(({ kind, index, cacheable, role }) => ({ + kind, + index, + cacheable, + ...(role ? { role } : {}), + })), + [ + { kind: 'tool_schema', index: 0, cacheable: true }, + { kind: 'system_prompt', index: 0, cacheable: true }, + { kind: 'message', index: 0, cacheable: true, role: 'user' }, + { kind: 'provider_options', index: 0, cacheable: false }, + ], + ); + assert.match(result.requestHash, /^sha256:[a-f0-9]{64}$/); + assert.equal(result.requestBytes, Buffer.byteLength(result.serializedRequest, 'utf8')); + assert.ok(result.segments.every((segment) => segment.bytes > 0)); + assert.ok(result.segments.every((segment) => /^sha256:[a-f0-9]{64}$/.test(segment.hash))); + }); + + test('finds the first changed cacheable segment by exact content hash', () => { + const capture = requestShape.capturePreparedProviderRequest; + const findFirstChanged = Reflect.get(requestShape, 'findFirstChangedCacheableSegment') as + | (( + current: ReturnType, + prior: ReturnType, + ) => { kind: string; index: number; role?: string } | undefined) + | undefined; + assert.equal(typeof findFirstChanged, 'function'); + + const prior = capture({ + providerId: 'openai', + modelId: 'gpt-test', + instructions: 'system', + messages: [{ role: 'user', content: 'alpha' }], + tools: [{ name: 'Read', inputSchema: { type: 'object' } }], + providerOptions: { openai: { reasoningEffort: 'low' } }, + }); + const changedMessage = capture({ + providerId: 'openai', + modelId: 'gpt-test', + instructions: 'system', + messages: [{ role: 'user', content: 'bravo' }], + tools: [{ name: 'Read', inputSchema: { type: 'object' } }], + providerOptions: { openai: { reasoningEffort: 'low' } }, + }); + assert.deepEqual(findFirstChanged!(changedMessage, prior), { + kind: 'message', + index: 0, + role: 'user', + }); + + const onlyOptionsChanged = capture({ + providerId: 'openai', + modelId: 'gpt-test', + instructions: 'system', + messages: [{ role: 'user', content: 'alpha' }], + tools: [{ name: 'Read', inputSchema: { type: 'object' } }], + providerOptions: { openai: { reasoningEffort: 'high' } }, + }); + assert.equal(findFirstChanged!(onlyOptionsChanged, prior), undefined); + + const appendedMessage = capture({ + providerId: 'openai', + modelId: 'gpt-test', + instructions: 'system', + messages: [ + { role: 'user', content: 'alpha' }, + { role: 'assistant', content: 'done' }, + ], + tools: [{ name: 'Read', inputSchema: { type: 'object' } }], + providerOptions: { openai: { reasoningEffort: 'low' } }, + }); + assert.deepEqual(findFirstChanged!(appendedMessage, prior), { + kind: 'message', + index: 1, + role: 'assistant', + }); + }); +}); diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index a9b1f313ab..802dad4c9a 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -7679,6 +7679,137 @@ describe('SessionManager permission mode updates', () => { expect(JSON.stringify(events).includes('sk-live-secret-token-value')).toBe(false); }); + test('durable run ledger records provider request capture metadata and complete attempt segments', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + backends.register('fake', (ctx) => new ProviderRequestTraceBackend(ctx)); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + newId: nextId(), + now: nextNow(12_760), + }); + const session = await manager.createSession(makeInput()); + + await drain(manager.sendMessage(session.id, { turnId: 'turn-1', text: 'hello' })); + + const [run] = await runStore.listSessionRuns(session.id); + const events = await runStore.readEvents(session.id, run!.runId); + const captured = events.find((event) => event.type === 'provider_request_captured'); + const attempt = events.find((event) => event.type === 'provider_request_attempt_recorded'); + expect(captured?.data?.artifactId).toBe('artifact-capture'); + expect(captured?.data?.requestHash).toBe('sha256:request'); + expect(attempt?.data?.captureId).toBe('capture-1'); + expect((attempt?.data?.segments as unknown[])?.length).toBe(75); + }); + + test('required capture and later attempt still write after an attempt append fails', async () => { + const store = new MemorySessionStore(); + const attemptFailureRecorded = makeGate(); + const captureOutcomes: string[] = []; + let failAttemptOnce = true; + const runStore = new MemoryAgentRunStore({ + beforeAgentRunEventAppend: async (_sessionId, _runId, event) => { + if (event.type === 'provider_request_attempt_recorded' && failAttemptOnce) { + failAttemptOnce = false; + throw new Error('diagnostic attempt append failed'); + } + if (event.type === 'trace_write_failed') attemptFailureRecorded.release(); + }, + }); + const backends = new BackendRegistry(); + backends.register( + 'fake', + (ctx) => + new ProviderCaptureAfterAttemptFailureBackend( + ctx, + attemptFailureRecorded.promise, + captureOutcomes, + ), + ); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + newId: nextId(), + now: nextNow(12_762), + }); + const session = await manager.createSession(makeInput()); + + await drain(manager.sendMessage(session.id, { turnId: 'turn-1', text: 'hello' })); + + expect(captureOutcomes).toEqual(['fulfilled']); + const [run] = await runStore.listSessionRuns(session.id); + const events = await runStore.readEvents(session.id, run!.runId); + expect(events.some((event) => event.type === 'provider_request_captured')).toBe(true); + expect(events.some((event) => event.id === 'attempt-2')).toBe(true); + }); + + test('finalizes the run when a required provider capture append fails', async () => { + const store = new MemorySessionStore(); + let providerDispatches = 0; + let failCaptureOnce = true; + const runStore = new MemoryAgentRunStore({ + beforeAgentRunEventAppend: async (_sessionId, _runId, event) => { + if (event.type === 'provider_request_captured' && failCaptureOnce) { + failCaptureOnce = false; + throw new Error('required capture append failed'); + } + }, + }); + const backends = new BackendRegistry(); + backends.register( + 'fake', + (ctx) => new ProviderCaptureGateBackend(ctx, () => (providerDispatches += 1)), + ); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends, + newId: nextId(), + now: nextNow(12_764), + }); + const session = await manager.createSession(makeInput()); + + await drain(manager.sendMessage(session.id, { turnId: 'turn-1', text: 'hello' })).catch( + () => {}, + ); + + expect(providerDispatches).toBe(0); + const [run] = await runStore.listSessionRuns(session.id); + expect(run?.status).toBe('failed'); + expect(run?.completedAt).toBeDefined(); + }); + + test('omits provider request telemetry hooks when no run store is configured', async () => { + const store = new MemorySessionStore(); + const backends = new BackendRegistry(); + let captureHook: BackendFactoryContext['recordProviderRequestCapture']; + let attemptHook: BackendFactoryContext['recordProviderRequestAttempt']; + backends.register('fake', (ctx) => { + captureHook = ctx.recordProviderRequestCapture; + attemptHook = ctx.recordProviderRequestAttempt; + return new FakeBackend(ctx); + }); + const manager = new SessionManager({ + store, + backends, + newId: nextId(), + now: nextNow(12_765), + }); + const session = await manager.createSession(makeInput()); + + await drain(manager.sendMessage(session.id, { turnId: 'turn-1', text: 'hello' })); + + expect(captureHook).toBeUndefined(); + expect(attemptHook).toBeUndefined(); + }); + test('durable run ledger records full active compact blocks asynchronously', async () => { const store = new MemorySessionStore(); const runStore = new MemoryAgentRunStore(); @@ -11623,6 +11754,189 @@ class TraceBackend implements AgentBackend { async dispose(): Promise {} } +class ProviderRequestTraceBackend implements AgentBackend { + readonly kind = 'fake' as const; + readonly sessionId: string; + + constructor(private readonly ctx: BackendFactoryContext) { + this.sessionId = ctx.sessionId; + } + + async *send(input: BackendSendInput): AsyncIterable { + await this.ctx.recordProviderRequestCapture?.({ + schemaVersion: 1, + traceId: 'provider-trace-1', + captureId: 'capture-1', + turnId: input.turnId, + step: 0, + providerId: 'fake', + modelId: 'fake-model', + requestHash: 'sha256:request', + requestBytes: 100, + segments: [], + artifactId: 'artifact-capture', + }); + await this.ctx.recordProviderRequestAttempt?.({ + traceId: 'provider-trace-1', + attemptId: 'attempt-1', + turnId: input.turnId, + step: 0, + attempt: 1, + captureId: 'capture-1', + captureArtifactId: 'artifact-capture', + providerId: 'fake', + modelId: 'fake-model', + requestHash: 'sha256:request', + requestBytes: 100, + segments: Array.from({ length: 75 }, (_, index) => ({ + kind: 'message' as const, + index, + cacheable: true, + hash: `sha256:${index}`, + bytes: 1, + })), + startedAt: 1, + completedAt: 2, + status: 'completed', + finishReason: 'stop', + latencyMs: 1, + }); + yield { + type: 'complete', + id: `${input.turnId}-complete`, + turnId: input.turnId, + ts: 3, + stopReason: 'end_turn', + }; + } + + async stop(): Promise {} + async respondToPermission(_decision: PermissionDecision): Promise {} + async dispose(): Promise {} +} + +class ProviderCaptureGateBackend implements AgentBackend { + readonly kind = 'fake' as const; + readonly sessionId: string; + + constructor( + private readonly ctx: BackendFactoryContext, + private readonly dispatch: () => void, + ) { + this.sessionId = ctx.sessionId; + } + + async *send(input: BackendSendInput): AsyncIterable { + await this.ctx.recordProviderRequestCapture?.({ + schemaVersion: 1, + traceId: 'provider-trace-gated', + captureId: 'capture-gated', + turnId: input.turnId, + step: 0, + providerId: 'fake', + modelId: 'fake-model', + requestHash: 'sha256:gated', + requestBytes: 100, + segments: [], + artifactId: 'artifact-gated', + }); + this.dispatch(); + yield { + type: 'complete', + id: `${input.turnId}-complete`, + turnId: input.turnId, + ts: 3, + stopReason: 'end_turn', + }; + } + + async stop(): Promise {} + async respondToPermission(_decision: PermissionDecision): Promise {} + async dispose(): Promise {} +} + +class ProviderCaptureAfterAttemptFailureBackend implements AgentBackend { + readonly kind = 'fake' as const; + readonly sessionId: string; + + constructor( + private readonly ctx: BackendFactoryContext, + private readonly attemptFailureRecorded: Promise, + private readonly captureOutcomes: string[], + ) { + this.sessionId = ctx.sessionId; + } + + async *send(input: BackendSendInput): AsyncIterable { + await this.ctx.recordProviderRequestAttempt?.({ + traceId: 'provider-trace-1', + attemptId: 'attempt-1', + turnId: input.turnId, + step: 0, + attempt: 1, + captureId: 'capture-1', + captureArtifactId: 'artifact-capture-1', + providerId: 'fake', + modelId: 'fake-model', + requestHash: 'sha256:request-1', + requestBytes: 100, + segments: [], + startedAt: 1, + completedAt: 2, + status: 'completed', + latencyMs: 1, + }); + await this.attemptFailureRecorded; + try { + await this.ctx.recordProviderRequestCapture?.({ + schemaVersion: 1, + traceId: 'provider-trace-1', + captureId: 'capture-2', + turnId: input.turnId, + step: 1, + providerId: 'fake', + modelId: 'fake-model', + requestHash: 'sha256:request-2', + requestBytes: 120, + segments: [], + artifactId: 'artifact-capture-2', + }); + this.captureOutcomes.push('fulfilled'); + } catch { + this.captureOutcomes.push('rejected'); + } + await this.ctx.recordProviderRequestAttempt?.({ + traceId: 'provider-trace-1', + attemptId: 'attempt-2', + turnId: input.turnId, + step: 1, + attempt: 1, + captureId: 'capture-2', + captureArtifactId: 'artifact-capture-2', + providerId: 'fake', + modelId: 'fake-model', + requestHash: 'sha256:request-2', + requestBytes: 120, + segments: [], + startedAt: 2, + completedAt: 3, + status: 'completed', + latencyMs: 1, + }); + yield { + type: 'complete', + id: `${input.turnId}-complete`, + turnId: input.turnId, + ts: 3, + stopReason: 'end_turn', + }; + } + + async stop(): Promise {} + async respondToPermission(_decision: PermissionDecision): Promise {} + async dispose(): Promise {} +} + class ActiveCompactBlockBackend implements AgentBackend { readonly kind = 'fake' as const; readonly sessionId: string; diff --git a/packages/runtime/src/agent-run.ts b/packages/runtime/src/agent-run.ts index 38626b474b..1b948eb867 100644 --- a/packages/runtime/src/agent-run.ts +++ b/packages/runtime/src/agent-run.ts @@ -41,6 +41,10 @@ import { AiSdkFlow } from './ai-sdk-flow.js'; import type { InvocationContext } from './invocation-context.js'; import { buildInitialUserRuntimeEvent } from './runtime-runner.js'; import type { RuntimeContinuation } from './runtime-resume.js'; +import type { + ProviderRequestAttemptRecord, + ProviderRequestCaptureLedgerRecord, +} from './provider-request-telemetry.js'; export interface AgentRunActiveSession { sessionId: string; @@ -241,6 +245,46 @@ export class AgentRun { }); } + recordProviderRequestCapture(capture: ProviderRequestCaptureLedgerRecord): Promise { + if (!this.input.runStore) return Promise.reject(new Error('AgentRun store is not configured')); + return this.enqueueRequiredProviderCapture('append provider request capture', async () => { + const { + schemaVersion, + serializedRequest: _serializedRequest, + ...data + } = capture as ProviderRequestCaptureLedgerRecord & { serializedRequest?: string }; + await this.input.runStore?.appendEvent( + this.sessionId, + this.runId, + { + type: 'provider_request_captured', + id: capture.captureId, + runId: this.runId, + sessionId: this.sessionId, + turnId: capture.turnId, + ts: this.input.now(), + data: { schemaVersion, ...data }, + }, + { durable: true }, + ); + }); + } + + recordProviderRequestAttempt(attempt: ProviderRequestAttemptRecord): void { + if (!this.input.runStore) return; + this.enqueueBestEffortProviderAttempt('append provider request attempt', async () => { + await this.input.runStore?.appendEvent(this.sessionId, this.runId, { + type: 'provider_request_attempt_recorded', + id: attempt.attemptId, + runId: this.runId, + sessionId: this.sessionId, + turnId: attempt.turnId, + ts: attempt.completedAt, + data: { ...attempt }, + }); + }); + } + recordActiveFullCompactBlock(block: ActiveFullCompactBlock): void { if (!this.input.runStore || !this.runStoreAvailable) return; this.enqueueRunStore('append active full compact block', async () => { @@ -1285,6 +1329,37 @@ export class AgentRun { return next; } + /** + * Each physical provider request gets its own best-effort diagnostic row. + * One failed attempt append must not suppress later attempts or poison the + * general AgentRun store latch; a required capture independently gates every + * provider dispatch. + */ + private enqueueBestEffortProviderAttempt(label: string, operation: () => Promise): void { + const next = this.traceQueue + .then(operation, operation) + .catch((error) => this.enqueueTraceWriteFailure(error, label)); + this.traceQueue = next.catch(() => {}); + } + + /** + * A prepared-request capture is a dispatch gate, not diagnostic telemetry. + * Always attempt its durable append even when an earlier best-effort run + * trace write marked the general run ledger unavailable; only this append's + * own outcome may decide whether the provider request can be dispatched. + */ + private enqueueRequiredProviderCapture( + label: string, + operation: () => Promise, + ): Promise { + const next = this.traceQueue.then(operation, operation).catch(async (error) => { + await this.enqueueTraceWriteFailure(error, label); + throw error; + }); + this.traceQueue = next.catch(() => {}); + return next; + } + /** * Read-back disambiguation for a failed durability-required append: true * only when the ledger demonstrably contains the event. Any doubt (no diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 6298d5927a..0cca4c8d54 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -152,6 +152,11 @@ import { toolSchemaCharsForDiagnostics, type RequestShapeDiagnostic, } from './request-shape.js'; +import { + ProviderRequestTracker, + type ProviderRequestAttemptRecord, + type ProviderRequestCaptureRecord, +} from './provider-request-telemetry.js'; import { ToolAvailabilityRuntime, type ToolAvailabilityConfig } from './tool-availability.js'; import { applyRuntimeEventContextBudget, @@ -495,6 +500,15 @@ export interface AiSdkBackendInput { }) => Promise; /** Optional diagnostic trace hook for explaining a runtime turn without changing renderer events. */ recordRunTrace?: RunTraceRecorder; + /** + * Durable prepared-request capture boundary. When configured, rejection + * prevents the corresponding provider request from being dispatched. + */ + recordProviderRequestCapture?: ( + capture: ProviderRequestCaptureRecord, + ) => Promise<{ artifactId: string }>; + /** Best-effort durable row for one physical provider request attempt. */ + recordProviderRequestAttempt?: (attempt: ProviderRequestAttemptRecord) => void | Promise; /** * Optional artifact recorder. Runtime derives only deterministic candidates * from structured tool results / explicit redirects; desktop main owns @@ -866,6 +880,18 @@ export class AiSdkBackend implements AgentBackend { }); this.currentRunTrace = trace; trace.turnStarted(); + const recordProviderRequestCapture = this.input.recordProviderRequestCapture; + const providerRequestTraceId = recordProviderRequestCapture ? this.newId() : undefined; + const providerRequestTracker = providerRequestTraceId + ? new ProviderRequestTracker({ + traceId: providerRequestTraceId, + turnId, + now: this.now, + newId: this.newId, + persistCapture: recordProviderRequestCapture!, + recordAttempt: this.input.recordProviderRequestAttempt ?? (() => {}), + }) + : undefined; // --- Resolve model (API key already attached at construct time) --- let model: unknown; @@ -1269,6 +1295,7 @@ export class AiSdkBackend implements AgentBackend { // boundary even when no shaping hook is configured: a transient // transport retry can resend it without replaying completed tools. attemptObservedSteps = options.steps; + providerRequestTracker?.setStep(attemptStepBase + options.stepNumber); // Step boundary: lease the caller's queued steering, echo each as a // user event, ack only after it is durably persisted AND in the // injection set (nack on any failure so the queue reclaims it). @@ -1349,6 +1376,7 @@ export class AiSdkBackend implements AgentBackend { abortSignal: this.abortController!.signal, stopAfterStep: () => this.stopAfterStepRequested, prepareStep: sendScopedPrepareStep, + ...(providerRequestTracker ? { providerRequestTracker } : {}), ...(remainingStepBudget !== undefined ? { maxSteps: remainingStepBudget } : {}), }); @@ -1621,6 +1649,7 @@ export class AiSdkBackend implements AgentBackend { requestShapeChangeReason: turnDiagnostics.requestShape.requestShapeChangeReason, promptSegments: turnDiagnostics.promptSegments, ...(contextBudgetForUsage ? { contextBudget: contextBudgetForUsage } : {}), + ...(providerRequestTraceId ? { providerRequestTraceId } : {}), }; await this.input.appendMessage(tu).catch(() => {}); if ( @@ -1695,6 +1724,7 @@ export class AiSdkBackend implements AgentBackend { ...(contextRemainingForUsage !== undefined ? { contextRemaining: contextRemainingForUsage } : {}), + ...(providerRequestTraceId ? { providerRequestTraceId } : {}), } satisfies TokenUsageEvent); } } catch { diff --git a/packages/runtime/src/ai-sdk-flow.ts b/packages/runtime/src/ai-sdk-flow.ts index 6c7f557db6..f1c54618a3 100644 --- a/packages/runtime/src/ai-sdk-flow.ts +++ b/packages/runtime/src/ai-sdk-flow.ts @@ -529,6 +529,9 @@ function mapBackendSessionEvent( ...(event.contextBudget !== undefined ? { contextBudget: event.contextBudget } : {}), }, }, + ...(event.providerRequestTraceId !== undefined + ? { refs: { providerRequestTraceId: event.providerRequestTraceId } } + : {}), }; // ── Error ───────────────────────────────────────────────────────────── diff --git a/packages/runtime/src/index.ts b/packages/runtime/src/index.ts index 5abe1a1691..d4ddc8f9c9 100644 --- a/packages/runtime/src/index.ts +++ b/packages/runtime/src/index.ts @@ -109,6 +109,20 @@ export type { } from './filesystem-worker/index.js'; export { AiSdkBackend } from './ai-sdk-backend.js'; +export { findFirstChangedCacheableSegment } from './request-shape.js'; +export { createProviderRequestCaptureRecorder } from './provider-request-telemetry.js'; +export type { + PreparedProviderRequestCapture, + PreparedRequestSegment, + PreparedRequestSegmentRef, +} from './request-shape.js'; +export type { + ProviderRequestAttemptRecord, + ProviderRequestCaptureLedgerRecord, + ProviderRequestCaptureRecord, + ProviderRequestCaptureRecorderInput, + ProviderRequestUsage, +} from './provider-request-telemetry.js'; export type { MakaTool, MakaToolContext } from './tool-runtime.js'; export { buildMcpTools, mcpProxyToolName } from './mcp-tools.js'; export type { McpToolProvider, BuildMcpToolsOptions } from './mcp-tools.js'; diff --git a/packages/runtime/src/model-adapter.ts b/packages/runtime/src/model-adapter.ts index c52ecc5474..4a158697ea 100644 --- a/packages/runtime/src/model-adapter.ts +++ b/packages/runtime/src/model-adapter.ts @@ -14,6 +14,7 @@ import type { ModelMessage } from 'ai'; import type { AsyncEventQueue } from './async-queue.js'; import { resolveModelRuntime } from './model-runtime.js'; import { classifyError, errorPresentationFromClass } from './provider-error-classification.js'; +import type { ProviderRequestTracker } from './provider-request-telemetry.js'; /** * Build an ai-sdk LanguageModel from a single input object. @@ -130,6 +131,8 @@ export interface ModelAdapterStreamInput { maxSteps?: number; /** Stop the SDK tool loop after the current provider step completes. */ stopAfterStep?: () => boolean; + /** Main-agent provider-call tracker. Auxiliary model calls intentionally omit it. */ + providerRequestTracker?: ProviderRequestTracker; } export interface ModelAdapterStreamCallbacks { @@ -147,6 +150,16 @@ export interface ModelAdapterStreamCallbacks { onThinkingSignature: (signature: string) => void; } +interface ProviderMiddlewareStreamInput { + doStream: () => PromiseLike<{ + stream: ReadableStream; + request?: unknown; + response?: unknown; + }>; + params: Record & { abortSignal?: AbortSignal }; + model: { provider: string; modelId: string }; +} + export class ModelAdapter { constructor(private readonly input: ModelAdapterInput) {} @@ -176,10 +189,11 @@ export class ModelAdapter { `Failed to load 'ai' package. Run \`npm install ai\`. Inner: ${(err as Error).message}`, ); }); - const { streamText, isStepCount, isLoopFinished } = ai as unknown as { + const { streamText, isStepCount, isLoopFinished, wrapLanguageModel } = ai as unknown as { streamText: (opts: Record) => StreamTextResult; isStepCount: (n: number) => unknown; isLoopFinished: () => unknown; + wrapLanguageModel: (input: Record) => unknown; }; const maxSteps = input.maxSteps ?? this.input.maxSteps; @@ -190,8 +204,23 @@ export class ModelAdapter { ); const configuredStop = maxSteps === undefined ? isLoopFinished() : isStepCount(maxSteps); const stopAfterStep = input.stopAfterStep; + const trackedModel = input.providerRequestTracker + ? wrapLanguageModel({ + model: input.model, + middleware: { + wrapStream: async ({ doStream, params, model }: ProviderMiddlewareStreamInput) => + await input.providerRequestTracker!.trackStream({ + providerId: model.provider, + modelId: model.modelId, + params, + abortSignal: input.abortSignal, + doStream, + }), + }, + }) + : input.model; return streamText({ - model: input.model, + model: trackedModel, messages: input.messages, tools: input.tools, activeTools: input.activeTools, diff --git a/packages/runtime/src/provider-request-telemetry.ts b/packages/runtime/src/provider-request-telemetry.ts new file mode 100644 index 0000000000..6ba40e8c97 --- /dev/null +++ b/packages/runtime/src/provider-request-telemetry.ts @@ -0,0 +1,516 @@ +import { + capturePreparedProviderRequest, + type PreparedProviderRequestCapture, + type PreparedRequestSegment, +} from './request-shape.js'; + +export type ProviderRequestCacheValueSource = 'provider' | 'derived'; + +export interface ProviderRequestUsage { + inputTokens?: number; + cacheReadInputTokens?: number; + cacheReadInputSource?: ProviderRequestCacheValueSource; + cacheMissInputTokens?: number; + cacheMissInputSource?: ProviderRequestCacheValueSource; + cacheWriteInputTokens?: number; + cacheWriteInputSource?: ProviderRequestCacheValueSource; + outputTokens?: number; + reasoningTokens?: number; +} + +export interface ProviderRequestUsageLike { + inputTokens?: + | number + | { total?: number; noCache?: number; cacheRead?: number; cacheWrite?: number }; + outputTokens?: number | { total?: number; text?: number; reasoning?: number }; + raw?: Record; +} + +export type ProviderRequestAttemptStatus = 'completed' | 'failed' | 'interrupted' | 'aborted'; + +export interface ProviderRequestCaptureRecord extends PreparedProviderRequestCapture { + traceId: string; + captureId: string; + turnId: string; + step: number; + providerId: string; + modelId: string; +} + +export interface ProviderRequestCaptureRef { + captureId: string; + artifactId: string; +} + +export type ProviderRequestCaptureLedgerRecord = Omit< + ProviderRequestCaptureRecord, + 'serializedRequest' +> & { + artifactId: string; +}; + +export interface ProviderRequestAttemptRecord extends ProviderRequestUsage { + traceId: string; + attemptId: string; + turnId: string; + step: number; + attempt: number; + captureId: string; + captureArtifactId: string; + providerId: string; + modelId: string; + requestHash: string; + requestBytes: number; + segments: PreparedRequestSegment[]; + startedAt: number; + completedAt: number; + status: ProviderRequestAttemptStatus; + finishReason?: string; + latencyMs: number; + timeToFirstTokenMs?: number; +} + +export interface ProviderRequestTrackerInput { + traceId: string; + turnId: string; + now: () => number; + newId: () => string; + persistCapture: ( + capture: ProviderRequestCaptureRecord, + ) => Promise>; + recordAttempt: (attempt: ProviderRequestAttemptRecord) => void | Promise; +} + +export interface ProviderRequestCaptureRecorderInput { + persistArtifact: ( + capture: ProviderRequestCaptureRecord, + ) => Promise>; + recordLedger: (capture: ProviderRequestCaptureLedgerRecord) => Promise; +} + +export interface TrackProviderStreamInput { + providerId: string; + modelId: string; + params: Record; + abortSignal?: AbortSignal; + doStream: () => PromiseLike; +} + +export function createProviderRequestCaptureRecorder( + input: ProviderRequestCaptureRecorderInput, +): ( + capture: ProviderRequestCaptureRecord, +) => Promise> { + return async (capture) => { + const artifact = await input.persistArtifact(capture); + const { serializedRequest: _serializedRequest, ...metadata } = capture; + await input.recordLedger({ ...metadata, artifactId: artifact.artifactId }); + return artifact; + }; +} + +export interface ProviderStreamResult { + stream: ReadableStream; + request?: unknown; + response?: unknown; +} + +interface StoredCapture { + capture: ProviderRequestCaptureRecord; + ref: ProviderRequestCaptureRef; +} + +export class ProviderRequestTracker { + private step = 0; + private readonly attemptsByStep = new Map(); + private readonly captures = new Map>(); + + constructor(private readonly input: ProviderRequestTrackerInput) {} + + get traceId(): string { + return this.input.traceId; + } + + setStep(step: number): void { + this.step = step; + } + + async trackStream(input: TrackProviderStreamInput): Promise { + throwIfAbortedBeforeDispatch(input.abortSignal); + const step = this.step; + const capture = await this.capture(step, input); + throwIfAbortedBeforeDispatch(input.abortSignal); + const attempt = (this.attemptsByStep.get(step) ?? 0) + 1; + this.attemptsByStep.set(step, attempt); + const attemptId = this.input.newId(); + const startedAt = this.input.now(); + let sawOutput = false; + let timeToFirstTokenMs: number | undefined; + let finished = false; + let abortListener: (() => void) | undefined; + + const finalize = async ( + status: ProviderRequestAttemptStatus, + finish?: { reason?: string; usage?: ProviderRequestUsageLike }, + ): Promise => { + if (finished) return; + finished = true; + if (abortListener) input.abortSignal?.removeEventListener('abort', abortListener); + const completedAt = this.input.now(); + const usage = strictProviderRequestUsage(finish?.usage); + const record: ProviderRequestAttemptRecord = { + traceId: this.input.traceId, + attemptId, + turnId: this.input.turnId, + step, + attempt, + captureId: capture.ref.captureId, + captureArtifactId: capture.ref.artifactId, + providerId: input.providerId, + modelId: input.modelId, + requestHash: capture.capture.requestHash, + requestBytes: capture.capture.requestBytes, + segments: capture.capture.segments, + startedAt, + completedAt, + status, + ...(finish?.reason !== undefined ? { finishReason: finish.reason } : {}), + latencyMs: Math.max(0, completedAt - startedAt), + ...(timeToFirstTokenMs !== undefined ? { timeToFirstTokenMs } : {}), + ...(usage ?? {}), + }; + try { + await this.input.recordAttempt(record); + } catch { + // Attempt telemetry is diagnostic. The provider outcome remains authoritative. + } + }; + + if (input.abortSignal) { + abortListener = () => { + void finalize('aborted'); + }; + if (input.abortSignal.aborted) await finalize('aborted'); + else input.abortSignal.addEventListener('abort', abortListener, { once: true }); + } + + let result: ProviderStreamResult; + try { + result = await input.doStream(); + } catch (error) { + await finalize(abortStatus(input.abortSignal, error)); + throw error; + } + + const reader = result.stream.getReader(); + const stream = new ReadableStream({ + pull: async (controller) => { + try { + const next = await reader.read(); + if (next.done) { + await finalize(input.abortSignal?.aborted ? 'aborted' : 'interrupted'); + controller.close(); + return; + } + const part = asRecord(next.value); + if (part && isOutputPart(part.type)) { + sawOutput = true; + if (timeToFirstTokenMs === undefined) { + timeToFirstTokenMs = Math.max(0, this.input.now() - startedAt); + } + } + if (part?.type === 'finish') { + await finalize(input.abortSignal?.aborted ? 'aborted' : 'completed', { + reason: finishReason(part.finishReason), + usage: asUsage(part.usage), + }); + } else if (part?.type === 'error') { + await finalize( + input.abortSignal?.aborted ? 'aborted' : sawOutput ? 'interrupted' : 'failed', + ); + } + controller.enqueue(next.value); + } catch (error) { + await finalize( + input.abortSignal?.aborted + ? 'aborted' + : sawOutput + ? 'interrupted' + : abortStatus(input.abortSignal, error), + ); + controller.error(error); + } + }, + cancel: async (reason) => { + try { + await reader.cancel(reason); + } finally { + await finalize(input.abortSignal?.aborted ? 'aborted' : 'interrupted'); + } + }, + }); + return { ...result, stream }; + } + + private async capture(step: number, input: TrackProviderStreamInput): Promise { + const prepared = preparedCapture(input.providerId, input.modelId, input.params); + const key = `${step}:${prepared.requestHash}`; + const existing = this.captures.get(key); + if (existing) return await existing; + + const pending = (async (): Promise => { + const captureId = this.input.newId(); + const capture: ProviderRequestCaptureRecord = { + ...prepared, + traceId: this.input.traceId, + captureId, + turnId: this.input.turnId, + step, + providerId: input.providerId, + modelId: input.modelId, + }; + const persisted = await this.input.persistCapture(capture); + return { capture, ref: { captureId, artifactId: persisted.artifactId } }; + })(); + this.captures.set(key, pending); + try { + return await pending; + } catch (error) { + this.captures.delete(key); + throw error; + } + } +} + +function throwIfAbortedBeforeDispatch(signal: AbortSignal | undefined): void { + if (signal?.aborted) { + throw new DOMException('The provider request was cancelled before dispatch', 'AbortError'); + } +} + +function preparedCapture( + providerId: string, + modelId: string, + params: Record, +): PreparedProviderRequestCapture { + const prompt = Array.isArray(params.prompt) ? params.prompt : []; + const instructions: unknown[] = []; + const messages: unknown[] = []; + for (const item of prompt) { + const record = asRecord(item); + if (record?.role === 'system') instructions.push(record.content); + else messages.push(item); + } + const tools = Array.isArray(params.tools) ? params.tools : []; + const providerOptions = asRecord(params.providerOptions); + return capturePreparedProviderRequest({ + providerId, + modelId, + instructions, + messages, + tools, + ...(providerOptions ? { providerOptions } : {}), + requestPayload: secretFreeParams(params), + }); +} + +function secretFreeParams(params: Record): Record { + const { abortSignal: _abortSignal, headers: _headers, ...safe } = params; + return safe; +} + +function abortStatus(signal: AbortSignal | undefined, error: unknown): 'failed' | 'aborted' { + if (signal?.aborted) return 'aborted'; + return error instanceof Error && error.name === 'AbortError' ? 'aborted' : 'failed'; +} + +function finishReason(value: unknown): string | undefined { + if (typeof value === 'string') return value; + const reason = asRecord(value); + if (typeof reason?.raw === 'string') return reason.raw; + return typeof reason?.unified === 'string' ? reason.unified : undefined; +} + +function isOutputPart(type: unknown): boolean { + return ( + typeof type === 'string' && + ![ + 'stream-start', + 'response-metadata', + 'raw', + 'finish', + 'error', + 'text-start', + 'text-end', + 'reasoning-start', + 'reasoning-end', + 'tool-input-start', + 'tool-input-end', + ].includes(type) + ); +} + +function asUsage(value: unknown): ProviderRequestUsageLike | undefined { + return asRecord(value) as ProviderRequestUsageLike | undefined; +} + +function asRecord(value: unknown): Record | undefined { + return value !== null && typeof value === 'object' + ? (value as Record) + : undefined; +} + +/** + * Extract provider-request usage without inheriting adapter-filled zeroes. + * Cache evidence is read from the raw provider payload; only cache miss may be + * derived, and only when the total plus every cache component needed for the + * subtraction was explicitly reported. + */ +export function strictProviderRequestUsage( + usage: ProviderRequestUsageLike | undefined, +): ProviderRequestUsage | undefined { + if (!usage) return undefined; + const raw = usage.raw; + const normalizedInputTokens = tokenTotal(usage.inputTokens); + const normalizedOutputTokens = tokenTotal(usage.outputTokens); + const inputTokens = canUseNormalizedTotal(raw, [ + 'prompt_tokens', + 'input_tokens', + 'promptTokenCount', + ]) + ? normalizedInputTokens + : undefined; + const outputTokens = canUseNormalizedTotal(raw, [ + 'completion_tokens', + 'output_tokens', + 'candidatesTokenCount', + ]) + ? normalizedOutputTokens + : undefined; + const result: ProviderRequestUsage = { + ...(inputTokens !== undefined ? { inputTokens } : {}), + ...(outputTokens !== undefined ? { outputTokens } : {}), + }; + + if (raw) { + const normalizedCacheMiss = + typeof usage.inputTokens === 'object' && usage.inputTokens !== null + ? finiteToken(usage.inputTokens.noCache) + : undefined; + applyAnthropicCacheUsage(result, raw, normalizedCacheMiss); + applyOpenAiCacheUsage(result, raw); + applyGoogleCacheUsage(result, raw); + const reasoningTokens = firstToken( + nestedToken(raw, 'completion_tokens_details', 'reasoning_tokens'), + nestedToken(raw, 'output_tokens_details', 'reasoning_tokens'), + ownToken(raw, 'thoughtsTokenCount'), + ); + if (reasoningTokens !== undefined) result.reasoningTokens = reasoningTokens; + } + + return Object.keys(result).length > 0 ? result : undefined; +} + +function applyGoogleCacheUsage(result: ProviderRequestUsage, raw: Record): void { + const totalInput = ownToken(raw, 'promptTokenCount'); + const cacheRead = ownToken(raw, 'cachedContentTokenCount'); + if (cacheRead === undefined) return; + result.cacheReadInputTokens = cacheRead; + result.cacheReadInputSource = 'provider'; + if (totalInput === undefined || cacheRead > totalInput) return; + result.cacheMissInputTokens = totalInput - cacheRead; + result.cacheMissInputSource = 'derived'; +} + +function applyAnthropicCacheUsage( + result: ProviderRequestUsage, + raw: Record, + normalizedCacheMiss: number | undefined, +): void { + const cacheRead = ownToken(raw, 'cache_read_input_tokens'); + const cacheWrite = ownToken(raw, 'cache_creation_input_tokens'); + if (cacheRead !== undefined) { + result.cacheReadInputTokens = cacheRead; + result.cacheReadInputSource = 'provider'; + } + if (cacheWrite !== undefined) { + result.cacheWriteInputTokens = cacheWrite; + result.cacheWriteInputSource = 'provider'; + } + // Anthropic defines input_tokens as the non-cached input component. Treat it + // as cache-miss evidence only when this is recognizably an Anthropic cache + // usage object, rather than an OpenAI Responses usage object with the same + // top-level input_tokens spelling. + if (cacheRead !== undefined || cacheWrite !== undefined) { + const rawCacheMiss = ownToken(raw, 'input_tokens'); + if (rawCacheMiss !== undefined) { + result.cacheMissInputTokens = normalizedCacheMiss ?? rawCacheMiss; + result.cacheMissInputSource = 'provider'; + } + } +} + +function applyOpenAiCacheUsage(result: ProviderRequestUsage, raw: Record): void { + const promptInput = ownToken(raw, 'prompt_tokens'); + const responsesInput = ownToken(raw, 'input_tokens'); + const cacheRead = firstToken( + nestedToken(raw, 'prompt_tokens_details', 'cached_tokens'), + nestedToken(raw, 'input_tokens_details', 'cached_tokens'), + ); + const cacheWrite = firstToken( + nestedToken(raw, 'prompt_tokens_details', 'cache_write_tokens'), + nestedToken(raw, 'input_tokens_details', 'cache_write_tokens'), + ); + if (cacheRead === undefined && cacheWrite === undefined) return; + + if (cacheRead !== undefined) { + result.cacheReadInputTokens = cacheRead; + result.cacheReadInputSource = 'provider'; + } + if (cacheWrite !== undefined) { + result.cacheWriteInputTokens = cacheWrite; + result.cacheWriteInputSource = 'provider'; + } + const totalInput = promptInput ?? responsesInput; + if (totalInput === undefined || cacheRead === undefined) return; + const accountedInput = cacheRead + (cacheWrite ?? 0); + if (accountedInput > totalInput) return; + result.cacheMissInputTokens = totalInput - accountedInput; + result.cacheMissInputSource = 'derived'; +} + +function canUseNormalizedTotal( + raw: Record | undefined, + keys: readonly string[], +): boolean { + return raw === undefined || keys.some((key) => ownToken(raw, key) !== undefined); +} + +function tokenTotal( + value: ProviderRequestUsageLike['inputTokens'] | ProviderRequestUsageLike['outputTokens'], +): number | undefined { + return finiteToken(typeof value === 'object' && value !== null ? value.total : value); +} + +function ownToken(value: Record, key: string): number | undefined { + return Object.hasOwn(value, key) ? finiteToken(value[key]) : undefined; +} + +function nestedToken( + value: Record, + key: string, + nestedKey: string, +): number | undefined { + if (!Object.hasOwn(value, key)) return undefined; + const nested = value[key]; + if (!nested || typeof nested !== 'object' || !Object.hasOwn(nested, nestedKey)) return undefined; + return finiteToken((nested as Record)[nestedKey]); +} + +function firstToken(...values: Array): number | undefined { + return values.find((value) => value !== undefined); +} + +function finiteToken(value: unknown): number | undefined { + return typeof value === 'number' && Number.isFinite(value) && value >= 0 ? value : undefined; +} diff --git a/packages/runtime/src/request-shape.ts b/packages/runtime/src/request-shape.ts index 35d97e5056..70c303d5fd 100644 --- a/packages/runtime/src/request-shape.ts +++ b/packages/runtime/src/request-shape.ts @@ -49,6 +49,42 @@ export interface RequestShapeDiagnostic { toolAvailability?: ToolAvailabilityDiagnostic; } +export type PreparedRequestSegmentKind = + | 'tool_schema' + | 'system_prompt' + | 'message' + | 'provider_options'; + +export interface PreparedRequestSegment { + kind: PreparedRequestSegmentKind; + index: number; + cacheable: boolean; + hash: string; + bytes: number; + role?: string; +} + +export interface PreparedProviderRequestInput { + providerId: string; + modelId: string; + instructions?: unknown; + messages: readonly unknown[]; + tools?: readonly unknown[]; + providerOptions?: Record; + /** Exact secret-free model-call parameters captured at the provider seam. */ + requestPayload?: unknown; +} + +export interface PreparedProviderRequestCapture { + schemaVersion: 1; + requestHash: string; + requestBytes: number; + serializedRequest: string; + segments: PreparedRequestSegment[]; +} + +export type PreparedRequestSegmentRef = Pick; + /** * Split the registry into the full dispatch set (`providerTools`) and the * model-visible subset (`activeTools`). @@ -138,6 +174,89 @@ export function toolSchemaCharsForDiagnostics( }).length; } +/** + * Capture the standardized request immediately before the provider call. + * + * Segment order follows the stable Maka request-prefix model used for cache + * diagnostics: tools, system instructions, then conversation messages. + * Provider options are retained for exact replay evidence, but are not claimed + * to be a provider-cacheable prefix segment. + */ +export function capturePreparedProviderRequest( + input: PreparedProviderRequestInput, +): PreparedProviderRequestCapture { + const payload = input.requestPayload ?? { + instructions: input.instructions, + messages: input.messages, + tools: input.tools ?? [], + providerOptions: input.providerOptions ?? {}, + }; + // This is the evidence body, not the hash canonicalizer: preserve the exact + // JSON ordering and values presented at the model-call seam. + const serializedRequest = JSON.stringify(payload); + const segments: PreparedRequestSegment[] = []; + + for (const [index, tool] of (input.tools ?? []).entries()) { + segments.push(preparedSegment('tool_schema', index, tool, true)); + } + if (input.instructions !== undefined) { + const instructions = Array.isArray(input.instructions) + ? input.instructions + : [input.instructions]; + for (const [index, instruction] of instructions.entries()) { + segments.push(preparedSegment('system_prompt', index, instruction, true)); + } + } + for (const [index, message] of input.messages.entries()) { + const role = + isObjectLike(message) && typeof message.role === 'string' ? message.role : undefined; + segments.push(preparedSegment('message', index, message, true, role)); + } + if (input.providerOptions !== undefined) { + segments.push(preparedSegment('provider_options', 0, input.providerOptions, false)); + } + + return { + schemaVersion: 1, + requestHash: stableHash({ + providerId: input.providerId, + modelId: input.modelId, + payload, + }), + requestBytes: Buffer.byteLength(serializedRequest, 'utf8'), + serializedRequest, + segments, + }; +} + +export function findFirstChangedCacheableSegment( + current: Pick, + prior: Pick, +): PreparedRequestSegmentRef | undefined { + const currentSegments = current.segments.filter((segment) => segment.cacheable); + const priorSegments = prior.segments.filter((segment) => segment.cacheable); + const segmentCount = Math.max(currentSegments.length, priorSegments.length); + for (let position = 0; position < segmentCount; position += 1) { + const currentSegment = currentSegments[position]; + const priorSegment = priorSegments[position]; + if ( + currentSegment?.kind === priorSegment?.kind && + currentSegment?.index === priorSegment?.index && + currentSegment?.hash === priorSegment?.hash + ) { + continue; + } + const changed = currentSegment ?? priorSegment; + if (!changed) return undefined; + return { + kind: changed.kind, + index: changed.index, + ...(changed.role !== undefined ? { role: changed.role } : {}), + }; + } + return undefined; +} + /** The provider-visible tools — the active subset actually serialized on the wire. */ function providerVisibleTools( providerTools: readonly MakaTool[], @@ -147,6 +266,24 @@ function providerVisibleTools( return providerTools.filter((tool) => active.has(tool.name)); } +function preparedSegment( + kind: PreparedRequestSegmentKind, + index: number, + value: unknown, + cacheable: boolean, + role?: string, +): PreparedRequestSegment { + const serialized = stableStringify(value); + return { + kind, + index, + cacheable, + hash: stableHash(value), + bytes: Buffer.byteLength(serialized, 'utf8'), + ...(role !== undefined ? { role } : {}), + }; +} + export function stableHash(value: unknown): string { return `sha256:${createHash('sha256').update(stableStringify(value)).digest('hex')}`; } diff --git a/packages/runtime/src/runtime-kernel.ts b/packages/runtime/src/runtime-kernel.ts index 95261389e6..f90417518e 100644 --- a/packages/runtime/src/runtime-kernel.ts +++ b/packages/runtime/src/runtime-kernel.ts @@ -1345,6 +1345,20 @@ export class RuntimeKernel implements RuntimeKernelLike { }, ...(this.deps.runStore ? { + recordProviderRequestCapture: (capture) => { + const active = this.active.get(sessionId); + const runId = active?.turnToRunId.get(capture.turnId); + const run = runId ? active?.activeRuns.get(runId) : undefined; + if (!run) + return Promise.reject(new Error('No active AgentRun for provider request capture')); + return run.recordProviderRequestCapture(capture); + }, + recordProviderRequestAttempt: (attempt) => { + const active = this.active.get(sessionId); + const runId = active?.turnToRunId.get(attempt.turnId); + const run = runId ? active?.activeRuns.get(runId) : undefined; + run?.recordProviderRequestAttempt(attempt); + }, loadHistoryCompactCheckpoint: () => this.loadHistoryCompactCheckpoint(sessionId), recordHistoryCompactCheckpoint: ( checkpoint: HistoryCompactCheckpoint, @@ -1421,6 +1435,20 @@ export class RuntimeKernel implements RuntimeKernelLike { }, ...(this.deps.runStore ? { + recordProviderRequestCapture: (capture) => { + const active = this.childActive.get(activeKey); + const runId = active?.turnToRunId.get(capture.turnId); + const run = runId ? active?.activeRuns.get(runId) : undefined; + if (!run) + return Promise.reject(new Error('No active AgentRun for provider request capture')); + return run.recordProviderRequestCapture(capture); + }, + recordProviderRequestAttempt: (attempt) => { + const active = this.childActive.get(activeKey); + const runId = active?.turnToRunId.get(attempt.turnId); + const run = runId ? active?.activeRuns.get(runId) : undefined; + run?.recordProviderRequestAttempt(attempt); + }, loadHistoryCompactCheckpoint: () => this.loadHistoryCompactCheckpoint(sessionId), recordHistoryCompactCheckpoint: ( checkpoint: HistoryCompactCheckpoint, diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index 99ef343881..8e5fc9b1b3 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -80,6 +80,10 @@ import { import type { AgentBackend, BackendStopMode } from '@maka/core/backend-types'; import type { AgentTeamExecutionContext, MakaTool } from './tool-runtime.js'; import type { RunTraceRecorder } from './run-trace.js'; +import type { + ProviderRequestAttemptRecord, + ProviderRequestCaptureLedgerRecord, +} from './provider-request-telemetry.js'; import type { ShellRunProcessManager } from './shell-run-manager.js'; import type { ActiveFullCompactBlock } from './active-full-compact.js'; import type { SemanticCompactBlock } from './semantic-compact.js'; @@ -254,6 +258,10 @@ export interface BackendFactoryContext { /** Trusted child expert-team identity. Main-session factories leave this undefined. */ agentTeam?: AgentTeamExecutionContext; recordRunTrace?: RunTraceRecorder; + /** Durable AgentRun metadata row written after the private capture artifact. */ + recordProviderRequestCapture?: (capture: ProviderRequestCaptureLedgerRecord) => Promise; + /** Best-effort AgentRun row for one physical provider call. */ + recordProviderRequestAttempt?: (attempt: ProviderRequestAttemptRecord) => void; loadHistoryCompactCheckpoint?: () => Promise; recordHistoryCompactCheckpoint?: ( checkpoint: HistoryCompactCheckpoint, diff --git a/packages/storage/src/__tests__/provider-request-capture-artifact.test.ts b/packages/storage/src/__tests__/provider-request-capture-artifact.test.ts new file mode 100644 index 0000000000..97fd34ede2 --- /dev/null +++ b/packages/storage/src/__tests__/provider-request-capture-artifact.test.ts @@ -0,0 +1,34 @@ +import { mkdtemp } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import assert from 'node:assert/strict'; + +import { createArtifactStore } from '../artifact-store.js'; +import * as storage from '../index.js'; + +test('persists the exact prepared request as a private artifact', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-provider-capture-')); + const store = createArtifactStore(root); + const persist = Reflect.get(storage, 'persistProviderRequestCaptureArtifact') as unknown as + | (( + store: ReturnType, + input: Record, + ) => Promise<{ id: string; source?: string; sizeBytes: number }>) + | undefined; + assert.equal(typeof persist, 'function'); + const serializedRequest = '{"messages":[{"role":"user","content":"exact"}]}'; + + const artifact = await persist!(store, { + sessionId: 'session-1', + turnId: 'turn-1', + captureId: 'capture-1', + step: 2, + serializedRequest, + now: 1, + }); + + assert.equal(artifact.source, 'provider_request_capture'); + assert.equal(artifact.sizeBytes, Buffer.byteLength(serializedRequest)); + assert.deepEqual(await store.readText(artifact.id), { ok: true, text: serializedRequest }); +}); diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index 64156e6a87..5eaea85667 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -20,6 +20,7 @@ export * from './settings-store.js'; export * from './telemetry-repo.js'; export * from './artifact-store.js'; export * from './artifact-attachments.js'; +export * from './provider-request-capture-artifact.js'; export * from './plan-reminder-store.js'; export * from './task-ledger-store.js'; export * from './foreign-session-store.js'; diff --git a/packages/storage/src/provider-request-capture-artifact.ts b/packages/storage/src/provider-request-capture-artifact.ts new file mode 100644 index 0000000000..67f7985d92 --- /dev/null +++ b/packages/storage/src/provider-request-capture-artifact.ts @@ -0,0 +1,29 @@ +import type { ArtifactRecord } from '@maka/core'; + +import type { ArtifactStore } from './artifact-store.js'; + +export interface PersistProviderRequestCaptureArtifactInput { + sessionId: string; + turnId: string; + captureId: string; + step: number; + serializedRequest: string; + now?: number; +} + +export function persistProviderRequestCaptureArtifact( + store: ArtifactStore, + input: PersistProviderRequestCaptureArtifactInput, +): Promise { + return store.create({ + sessionId: input.sessionId, + turnId: input.turnId, + name: `provider-request-step-${input.step}-${input.captureId}.json`, + kind: 'file', + content: input.serializedRequest, + mimeType: 'application/json', + source: 'provider_request_capture', + summary: `Prepared provider request for step ${input.step}`, + ...(input.now !== undefined ? { now: input.now } : {}), + }); +}