diff --git a/packages/headless/src/__tests__/harbor-cell.test.ts b/packages/headless/src/__tests__/harbor-cell.test.ts index 2edbf706f5..ece43588dd 100644 --- a/packages/headless/src/__tests__/harbor-cell.test.ts +++ b/packages/headless/src/__tests__/harbor-cell.test.ts @@ -3124,7 +3124,7 @@ describe('runHarborCell', () => { assert.ok(backendInput.recordProviderRequestCapture); await backendInput.loadSynthesisCache({ sessionId: 'session-1' }); const capture: ProviderRequestCaptureRecord = { - schemaVersion: 1, + schemaVersion: 2, traceId: 'trace-1', captureId: 'capture-1', turnId: 'turn-1', @@ -3132,6 +3132,7 @@ describe('runHarborCell', () => { providerId: 'openai', modelId: 'gpt-4o-mini', requestHash: 'sha256:request', + requestPayloadWithoutProviderOptionsHash: 'sha256:shared-request', requestBytes: 18, segments: [], serializedRequest: '{"prompt":"hello"}', diff --git a/packages/headless/src/__tests__/provider-request-trace.test.ts b/packages/headless/src/__tests__/provider-request-trace.test.ts index 511cbe2b9f..aab1751628 100644 --- a/packages/headless/src/__tests__/provider-request-trace.test.ts +++ b/packages/headless/src/__tests__/provider-request-trace.test.ts @@ -1,9 +1,13 @@ -import { mkdtemp, writeFile } from 'node:fs/promises'; +import { mkdir, mkdtemp, readFile, rm, 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 type { AgentRunEvent, AgentRunHeader } from '@maka/core'; +import type { InvocationResult } from '@maka/runtime'; +import { writeHarborTaskRunTrace } from '../harbor-cell.js'; +import { openHeadlessStorageForWrite } from '../headless-storage.js'; import * as traceAnalysis from '../provider-request-trace.js'; test('derives the first changed cacheable segment from the existing AgentRun trace', async () => { @@ -62,7 +66,16 @@ test('derives the first changed cacheable segment from the existing AgentRun tra const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); assert.equal(result.traceId, 'provider-trace-1'); + assert.deepEqual(result.identity, { + runId: 'run-1', + sessionId: 'session-1', + turnId: 'turn-1', + }); assert.equal(result.captures.length, 2); + assert.equal(result.captures[0]?.schemaVersion, 1); + assert.equal(result.captures[0]?.requestPayloadWithoutProviderOptionsHash, undefined); + assert.deepEqual(result.attempts, []); + assert.deepEqual(result.diagnostics, []); assert.deepEqual(result.captures[1]?.firstChangedCacheableSegment, { kind: 'message', index: 1, @@ -103,4 +116,934 @@ test('keeps complete provider captures when the AgentRun trace ends with a torn result.captures.map((capture) => capture.captureId), ['capture-1'], ); + assert.deepEqual( + result.diagnostics.map((diagnostic) => diagnostic.code), + ['invalid_json'], + ); + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /line 2.*valid JSON/i, + ); }); + +test('exports the Run-header trace failure latch when its sentinel event is missing', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-provider-trace-export-')); + const storageRoot = join(root, 'storage'); + const outputDir = join(root, 'output'); + const identity = { runId: 'run-1', sessionId: 'session-1', turnId: 'turn-1' }; + const header: AgentRunHeader = { + ...identity, + status: 'completed', + backendKind: 'ai-sdk', + llmConnectionSlug: 'kimi-coding-plan', + modelId: 'k3', + cwd: root, + permissionMode: 'execute', + createdAt: 1, + updatedAt: 4, + completedAt: 4, + }; + const invocation: InvocationResult = { + invocationId: 'invocation-1', + ...identity, + status: 'completed', + events: [], + startedAt: 1, + finishedAt: 4, + }; + + try { + await mkdir(outputDir, { recursive: true }); + const storage = await openHeadlessStorageForWrite(storageRoot); + const runStore = storage.executionStores.agentRunStore; + await runStore.createRun(header); + await runStore.appendEvent(identity.sessionId, identity.runId, { + type: 'provider_request_captured', + id: 'capture-event-1', + ...identity, + ts: 2, + data: { + schemaVersion: 2, + traceId: 'provider-trace-1', + captureId: 'capture-1', + artifactId: 'artifact-capture-1', + turnId: identity.turnId, + step: 0, + providerId: 'openai', + modelId: 'k3', + requestHash: 'sha256:request-1', + requestPayloadWithoutProviderOptionsHash: 'sha256:shared-request-1', + requestBytes: 100, + segments: [], + }, + }); + await runStore.appendEvent(identity.sessionId, identity.runId, { + type: 'provider_request_attempt_recorded', + id: 'attempt-event-1', + ...identity, + ts: 3, + data: { + traceId: 'provider-trace-1', + attemptId: 'attempt-1', + turnId: identity.turnId, + step: 0, + attempt: 1, + captureId: 'capture-1', + captureArtifactId: 'artifact-capture-1', + providerId: 'openai', + modelId: 'k3', + requestHash: 'sha256:request-1', + requestBytes: 100, + segments: [], + startedAt: 2, + completedAt: 3, + status: 'completed', + latencyMs: 1, + }, + }); + await runStore.updateRun(identity.sessionId, identity.runId, { + traceWriteError: 'append provider request attempt: disk full', + updatedAt: 4, + }); + + const traceEventsPath = await writeHarborTaskRunTrace({ + outputDir, + storage, + invocations: [invocation], + }); + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /trace write failed evidence/i, + ); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test('exports a torn AgentRun tail as incomplete provider-request evidence', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-provider-trace-export-')); + const storageRoot = join(root, 'storage'); + const outputDir = join(root, 'output'); + const runId = 'run-torn-export'; + const identity = { + runId, + sessionId: `session-${runId}`, + turnId: `turn-${runId}`, + }; + const header: AgentRunHeader = { + ...identity, + status: 'completed', + backendKind: 'ai-sdk', + llmConnectionSlug: 'kimi-coding-plan', + modelId: 'k3', + cwd: root, + permissionMode: 'execute', + createdAt: 1, + updatedAt: 4, + completedAt: 4, + }; + const invocation: InvocationResult = { + invocationId: 'invocation-torn-export', + ...identity, + status: 'completed', + events: [], + startedAt: 1, + finishedAt: 4, + }; + const { capture, attempt } = completeTraceRows(runId); + + try { + await mkdir(outputDir, { recursive: true }); + const storage = await openHeadlessStorageForWrite(storageRoot); + const runStore = storage.executionStores.agentRunStore; + await runStore.createRun(header); + await runStore.appendEvent(identity.sessionId, identity.runId, capture as AgentRunEvent); + await runStore.appendEvent(identity.sessionId, identity.runId, attempt as AgentRunEvent); + await writeFile( + join(storageRoot, 'sessions', identity.sessionId, 'runs', identity.runId, 'events.jsonl'), + '{"type":"provider_request_attempt_recorded"', + { flag: 'a' }, + ); + + const traceEventsPath = await writeHarborTaskRunTrace({ + outputDir, + storage, + invocations: [invocation], + }); + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual( + result.diagnostics.map((diagnostic) => diagnostic.code), + ['event_corrupt'], + ); + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /event corrupt evidence/i, + ); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test('also diagnoses missing provider evidence when the only run event is corrupt', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-provider-trace-export-')); + const storageRoot = join(root, 'storage'); + const outputDir = join(root, 'output'); + const identity = { + runId: 'run-only-corrupt', + sessionId: 'session-only-corrupt', + turnId: 'turn-only-corrupt', + }; + const invocation: InvocationResult = { + invocationId: 'invocation-only-corrupt', + ...identity, + status: 'completed', + events: [], + startedAt: 1, + finishedAt: 4, + }; + + try { + await mkdir(outputDir, { recursive: true }); + const storage = await openHeadlessStorageForWrite(storageRoot); + await storage.executionStores.agentRunStore.createRun({ + ...identity, + status: 'completed', + backendKind: 'ai-sdk', + llmConnectionSlug: 'kimi-coding-plan', + modelId: 'k3', + cwd: root, + permissionMode: 'execute', + createdAt: 1, + updatedAt: 4, + completedAt: 4, + }); + await writeFile( + join(storageRoot, 'sessions', identity.sessionId, 'runs', identity.runId, 'events.jsonl'), + '{"type":"tool_failed"', + ); + + const traceEventsPath = await writeHarborTaskRunTrace({ + outputDir, + storage, + invocations: [invocation], + }); + const exportedEvents = (await readFile(traceEventsPath, 'utf8')) + .trim() + .split('\n') + .map((line) => JSON.parse(line) as AgentRunEvent); + + assert.deepEqual( + exportedEvents.map((event) => event.type), + ['event_corrupt', 'event_corrupt'], + ); + assert.equal(exportedEvents[1]?.data?.reason, 'missing_provider_request_evidence'); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test('exports missing provider-request evidence for every continuation invocation', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-provider-trace-export-')); + const storageRoot = join(root, 'storage'); + const outputDir = join(root, 'output'); + const identities = [ + { runId: 'run-1', sessionId: 'session-1', turnId: 'turn-1' }, + { runId: 'run-2', sessionId: 'session-1', turnId: 'turn-2' }, + ]; + const invocations: InvocationResult[] = identities.map((identity, index) => ({ + invocationId: `invocation-${index + 1}`, + ...identity, + status: 'completed', + events: [], + startedAt: index * 10 + 1, + finishedAt: index * 10 + 4, + })); + + try { + await mkdir(outputDir, { recursive: true }); + const storage = await openHeadlessStorageForWrite(storageRoot); + const runStore = storage.executionStores.agentRunStore; + for (const [index, identity] of identities.entries()) { + const timestamp = index * 10 + 1; + await runStore.createRun({ + ...identity, + status: 'completed', + backendKind: 'ai-sdk', + llmConnectionSlug: 'kimi-coding-plan', + modelId: 'k3', + cwd: root, + permissionMode: 'execute', + createdAt: timestamp, + updatedAt: timestamp + 3, + completedAt: timestamp + 3, + }); + } + + const completeIdentity = identities[1]!; + const { capture, attempt } = completeTraceRows(completeIdentity.runId); + Object.assign(capture, completeIdentity); + Object.assign(attempt, completeIdentity); + capture.data.turnId = completeIdentity.turnId; + attempt.data.turnId = completeIdentity.turnId; + await runStore.appendEvent( + completeIdentity.sessionId, + completeIdentity.runId, + capture as AgentRunEvent, + ); + await runStore.appendEvent( + completeIdentity.sessionId, + completeIdentity.runId, + attempt as AgentRunEvent, + ); + + const traceEventsPath = await writeHarborTaskRunTrace({ + outputDir, + storage, + invocations, + }); + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual(result.identities, identities); + assert.throws( + () => + traceAnalysis.assertProviderRequestTraceComplete(result, { + expectedIdentity: completeIdentity, + }), + /event corrupt evidence/i, + ); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test('preserves existing run events when exporting a missing-evidence diagnostic', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-provider-trace-export-')); + const storageRoot = join(root, 'storage'); + const outputDir = join(root, 'output'); + const identity = { runId: 'run-tool-failed', sessionId: 'session-1', turnId: 'turn-1' }; + const invocation: InvocationResult = { + invocationId: 'invocation-tool-failed', + ...identity, + status: 'completed', + events: [], + startedAt: 1, + finishedAt: 4, + }; + + try { + await mkdir(outputDir, { recursive: true }); + const storage = await openHeadlessStorageForWrite(storageRoot); + const runStore = storage.executionStores.agentRunStore; + await runStore.createRun({ + ...identity, + status: 'completed', + backendKind: 'ai-sdk', + llmConnectionSlug: 'kimi-coding-plan', + modelId: 'k3', + cwd: root, + permissionMode: 'execute', + createdAt: 1, + updatedAt: 4, + completedAt: 4, + }); + await runStore.appendEvent(identity.sessionId, identity.runId, { + type: 'tool_failed', + id: 'tool-failed-1', + ...identity, + ts: 3, + message: 'tool failed before the provider trace was recorded', + }); + + const traceEventsPath = await writeHarborTaskRunTrace({ + outputDir, + storage, + invocations: [invocation], + }); + const exportedEvents = (await readFile(traceEventsPath, 'utf8')) + .trim() + .split('\n') + .map((line) => JSON.parse(line) as AgentRunEvent); + + assert.deepEqual( + exportedEvents.map((event) => event.type), + ['tool_failed', 'event_corrupt'], + ); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +for (const backendKind of ['fake', 'pi-agent'] as const) { + test(`does not diagnose missing provider-request evidence for ${backendKind} runs`, async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-provider-trace-export-')); + const storageRoot = join(root, 'storage'); + const outputDir = join(root, 'output'); + const identity = { + runId: `run-${backendKind}`, + sessionId: 'session-1', + turnId: 'turn-1', + }; + const invocation: InvocationResult = { + invocationId: `invocation-${backendKind}`, + ...identity, + status: 'completed', + events: [], + startedAt: 1, + finishedAt: 4, + }; + + try { + await mkdir(outputDir, { recursive: true }); + const storage = await openHeadlessStorageForWrite(storageRoot); + await storage.executionStores.agentRunStore.createRun({ + ...identity, + status: 'completed', + backendKind, + llmConnectionSlug: 'test-connection', + modelId: 'test-model', + cwd: root, + permissionMode: 'execute', + createdAt: 1, + updatedAt: 4, + completedAt: 4, + }); + + const traceEventsPath = await writeHarborTaskRunTrace({ + outputDir, + storage, + invocations: [invocation], + }); + + assert.equal(await readFile(traceEventsPath, 'utf8'), ''); + } finally { + await rm(root, { recursive: true, force: true }); + } + }); +} + +test('reads a v2 capture and attempt and binds them to the expected execution identity', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const identity = { runId: 'run-2', sessionId: 'session-2', turnId: 'turn-2' }; + const capture = { + type: 'provider_request_captured', + id: 'capture-event-1', + ...identity, + ts: 1, + data: { + schemaVersion: 2, + traceId: 'provider-trace-2', + captureId: 'capture-1', + artifactId: 'artifact-capture-1', + turnId: identity.turnId, + step: 0, + providerId: 'openai', + modelId: 'k3', + requestHash: 'sha256:request-1', + requestPayloadWithoutProviderOptionsHash: 'sha256:shared-request-1', + requestBytes: 100, + segments: [], + }, + }; + const attempt = { + type: 'provider_request_attempt_recorded', + id: 'attempt-event-1', + ...identity, + ts: 4, + data: { + traceId: 'provider-trace-2', + attemptId: 'attempt-1', + turnId: identity.turnId, + step: 0, + attempt: 1, + captureId: 'capture-1', + captureArtifactId: 'artifact-capture-1', + providerId: 'openai', + modelId: 'k3', + requestHash: 'sha256:request-1', + requestBytes: 100, + segments: [], + startedAt: 2, + completedAt: 4, + status: 'completed', + finishReason: 'stop', + latencyMs: 2, + inputTokens: 12, + outputTokens: 3, + }, + }; + await writeFile(traceEventsPath, `${JSON.stringify(capture)}\n${JSON.stringify(attempt)}\n`); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.equal(result.captures[0]?.schemaVersion, 2); + assert.equal( + result.captures[0]?.requestPayloadWithoutProviderOptionsHash, + 'sha256:shared-request-1', + ); + assert.equal(result.attempts.length, 1); + assert.doesNotThrow(() => + traceAnalysis.assertProviderRequestTraceComplete(result, { expectedIdentity: identity }), + ); + assert.throws( + () => + traceAnalysis.assertProviderRequestTraceComplete(result, { + expectedIdentity: { ...identity, runId: 'run-unrelated' }, + }), + /runId.*run-unrelated.*run-2/i, + ); +}); + +test('validates every request across same-session continuation executions', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-continuation-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const rows: Record[] = []; + const identities = [ + { runId: 'run-1', sessionId: 'session-1', turnId: 'turn-1' }, + { runId: 'run-2', sessionId: 'session-1', turnId: 'turn-2' }, + ]; + for (const [index, identity] of identities.entries()) { + const suffix = String(index + 1); + const traceId = `provider-trace-${suffix}`; + const captureId = `capture-${suffix}`; + const artifactId = `artifact-${suffix}`; + rows.push( + { + type: 'provider_request_captured', + id: `capture-event-${suffix}`, + ...identity, + ts: index * 10 + 1, + data: { + schemaVersion: 2, + traceId, + captureId, + artifactId, + turnId: identity.turnId, + step: 0, + providerId: 'openai', + modelId: 'k3', + requestHash: `sha256:request-${suffix}`, + requestPayloadWithoutProviderOptionsHash: `sha256:shared-request-${suffix}`, + requestBytes: 100, + segments: [ + { + kind: 'message', + index: 0, + role: 'user', + cacheable: true, + hash: `sha256:message-${suffix}`, + bytes: 10, + }, + ], + }, + }, + { + type: 'provider_request_attempt_recorded', + id: `attempt-event-${suffix}`, + ...identity, + ts: index * 10 + 3, + data: { + traceId, + attemptId: `attempt-${suffix}`, + turnId: identity.turnId, + step: 0, + attempt: 1, + captureId, + captureArtifactId: artifactId, + providerId: 'openai', + modelId: 'k3', + requestHash: `sha256:request-${suffix}`, + requestBytes: 100, + segments: [ + { + kind: 'message', + index: 0, + role: 'user', + cacheable: true, + hash: `sha256:message-${suffix}`, + bytes: 10, + }, + ], + startedAt: index * 10 + 2, + completedAt: index * 10 + 3, + status: 'completed', + latencyMs: 1, + }, + }, + ); + } + await writeFile(traceEventsPath, `${rows.map((row) => JSON.stringify(row)).join('\n')}\n`); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.equal(result.captures.length, 2); + assert.equal(result.captures[1]?.firstChangedCacheableSegment, undefined); + assert.doesNotThrow(() => + traceAnalysis.assertProviderRequestTraceComplete(result, { + expectedIdentity: identities[1], + }), + ); +}); + +test('reads autonomous retry traces across distinct session identities', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-retries-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const identities = [ + { runId: 'run-retry-1', sessionId: 'session-retry-1', turnId: 'turn-retry-1' }, + { runId: 'run-retry-2', sessionId: 'session-retry-2', turnId: 'turn-retry-2' }, + ]; + const rows = identities.flatMap((identity, index) => { + const suffix = String(index + 1); + const { capture, attempt } = completeTraceRows(identity.runId); + Object.assign(capture, identity, { id: `capture-event-retry-${suffix}` }); + Object.assign(attempt, identity, { id: `attempt-event-retry-${suffix}` }); + capture.data.captureId = `capture-retry-${suffix}`; + capture.data.artifactId = `artifact-retry-${suffix}`; + capture.data.turnId = identity.turnId; + attempt.data.attemptId = `attempt-retry-${suffix}`; + attempt.data.captureId = capture.data.captureId; + attempt.data.captureArtifactId = capture.data.artifactId; + attempt.data.turnId = identity.turnId; + return [capture, attempt]; + }); + await writeFile(traceEventsPath, `${rows.map((row) => JSON.stringify(row)).join('\n')}\n`); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual(result.identities, identities); + assert.deepEqual(result.diagnostics, []); + assert.doesNotThrow(() => + traceAnalysis.assertProviderRequestTraceComplete(result, { expectedIdentities: identities }), + ); + assert.throws( + () => + traceAnalysis.assertProviderRequestTraceComplete(result, { + expectedIdentities: [identities[1]!], + }), + /execution identity set.*run-retry-1/i, + ); +}); + +test('fails completeness when a later request attempt record is torn', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const identity = { runId: 'run-3', sessionId: 'session-3', turnId: 'turn-3' }; + const capture = { + type: 'provider_request_captured', + id: 'capture-event-1', + ...identity, + ts: 1, + data: { + schemaVersion: 2, + traceId: 'provider-trace-3', + captureId: 'capture-1', + artifactId: 'artifact-capture-1', + turnId: identity.turnId, + step: 0, + providerId: 'openai', + modelId: 'k3', + requestHash: 'sha256:request-1', + requestPayloadWithoutProviderOptionsHash: 'sha256:shared-request-1', + requestBytes: 100, + segments: [], + }, + }; + const firstAttempt = { + type: 'provider_request_attempt_recorded', + id: 'attempt-event-1', + ...identity, + ts: 3, + data: { + traceId: 'provider-trace-3', + attemptId: 'attempt-1', + turnId: identity.turnId, + step: 0, + attempt: 1, + captureId: 'capture-1', + captureArtifactId: 'artifact-capture-1', + providerId: 'openai', + modelId: 'k3', + requestHash: 'sha256:request-1', + requestBytes: 100, + segments: [], + startedAt: 2, + completedAt: 3, + status: 'failed', + latencyMs: 1, + }, + }; + await writeFile( + traceEventsPath, + `${JSON.stringify(capture)}\n${JSON.stringify(firstAttempt)}\n{"type":"provider_request_attempt_recorded"`, + ); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.equal(result.attempts.length, 1); + assert.deepEqual( + result.diagnostics.map((diagnostic) => diagnostic.code), + ['invalid_json'], + ); + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /line 3.*valid JSON/i, + ); +}); + +test('diagnoses a complete but malformed request attempt row', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const { capture, attempt } = completeTraceRows('run-4'); + Reflect.deleteProperty(attempt.data, 'requestHash'); + await writeFile(traceEventsPath, `${JSON.stringify(capture)}\n${JSON.stringify(attempt)}\n`); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual( + result.diagnostics.map((diagnostic) => diagnostic.code), + ['invalid_attempt'], + ); + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /line 2.*attempt data is invalid/i, + ); +}); + +test('accepts request attempt timing emitted across a non-monotonic clock adjustment', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const { capture, attempt } = completeTraceRows('run-clock-adjustment'); + attempt.data.startedAt = 4; + attempt.data.completedAt = 3; + attempt.data.latencyMs = 0; + await writeFile(traceEventsPath, `${JSON.stringify(capture)}\n${JSON.stringify(attempt)}\n`); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual(result.diagnostics, []); + assert.doesNotThrow(() => traceAnalysis.assertProviderRequestTraceComplete(result)); +}); + +test('accepts finite fractional timing emitted by the Runtime attempt contract', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const { capture, attempt } = completeTraceRows('run-fractional-clock'); + attempt.data.startedAt = 2.25; + attempt.data.completedAt = 4.75; + attempt.data.latencyMs = 2.5; + Reflect.set(attempt.data, 'timeToFirstTokenMs', 0.5); + await writeFile(traceEventsPath, `${JSON.stringify(capture)}\n${JSON.stringify(attempt)}\n`); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual(result.diagnostics, []); + assert.doesNotThrow(() => traceAnalysis.assertProviderRequestTraceComplete(result)); +}); + +test('fails completeness when the execution contains a trace failure sentinel', async () => { + for (const type of ['trace_write_failed', 'event_corrupt'] as const) { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const { capture, attempt } = completeTraceRows(`run-${type}`); + const sentinel = { + type, + id: `${type}-event`, + runId: capture.runId, + sessionId: capture.sessionId, + turnId: capture.turnId, + ts: 5, + message: `${type} fixture`, + }; + await writeFile( + traceEventsPath, + `${[capture, attempt, sentinel].map((event) => JSON.stringify(event)).join('\n')}\n`, + ); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual( + result.diagnostics.map((diagnostic) => diagnostic.code), + [type], + ); + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + new RegExp(`line 3.*${type.replaceAll('_', ' ')}`, 'i'), + ); + } +}); + +test('fails completeness when provider request rows mix execution identities', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const { capture, attempt } = completeTraceRows('run-identity-a'); + attempt.runId = 'run-identity-b'; + await writeFile(traceEventsPath, `${JSON.stringify(capture)}\n${JSON.stringify(attempt)}\n`); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual( + result.diagnostics.map((diagnostic) => diagnostic.code), + ['mixed_identity'], + ); + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /line 2.*identity.*run-identity-b.*run-identity-a/i, + ); +}); + +test('fails completeness when an attempt does not match its request capture', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const { capture, attempt } = completeTraceRows('run-mismatched-capture'); + attempt.data.requestHash = 'sha256:another-request'; + await writeFile(traceEventsPath, `${JSON.stringify(capture)}\n${JSON.stringify(attempt)}\n`); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual(result.diagnostics, []); + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /attempt attempt-1 does not match its request capture/i, + ); +}); + +test('fails completeness when capture ids are duplicated', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const { capture, attempt } = completeTraceRows('run-duplicate-capture'); + const duplicateCapture = structuredClone(capture); + duplicateCapture.id = 'capture-event-2'; + await writeFile( + traceEventsPath, + `${[capture, duplicateCapture, attempt].map((event) => JSON.stringify(event)).join('\n')}\n`, + ); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /duplicate capture id capture-1/i, + ); +}); + +test('fails completeness when attempt ids are duplicated', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const { capture, attempt } = completeTraceRows('run-duplicate-attempt'); + const duplicateAttempt = structuredClone(attempt); + duplicateAttempt.id = 'attempt-event-2'; + duplicateAttempt.data.attempt = 2; + await writeFile( + traceEventsPath, + `${[capture, attempt, duplicateAttempt].map((event) => JSON.stringify(event)).join('\n')}\n`, + ); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /duplicate attempt id attempt-1/i, + ); +}); + +test('fails completeness when a capture has no request attempt', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const { capture, attempt } = completeTraceRows('run-unattempted-capture'); + const unattemptedCapture = structuredClone(capture); + unattemptedCapture.id = 'capture-event-2'; + unattemptedCapture.ts = 5; + unattemptedCapture.data.captureId = 'capture-2'; + unattemptedCapture.data.artifactId = 'artifact-capture-2'; + unattemptedCapture.data.step = 1; + unattemptedCapture.data.requestHash = 'sha256:request-2'; + await writeFile( + traceEventsPath, + `${[capture, attempt, unattemptedCapture].map((event) => JSON.stringify(event)).join('\n')}\n`, + ); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /capture capture-2 has no request attempt/i, + ); +}); + +test('fails completeness when individually valid attempts have an ordinal gap', async () => { + const dir = await mkdtemp(join(tmpdir(), 'maka-provider-trace-')); + const traceEventsPath = join(dir, 'events.jsonl'); + const { capture, attempt } = completeTraceRows('run-5'); + const thirdAttempt = structuredClone(attempt); + thirdAttempt.id = 'attempt-event-3'; + thirdAttempt.ts = 6; + thirdAttempt.data.attemptId = 'attempt-3'; + thirdAttempt.data.attempt = 3; + thirdAttempt.data.startedAt = 5; + thirdAttempt.data.completedAt = 6; + thirdAttempt.data.latencyMs = 1; + await writeFile( + traceEventsPath, + `${[capture, attempt, thirdAttempt].map((event) => JSON.stringify(event)).join('\n')}\n`, + ); + + const result = await traceAnalysis.readProviderRequestTrace(traceEventsPath); + + assert.deepEqual(result.diagnostics, []); + assert.throws( + () => traceAnalysis.assertProviderRequestTraceComplete(result), + /step 0 request attempt sequence is incomplete/i, + ); +}); + +function completeTraceRows(runId: string) { + const identity = { runId, sessionId: `session-${runId}`, turnId: `turn-${runId}` }; + const capture = { + type: 'provider_request_captured', + id: 'capture-event-1', + ...identity, + ts: 1, + data: { + schemaVersion: 2, + traceId: `trace-${runId}`, + captureId: 'capture-1', + artifactId: 'artifact-capture-1', + turnId: identity.turnId, + step: 0, + providerId: 'openai', + modelId: 'k3', + requestHash: 'sha256:request-1', + requestPayloadWithoutProviderOptionsHash: 'sha256:shared-request-1', + requestBytes: 100, + segments: [], + }, + }; + const attempt = { + type: 'provider_request_attempt_recorded', + id: 'attempt-event-1', + ...identity, + ts: 4, + data: { + traceId: `trace-${runId}`, + attemptId: 'attempt-1', + turnId: identity.turnId, + step: 0, + attempt: 1, + captureId: 'capture-1', + captureArtifactId: 'artifact-capture-1', + providerId: 'openai', + modelId: 'k3', + requestHash: 'sha256:request-1', + requestBytes: 100, + segments: [], + startedAt: 2, + completedAt: 4, + status: 'completed', + latencyMs: 2, + }, + }; + return { capture, attempt }; +} diff --git a/packages/headless/src/harbor-cell.ts b/packages/headless/src/harbor-cell.ts index 4b29b15d8d..b299c33a00 100644 --- a/packages/headless/src/harbor-cell.ts +++ b/packages/headless/src/harbor-cell.ts @@ -2,6 +2,7 @@ import { mkdir, readFile, rename, rm, writeFile } from 'node:fs/promises'; import { randomUUID } from 'node:crypto'; import { join } from 'node:path'; import type { + AgentRunEvent, AgentRunHeader, BackendKind, PricingConfig, @@ -546,6 +547,7 @@ async function readHarborCellUsageCheckpoint( } } +/** Export top-level invocation ledgers for provider-request benchmark evidence. */ export async function writeHarborTaskRunTrace(input: { outputDir: string; storage: HeadlessStorageWriter; @@ -553,9 +555,49 @@ export async function writeHarborTaskRunTrace(input: { }): Promise { const storage = authenticateHeadlessStorageWriter(input.storage); const eventGroups = await Promise.all( - input.invocations.map((invocation) => - storage.executionStores.agentRunStore.readEvents(invocation.sessionId, invocation.runId), - ), + input.invocations.map(async (invocation) => { + const [header, events] = await Promise.all([ + storage.executionStores.agentRunStore.readRun(invocation.sessionId, invocation.runId), + storage.executionStores.agentRunStore.readEventsForEvidence( + invocation.sessionId, + invocation.runId, + ), + ]); + const evidenceEvents = + header.traceWriteError && !events.some((event) => event.type === 'trace_write_failed') + ? [ + ...events, + { + type: 'trace_write_failed', + id: `run-header-trace-write-failed-${header.runId}`, + runId: header.runId, + sessionId: header.sessionId, + turnId: header.turnId, + ts: header.updatedAt, + message: header.traceWriteError, + } satisfies AgentRunEvent, + ] + : events; + if (header.backendKind !== 'ai-sdk' || evidenceEvents.some(isProviderRequestTraceEvidence)) { + return evidenceEvents; + } + return [ + ...evidenceEvents, + { + type: 'event_corrupt', + id: `run-provider-request-evidence-missing-${header.runId}`, + runId: header.runId, + sessionId: header.sessionId, + turnId: header.turnId, + ts: header.updatedAt, + message: `Provider request trace evidence is missing for invocation ${invocation.invocationId}`, + data: { + reason: 'missing_provider_request_evidence', + invocationId: invocation.invocationId, + }, + } satisfies AgentRunEvent, + ]; + }), ); const chunks = eventGroups.map((events) => events.map((event) => JSON.stringify(event)).join('\n'), @@ -569,6 +611,14 @@ export async function writeHarborTaskRunTrace(input: { return traceEventsPath; } +function isProviderRequestTraceEvidence(event: AgentRunEvent): boolean { + return ( + event.type === 'provider_request_captured' || + event.type === 'provider_request_attempt_recorded' || + event.type === 'trace_write_failed' + ); +} + export async function runHarborCellFromEnv( env: RunHarborCellEnv = process.env, options: RunHarborCellFromEnvOptions = {}, diff --git a/packages/headless/src/index.ts b/packages/headless/src/index.ts index 89a2a6f656..1dca2e379b 100644 --- a/packages/headless/src/index.ts +++ b/packages/headless/src/index.ts @@ -4,10 +4,18 @@ // 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 { + assertProviderRequestTraceComplete, + readProviderRequestTrace, +} from './provider-request-trace.js'; export type { + AssertProviderRequestTraceCompleteOptions, ProviderRequestTraceAnalysis, + ProviderRequestTraceAttemptAnalysis, ProviderRequestTraceCaptureAnalysis, + ProviderRequestTraceDiagnostic, + ProviderRequestTraceDiagnosticCode, + ProviderRequestTraceIdentity, } from './provider-request-trace.js'; export type { MakaChangeAuditRecord } from './change-audit.js'; export type { diff --git a/packages/headless/src/provider-request-trace.ts b/packages/headless/src/provider-request-trace.ts index 0ef1589b54..0ba5d3bc44 100644 --- a/packages/headless/src/provider-request-trace.ts +++ b/packages/headless/src/provider-request-trace.ts @@ -4,9 +4,17 @@ import { findFirstChangedCacheableSegment, type PreparedRequestSegment, type PreparedRequestSegmentRef, + type ProviderRequestAttemptRecord, } from '@maka/runtime'; +export interface ProviderRequestTraceIdentity { + runId: string; + sessionId: string; + turnId: string; +} + export interface ProviderRequestTraceCaptureAnalysis { + schemaVersion: 1 | 2; traceId: string; captureId: string; artifactId: string; @@ -15,14 +23,43 @@ export interface ProviderRequestTraceCaptureAnalysis { providerId: string; modelId: string; requestHash: string; + requestPayloadWithoutProviderOptionsHash?: string; requestBytes: number; segments: PreparedRequestSegment[]; firstChangedCacheableSegment?: PreparedRequestSegmentRef; } +export type ProviderRequestTraceAttemptAnalysis = ProviderRequestAttemptRecord; + +export type ProviderRequestTraceDiagnosticCode = + | 'invalid_json' + | 'invalid_agent_run_event' + | 'invalid_capture' + | 'invalid_attempt' + | 'trace_write_failed' + | 'event_corrupt' + | 'mixed_identity'; + +export interface ProviderRequestTraceDiagnostic { + code: ProviderRequestTraceDiagnosticCode; + line: number; + message: string; +} + export interface ProviderRequestTraceAnalysis { + identity?: ProviderRequestTraceIdentity; + /** Every exported top-level execution represented by this task trace. */ + identities?: ProviderRequestTraceIdentity[]; traceId?: string; captures: ProviderRequestTraceCaptureAnalysis[]; + attempts: ProviderRequestTraceAttemptAnalysis[]; + diagnostics: ProviderRequestTraceDiagnostic[]; +} + +export interface AssertProviderRequestTraceCompleteOptions { + expectedIdentity?: ProviderRequestTraceIdentity; + expectedIdentities?: readonly ProviderRequestTraceIdentity[]; + label?: string; } /** Read Harbor's existing AgentRun events.jsonl; no provider-proxy sidecar is required. */ @@ -31,41 +68,256 @@ export async function readProviderRequestTrace( ): Promise { const text = await readFile(traceEventsPath, 'utf8'); const captures: ProviderRequestTraceCaptureAnalysis[] = []; - for (const line of text.split('\n')) { + const attempts: ProviderRequestTraceAttemptAnalysis[] = []; + const diagnostics: ProviderRequestTraceDiagnostic[] = []; + let identity: ProviderRequestTraceIdentity | undefined; + const identities: ProviderRequestTraceIdentity[] = []; + const identityByRunId = new Map(); + const identityByTurnId = new Map(); + let traceId: string | undefined; + const lastCaptureByTraceId = new Map(); + + for (const [index, line] of text.split('\n').entries()) { if (!line.trim()) continue; + const lineNumber = index + 1; + let value: unknown; + try { + value = JSON.parse(line); + } catch { + diagnostics.push({ + code: 'invalid_json', + line: lineNumber, + message: 'provider request trace row is not valid JSON', + }); + continue; + } + let event: ReturnType; try { - event = decodeAgentRunEvent(JSON.parse(line)); + event = decodeAgentRunEvent(value); } catch { + diagnostics.push({ + code: 'invalid_agent_run_event', + line: lineNumber, + message: 'provider request trace row is not a valid AgentRun event', + }); 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), - } - : {}), - }); + if ( + event.type !== 'provider_request_captured' && + event.type !== 'provider_request_attempt_recorded' && + event.type !== 'trace_write_failed' && + event.type !== 'event_corrupt' + ) { + continue; + } + + const eventIdentity = identityFromEvent(event); + if (!identity) { + identity = eventIdentity; + } + const conflictingIdentity = + identityByRunId.get(eventIdentity.runId) ?? identityByTurnId.get(eventIdentity.turnId); + if (conflictingIdentity && !sameIdentity(conflictingIdentity, eventIdentity)) { + diagnostics.push({ + code: 'mixed_identity', + line: lineNumber, + message: `provider request trace row identity ${formatIdentity(eventIdentity)} differs from ${formatIdentity(conflictingIdentity)}`, + }); + } else if (!identities.some((candidate) => sameIdentity(candidate, eventIdentity))) { + identities.push(eventIdentity); + identityByRunId.set(eventIdentity.runId, eventIdentity); + identityByTurnId.set(eventIdentity.turnId, eventIdentity); + } + + if (event.type === 'trace_write_failed' || event.type === 'event_corrupt') { + diagnostics.push({ + code: event.type, + line: lineNumber, + message: `provider request trace contains ${event.type.replaceAll('_', ' ')} evidence`, + }); + continue; + } + + if (event.type === 'provider_request_captured') { + const parsed = captureFromEvent(event.turnId, event.data); + if ('error' in parsed) { + diagnostics.push({ code: 'invalid_capture', line: lineNumber, message: parsed.error }); + continue; + } + traceId ??= parsed.value.traceId; + const prior = lastCaptureByTraceId.get(parsed.value.traceId); + captures.push({ + ...parsed.value, + ...(prior + ? { + firstChangedCacheableSegment: findFirstChangedCacheableSegment(parsed.value, prior), + } + : {}), + }); + lastCaptureByTraceId.set(parsed.value.traceId, parsed.value); + continue; + } + + const parsed = attemptFromEvent(event.turnId, event.data); + if ('error' in parsed) { + diagnostics.push({ code: 'invalid_attempt', line: lineNumber, message: parsed.error }); + continue; + } + traceId ??= parsed.value.traceId; + attempts.push(parsed.value); } + return { - ...(captures[0] ? { traceId: captures[0].traceId } : {}), + ...(identity ? { identity } : {}), + ...(identities.length > 0 ? { identities } : {}), + ...(traceId ? { traceId } : {}), captures, + attempts, + diagnostics, + }; +} + +/** + * Validate tracked requests for exported top-level Harbor invocations. + * Semantic-compaction and child-agent provider dispatches are outside this + * artifact contract. A pre-dispatch cancellation is intentionally incomplete + * because it is indistinguishable from a missing attempt row. + */ +export function assertProviderRequestTraceComplete( + trace: ProviderRequestTraceAnalysis, + options: AssertProviderRequestTraceCompleteOptions = {}, +): void { + const label = options.label ?? 'Provider request trace'; + const fail = (message: string): never => { + throw new Error(`${label}: ${message}`); }; + const diagnostic = trace.diagnostics[0]; + if (diagnostic) fail(`line ${diagnostic.line}: ${diagnostic.message}`); + const identities = + trace.identities && trace.identities.length > 0 + ? trace.identities + : trace.identity + ? [trace.identity] + : fail('has no execution identity'); + const identity = trace.identity ?? identities[0]!; + if (options.expectedIdentity && options.expectedIdentities) { + fail('cannot validate both expectedIdentity and expectedIdentities'); + } + if (options.expectedIdentities) { + const missing = options.expectedIdentities.find( + (expected) => !identities.some((candidate) => sameIdentity(candidate, expected)), + ); + const unexpected = identities.find( + (candidate) => + !options.expectedIdentities!.some((expected) => sameIdentity(candidate, expected)), + ); + if (missing || unexpected || identities.length !== options.expectedIdentities.length) { + fail( + `execution identity set differs from expected; observed ${identities + .map(formatIdentity) + .join(', ')}`, + ); + } + } else if (options.expectedIdentity) { + if (identities.length === 1) { + for (const key of ['runId', 'sessionId', 'turnId'] as const) { + if (identity[key] !== options.expectedIdentity[key]) { + fail(`${key} expected ${options.expectedIdentity[key]}, observed ${identity[key]}`); + } + } + } else { + const foreignSession = identities.find( + (candidate) => candidate.sessionId !== options.expectedIdentity!.sessionId, + ); + if (foreignSession) { + fail( + `sessionId expected ${options.expectedIdentity.sessionId}, observed ${foreignSession.sessionId}`, + ); + } + if (!identities.some((candidate) => sameIdentity(candidate, options.expectedIdentity!))) { + fail( + `does not contain expected execution identity ${formatIdentity(options.expectedIdentity)}`, + ); + } + } + } + trace.traceId ?? fail('has no provider trace id'); + if (trace.captures.length === 0 || trace.attempts.length === 0) { + fail('has incomplete provider request telemetry'); + } + + const captures = new Map(); + const turnIdByTraceId = new Map(); + for (const capture of trace.captures) { + if (captures.has(capture.captureId)) fail(`has duplicate capture id ${capture.captureId}`); + if (!identities.some((candidate) => candidate.turnId === capture.turnId)) { + fail(`capture ${capture.captureId} has another turn id`); + } + const traceTurnId = turnIdByTraceId.get(capture.traceId); + if (traceTurnId !== undefined && traceTurnId !== capture.turnId) { + fail(`capture ${capture.captureId} has another turn id for trace ${capture.traceId}`); + } + turnIdByTraceId.set(capture.traceId, capture.turnId); + captures.set(capture.captureId, capture); + } + + const attemptIds = new Set(); + const referencedCaptureIds = new Set(); + const attemptNumbersByStep = new Map(); + for (const attempt of trace.attempts) { + if (attemptIds.has(attempt.attemptId)) fail(`has duplicate attempt id ${attempt.attemptId}`); + attemptIds.add(attempt.attemptId); + if (!identities.some((candidate) => candidate.turnId === attempt.turnId)) { + fail(`attempt ${attempt.attemptId} has another turn id`); + } + const capture = + captures.get(attempt.captureId) ?? + fail(`attempt ${attempt.attemptId} does not match its request capture`); + if (!attemptMatchesCapture(attempt, capture)) { + fail(`attempt ${attempt.attemptId} does not match its request capture`); + } + referencedCaptureIds.add(capture.captureId); + const stepKey = `${attempt.traceId}\u0000${attempt.step}`; + const numbers = attemptNumbersByStep.get(stepKey) ?? []; + numbers.push(attempt.attempt); + attemptNumbersByStep.set(stepKey, numbers); + } + + for (const captureId of captures.keys()) { + if (!referencedCaptureIds.has(captureId)) fail(`capture ${captureId} has no request attempt`); + } + for (const [stepKey, numbers] of attemptNumbersByStep) { + numbers.sort((left, right) => left - right); + for (let index = 0; index < numbers.length; index += 1) { + if (numbers[index] !== index + 1) { + const step = stepKey.slice(stepKey.lastIndexOf('\u0000') + 1); + fail(`step ${step} request attempt sequence is incomplete`); + } + } + } } +type ParseResult = { value: T } | { error: string }; + 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) - : []; +): ParseResult { + if (!data) return { error: 'provider request capture has no data' }; + if (data.schemaVersion !== 1 && data.schemaVersion !== 2) { + return { error: 'provider request capture has an unsupported schema version' }; + } + if (data.turnId !== undefined && data.turnId !== turnId) { + return { error: 'provider request capture turn id differs from its event envelope' }; + } + if (!Array.isArray(data.segments)) { + return { error: 'provider request capture segments are missing' }; + } + const segments = data.segments.map(segmentFromValue); + if (segments.some((segment) => segment === undefined)) { + return { error: 'provider request capture contains an invalid segment' }; + } if ( typeof data.traceId !== 'string' || typeof data.captureId !== 'string' || @@ -75,24 +327,176 @@ function captureFromEvent( typeof data.modelId !== 'string' || typeof data.requestHash !== 'string' || !isNonNegativeInteger(data.requestBytes) || - segments.length !== (Array.isArray(data.segments) ? data.segments.length : 0) + (data.requestPayloadWithoutProviderOptionsHash !== undefined && + typeof data.requestPayloadWithoutProviderOptionsHash !== 'string') || + (data.schemaVersion === 2 && typeof data.requestPayloadWithoutProviderOptionsHash !== 'string') ) { - return undefined; + return { error: 'provider request capture data is invalid' }; } 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, + value: { + schemaVersion: data.schemaVersion, + traceId: data.traceId, + captureId: data.captureId, + artifactId: data.artifactId, + turnId, + step: data.step, + providerId: data.providerId, + modelId: data.modelId, + requestHash: data.requestHash, + ...(data.requestPayloadWithoutProviderOptionsHash !== undefined + ? { + requestPayloadWithoutProviderOptionsHash: data.requestPayloadWithoutProviderOptionsHash, + } + : {}), + requestBytes: data.requestBytes, + segments: segments as PreparedRequestSegment[], + }, }; } +function attemptFromEvent( + turnId: string, + data: Record | undefined, +): ParseResult { + if (!data) return { error: 'provider request attempt has no data' }; + if (data.turnId !== turnId) { + return { error: 'provider request attempt turn id differs from its event envelope' }; + } + if (!Array.isArray(data.segments)) { + return { error: 'provider request attempt segments are missing' }; + } + const segments = data.segments.map(segmentFromValue); + if (segments.some((segment) => segment === undefined)) { + return { error: 'provider request attempt contains an invalid segment' }; + } + const requiredStrings = [ + 'traceId', + 'attemptId', + 'captureId', + 'captureArtifactId', + 'providerId', + 'modelId', + 'requestHash', + ] as const; + if ( + requiredStrings.some((key) => typeof data[key] !== 'string') || + !isNonNegativeInteger(data.step) || + !isPositiveInteger(data.attempt) || + !isNonNegativeInteger(data.requestBytes) || + !isNonNegativeFiniteNumber(data.startedAt) || + !isNonNegativeFiniteNumber(data.completedAt) || + !isAttemptStatus(data.status) || + !isNonNegativeFiniteNumber(data.latencyMs) || + (data.finishReason !== undefined && typeof data.finishReason !== 'string') || + (data.timeToFirstTokenMs !== undefined && !isNonNegativeFiniteNumber(data.timeToFirstTokenMs)) + ) { + return { error: 'provider request attempt data is invalid' }; + } + const optionalTokens = [ + 'inputTokens', + 'cacheReadInputTokens', + 'cacheMissInputTokens', + 'cacheWriteInputTokens', + 'outputTokens', + 'reasoningTokens', + ] as const; + if ( + optionalTokens.some((key) => data[key] !== undefined && !isNonNegativeInteger(data[key])) || + !validSource(data.cacheReadInputSource) || + !validSource(data.cacheMissInputSource) || + !validSource(data.cacheWriteInputSource) + ) { + return { error: 'provider request attempt usage is invalid' }; + } + const inputTokens = data.inputTokens as number | undefined; + const cacheReadInputTokens = data.cacheReadInputTokens as number | undefined; + const cacheMissInputTokens = data.cacheMissInputTokens as number | undefined; + const cacheWriteInputTokens = data.cacheWriteInputTokens as number | undefined; + const outputTokens = data.outputTokens as number | undefined; + const reasoningTokens = data.reasoningTokens as number | undefined; + const cacheReadInputSource = data.cacheReadInputSource as + | ProviderRequestAttemptRecord['cacheReadInputSource'] + | undefined; + const cacheMissInputSource = data.cacheMissInputSource as + | ProviderRequestAttemptRecord['cacheMissInputSource'] + | undefined; + const cacheWriteInputSource = data.cacheWriteInputSource as + | ProviderRequestAttemptRecord['cacheWriteInputSource'] + | undefined; + return { + value: { + traceId: data.traceId as string, + attemptId: data.attemptId as string, + turnId, + step: data.step, + attempt: data.attempt, + captureId: data.captureId as string, + captureArtifactId: data.captureArtifactId as string, + providerId: data.providerId as string, + modelId: data.modelId as string, + requestHash: data.requestHash as string, + requestBytes: data.requestBytes, + segments: segments as PreparedRequestSegment[], + startedAt: data.startedAt, + completedAt: data.completedAt, + status: data.status, + ...(data.finishReason !== undefined ? { finishReason: data.finishReason } : {}), + latencyMs: data.latencyMs, + ...(data.timeToFirstTokenMs !== undefined + ? { timeToFirstTokenMs: data.timeToFirstTokenMs } + : {}), + ...(inputTokens !== undefined ? { inputTokens } : {}), + ...(cacheReadInputTokens !== undefined ? { cacheReadInputTokens } : {}), + ...(cacheReadInputSource !== undefined ? { cacheReadInputSource } : {}), + ...(cacheMissInputTokens !== undefined ? { cacheMissInputTokens } : {}), + ...(cacheMissInputSource !== undefined ? { cacheMissInputSource } : {}), + ...(cacheWriteInputTokens !== undefined ? { cacheWriteInputTokens } : {}), + ...(cacheWriteInputSource !== undefined ? { cacheWriteInputSource } : {}), + ...(outputTokens !== undefined ? { outputTokens } : {}), + ...(reasoningTokens !== undefined ? { reasoningTokens } : {}), + }, + }; +} + +function attemptMatchesCapture( + attempt: ProviderRequestTraceAttemptAnalysis, + capture: ProviderRequestTraceCaptureAnalysis, +): boolean { + return ( + attempt.traceId === capture.traceId && + attempt.captureArtifactId === capture.artifactId && + attempt.turnId === capture.turnId && + attempt.step === capture.step && + attempt.providerId === capture.providerId && + attempt.modelId === capture.modelId && + attempt.requestHash === capture.requestHash && + attempt.requestBytes === capture.requestBytes && + segmentsEqual(attempt.segments, capture.segments) + ); +} + +function segmentsEqual( + left: readonly PreparedRequestSegment[], + right: readonly PreparedRequestSegment[], +): boolean { + return ( + left.length === right.length && + left.every((segment, index) => { + const other = right[index]; + return ( + other !== undefined && + segment.kind === other.kind && + segment.index === other.index && + segment.cacheable === other.cacheable && + segment.hash === other.hash && + segment.bytes === other.bytes && + segment.role === other.role + ); + }) + ); +} + function segmentFromValue(value: unknown): PreparedRequestSegment | undefined { if (!value || typeof value !== 'object') return undefined; const segment = value as Record; @@ -111,6 +515,45 @@ function segmentFromValue(value: unknown): PreparedRequestSegment | undefined { return segment as unknown as PreparedRequestSegment; } +function identityFromEvent( + event: ReturnType, +): ProviderRequestTraceIdentity { + return { runId: event.runId, sessionId: event.sessionId, turnId: event.turnId }; +} + +function sameIdentity( + left: ProviderRequestTraceIdentity, + right: ProviderRequestTraceIdentity, +): boolean { + return ( + left.runId === right.runId && left.sessionId === right.sessionId && left.turnId === right.turnId + ); +} + +function formatIdentity(identity: ProviderRequestTraceIdentity): string { + return `${identity.runId}/${identity.sessionId}/${identity.turnId}`; +} + function isNonNegativeInteger(value: unknown): value is number { return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; } + +function isPositiveInteger(value: unknown): value is number { + return isNonNegativeInteger(value) && value > 0; +} + +function isNonNegativeFiniteNumber(value: unknown): value is number { + return typeof value === 'number' && Number.isFinite(value) && value >= 0; +} + +function isAttemptStatus(value: unknown): value is ProviderRequestAttemptRecord['status'] { + return ( + value === 'completed' || value === 'failed' || value === 'interrupted' || value === 'aborted' + ); +} + +function validSource( + value: unknown, +): value is ProviderRequestAttemptRecord['cacheReadInputSource'] { + return value === undefined || value === 'provider' || value === 'derived'; +} diff --git a/packages/runtime/src/__tests__/provider-request-telemetry.test.ts b/packages/runtime/src/__tests__/provider-request-telemetry.test.ts index ad57cca754..76abdd1736 100644 --- a/packages/runtime/src/__tests__/provider-request-telemetry.test.ts +++ b/packages/runtime/src/__tests__/provider-request-telemetry.test.ts @@ -199,7 +199,7 @@ describe('provider request capture commit', () => { }); const result = await recordCapture({ - schemaVersion: 1, + schemaVersion: 2, traceId: 'trace-1', captureId: 'capture-1', turnId: 'turn-1', @@ -207,6 +207,7 @@ describe('provider request capture commit', () => { providerId: 'openai', modelId: 'gpt-test', requestHash: 'sha256:request', + requestPayloadWithoutProviderOptionsHash: 'sha256:shared-request', requestBytes: 2, segments: [], serializedRequest: '{}', @@ -242,7 +243,7 @@ describe('provider request capture commit', () => { await assert.rejects( recordCapture({ - schemaVersion: 1, + schemaVersion: 2, traceId: 'trace-1', captureId: 'capture-1', turnId: 'turn-1', @@ -250,6 +251,7 @@ describe('provider request capture commit', () => { providerId: 'openai', modelId: 'gpt-test', requestHash: 'sha256:request', + requestPayloadWithoutProviderOptionsHash: 'sha256:shared-request', requestBytes: 2, segments: [], serializedRequest: '{}', diff --git a/packages/runtime/src/__tests__/request-shape.test.ts b/packages/runtime/src/__tests__/request-shape.test.ts index 8a6a0d7b0b..be4acd2bba 100644 --- a/packages/runtime/src/__tests__/request-shape.test.ts +++ b/packages/runtime/src/__tests__/request-shape.test.ts @@ -170,6 +170,378 @@ describe('prepared provider request capture', () => { assert.ok(result.segments.every((segment) => /^sha256:[a-f0-9]{64}$/.test(segment.hash))); }); + test('versions and hashes non-provider-options request parameters for comparison', () => { + const capture = (providerOptions: Record, maxOutputTokens?: number) => + requestShape.capturePreparedProviderRequest({ + providerId: 'provider', + modelId: 'k3', + instructions: 'system', + messages: [{ role: 'user', content: 'hello' }], + tools: [{ name: 'Read', inputSchema: { type: 'object' } }], + providerOptions, + requestPayload: { + prompt: [{ role: 'user', content: 'hello' }], + tools: [{ name: 'Read', inputSchema: { type: 'object' } }], + providerOptions, + ...(maxOutputTokens !== undefined ? { maxOutputTokens } : {}), + }, + }); + + const anthropic = capture({ anthropic: { effort: 'max' } }, 131_072); + const openai = capture({ kimiCodingPlan: { reasoningEffort: 'max' } }, 131_072); + const changedSharedParameter = capture({ kimiCodingPlan: { reasoningEffort: 'max' } }, 32_768); + + assert.equal(anthropic.schemaVersion, 2); + assert.equal( + anthropic.requestPayloadWithoutProviderOptionsHash, + openai.requestPayloadWithoutProviderOptionsHash, + ); + assert.notEqual( + anthropic.requestPayloadWithoutProviderOptionsHash, + changedSharedParameter.requestPayloadWithoutProviderOptionsHash, + ); + }); + + test('keeps reasoning effort across provider namespaces in the protocol-independent hash', () => { + const hash = (providerOptions: Record, maxOutputTokens: number) => + requestShape.capturePreparedProviderRequest({ + providerId: 'kimi-coding-plan', + modelId: 'kimi-for-coding', + messages: [{ role: 'user', content: 'hello' }], + tools: [], + providerOptions, + requestPayload: { + prompt: [{ role: 'user', content: 'hello' }], + maxOutputTokens, + providerOptions, + }, + }).requestPayloadWithoutProviderOptionsHash; + + const anthropicMax = hash( + { + anthropic: { + effort: 'max', + thinking: { type: 'enabled', budgetTokens: 1_024 }, + }, + }, + 31_744, + ); + const openaiMax = hash({ kimiCodingPlan: { reasoningEffort: 'max' } }, 32_768); + const nativeOpenaiMax = hash({ openai: { reasoningEffort: 'max' } }, 32_768); + const nativeOpenaiHigh = hash({ openai: { reasoningEffort: 'high' } }, 32_768); + const zaiHigh = hash({ 'zai-coding-plan': { reasoningEffort: 'high' } }, 32_768); + const zaiLow = hash({ 'zai-coding-plan': { reasoningEffort: 'low' } }, 32_768); + + assert.equal(anthropicMax, openaiMax); + assert.equal(anthropicMax, nativeOpenaiMax); + assert.equal(nativeOpenaiHigh, zaiHigh); + assert.notEqual(zaiHigh, zaiLow); + assert.notEqual(anthropicMax, hash({ kimiCodingPlan: { reasoningEffort: 'low' } }, 32_768)); + assert.notEqual(anthropicMax, hash({ kimiCodingPlan: { reasoningEffort: 'none' } }, 32_768)); + }); + + test('normalizes disabled Anthropic reasoning to OpenAI none', () => { + const hash = (providerOptions: Record) => + requestShape.capturePreparedProviderRequest({ + providerId: 'provider', + modelId: 'model', + messages: [{ role: 'user', content: 'hello' }], + tools: [], + providerOptions, + requestPayload: { + prompt: [{ role: 'user', content: 'hello' }], + maxOutputTokens: 32_768, + providerOptions, + }, + }).requestPayloadWithoutProviderOptionsHash; + + assert.equal( + hash({ anthropic: { thinking: { type: 'disabled' } } }), + hash({ openai: { reasoningEffort: 'none' } }), + ); + }); + + test('normalizes Google thinking level to OpenAI reasoning effort', () => { + const hash = (providerOptions: Record) => + requestShape.capturePreparedProviderRequest({ + providerId: 'provider', + modelId: 'model', + messages: [{ role: 'user', content: 'hello' }], + tools: [], + providerOptions, + requestPayload: { + prompt: [{ role: 'user', content: 'hello' }], + maxOutputTokens: 32_768, + providerOptions, + }, + }).requestPayloadWithoutProviderOptionsHash; + + assert.equal( + hash({ google: { thinkingConfig: { includeThoughts: true, thinkingLevel: 'high' } } }), + hash({ openai: { reasoningEffort: 'high' } }), + ); + }); + + test('normalizes zero Google thinking budget to OpenAI none', () => { + const hash = (providerOptions: Record) => + requestShape.capturePreparedProviderRequest({ + providerId: 'provider', + modelId: 'model', + messages: [{ role: 'user', content: 'hello' }], + tools: [], + providerOptions, + requestPayload: { + prompt: [{ role: 'user', content: 'hello' }], + maxOutputTokens: 32_768, + providerOptions, + }, + }).requestPayloadWithoutProviderOptionsHash; + + assert.equal( + hash({ google: { thinkingConfig: { thinkingBudget: 0 } } }), + hash({ openai: { reasoningEffort: 'none' } }), + ); + }); + + test('normalizes disabled Cloudflare thinking to OpenAI none', () => { + const hash = (providerOptions: Record) => + requestShape.capturePreparedProviderRequest({ + providerId: 'provider', + modelId: 'model', + messages: [{ role: 'user', content: 'hello' }], + tools: [], + providerOptions, + requestPayload: { + prompt: [{ role: 'user', content: 'hello' }], + maxOutputTokens: 32_768, + providerOptions, + }, + }).requestPayloadWithoutProviderOptionsHash; + + assert.equal( + hash({ + 'cloudflare-workers-ai': { chat_template_kwargs: { thinking: false } }, + }), + hash({ openai: { reasoningEffort: 'none' } }), + ); + }); + + test('normalizes Anthropic thinking budget into the protocol-independent output limit', () => { + const capture = (providerOptions: Record, maxOutputTokens: number) => + requestShape.capturePreparedProviderRequest({ + providerId: 'kimi-coding-plan', + modelId: 'kimi-for-coding', + instructions: 'system', + messages: [{ role: 'user', content: 'hello' }], + tools: [], + providerOptions, + requestPayload: { + prompt: [{ role: 'user', content: 'hello' }], + maxOutputTokens, + providerOptions, + }, + }); + + const anthropic = capture( + { anthropic: { thinking: { type: 'enabled', budgetTokens: 1_024 } } }, + 31_744, + ); + const openai = capture({ maka: { kimiReasoningField: 'reasoning_content' } }, 32_768); + + assert.equal( + anthropic.requestPayloadWithoutProviderOptionsHash, + openai.requestPayloadWithoutProviderOptionsHash, + ); + assert.notEqual(anthropic.requestHash, openai.requestHash); + }); + + test('excludes provider metadata nested in prompt messages and parts', () => { + const capture = (prompt: unknown[], tools: unknown[] = []) => + requestShape.capturePreparedProviderRequest({ + providerId: 'provider', + modelId: 'model', + messages: prompt, + tools, + requestPayload: { prompt, tools }, + }); + const sharedPrompt = [ + { + role: 'assistant', + content: [{ type: 'reasoning', text: 'analysis' }], + }, + ]; + const anthropicPrompt = [ + { + role: 'assistant', + providerOptions: { anthropic: { cacheControl: { type: 'ephemeral' } } }, + content: [ + { + type: 'reasoning', + text: 'analysis', + providerOptions: { anthropic: { signature: 'signed-reasoning' } }, + }, + ], + }, + ]; + + assert.equal( + capture(anthropicPrompt).requestPayloadWithoutProviderOptionsHash, + capture(sharedPrompt).requestPayloadWithoutProviderOptionsHash, + ); + + const sharedToolPrompt = [ + { + role: 'tool', + content: [ + { + type: 'tool-result', + toolCallId: 'call-1', + toolName: 'Inspect', + output: { type: 'content', value: [{ type: 'text', text: 'done' }] }, + }, + ], + }, + ]; + const providerToolPrompt = [ + { + ...sharedToolPrompt[0], + providerOptions: { anthropic: { cacheControl: { type: 'ephemeral' } } }, + content: [ + { + ...sharedToolPrompt[0]!.content[0], + providerOptions: { anthropic: { toolUseId: 'provider-call-1' } }, + output: { + type: 'content', + providerOptions: { anthropic: { resultId: 'provider-result-1' } }, + value: [ + { + type: 'text', + text: 'done', + providerOptions: { anthropic: { blockId: 'provider-block-1' } }, + }, + ], + }, + }, + ], + }, + ]; + const sharedTools = [{ type: 'function', name: 'Inspect', inputSchema: { type: 'object' } }]; + const providerTools = [ + { + ...sharedTools[0], + providerOptions: { anthropic: { deferLoading: true } }, + }, + ]; + + assert.equal( + capture(providerToolPrompt, providerTools).requestPayloadWithoutProviderOptionsHash, + capture(sharedToolPrompt, sharedTools).requestPayloadWithoutProviderOptionsHash, + ); + }); + + test('normalizes provider-local tool and approval bookkeeping', () => { + const hash = (prompt: unknown[]) => + requestShape.capturePreparedProviderRequest({ + providerId: 'provider', + modelId: 'model', + messages: prompt, + tools: [], + requestPayload: { prompt, tools: [] }, + }).requestPayloadWithoutProviderOptionsHash; + const prompt = (suffix: string, approved: boolean) => [ + { + role: 'assistant', + content: [ + { + type: 'tool-call', + toolCallId: `call-${suffix}`, + toolName: 'Inspect', + input: { path: 'README.md' }, + providerExecuted: suffix === 'anthropic', + }, + { + type: 'tool-approval-request', + approvalId: `approval-${suffix}`, + toolCallId: `call-${suffix}`, + isAutomatic: suffix === 'anthropic', + signature: `signature-${suffix}`, + }, + ], + }, + { + role: 'tool', + content: [ + { + type: 'tool-result', + toolCallId: `call-${suffix}`, + toolName: 'Inspect', + output: { type: 'text', value: 'done' }, + }, + { + type: 'tool-approval-response', + approvalId: `approval-${suffix}`, + approved, + providerExecuted: suffix === 'anthropic', + }, + ], + }, + ]; + + assert.equal(hash(prompt('anthropic', true)), hash(prompt('openai', true))); + assert.notEqual(hash(prompt('anthropic', true)), hash(prompt('openai', false))); + }); + + test('preserves same-named fields inside user data and tool schemas', () => { + const hash = (prompt: unknown[], tools: unknown[] = []) => + requestShape.capturePreparedProviderRequest({ + providerId: 'provider', + modelId: 'model', + messages: prompt, + tools, + requestPayload: { prompt, tools }, + }).requestPayloadWithoutProviderOptionsHash; + const toolCall = (value: string) => [ + { + role: 'assistant', + content: [ + { + type: 'tool-call', + toolCallId: 'call-1', + toolName: 'Inspect', + input: { providerOptions: value }, + }, + ], + }, + ]; + const toolResult = (value: string) => [ + { + role: 'tool', + content: [ + { + type: 'tool-result', + toolCallId: 'call-1', + toolName: 'Inspect', + output: { type: 'json', value: { providerOptions: value } }, + }, + ], + }, + ]; + const tool = (description: string) => [ + { + type: 'function', + name: 'Inspect', + inputSchema: { + type: 'object', + properties: { providerOptions: { type: 'string', description } }, + }, + }, + ]; + + assert.notEqual(hash(toolCall('alpha')), hash(toolCall('bravo'))); + assert.notEqual(hash(toolResult('alpha')), hash(toolResult('bravo'))); + assert.notEqual(hash([], tool('alpha')), hash([], tool('bravo'))); + }); + test('finds the first changed cacheable segment by exact content hash', () => { const capture = requestShape.capturePreparedProviderRequest; const findFirstChanged = Reflect.get(requestShape, 'findFirstChangedCacheableSegment') as diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index 2d0c8ab71a..51f13e875e 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -10341,6 +10341,52 @@ describe('SessionManager permission mode updates', () => { expect(events.some((event) => event.id === 'attempt-2')).toBe(true); }); + test('carries a provider attempt failure latch into the terminal run header', async () => { + const store = new MemorySessionStore(); + let failAttemptAppend = true; + let failFailureLatch = true; + let failFailureSentinel = true; + const runStore = new MemoryAgentRunStore({ + beforeAgentRunUpdate: async (_sessionId, _runId, patch) => { + if (patch.traceWriteError && failFailureLatch) { + failFailureLatch = false; + throw new Error('trace failure latch update failed'); + } + }, + beforeAgentRunEventAppend: async (_sessionId, _runId, event) => { + if (event.type === 'provider_request_attempt_recorded' && failAttemptAppend) { + failAttemptAppend = false; + throw new Error('provider attempt append failed'); + } + if (event.type === 'trace_write_failed' && failFailureSentinel) { + failFailureSentinel = false; + throw new Error('trace failure sentinel append failed'); + } + }, + }); + 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_763), + }); + const session = await manager.createSession(makeInput()); + + await drain(manager.sendMessage(session.id, { turnId: 'turn-1', text: 'hello' })); + + const [run] = await runStore.listSessionRuns(session.id); + expect(run?.status).toBe('completed'); + expect(run?.traceWriteError).toMatch( + /append provider request attempt: provider attempt append failed/, + ); + const events = await runStore.readEvents(session.id, run!.runId); + expect(events.some((event) => event.type === 'trace_write_failed')).toBe(false); + }); + test('finalizes the run when a required provider capture append fails', async () => { const store = new MemorySessionStore(); let providerDispatches = 0; @@ -14765,7 +14811,7 @@ class ProviderRequestTraceBackend implements AgentBackend { async *send(input: BackendSendInput): AsyncIterable { await this.ctx.recordProviderRequestCapture?.({ - schemaVersion: 1, + schemaVersion: 2, traceId: 'provider-trace-1', captureId: 'capture-1', turnId: input.turnId, @@ -14773,6 +14819,7 @@ class ProviderRequestTraceBackend implements AgentBackend { providerId: 'fake', modelId: 'fake-model', requestHash: 'sha256:request', + requestPayloadWithoutProviderOptionsHash: 'sha256:shared-request', requestBytes: 100, segments: [], artifactId: 'artifact-capture', @@ -14829,7 +14876,7 @@ class ProviderCaptureGateBackend implements AgentBackend { async *send(input: BackendSendInput): AsyncIterable { await this.ctx.recordProviderRequestCapture?.({ - schemaVersion: 1, + schemaVersion: 2, traceId: 'provider-trace-gated', captureId: 'capture-gated', turnId: input.turnId, @@ -14837,6 +14884,7 @@ class ProviderCaptureGateBackend implements AgentBackend { providerId: 'fake', modelId: 'fake-model', requestHash: 'sha256:gated', + requestPayloadWithoutProviderOptionsHash: 'sha256:shared-gated', requestBytes: 100, segments: [], artifactId: 'artifact-gated', @@ -14890,7 +14938,7 @@ class ProviderCaptureAfterAttemptFailureBackend implements AgentBackend { await this.attemptFailureRecorded; try { await this.ctx.recordProviderRequestCapture?.({ - schemaVersion: 1, + schemaVersion: 2, traceId: 'provider-trace-1', captureId: 'capture-2', turnId: input.turnId, @@ -14898,6 +14946,7 @@ class ProviderCaptureAfterAttemptFailureBackend implements AgentBackend { providerId: 'fake', modelId: 'fake-model', requestHash: 'sha256:request-2', + requestPayloadWithoutProviderOptionsHash: 'sha256:shared-request-2', requestBytes: 120, segments: [], artifactId: 'artifact-capture-2', @@ -15700,6 +15749,11 @@ class MemoryAgentRunStore implements AgentRunStore, RuntimeEventStore { event: AgentRunEvent, ) => Promise | void; beforeAgentRunEventRead?: (sessionId: string, runId: string) => Promise | void; + beforeAgentRunUpdate?: ( + sessionId: string, + runId: string, + patch: Partial, + ) => Promise | void; } = {}, ) {} @@ -15716,6 +15770,7 @@ class MemoryAgentRunStore implements AgentRunStore, RuntimeEventStore { runId: string, patch: Partial, ): Promise { + await this.options.beforeAgentRunUpdate?.(sessionId, runId, patch); if (this.options.failUpdateRunOnce) { this.options.failUpdateRunOnce = false; throw new Error('update run failed'); diff --git a/packages/runtime/src/agent-run.ts b/packages/runtime/src/agent-run.ts index f94f9626ed..c0a8ee6915 100644 --- a/packages/runtime/src/agent-run.ts +++ b/packages/runtime/src/agent-run.ts @@ -163,6 +163,7 @@ export class AgentRun { private runStoreAvailable = true; private runtimeEventStoreAvailable = true; private runtimeEventStoreFailure: unknown; + private traceWriteError: string | undefined; private failureClass: string | undefined; private failureMessage: string | undefined; private lastTs = 0; @@ -1204,6 +1205,7 @@ export class AgentRun { ? { failureClass: this.failureClass ?? finalStatus?.blockedReason } : {}), ...(this.failureMessage ? { failureMessage: this.failureMessage } : {}), + ...(this.traceWriteError ? { traceWriteError: this.traceWriteError } : {}), ...(this.abortSource || fallbackStatus === 'cancelled' ? { abortSource: this.abortSource ?? 'user_stop' } : {}), @@ -1343,11 +1345,16 @@ export class AgentRun { label = 'agent run store write', ): Promise { const message = errorMessage(error); + this.traceWriteError ??= `${label}: ${message}`; try { await this.input.runStore?.updateRun(this.sessionId, this.runId, { - traceWriteError: `${label}: ${message}`, + traceWriteError: this.traceWriteError, updatedAt: this.input.now(), }); + } catch { + // The terminal header commit retries the in-memory latch. + } + try { await this.input.runStore?.appendEvent(this.sessionId, this.runId, { type: 'trace_write_failed', id: this.input.newId(), @@ -1358,7 +1365,7 @@ export class AgentRun { message, }); } catch { - // Diagnostic persistence failed too; never perturb model/tool execution. + // Diagnostic persistence is best effort; never perturb model/tool execution. } } } diff --git a/packages/runtime/src/request-shape.ts b/packages/runtime/src/request-shape.ts index 9dbf9be9c3..ab7db1b916 100644 --- a/packages/runtime/src/request-shape.ts +++ b/packages/runtime/src/request-shape.ts @@ -76,8 +76,10 @@ export interface PreparedProviderRequestInput { } export interface PreparedProviderRequestCapture { - schemaVersion: 1; + schemaVersion: 2; requestHash: string; + /** Hash of protocol-independent model-call semantics for cross-protocol comparison. */ + requestPayloadWithoutProviderOptionsHash: string; requestBytes: number; serializedRequest: string; segments: PreparedRequestSegment[]; @@ -217,18 +219,184 @@ export function capturePreparedProviderRequest( } return { - schemaVersion: 1, + schemaVersion: 2, requestHash: stableHash({ providerId: input.providerId, modelId: input.modelId, payload, }), + requestPayloadWithoutProviderOptionsHash: stableHash( + protocolIndependentRequestPayload(payload), + ), requestBytes: Buffer.byteLength(serializedRequest, 'utf8'), serializedRequest, segments, }; } +function protocolIndependentRequestPayload(payload: unknown): unknown { + if (!isObjectLike(payload)) return payload; + const { providerOptions, ...shared } = payload; + const identities: ProtocolIndependentRequestIdentities = { + approvalIds: new Map(), + toolCallIds: new Map(), + }; + const reasoningEffort = protocolIndependentReasoningEffort(providerOptions); + const protocolIndependent: Record = { + ...shared, + ...(Array.isArray(shared.prompt) + ? { + prompt: shared.prompt.map((message) => withoutPromptProviderOptions(message, identities)), + } + : {}), + ...(Array.isArray(shared.messages) + ? { + messages: shared.messages.map((message) => + withoutPromptProviderOptions(message, identities), + ), + } + : {}), + ...(Array.isArray(shared.tools) + ? { tools: shared.tools.map(withoutObjectProviderOptions) } + : {}), + ...(reasoningEffort !== undefined ? { reasoningEffort } : {}), + }; + const thinkingBudget = anthropicThinkingBudget(providerOptions); + if ( + thinkingBudget === undefined || + !isNonNegativeSafeInteger(protocolIndependent.maxOutputTokens) + ) { + return protocolIndependent; + } + const wireOutputLimit = protocolIndependent.maxOutputTokens + thinkingBudget; + return Number.isSafeInteger(wireOutputLimit) + ? { ...protocolIndependent, maxOutputTokens: wireOutputLimit } + : protocolIndependent; +} + +function protocolIndependentReasoningEffort( + providerOptions: unknown, +): string | string[] | undefined { + if (!isObjectLike(providerOptions)) return undefined; + const efforts = new Set(); + const anthropic = providerOptions.anthropic; + if (isObjectLike(anthropic) && typeof anthropic.effort === 'string') { + efforts.add(anthropic.effort); + } + for (const namespace of Object.values(providerOptions)) { + if (!isObjectLike(namespace)) continue; + if (typeof namespace.reasoningEffort === 'string') efforts.add(namespace.reasoningEffort); + if (isObjectLike(namespace.thinking) && namespace.thinking.type === 'disabled') { + efforts.add('none'); + } + const thinkingConfig = namespace.thinkingConfig; + if (isObjectLike(thinkingConfig) && typeof thinkingConfig.thinkingLevel === 'string') { + efforts.add(thinkingConfig.thinkingLevel); + } + if (isObjectLike(thinkingConfig) && thinkingConfig.thinkingBudget === 0) { + efforts.add('none'); + } + const chatTemplateKwargs = namespace.chat_template_kwargs; + if (isObjectLike(chatTemplateKwargs) && chatTemplateKwargs.thinking === false) { + efforts.add('none'); + } + } + const normalized = [...efforts].sort(); + return normalized.length > 1 ? normalized : normalized[0]; +} + +interface ProtocolIndependentRequestIdentities { + approvalIds: Map; + toolCallIds: Map; +} + +function withoutPromptProviderOptions( + value: unknown, + identities: ProtocolIndependentRequestIdentities, +): unknown { + const message = withoutObjectProviderOptions(value); + if (!isObjectLike(message) || !Array.isArray(message.content)) return message; + return { + ...message, + content: message.content.map((part) => withoutPromptPartProviderOptions(part, identities)), + }; +} + +function withoutPromptPartProviderOptions( + value: unknown, + identities: ProtocolIndependentRequestIdentities, +): unknown { + const part = withoutObjectProviderOptions(value); + if (!isObjectLike(part)) return part; + if (part.type === 'tool-call') { + const { providerExecuted: _providerExecuted, ...shared } = part; + return { + ...shared, + toolCallId: protocolIndependentId(part.toolCallId, identities.toolCallIds, 'tool-call'), + }; + } + if (part.type === 'tool-approval-request') { + const { isAutomatic: _isAutomatic, signature: _signature, ...shared } = part; + return { + ...shared, + approvalId: protocolIndependentId(part.approvalId, identities.approvalIds, 'approval'), + toolCallId: protocolIndependentId(part.toolCallId, identities.toolCallIds, 'tool-call'), + }; + } + if (part.type === 'tool-approval-response') { + const { providerExecuted: _providerExecuted, ...shared } = part; + return { + ...shared, + approvalId: protocolIndependentId(part.approvalId, identities.approvalIds, 'approval'), + }; + } + if (part.type !== 'tool-result') return part; + const output = withoutObjectProviderOptions(part.output); + const normalizedPart = { + ...part, + toolCallId: protocolIndependentId(part.toolCallId, identities.toolCallIds, 'tool-call'), + }; + if (!isObjectLike(output) || output.type !== 'content' || !Array.isArray(output.value)) { + return { ...normalizedPart, output }; + } + return { + ...normalizedPart, + output: { ...output, value: output.value.map(withoutObjectProviderOptions) }, + }; +} + +function protocolIndependentId( + value: unknown, + identities: Map, + prefix: string, +): unknown { + if (typeof value !== 'string') return value; + const existing = identities.get(value); + if (existing) return existing; + const normalized = `${prefix}-${identities.size + 1}`; + identities.set(value, normalized); + return normalized; +} + +function withoutObjectProviderOptions(value: unknown): unknown { + if (!isObjectLike(value)) return value; + const { providerOptions: _providerOptions, ...shared } = value; + return shared; +} + +function anthropicThinkingBudget(providerOptions: unknown): number | undefined { + if (!isObjectLike(providerOptions)) return undefined; + const anthropic = providerOptions.anthropic; + if (!isObjectLike(anthropic)) return undefined; + const thinking = anthropic.thinking; + if (!isObjectLike(thinking) || thinking.type !== 'enabled') return undefined; + return isNonNegativeSafeInteger(thinking.budgetTokens) ? thinking.budgetTokens : undefined; +} + +function isNonNegativeSafeInteger(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; +} + export function findFirstChangedCacheableSegment( current: Pick, prior: Pick, diff --git a/packages/runtime/src/terminal-run-commit.ts b/packages/runtime/src/terminal-run-commit.ts index e090b7f19e..eadd80f50f 100644 --- a/packages/runtime/src/terminal-run-commit.ts +++ b/packages/runtime/src/terminal-run-commit.ts @@ -71,6 +71,7 @@ export interface CommitTerminalRunWithRuntimeFactInput { terminalEvent: RuntimeEvent; failureClass?: string; failureMessage?: string; + traceWriteError?: string; abortSource?: string; runEventData?: Record; runEventMessage?: string; @@ -122,6 +123,7 @@ async function commitTerminalRunProjection( completedAt: input.ts, ...(failureClass ? { failureClass } : {}), ...(input.failureMessage ? { failureMessage: input.failureMessage } : {}), + ...(input.traceWriteError ? { traceWriteError: input.traceWriteError } : {}), ...(abortSource ? { abortSource } : {}), }, { durable: true }, diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index 71f5ba43c2..5b0c44ee1a 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -111,6 +111,7 @@ export interface RootTurnAdmissionStore { export interface DurableAgentRunStore extends AgentRunStore, RootTurnAdmissionStore { listSessionRunsForRecovery(sessionId: string): Promise; readEventsForRecovery(sessionId: string, runId: string): Promise; + readEventsForEvidence(sessionId: string, runId: string): Promise; readEventProjection( sessionId: string, type: AgentRunEventType, @@ -481,10 +482,15 @@ class FileAgentRunStore implements DurableAgentRunStore { return this.readEventsWithPolicy(sessionId, runId, true); } + async readEventsForEvidence(sessionId: string, runId: string): Promise { + return this.readEventsWithPolicy(sessionId, runId, false, true); + } + private async readEventsWithPolicy( sessionId: string, runId: string, strict: boolean, + preserveIncompleteTail = false, ): Promise { assertSafeId(sessionId, 'Invalid session id'); assertSafeId(runId, 'Invalid run id'); @@ -508,12 +514,11 @@ class FileAgentRunStore implements DurableAgentRunStore { try { parsed = JSON.parse(entry.line); } catch (error) { - if ( + const incompleteTail = !endsWithNewline && entry.lineNumber === lastLineNumber && - classifyJsonRecord(entry.line) === 'incomplete-prefix' - ) - continue; + classifyJsonRecord(entry.line) === 'incomplete-prefix'; + if (incompleteTail && !preserveIncompleteTail) continue; if (strict) { const detail = error instanceof Error ? error.message : String(error); throw new Error( @@ -527,7 +532,11 @@ class FileAgentRunStore implements DurableAgentRunStore { sessionId, turnId: header.turnId, ts: header.updatedAt, - message: error instanceof Error ? error.message : 'Invalid AgentRun event JSONL line', + message: incompleteTail + ? 'Incomplete AgentRun event JSONL tail' + : error instanceof Error + ? error.message + : 'Invalid AgentRun event JSONL line', data: { lineNumber: entry.lineNumber }, }); continue; diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index c367084e18..6e4d4229e5 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -198,6 +198,8 @@ async function openExecutionStoresForWrite( readEvents: (sessionId, runId) => run(() => agentRunStore.readEvents(sessionId, runId)), readEventsForRecovery: (sessionId, runId) => run(() => agentRunStore.readEventsForRecovery(sessionId, runId)), + readEventsForEvidence: (sessionId, runId) => + run(() => agentRunStore.readEventsForEvidence(sessionId, runId)), readEventProjection: (sessionId, type) => run(() => agentRunStore.readEventProjection(sessionId, type)), repairEventProjection: (sessionId, type, event, options) =>