From d7235acef039b9945ccbae2abd8fb5de530e376d Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 11:36:16 +0800 Subject: [PATCH 1/8] perf(runtime): stop storing a copy of every prepared provider request MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each dispatched provider request was serialized whole and written to the artifact store. The request is built from the conversation the run already holds, so every capture was another copy of the same messages, and each one grew with the conversation: one session here reached 772 MB of captures carrying 4.8 MB of distinct content. The bytes were the smaller cost. Captures were 87% of the artifact population, and every artifact write paid for the whole population, so the capture sink is what turned a growing conversation into quadratic write amplification. Nothing read them. The reader shipped with the capture in #1277 and was deleted by #2605; the producer stayed. What the panels and diagnostics actually read is the bounded observation on the canonical ModelCallAttempt, which is unchanged. The request is still serialized in memory to size and identify it, and is then dropped. `PreparedRequestMaterial` collapses into the observation it wrapped, and the tracker's per-step capture memo goes with it: its key was the digest, so it never saved the work it appeared to cache. Decoders stay. `captureArtifactId`, the `provider_request_captured` event and the `provider_request_capture` artifact source all still resolve, so attempts and sessions already on disk keep decoding and keep copying. Removing them would fail exactly the records this change is meant to stop producing more of. Tests that used the sink as a hook now use the dispatch gate, and the ones that used it to inspect the outgoing request assert against the provider request bodies instead — the stronger evidence of the two. Closes #4082 Generated-by: Claude Code --- packages/core/src/artifacts.ts | 2 + packages/core/src/model-call-attempt.ts | 14 +- .../execution-model-composition.test.ts | 43 +--- .../src/server/execution-model-composition.ts | 18 -- .../src/__tests__/ai-sdk-backend.test.ts | 54 +--- .../computer-use-provider-protocol.test.ts | 19 +- .../history-compact-summarizer.test.ts | 2 - .../__tests__/latest-context-commit.test.ts | 8 +- .../mid-turn-capacity-backend.test.ts | 3 - .../src/__tests__/prompt-composition.test.ts | 6 +- .../provider-request-telemetry.test.ts | 232 ++---------------- .../src/__tests__/request-shape.test.ts | 42 ++-- packages/runtime/src/ai-sdk-backend.ts | 14 +- .../runtime/src/provider-request-telemetry.ts | 90 +------ packages/runtime/src/request-shape.ts | 23 +- .../provider-request-capture-artifact.test.ts | 56 ----- packages/storage/src/artifact-stores.ts | 1 - .../src/provider-request-capture-artifact.ts | 48 ---- 18 files changed, 84 insertions(+), 591 deletions(-) delete mode 100644 packages/storage/src/__tests__/provider-request-capture-artifact.test.ts delete mode 100644 packages/storage/src/provider-request-capture-artifact.ts diff --git a/packages/core/src/artifacts.ts b/packages/core/src/artifacts.ts index 0e203c9f92..9fd6bcf5c2 100644 --- a/packages/core/src/artifacts.ts +++ b/packages/core/src/artifacts.ts @@ -166,6 +166,8 @@ const ARTIFACT_SOURCE_POLICIES = { synthesis_cache_block: { userDeletable: true, userVisible: false, sharedReadable: false }, history_compact_block: { userDeletable: true, userVisible: false, sharedReadable: false }, history_compact_source: { userDeletable: true, userVisible: false, sharedReadable: false }, + // Historical only: nothing produces these any more. The policy stays so the + // records already on disk keep decoding and stay deletable. provider_request_capture: { userDeletable: true, userVisible: false, sharedReadable: false }, subagent_writeback: { userDeletable: false, userVisible: true, sharedReadable: false }, deep_research: { userDeletable: false, userVisible: true, sharedReadable: false }, diff --git a/packages/core/src/model-call-attempt.ts b/packages/core/src/model-call-attempt.ts index efac9504c7..bbb9c03015 100644 --- a/packages/core/src/model-call-attempt.ts +++ b/packages/core/src/model-call-attempt.ts @@ -112,9 +112,8 @@ export interface PreparedRequestObservationSegment { /** * Bounded, secret-free observation of one prepared semantic model request. * - * This is not the provider wire body. The full secret-free serialization stays - * in the private request artifact referenced by `captureArtifactId` when that - * sink is available. + * This is not the provider wire body, and no copy of that body is kept: the + * request is built from the conversation the run already stores. */ export interface PreparedRequestObservation { schemaVersion: typeof PREPARED_REQUEST_OBSERVATION_SCHEMA_VERSION; @@ -165,7 +164,14 @@ export interface ModelCallAttempt { providerId: string; modelId: string; contextWindow?: number; - /** Join key for the private prepared-request artifact, when best-effort persistence won the race. */ + /** + * Join key for the private prepared-request artifact. + * + * Historical only: nothing writes it any more. Every capture was a copy of + * the conversation the run already stores, and the copies grew with the + * conversation. Attempts recorded before that sink was removed still carry + * the key, so it stays decodable. + */ captureArtifactId?: string; /** Semantic request actually prepared for this dispatched physical attempt. */ requestObservation?: PreparedRequestObservation; diff --git a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts index 67abcacd27..d7394f75a0 100644 --- a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts @@ -2047,22 +2047,6 @@ test('production Host executes a canonical ai-sdk Session against a real provide } const artifacts = await openInteractiveArtifactStoreForWrite(owner.lease); - const captureArtifacts = await waitForCaptureArtifacts( - artifacts, - session.id, - capturedRequestCount, - ); - assert.equal(captureArtifacts.length, capturedRequestCount); - let summaryCaptureFound = false; - for (const artifact of captureArtifacts) { - const read = await artifacts.readTextInSession(session.id, artifact.id); - if (read.ok && /context summarization assistant/.test(read.text)) { - summaryCaptureFound = true; - break; - } - } - assert.equal(summaryCaptureFound, true); - const streamRequestsBeforeArtifactFailure = provider.requests.filter( (request) => request.body.stream === true, ).length; @@ -2469,8 +2453,7 @@ test('production Host executes a durable runnable child with an exact tool ceili ); const artifacts = await openInteractiveArtifactStoreForWrite(owner.lease); const childArtifacts = await artifacts.listTurnArtifacts(child.id, childRuns[0]!.turnId); - assert.equal(childArtifacts.length, 1); - assert.equal(childArtifacts[0]?.source, 'provider_request_capture'); + assert.equal(childArtifacts.length, 0, 'a child turn no longer stores anything of its own'); const parentRuntimeEvents = await execution.runtimeEventStore.readRuntimeEvents( parent.id, terminal.runId, @@ -2483,7 +2466,7 @@ test('production Host executes a durable runnable child with an exact tool ceili const typedSpawnResult = decodeCanonicalToolResultContent(spawnResult.content.result); assert.equal(typedSpawnResult.kind, 'subagent'); assert.deepEqual( - (typedSpawnResult as { artifactIds?: readonly string[] }).artifactIds, + (typedSpawnResult as { artifactIds?: readonly string[] }).artifactIds ?? [], childArtifacts.map((artifact) => artifact.id), ); } finally { @@ -2686,11 +2669,7 @@ test('production Host publishes and retires an implementation child patch', asyn ); const artifacts = await openInteractiveArtifactStoreForWrite(owner.lease); const childArtifacts = await artifacts.listTurnArtifacts(child.id, childRuns[0]!.turnId); - assert.equal(childArtifacts.length, childRequests.length + 2); - assert.equal( - childArtifacts.filter((artifact) => artifact.source === 'provider_request_capture').length, - childRequests.length, - ); + assert.equal(childArtifacts.length, 2); assert.ok( childArtifacts.some( (artifact) => artifact.source === 'tool_result' && artifact.name === 'implementation.txt', @@ -3905,22 +3884,6 @@ async function waitForCanonicalAttempts( ); } -async function waitForCaptureArtifacts( - artifacts: Awaited>, - sessionId: string, - expectedRequests: number, -) { - for (let attempt = 0; attempt < 100; attempt += 1) { - const page = await artifacts.listPage(sessionId, { offset: 0, limit: 100 }); - const captures = page.records.filter( - (artifact) => artifact.source === 'provider_request_capture', - ); - if (captures.length >= expectedRequests) return captures; - await new Promise((resolve) => setTimeout(resolve, 10)); - } - throw new Error(`Hosted request artifacts did not reach ${expectedRequests}`); -} - async function waitForAutomaticMemoryRequestsToSettle( requests: readonly ProviderRequest[], ): Promise { diff --git a/packages/runtime-host/src/server/execution-model-composition.ts b/packages/runtime-host/src/server/execution-model-composition.ts index 667b597af4..cc495950e2 100644 --- a/packages/runtime-host/src/server/execution-model-composition.ts +++ b/packages/runtime-host/src/server/execution-model-composition.ts @@ -52,7 +52,6 @@ import { type RuntimeCommitSink } from '@maka/runtime/runtime-commit-sink'; import { createAttachmentByteReader, createReadImageSnapshotPlanner, - persistProviderRequestCaptureArtifact, type InteractiveArtifactStoreWriter, } from '@maka/storage/artifact-stores'; import type { InteractiveContextOffloadReader } from '@maka/storage/context-offload-store'; @@ -292,22 +291,6 @@ async function buildHostAiSdkBackend( throw new Error('Canonical model-call accounting authority is unavailable'); } }; - const persistPreparedRequestArtifact = async (capture: { - turnId: string; - captureId: string; - step: number; - serializedRequest: string; - }): Promise<{ artifactId: string }> => { - const artifact = await persistProviderRequestCaptureArtifact(input.artifacts, { - sessionId: input.context.sessionId, - turnId: capture.turnId, - captureId: capture.captureId, - step: capture.step, - serializedRequest: capture.serializedRequest, - now: Date.now(), - }); - return { artifactId: artifact.id }; - }; const resolveRunPrompt = async (context: { readonly turnId: string; readonly emitSkillCatalogTrace?: (message: string, data?: Record) => void; @@ -468,7 +451,6 @@ async function buildHostAiSdkBackend( lookupPricing: pricing, recordModelCallAttempt, assertModelCallAccountingReady, - persistPreparedRequestArtifact, recordToolInvocation: (event) => recordToolInvocation({ repo: telemetry }, event), ...(input.runtimeCommitSink ? { runtimeCommitSink: input.runtimeCommitSink } : {}), newId: randomUUID, diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index f9693fd817..37c4c8b435 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -81,7 +81,6 @@ import { } from '../sandbox-boundary-declaration.js'; import { FilesystemWorkerClientError } from '../filesystem-worker/client.js'; import { RunTrace } from '../run-trace.js'; -import type { PreparedRequestArtifactInput } from '../provider-request-telemetry.js'; import { decodeModelCallAttempt, type ModelCallAttempt } from '@maka/core/model-call-attempt'; import { buildLlmHistorySummarizer } from '../history-compact-summarizer.js'; import { createToolResultArchiveCapability } from '../tool-result-archive-capability.js'; @@ -8969,7 +8968,6 @@ 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: PreparedRequestArtifactInput[] = []; const attempts: ModelCallAttempt[] = []; const durable = durableTurnHarness('turn-1', 'hi'); let calls = 0; @@ -9069,10 +9067,6 @@ describe('AiSdkBackend RunTrace', () => { loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents, newId: idGenerator(), now: monotonicClock(), - persistPreparedRequestArtifact: async (capture) => { - captures.push(capture); - return { artifactId: `artifact-${captures.length}` }; - }, recordModelCallAttempt: ({ attempt }) => { attempts.push(attempt); }, @@ -9080,7 +9074,6 @@ describe('AiSdkBackend RunTrace', () => { const events = await drainDurably(backend.send(durable.input({ runId: 'run-1' })), durable); - assert.equal(captures.length, 2); assert.deepEqual( attempts.map(({ step, attempt, status }) => ({ step, attempt, status })), [ @@ -9107,15 +9100,9 @@ describe('AiSdkBackend RunTrace', () => { } test('observes the prepared request at dispatch and records its canonical attempt', async () => { - const captures: PreparedRequestArtifactInput[] = []; const attempts: ModelCallAttempt[] = []; const model = new MockLanguageModelV4({ doStream: async () => { - assert.equal( - captures.length, - 1, - 'artifact persistence must start before provider dispatch', - ); return { stream: simulateReadableStream({ chunks: [ @@ -9154,10 +9141,6 @@ describe('AiSdkBackend RunTrace', () => { tools: [], newId: idGenerator(), now: monotonicClock(), - persistPreparedRequestArtifact: async (capture) => { - captures.push(capture); - return { artifactId: `artifact-${captures.length}` }; - }, recordModelCallAttempt: async ({ attempt }) => { attempts.push(attempt); }, @@ -9173,7 +9156,6 @@ describe('AiSdkBackend RunTrace', () => { events.push(event); } - assert.equal(captures.length, 1); assert.equal(attempts.length, 1); assert.equal(attempts[0]?.step, 0); assert.equal(attempts[0]?.attempt, 0); @@ -9182,7 +9164,7 @@ describe('AiSdkBackend RunTrace', () => { assert.equal(attempts[0]?.cacheMissInputTokens, 4); assert.equal( events.find((event) => event.type === 'token_usage')?.providerRequestTraceId, - captures[0]?.traceId, + attempts[0]?.traceId, ); }); @@ -9288,36 +9270,7 @@ describe('AiSdkBackend RunTrace', () => { assert.equal(stored?.type === 'token_usage' && 'contextRemaining' in stored, false); }); - test('continues the canonical call when private request persistence fails', async () => { - const model = completionModel(); - const backend = createTestAiSdkBackend({ - sessionId: 'session-1', - header: header(), - appendMessage: async () => {}, - connection: connection(), - apiKey: 'sk-test', - modelId: 'mock-model-id', - modelFactory: () => model, - tools: [], - newId: idGenerator(), - now: monotonicClock(), - persistPreparedRequestArtifact: async () => { - throw new Error('capture unavailable'); - }, - }); - - const events: SessionEvent[] = []; - for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { - events.push(event); - } - - assert.equal(model.doStreamCalls.length, 1); - assert.equal(events.at(-1)?.type, 'complete'); - assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'end_turn'); - }); - test('disables hidden AI SDK retries and traces the one explicit Runtime retry', async () => { - const captures: PreparedRequestArtifactInput[] = []; const attempts: ModelCallAttempt[] = []; let calls = 0; const model = new MockLanguageModelV4({ @@ -9362,10 +9315,6 @@ describe('AiSdkBackend RunTrace', () => { tools: [], newId: idGenerator(), now: monotonicClock(), - persistPreparedRequestArtifact: async (capture) => { - captures.push(capture); - return { artifactId: 'artifact-1' }; - }, recordModelCallAttempt: ({ attempt }) => { attempts.push(attempt); }, @@ -9375,7 +9324,6 @@ describe('AiSdkBackend RunTrace', () => { await drain(backend.send({ turnId: 'turn-1', runId: 'run-1', text: 'hi', context: [] })); assert.equal(calls, 2); - assert.equal(captures.length, 1); assert.deepEqual( attempts.map(({ attempt, status }) => ({ attempt, status })), [ diff --git a/packages/runtime/src/__tests__/computer-use-provider-protocol.test.ts b/packages/runtime/src/__tests__/computer-use-provider-protocol.test.ts index 916d397b8f..fc1e335d1c 100644 --- a/packages/runtime/src/__tests__/computer-use-provider-protocol.test.ts +++ b/packages/runtime/src/__tests__/computer-use-provider-protocol.test.ts @@ -37,7 +37,6 @@ import { type CuObservation, } from '../computer-use-tools.js'; import { buildProviderOptions, getAIModel } from '../model-factory.js'; -import type { PreparedRequestArtifactInput } from '../provider-request-telemetry.js'; import { backfillRuntimeEventsFromStoredMessages } from '../runtime-event-backfill.js'; import { createDurableTurnHarness } from './durable-turn-harness.js'; import { createTestAiSdkBackend } from './execution-boundary-test-helpers.js'; @@ -219,7 +218,6 @@ describe('Anthropic-compatible Computer Use product loops', () => { text: 'Set the fixture field to provider-loop.', }); const requestBodies: Array> = []; - const captures: PreparedRequestArtifactInput[] = []; const attempts: ModelCallAttempt[] = []; const server = await startJsonServer(async (request, response) => { assert.equal(request.method, 'POST'); @@ -278,10 +276,6 @@ describe('Anthropic-compatible Computer Use product loops', () => { loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents, newId: idGenerator(), now: monotonicClock(), - persistPreparedRequestArtifact: async (capture) => { - captures.push(capture); - return { artifactId: `capture-artifact-${captures.length}` }; - }, recordModelCallAttempt: ({ attempt }) => { attempts.push(attempt); }, @@ -306,7 +300,6 @@ describe('Anthropic-compatible Computer Use product loops', () => { ); assert.equal(events.at(-1)?.type, 'complete'); assert.equal(requestBodies.length, 4); - assert.equal(captures.length, 4); assert.equal(attempts.length, 4); assert.deepEqual(toolResults, [{ isError: false }, { isError: false }, { isError: false }]); assert.deepEqual( @@ -901,7 +894,6 @@ describe('OpenAI-compatible product loops', () => { text: 'Set the fixture field to provider-loop.', }); const requestBodies: Array> = []; - const captures: PreparedRequestArtifactInput[] = []; const attempts: ModelCallAttempt[] = []; const server = await startJsonServer(async (request, response) => { assert.equal(request.method, 'POST'); @@ -952,10 +944,6 @@ describe('OpenAI-compatible product loops', () => { loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents, newId: idGenerator(), now: monotonicClock(), - persistPreparedRequestArtifact: async (capture) => { - captures.push(capture); - return { artifactId: `capture-artifact-${captures.length}` }; - }, recordModelCallAttempt: ({ attempt }) => { attempts.push(attempt); }, @@ -981,13 +969,12 @@ describe('OpenAI-compatible product loops', () => { ); assert.equal(events.at(-1)?.type, 'complete'); assert.equal(requestBodies.length, 4); - assert.equal(captures.length, 4); assert.equal(attempts.length, 4); - for (const capture of captures) { + for (const body of requestBodies) { assert.doesNotMatch( - capture.serializedRequest, + JSON.stringify(body), /MAKA_(?:KIMI|OPENAI_CHAT)_EMPTY_REASONING/, - 'provider request evidence must not persist the SDK-only empty-reasoning marker', + 'the SDK-only empty-reasoning marker must not reach the provider', ); } for (const body of requestBodies) { diff --git a/packages/runtime/src/__tests__/history-compact-summarizer.test.ts b/packages/runtime/src/__tests__/history-compact-summarizer.test.ts index 16cbbd3698..9d93700b86 100644 --- a/packages/runtime/src/__tests__/history-compact-summarizer.test.ts +++ b/packages/runtime/src/__tests__/history-compact-summarizer.test.ts @@ -133,7 +133,6 @@ describe('buildLlmHistorySummarizer', () => { return now; }, newId: () => 'trace-id', - persistArtifact: async () => ({ artifactId: 'artifact-1' }), accounting: { sessionId: 'sess-1', resolveRunId: () => 'run-1', @@ -197,7 +196,6 @@ describe('buildLlmHistorySummarizer', () => { turnId: 'turn-1', now: () => 100 + id, newId: () => `request-${++id}`, - persistArtifact: async () => ({ artifactId: `artifact-${id}` }), accounting: { sessionId: 'sess-1', resolveRunId: () => 'run-1', diff --git a/packages/runtime/src/__tests__/latest-context-commit.test.ts b/packages/runtime/src/__tests__/latest-context-commit.test.ts index c04e4afc71..f3eb59d021 100644 --- a/packages/runtime/src/__tests__/latest-context-commit.test.ts +++ b/packages/runtime/src/__tests__/latest-context-commit.test.ts @@ -186,7 +186,7 @@ test('a real send seals its observation into SQLite and reconstructs it after re } }); -test('an artifact captured before abort does not create a canonical sent attempt', async () => { +test('a turn aborted before dispatch does not create a canonical sent attempt', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-aborted-request-chain-')); try { const sessionStore = createSessionStore(root); @@ -196,7 +196,6 @@ test('an artifact captured before abort does not create a canonical sent attempt let ids = 0; const newId = () => `abort-chain-${++ids}`; let providerCalls = 0; - let artifactWrites = 0; backends.register('ai-sdk', (ctx) => { let backend!: ReturnType; @@ -220,10 +219,8 @@ test('an artifact captured before abort does not create a canonical sent attempt }, }), tools: [], - persistPreparedRequestArtifact: async () => { - artifactWrites += 1; + beforeRunProviderDispatch: () => { void backend.stop('user_stop'); - return { artifactId: 'abandoned-artifact' }; }, ...(ctx.recordModelCallAttempt ? { recordModelCallAttempt: ctx.recordModelCallAttempt } @@ -260,7 +257,6 @@ test('an artifact captured before abort does not create a canonical sent attempt const events = ( await Promise.all(runIds.map((runId) => runStore.readEvents(session.id, runId))) ).flat(); - assert.equal(artifactWrites, 1); assert.equal(providerCalls, 0); assert.equal(events.filter((event) => event.type === 'model_call_attempt_recorded').length, 0); assert.deepEqual(await readLatestContextDiagnostics(runStore, session.id, runIds), { diff --git a/packages/runtime/src/__tests__/mid-turn-capacity-backend.test.ts b/packages/runtime/src/__tests__/mid-turn-capacity-backend.test.ts index b3e6e99556..8e0eff790d 100644 --- a/packages/runtime/src/__tests__/mid-turn-capacity-backend.test.ts +++ b/packages/runtime/src/__tests__/mid-turn-capacity-backend.test.ts @@ -597,9 +597,6 @@ function buildFixture(options: MidTurnFixtureOptions = {}): MidTurnFixture { }, ...(options.meteredSummarizer ? { - persistPreparedRequestArtifact: async () => ({ - artifactId: 'artifact-mid-turn-capture', - }), recordModelCallAttempt: (commit: ModelCallCommit) => { commits.push(commit); modelCalls.push(commit.attempt); diff --git a/packages/runtime/src/__tests__/prompt-composition.test.ts b/packages/runtime/src/__tests__/prompt-composition.test.ts index 6cff37a2ad..ce0f14af0f 100644 --- a/packages/runtime/src/__tests__/prompt-composition.test.ts +++ b/packages/runtime/src/__tests__/prompt-composition.test.ts @@ -132,7 +132,7 @@ describe('a real observation survives the whole chain into one fold', () => { providerOptions: { anthropic: { thinking: { type: 'enabled' } } }, }); - const composition = foldPromptComposition(material.observation.segments); + const composition = foldPromptComposition(material.segments); assert.deepEqual( composition?.tools?.map((tool) => tool.name), @@ -165,8 +165,8 @@ describe('a real observation survives the whole chain into one fold', () => { providerOptions: { anthropic: { thinking: { type: 'enabled' } } }, }); - assert.equal(material.observation.segments.length, 256); - const composition = foldPromptComposition(material.observation.segments); + assert.equal(material.segments.length, 256); + const composition = foldPromptComposition(material.segments); assert.equal(composition?.tools?.length, 64); assert.equal(composition?.remainingTools?.count, 189); assert.equal(composition?.unlabelledToolBytes, undefined); diff --git a/packages/runtime/src/__tests__/provider-request-telemetry.test.ts b/packages/runtime/src/__tests__/provider-request-telemetry.test.ts index 8aa2970afd..c6c881a7db 100644 --- a/packages/runtime/src/__tests__/provider-request-telemetry.test.ts +++ b/packages/runtime/src/__tests__/provider-request-telemetry.test.ts @@ -253,7 +253,6 @@ describe('provider request tracker', () => { contextWindow: 200_000, now: () => Date.now(), newId: () => 'id', - persistArtifact: async () => ({ artifactId: 'artifact' }), accounting: canonicalAccounting(attempts), }); @@ -276,7 +275,6 @@ describe('provider request tracker', () => { contextWindow: 0, now: () => Date.now(), newId: () => 'id', - persistArtifact: async () => ({ artifactId: 'artifact' }), accounting: canonicalAccounting(attempts), }); @@ -291,11 +289,7 @@ describe('provider request tracker', () => { assert.equal(attempts[0]?.contextWindow, undefined); }); - test('persists a logical capture before each physical attempt and reuses it for retries', async () => { - const captures: Array<{ - captureId: string; - serializedRequest: string; - }> = []; + test('counts a retry as another attempt of the same step', async () => { const attempts: ModelCallAttempt[] = []; let id = 0; const tracker = new telemetry.ProviderRequestTracker({ @@ -303,10 +297,6 @@ describe('provider request tracker', () => { turnId: 'turn-1', now: () => Date.now(), newId: () => `id-${++id}`, - persistArtifact: async (capture) => { - captures.push(capture); - return { artifactId: `artifact-${captures.length}` }; - }, accounting: canonicalAccounting(attempts), }); tracker.setStep(2); @@ -355,36 +345,20 @@ describe('provider request tracker', () => { }); await drain(result.stream); - assert.equal(captures.length, 1); - assert.deepEqual(JSON.parse(captures[0]!.serializedRequest), params); assert.deepEqual( - attempts.map(({ step, attempt, status, captureArtifactId }) => ({ - step, - attempt, - status, - captureArtifactId, - })), + attempts.map(({ step, attempt, status }) => ({ step, attempt, status })), [ - { - step: 2, - attempt: 0, - status: 'failed', - captureArtifactId: 'artifact-1', - }, - { - step: 2, - attempt: 1, - status: 'completed', - captureArtifactId: 'artifact-1', - }, + { step: 2, attempt: 0, status: 'failed' }, + { step: 2, attempt: 1, status: 'completed' }, ], ); + // Both attempts observed the same request, so they share its identity. + assert.equal(attempts[0]?.requestObservation?.digest, attempts[1]?.requestObservation?.digest); assert.equal(attempts[1]?.cacheReadInputTokens, 4); assert.equal(attempts[1]?.cacheMissInputTokens, 6); }); - test('captures and attributes a non-streaming physical provider call', async () => { - const captures: Array<{ captureId: string; serializedRequest: string }> = []; + test('attributes a non-streaming physical provider call', async () => { const attempts: ModelCallAttempt[] = []; let providerCalls = 0; let id = 0; @@ -393,10 +367,6 @@ describe('provider request tracker', () => { turnId: 'turn-history', now: () => 1_000 + id, newId: () => `history-${++id}`, - persistArtifact: async (capture) => { - captures.push(capture); - return { artifactId: 'history-artifact' }; - }, accounting: canonicalAccounting(attempts), }); const params = preparedParams('history summary'); @@ -421,40 +391,25 @@ describe('provider request tracker', () => { assert.equal(result.text, 'summary'); assert.equal(providerCalls, 1); - assert.equal(captures.length, 1); - assert.deepEqual(JSON.parse(captures[0]!.serializedRequest), params); assert.deepEqual( - attempts.map(({ status, finishReason, inputTokens, outputTokens, captureArtifactId }) => ({ + attempts.map(({ status, finishReason, inputTokens, outputTokens }) => ({ status, finishReason, inputTokens, outputTokens, - captureArtifactId, })), - [ - { - status: 'completed', - finishReason: 'stop', - inputTokens: 7, - outputTokens: 3, - captureArtifactId: 'history-artifact', - }, - ], + [{ status: 'completed', finishReason: 'stop', inputTokens: 7, outputTokens: 3 }], ); + assert.ok(attempts[0]?.requestObservation); }); - test('derives the artifact and canonical opaque observation from one redacted request', async () => { - const captures: telemetry.PreparedRequestArtifactInput[] = []; + test('derives a canonical opaque observation from a redacted request', async () => { const attempts: ModelCallAttempt[] = []; const tracker = new telemetry.ProviderRequestTracker({ traceId: 'compaction-trace', turnId: 'turn-compaction', now: () => 1_000, newId: () => 'compaction-id', - persistArtifact: async (capture) => { - captures.push(capture); - return { artifactId: 'compaction-artifact' }; - }, accounting: canonicalAccounting(attempts), }); const params = { @@ -502,49 +457,12 @@ describe('provider request tracker', () => { doGenerate: async () => ({ text: 'ok' }), }); - assert.equal(captures.length, 1); assert.equal(attempts.length, 1); - assert.deepEqual(attempts[0]?.requestObservation, captures[0]?.observation); assert.equal(attempts[0]?.requestObservation?.segments[0]?.comparison, 'opaque'); - assert.doesNotMatch(captures[0]!.serializedRequest, /cmp_secret|OPAQUE_ENCRYPTED_STATE/); assert.doesNotMatch(JSON.stringify(attempts[0]), /cmp_secret|OPAQUE_ENCRYPTED_STATE/); - assert.deepEqual(JSON.parse(captures[0]!.serializedRequest), { - image: 'https://example.com/provider-image.png', - prompt: [ - { - role: 'assistant', - content: [ - { - type: 'custom', - kind: 'openai.compaction', - providerOptions: { - openai: { safeMetadata: 'preserved', redacted: true }, - otherProvider: { cacheKey: 'preserved' }, - }, - }, - { - type: 'tool-call', - toolCallId: 'business-call', - toolName: 'echo', - input: { - type: 'custom', - kind: 'openai.compaction', - providerOptions: { - openai: { - itemId: 'BUSINESS_ITEM_ID', - encryptedContent: 'BUSINESS_OPAQUE_TEXT', - }, - }, - }, - }, - ], - }, - ], - }); }); test('awaits the durable dispatch gate before a non-streaming provider call', async () => { - let captured = false; let dispatched = false; const tracker = new telemetry.ProviderRequestTracker({ traceId: 'gated-history-trace', @@ -554,10 +472,6 @@ describe('provider request tracker', () => { beforeDispatch: async () => { throw new Error('Run Composition store unavailable'); }, - persistArtifact: async () => { - captured = true; - return { artifactId: 'unreachable-artifact' }; - }, }); await assert.rejects( @@ -573,95 +487,9 @@ describe('provider request tracker', () => { }), /Run Composition store unavailable/u, ); - assert.equal(captured, false); assert.equal(dispatched, false); }); - test('dispatches with its observation when private artifact persistence fails', async () => { - const captures: string[] = []; - let providerCalls = 0; - const tracker = new telemetry.ProviderRequestTracker({ - traceId: 'trace-2', - turnId: 'turn-2', - now: () => Date.now(), - newId: () => `capture-${captures.length + 1}`, - persistArtifact: async (capture) => { - captures.push(capture.observation.digest); - if (captures.length === 2) throw new Error('capture unavailable'); - return { artifactId: 'artifact-1' }; - }, - }); - 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); - - const withoutArtifact = await tracker.trackStream({ - providerId: 'anthropic', - modelId: 'claude-test', - params: preparedParams('after'), - abortSignal: new AbortController().signal, - doStream: async () => { - providerCalls += 1; - return { stream: streamOf([finishPart()]) }; - }, - }); - await drain(withoutArtifact.stream); - assert.equal(providerCalls, 2); - assert.equal(captures.length, 2); - assert.notEqual(captures[0], captures[1]); - }); - - test('does not wait for private artifact persistence before dispatch or accounting', async () => { - let releaseArtifact!: (value: { artifactId: string }) => void; - const artifactPending = new Promise<{ artifactId: string }>((resolve) => { - releaseArtifact = resolve; - }); - const recorded: ModelCallAttempt[] = []; - let providerCalls = 0; - const tracker = new telemetry.ProviderRequestTracker({ - traceId: 'trace-slow-artifact', - turnId: 'turn-slow-artifact', - now: () => 1_000, - newId: () => 'slow-artifact-id', - persistArtifact: () => artifactPending, - accounting: { - sessionId: 'session-1', - resolveRunId: () => 'run-1', - callKind: 'main', - record: ({ attempt }) => { - recorded.push(attempt); - }, - }, - }); - - const tracked = tracker.trackGenerate({ - providerId: 'anthropic', - modelId: 'claude-test', - params: preparedParams('hello'), - doGenerate: async () => { - providerCalls += 1; - return { finishReason: 'stop' }; - }, - }); - await new Promise((resolve) => setImmediate(resolve)); - assert.equal(providerCalls, 1, 'provider dispatch does not wait for artifact persistence'); - assert.equal(recorded.length, 1, 'canonical accounting does not wait for the artifact either'); - releaseArtifact({ artifactId: 'artifact-late' }); - await tracked; - - assert.equal(recorded[0]?.captureArtifactId, undefined); - assert.ok(recorded[0]?.requestObservation); - }); - test('records an errored stream after output as interrupted', async () => { const attempts: ModelCallAttempt[] = []; const tracker = new telemetry.ProviderRequestTracker({ @@ -669,7 +497,6 @@ describe('provider request tracker', () => { turnId: 'turn-3', now: () => Date.now(), newId: () => 'id', - persistArtifact: async () => ({ artifactId: 'artifact' }), accounting: canonicalAccounting(attempts), }); tracker.setStep(0); @@ -692,7 +519,6 @@ describe('provider request tracker', () => { turnId: 'turn-4', now: () => Date.now(), newId: () => 'id', - persistArtifact: async () => ({ artifactId: 'artifact' }), accounting: canonicalAccounting(attempts), }); tracker.setStep(0); @@ -710,8 +536,7 @@ describe('provider request tracker', () => { assert.equal(attempts[0]?.status, 'aborted'); }); - test('does not capture or record an attempt when cancellation predates dispatch', async () => { - let captures = 0; + test('does not record an attempt when cancellation predates dispatch', async () => { const attempts: ModelCallAttempt[] = []; let providerCalls = 0; const abort = new AbortController(); @@ -721,10 +546,6 @@ describe('provider request tracker', () => { turnId: 'turn-5', now: () => Date.now(), newId: () => 'id', - persistArtifact: async () => { - captures += 1; - return { artifactId: 'artifact' }; - }, accounting: canonicalAccounting(attempts), }); @@ -742,13 +563,11 @@ describe('provider request tracker', () => { { name: 'AbortError' }, ); - assert.equal(captures, 0); assert.equal(attempts.length, 0); assert.equal(providerCalls, 0); }); - test('does not dispatch or record an attempt when cancellation happens during capture', async () => { - let captures = 0; + test('does not dispatch or record an attempt when cancellation happens at the gate', async () => { const attempts: ModelCallAttempt[] = []; let providerCalls = 0; const abort = new AbortController(); @@ -757,10 +576,8 @@ describe('provider request tracker', () => { turnId: 'turn-6', now: () => Date.now(), newId: () => 'id', - persistArtifact: async () => { - captures += 1; + beforeDispatch: async () => { abort.abort(); - return { artifactId: 'artifact' }; }, accounting: canonicalAccounting(attempts), }); @@ -779,7 +596,6 @@ describe('provider request tracker', () => { { name: 'AbortError' }, ); - assert.equal(captures, 1); assert.equal(attempts.length, 0); assert.equal(providerCalls, 0); }); @@ -863,8 +679,6 @@ describe('canonical model-call accounting', () => { resolveCost?: telemetry.ModelCallAccountingInput['resolveCost']; assertReady?: () => void; resolveRunId?: () => string | undefined; - /** Models a deployment with request capture switched off. */ - withoutCapture?: boolean; callKind?: ModelCallAttempt['callKind']; historyCompactRoute?: ModelCallAttempt['historyCompactRoute']; }): telemetry.ProviderRequestTracker { @@ -874,9 +688,6 @@ describe('canonical model-call accounting', () => { turnId: 'turn-1', now: () => 1_000 + n, newId: () => `id-${++n}`, - ...(overrides.withoutCapture - ? {} - : { persistArtifact: async () => ({ artifactId: 'artifact-1' }) }), accounting: { sessionId: 'session-1', resolveRunId: overrides.resolveRunId ?? (() => 'run-1'), @@ -891,7 +702,7 @@ describe('canonical model-call accounting', () => { }); } - test('a capture abandoned before dispatch never enters the canonical sent sequence', async () => { + test('a call abandoned before dispatch never enters the canonical sent sequence', async () => { const recorded: ModelCallAttempt[] = []; let providerCalls = 0; const abort = new AbortController(); @@ -900,9 +711,8 @@ describe('canonical model-call accounting', () => { turnId: 'turn-abandoned-capture', now: () => 1_000, newId: () => 'capture-abandoned', - persistArtifact: async () => { + beforeDispatch: async () => { abort.abort(); - return { artifactId: 'artifact-abandoned' }; }, accounting: { sessionId: 'session-1', @@ -1093,13 +903,11 @@ describe('canonical model-call accounting', () => { assert.equal(attempt.costUsd, undefined); }); - test('metering survives a deployment with request capture switched off', async () => { - // Capture is a diagnostic. A record that cannot be joined to a stored - // request body is still a record of a call that really was billed, so the - // canonical seam must not be gated on the capture sink being configured. + test('carries the bounded request observation on the canonical attempt', async () => { + // The observation is the whole record of what was sent: nothing stores a + // copy of the request body for it to be joined against. const recorded: ModelCallAttempt[] = []; const tracker = accountingTracker({ - withoutCapture: true, record: ({ attempt }) => { recorded.push(attempt); }, @@ -1115,7 +923,7 @@ describe('canonical model-call accounting', () => { const attempt = decodeModelCallAttempt(recorded[0]); assert.equal(attempt.usageBasis, 'reported'); - assert.equal(attempt.captureArtifactId, undefined, 'there is no artifact to point at'); + assert.equal(attempt.captureArtifactId, undefined, 'nothing writes a capture join any more'); assert.match(attempt.requestObservation?.digest ?? '', /^sha256:[a-f0-9]{64}$/); assert.ok((attempt.requestObservation?.segments.length ?? 0) > 0); }); diff --git a/packages/runtime/src/__tests__/request-shape.test.ts b/packages/runtime/src/__tests__/request-shape.test.ts index 0286eb02e8..e102907374 100644 --- a/packages/runtime/src/__tests__/request-shape.test.ts +++ b/packages/runtime/src/__tests__/request-shape.test.ts @@ -17,8 +17,6 @@ * under the License. */ -import { Buffer } from 'node:buffer'; -import { createHash } from 'node:crypto'; import assert from 'node:assert/strict'; import { describe, test } from 'node:test'; @@ -64,17 +62,18 @@ describe('canonicalizeToolSet active allow-list', () => { }); describe('prepared request observation', () => { - test('derives the request digest and bytes from the private serialization', () => { + test('sizes and identifies the whole request, not just its named segments', () => { const material = prepareRequestObservation({ prompt: [{ role: 'user', content: 'hello' }], maxOutputTokens: 1_024, }); - assert.equal( - material.observation.digest, - `sha256:${createHash('sha256').update(material.serializedRequest).digest('hex')}`, + assert.match(material.digest, /^sha256:[a-f0-9]{64}$/); + // `maxOutputTokens` is part of the request but is not a segment, so the + // total has to exceed what the segments account for. + assert.ok( + material.bytes > material.segments.reduce((total, segment) => total + segment.bytes, 0), ); - assert.equal(material.observation.bytes, Buffer.byteLength(material.serializedRequest, 'utf8')); }); test('serializes non-JSON values without collapsing their semantic identity', () => { @@ -91,8 +90,7 @@ describe('prepared request observation', () => { headers: { 'x-observation': 'present' }, }); - assert.doesNotThrow(() => JSON.parse(observed.serializedRequest)); - assert.notEqual(observed.observation.digest, plain.observation.digest); + assert.notEqual(observed.digest, plain.digest); }); test('preserves the semantic identity of binary request content', () => { @@ -114,10 +112,9 @@ describe('prepared request observation', () => { const first = observe(1); const second = observe(2); - assert.notEqual(first.serializedRequest, second.serializedRequest); - assert.notEqual(first.observation.digest, second.observation.digest); - assert.notEqual(first.observation.segments[0]?.digest, second.observation.segments[0]?.digest); - assert.equal(first.observation.segments[0]?.comparison, 'exact'); + assert.notEqual(first.digest, second.digest); + assert.notEqual(first.segments[0]?.digest, second.segments[0]?.digest); + assert.equal(first.segments[0]?.comparison, 'exact'); }); test('marks redacted compaction content comparison-opaque', () => { @@ -136,8 +133,8 @@ describe('prepared request observation', () => { ], }); - assert.equal(material.observation.segments[0]?.kind, 'message'); - assert.equal(material.observation.segments[0]?.comparison, 'opaque'); + assert.equal(material.segments[0]?.kind, 'message'); + assert.equal(material.segments[0]?.comparison, 'opaque'); }); test('bounds ordered segments without dropping their count or bytes', () => { @@ -148,23 +145,20 @@ describe('prepared request observation', () => { const material = prepareRequestObservation({ prompt }); const expectedBytes = prompt.reduce( (total, message) => - total + prepareRequestObservation({ prompt: [message] }).observation.segments[0]!.bytes, + total + prepareRequestObservation({ prompt: [message] }).segments[0]!.bytes, 0, ); - assert.ok(material.observation.segments.length <= 256); + assert.ok(material.segments.length <= 256); assert.equal( - material.observation.segments.reduce((total, segment) => total + segment.bytes, 0), + material.segments.reduce((total, segment) => total + segment.bytes, 0), expectedBytes, ); assert.equal( - material.observation.segments.reduce( - (total, segment) => total + (segment.representedSegments ?? 1), - 0, - ), + material.segments.reduce((total, segment) => total + (segment.representedSegments ?? 1), 0), prompt.length, ); - assert.equal(material.observation.segments.at(-1)?.comparison, 'opaque'); + assert.equal(material.segments.at(-1)?.comparison, 'opaque'); }); test('records semantic segments in provider-prefix order and labels only tools', () => { @@ -178,7 +172,7 @@ describe('prepared request observation', () => { }); assert.deepEqual( - material.observation.segments.map(({ kind, index, cacheable, role, label }) => ({ + material.segments.map(({ kind, index, cacheable, role, label }) => ({ kind, index, cacheable, diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 8f641c56dd..f7144bef7b 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -231,7 +231,6 @@ import type { ModelCallAttempt, ModelCallKind } from '@maka/core/model-call-atte import { ProviderRequestTracker, type ModelCallAccountingInput, - type PreparedRequestArtifactInput, type ProviderRequestUsage, type ResolvedModelCallCost, } from './provider-request-telemetry.js'; @@ -758,10 +757,6 @@ export interface AiSdkBackendInput extends AiSdkCompactionCapabilities { readChildAgentOutput?: ToolRuntimeInput['readChildAgentOutput']; /** Optional diagnostic trace hook for explaining a runtime turn without changing renderer events. */ recordRunTrace?: RunTraceRecorder; - /** Optional private artifact sink for the secret-free prepared request. */ - persistPreparedRequestArtifact?: ( - input: PreparedRequestArtifactInput, - ) => Promise<{ artifactId: string }>; /** * Commits one settled provider request: the canonical attempt and, when it * is the completed main call, the derived latest-context row it authorises. @@ -3509,9 +3504,8 @@ export class AiSdkBackend implements AgentBackend { * this backend. Callers receive a ready tracker rather than the ingredients: * a half-wired tracker is what produces records nothing can attribute. * - * Absent only when there is nothing to feed: no artifact sink, canonical - * sink, or dispatch gate. Metering deliberately does not depend on artifact - * persistence: the observation is created in memory for every tracked call. + * Absent only when there is nothing to feed: no canonical sink and no + * dispatch gate. */ private createProviderRequestTracker(input: { turnId: string; @@ -3525,7 +3519,6 @@ export class AiSdkBackend implements AgentBackend { */ runId: string | undefined; }): ProviderRequestTracker | undefined { - const persistArtifact = this.input.persistPreparedRequestArtifact; const accounting = this.modelCallAccounting(input.callKind, { modelId: input.modelId, ...(input.runId ? { runId: input.runId } : {}), @@ -3542,14 +3535,13 @@ export class AiSdkBackend implements AgentBackend { runId, }) : undefined; - if (!persistArtifact && !accounting && !beforeDispatch) return undefined; + if (!accounting && !beforeDispatch) return undefined; return new ProviderRequestTracker({ traceId: this.newId(), turnId: input.turnId, contextWindow: resolveSelectedModelContextWindow(this.input.connection, input.modelId), now: this.now, newId: this.newId, - ...(persistArtifact ? { persistArtifact } : {}), ...(beforeDispatch ? { beforeDispatch } : {}), ...(accounting ? { accounting } : {}), }); diff --git a/packages/runtime/src/provider-request-telemetry.ts b/packages/runtime/src/provider-request-telemetry.ts index 49f9809caf..1171f697e3 100644 --- a/packages/runtime/src/provider-request-telemetry.ts +++ b/packages/runtime/src/provider-request-telemetry.ts @@ -26,7 +26,7 @@ import { type PreparedRequestObservation, } from '@maka/core/model-call-attempt'; import type { PricingConfig } from '@maka/core/usage-stats/types'; -import { prepareRequestObservation, type PreparedRequestMaterial } from './request-shape.js'; +import { prepareRequestObservation } from './request-shape.js'; import { rawFinishReasonString } from './model-protocol.js'; import { providerFailureDiagnostic, @@ -65,26 +65,12 @@ export interface ProviderRequestUsageLike { export type ProviderRequestAttemptStatus = 'completed' | 'failed' | 'interrupted' | 'aborted'; -export interface PreparedRequestArtifactInput extends PreparedRequestMaterial { - traceId: string; - captureId: string; - turnId: string; - step: number; - providerId: string; - modelId: string; -} - -export interface PreparedRequestArtifactRef { - artifactId: string; -} - interface SettledProviderAttempt extends ProviderRequestUsage { traceId: string; attemptId: string; turnId: string; step: number; attempt: number; - captureArtifactId?: string; providerId: string; modelId: string; contextWindow?: number; @@ -114,11 +100,6 @@ export interface ProviderRequestTrackerInput { contextWindow?: number; now: () => number; newId: () => string; - /** - * Optional private artifact sink. Failure leaves the canonical observation - * intact and the attempt explicitly has no artifact join. - */ - persistArtifact?: (input: PreparedRequestArtifactInput) => Promise; /** * Durable run metadata that must exist before any physical provider call. * Kept outside accounting because a dispatch gate is an execution contract, @@ -130,7 +111,7 @@ export interface ProviderRequestTrackerInput { * Canonical metering. Present as a unit or not at all: a `ModelCallAttempt` * without session, run, and kind is unattributable, so identity and sink are * wired together rather than as independently optional fields. Absent leaves - * the tracker purely diagnostic, which is what the capture-only tests use. + * the tracker purely diagnostic. */ accounting?: ModelCallAccountingInput; } @@ -285,12 +266,6 @@ export interface ProviderGenerateResult { [key: string]: unknown; } -interface StoredCapture { - material: PreparedRequestMaterial; - /** Absent when artifact persistence is unavailable or failed. */ - artifactId?: string; -} - const CANONICAL_USAGE_FIELDS = [ 'inputTokens', 'outputTokens', @@ -328,7 +303,6 @@ function modelCallUsageFields( export class ProviderRequestTracker { private step = 0; private readonly attemptsByStep = new Map(); - private readonly captures = new Map(); /** * One logical call per step. Retries of the same step are further attempts of * that call, not new calls, so they share this id. @@ -351,10 +325,10 @@ export class ProviderRequestTracker { throwIfAbortedBeforeDispatch(input.abortSignal); this.input.accounting?.assertReady?.(); const step = this.step; - const capture = this.capture(step, input); + const observation = this.observe(input); throwIfAbortedBeforeDispatch(input.abortSignal); let sawOutput = false; - const attempt = this.beginAttempt(step, capture, input); + const attempt = this.beginAttempt(step, observation, input); let result: ProviderStreamResult; try { @@ -420,9 +394,9 @@ export class ProviderRequestTracker { throwIfAbortedBeforeDispatch(input.abortSignal); this.input.accounting?.assertReady?.(); const step = this.step; - const capture = this.capture(step, input); + const observation = this.observe(input); throwIfAbortedBeforeDispatch(input.abortSignal); - const attempt = this.beginAttempt(step, capture, input); + const attempt = this.beginAttempt(step, observation, input); try { const result = await input.doGenerate(); await attempt.finalize(input.abortSignal?.aborted ? 'aborted' : 'completed', { @@ -438,7 +412,7 @@ export class ProviderRequestTracker { private beginAttempt( step: number, - capture: StoredCapture, + observation: PreparedRequestObservation, input: Pick< TrackProviderStreamInput | TrackProviderGenerateInput, 'providerId' | 'modelId' | 'abortSignal' | 'historyCompactBoundary' | 'historyCompactRoute' @@ -493,7 +467,6 @@ export class ProviderRequestTracker { turnId: this.input.turnId, step, attempt, - ...(capture.artifactId ? { captureArtifactId: capture.artifactId } : {}), providerId: input.providerId, modelId: input.modelId, ...(contextWindow !== undefined ? { contextWindow } : {}), @@ -511,7 +484,7 @@ export class ProviderRequestTracker { logicalCallId, usage, contextWindow, - requestObservation: capture.material.observation, + requestObservation: observation, // Frozen when THIS request was prepared, so a checkpoint published // mid-flight by another turn cannot be sealed into a prompt built // before it existed. @@ -592,9 +565,6 @@ export class ProviderRequestTracker { providerId: accounting.providerId ?? record.providerId, modelId: record.modelId, ...(context.contextWindow !== undefined ? { contextWindow: context.contextWindow } : {}), - ...(record.captureArtifactId !== undefined - ? { captureArtifactId: record.captureArtifactId } - : {}), requestObservation: context.requestObservation, startedAt: record.startedAt, completedAt: record.completedAt, @@ -639,43 +609,10 @@ export class ProviderRequestTracker { } } - private capture( - step: number, + private observe( input: TrackProviderStreamInput | TrackProviderGenerateInput, - ): StoredCapture { - const material = preparedRequestMaterial(input.params); - const key = `${step}:${input.providerId}:${input.modelId}:${material.observation.digest}`; - const existing = this.captures.get(key); - if (existing) return existing; - - const persistArtifact = this.input.persistArtifact; - const capture: StoredCapture = { material }; - this.captures.set(key, capture); - if (persistArtifact) { - const artifactInput: PreparedRequestArtifactInput = { - ...material, - traceId: this.input.traceId, - captureId: this.input.newId(), - turnId: this.input.turnId, - step, - providerId: input.providerId, - modelId: input.modelId, - }; - // Persist the private body in parallel. Dispatch and canonical accounting - // are both allowed to finish without it; the bounded observation already - // lives on the canonical attempt. If persistence wins the race, the - // attempt also carries the optional artifact join. - try { - void persistArtifact(artifactInput) - .then((ref) => { - capture.artifactId = ref.artifactId; - }) - .catch(() => undefined); - } catch { - // A synchronous adapter failure is the same optional-artifact miss. - } - } - return capture; + ): PreparedRequestObservation { + return prepareRequestObservation(secretFreeParams(input.params)); } } @@ -685,11 +622,6 @@ function throwIfAbortedBeforeDispatch(signal: AbortSignal | undefined): void { } } -function preparedRequestMaterial(params: Record): PreparedRequestMaterial { - const safeParams = secretFreeParams(params); - return prepareRequestObservation(safeParams); -} - function secretFreeParams(params: Record): Record { const { abortSignal: _abortSignal, headers: _headers, ...safe } = params; if (!Array.isArray(safe.prompt)) return safe; diff --git a/packages/runtime/src/request-shape.ts b/packages/runtime/src/request-shape.ts index 341ba344e4..33ac26ac5c 100644 --- a/packages/runtime/src/request-shape.ts +++ b/packages/runtime/src/request-shape.ts @@ -36,13 +36,6 @@ export interface CanonicalToolSet { activeTools: string[]; } -export interface PreparedRequestMaterial { - /** Full secret-free representation for the private request artifact. */ - serializedRequest: string; - /** Bounded public observation derived from that same representation. */ - observation: PreparedRequestObservation; -} - /** * Split the registry into the full dispatch set (`providerTools`) and the * model-visible subset (`activeTools`). @@ -93,8 +86,11 @@ export function toolSchemaCharsForDiagnostics( * exact request evidence, but are not claimed to be a provider-cacheable prefix * segment. None of this is presented as the provider's final wire body. */ -export function prepareRequestObservation(payload: unknown): PreparedRequestMaterial { +export function prepareRequestObservation(payload: unknown): PreparedRequestObservation { const normalizedPayload = normalizePreparedValue(payload); + // Serialized to size and identify the request, then dropped. Keeping the + // whole body was what filled the artifact store with re-serialized copies of + // the same conversation, one per step. const serializedRequest = JSON.stringify(normalizedPayload.value); const segments: PreparedRequestObservationSegment[] = []; const parts = semanticRequestParts(payload); @@ -120,13 +116,10 @@ export function prepareRequestObservation(payload: unknown): PreparedRequestMate } return { - serializedRequest, - observation: { - schemaVersion: PREPARED_REQUEST_OBSERVATION_SCHEMA_VERSION, - digest: hashSerialized(serializedRequest), - bytes: Buffer.byteLength(serializedRequest, 'utf8'), - segments: boundPreparedRequestSegments(segments), - }, + schemaVersion: PREPARED_REQUEST_OBSERVATION_SCHEMA_VERSION, + digest: hashSerialized(serializedRequest), + bytes: Buffer.byteLength(serializedRequest, 'utf8'), + segments: boundPreparedRequestSegments(segments), }; } diff --git a/packages/storage/src/__tests__/provider-request-capture-artifact.test.ts b/packages/storage/src/__tests__/provider-request-capture-artifact.test.ts deleted file mode 100644 index fcbe5ec943..0000000000 --- a/packages/storage/src/__tests__/provider-request-capture-artifact.test.ts +++ /dev/null @@ -1,56 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -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 { createSqliteArtifactStore as createArtifactStore } from '../artifact-store.js'; -import * as providerRequestCapture from '../provider-request-capture-artifact.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( - providerRequestCapture, - '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/artifact-stores.ts b/packages/storage/src/artifact-stores.ts index 1a7a46f3a4..cea55fc60a 100644 --- a/packages/storage/src/artifact-stores.ts +++ b/packages/storage/src/artifact-stores.ts @@ -46,7 +46,6 @@ export { type ArtifactAttachmentResourceReader, type ReadImageSnapshotPlan, } from './artifact-attachments.js'; -export { persistProviderRequestCaptureArtifact } from './provider-request-capture-artifact.js'; const writerBrand: unique symbol = Symbol('InteractiveArtifactStoreWriter'); const writers = new WeakSet(); diff --git a/packages/storage/src/provider-request-capture-artifact.ts b/packages/storage/src/provider-request-capture-artifact.ts deleted file mode 100644 index 2cd1d94ba8..0000000000 --- a/packages/storage/src/provider-request-capture-artifact.ts +++ /dev/null @@ -1,48 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -import type { ArtifactRecord } from '@maka/core/artifacts'; - -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: Pick, - 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 } : {}), - }); -} From 8191886abbaffd9fd92471f8a9613cfe4603d41e Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 11:54:02 +0800 Subject: [PATCH 2/8] perf(runtime): record what a prompt was made of, not every part it was made from MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every completed call stored a `PreparedRequestObservation`: up to 256 ordered segments, each with an index, a cacheable flag, a comparison mode, a sha256 digest, a byte count and a role. One reader existed, and it did one thing with all of it — `foldPromptComposition`, into four byte totals and a capped tool list. The other four fields per segment had no reader anywhere. So the fold moves to where the request is prepared, and the attempt carries its result. `PromptComposition` lives in core, next to the record that stores it, and the diagnostics types are now aliases of it rather than a second spelling kept in step by hand. What this stops doing per model call: serializing the entire request payload to hash it, hashing each of up to 256 segments, and writing that array into the run's event log. What it still answers is exactly what the panel and `/context` asked before. Attempts recorded before this still carry their segments, and folding them on read is the only way to say what those requests were made of, so that path stays. It is the same shape `readPromptCompositionEvent` already had for the generation before it. The 256-segment cap goes with the array. The fold's output was always the bound that mattered — four kinds and 64 named tools — and that constant now has one definition the producer and the decoder share. `hasRequestObservation` on the metering anchor becomes redundant once the compat fold happens inside it: it only ever meant "this anchor has no composition", which the composition itself now says. Closes #4082 Generated-by: Claude Code --- .../src/__tests__/model-call-attempt.test.ts | 51 +++++- packages/core/src/model-call-attempt.ts | 137 +++++++++++++++- .../execution-model-composition.test.ts | 2 +- .../__tests__/latest-context-commit.test.ts | 11 +- .../src/__tests__/prompt-composition.test.ts | 61 ------- .../provider-request-telemetry.test.ts | 19 ++- .../src/__tests__/request-shape.test.ts | 150 +++++++----------- packages/runtime/src/context-diagnostics.ts | 64 +++----- .../runtime/src/latest-context-snapshot.ts | 5 +- packages/runtime/src/prompt-composition.ts | 18 +-- .../runtime/src/provider-request-telemetry.ts | 26 +-- packages/runtime/src/request-shape.ts | 137 +++------------- 12 files changed, 327 insertions(+), 354 deletions(-) diff --git a/packages/core/src/__tests__/model-call-attempt.test.ts b/packages/core/src/__tests__/model-call-attempt.test.ts index 21b84414f9..11a6490c0f 100644 --- a/packages/core/src/__tests__/model-call-attempt.test.ts +++ b/packages/core/src/__tests__/model-call-attempt.test.ts @@ -23,6 +23,7 @@ import assert from 'node:assert/strict'; import { MODEL_CALL_DIAGNOSTIC_FIELD_MAX_LENGTH, MODEL_CALL_ATTEMPT_SCHEMA_VERSION, + PROMPT_COMPOSITION_MAX_TOOLS, decodeModelCallAttempt, groupModelCallAttempts, settledAttempt, @@ -59,7 +60,55 @@ function attempt(overrides: Partial = {}): ModelCallAttempt { } describe('ModelCallAttempt codec', () => { - test('accepts one bounded prepared-request observation on the canonical attempt', () => { + test('accepts the folded prompt composition on the canonical attempt', () => { + const decoded = decodeModelCallAttempt({ + ...attempt(), + promptComposition: { + segments: [ + { kind: 'system_instructions', bytes: 400 }, + { kind: 'tool_definitions', bytes: 300 }, + ], + tools: [{ name: 'Bash', bytes: 300 }], + remainingTools: { count: 2, bytes: 40 }, + unlabelledToolBytes: 10, + }, + }); + + assert.deepEqual(decoded.promptComposition?.tools, [{ name: 'Bash', bytes: 300 }]); + }); + + test('rejects a composition that names one bucket twice', () => { + // Two rows for one kind would let a reader's total disagree with the + // store's, and nothing downstream could tell which was meant. + assert.throws(() => + decodeModelCallAttempt({ + ...attempt(), + promptComposition: { + segments: [ + { kind: 'messages', bytes: 10 }, + { kind: 'messages', bytes: 20 }, + ], + }, + }), + ); + }); + + test('rejects a composition carrying more named tools than the fold can produce', () => { + assert.throws(() => + decodeModelCallAttempt({ + ...attempt(), + promptComposition: { + segments: [{ kind: 'tool_definitions', bytes: 650 }], + tools: Array.from({ length: PROMPT_COMPOSITION_MAX_TOOLS + 1 }, (_, index) => ({ + name: `tool-${index}`, + bytes: 10, + })), + }, + }), + ); + }); + + test('still decodes the prepared-request observation recorded before the fold', () => { const decoded = decodeModelCallAttempt({ ...attempt(), requestObservation: { diff --git a/packages/core/src/model-call-attempt.ts b/packages/core/src/model-call-attempt.ts index bbb9c03015..26b8854af7 100644 --- a/packages/core/src/model-call-attempt.ts +++ b/packages/core/src/model-call-attempt.ts @@ -109,11 +109,55 @@ export interface PreparedRequestObservationSegment { label?: string; } +export type PromptCompositionSegmentKind = + | 'system_instructions' + | 'tool_definitions' + | 'messages' + | 'other'; + +/** + * One part of a prepared request, measured in bytes of serialized request. + * + * Bytes only. `bytes / 4` is a rule of thumb over serialized JSON — wrong in a + * direction nobody here can correct for, badly so for an attachment's base64 — + * so the estimate is made where it is shown and labelled `≈` there. A figure + * rounded into this contract could no longer be labelled at all (#2323). + */ +export interface PromptCompositionSegment { + kind: PromptCompositionSegmentKind; + bytes: number; +} + +/** One tool's schema, sized on its own, so a reader knows which to remove. */ +export interface PromptCompositionTool { + name: string; + bytes: number; +} + +/** + * What a prepared request was made of, folded at the moment it was prepared. + * + * This is the whole durable answer to "what filled the context". The per-part + * detail it folds is not kept: every reader wanted these buckets, so storing + * the parts meant writing hundreds of rows per call for a fold nobody could + * do differently. + */ +export interface PromptComposition { + segments: PromptCompositionSegment[]; + /** The largest named tool schemas, largest first; bounded at the fold. */ + tools?: PromptCompositionTool[]; + /** Everything past the named rows, so the bytes still account for every tool. */ + remainingTools?: { count: number; bytes: number }; + /** Tool schemas the payload did not name, so their bytes are still counted. */ + unlabelledToolBytes?: number; +} + /** * Bounded, secret-free observation of one prepared semantic model request. * - * This is not the provider wire body, and no copy of that body is kept: the - * request is built from the conversation the run already stores. + * Historical only: `promptComposition` replaced it. Attempts recorded before + * that still carry it, and folding their segments is the only way to say what + * those requests were made of, so it stays decodable. */ export interface PreparedRequestObservation { schemaVersion: typeof PREPARED_REQUEST_OBSERVATION_SCHEMA_VERSION; @@ -173,7 +217,9 @@ export interface ModelCallAttempt { * the key, so it stays decodable. */ captureArtifactId?: string; - /** Semantic request actually prepared for this dispatched physical attempt. */ + /** What the request prepared for this dispatched physical attempt was made of. */ + promptComposition?: PromptComposition; + /** Replaced by `promptComposition`; still read on attempts recorded before it. */ requestObservation?: PreparedRequestObservation; startedAt: number; @@ -232,6 +278,7 @@ const MODEL_CALL_ATTEMPT_SHAPE = defineObjectShape()( 'historyCompactRoute', 'contextWindow', 'captureArtifactId', + 'promptComposition', 'requestObservation', 'timeToFirstTokenMs', 'finishReason', @@ -279,6 +326,41 @@ const PREPARED_REQUEST_SEGMENT_KINDS: readonly PreparedRequestObservationSegment 'provider_options', ]; +const PROMPT_COMPOSITION_SHAPE = defineObjectShape()( + ['segments'], + ['tools', 'remainingTools', 'unlabelledToolBytes'], +); + +const PROMPT_COMPOSITION_SEGMENT_SHAPE = defineObjectShape()( + ['kind', 'bytes'], + [], +); + +const PROMPT_COMPOSITION_TOOL_SHAPE = defineObjectShape()( + ['name', 'bytes'], + [], +); + +const PROMPT_COMPOSITION_REMAINING_TOOLS_SHAPE = defineObjectShape<{ + count: number; + bytes: number; +}>()(['count', 'bytes'], []); + +const PROMPT_COMPOSITION_SEGMENT_KINDS: readonly PromptCompositionSegmentKind[] = [ + 'system_instructions', + 'tool_definitions', + 'messages', + 'other', +]; + +/** + * The fold names one tool per row, so the row count is what bounds this record. + * Generous enough for a normal registry, small enough that a pathological one + * cannot make an attempt unbounded. Exported because the fold that produces + * these rows has to cut at the same number the decoder accepts. + */ +export const PROMPT_COMPOSITION_MAX_TOOLS = 64; + function isNonEmptyString(value: unknown): value is string { return typeof value === 'string' && value.length > 0; } @@ -345,6 +427,54 @@ function isPreparedRequestObservationSegment( ); } +function isPromptComposition(value: unknown): value is PromptComposition { + if (!isRecord(value) || !hasExactShape(value, PROMPT_COMPOSITION_SHAPE)) return false; + if (!Array.isArray(value.segments) || !value.segments.every(isPromptCompositionSegment)) { + return false; + } + // One kind per row: a fold that named the same bucket twice would let a + // reader's total disagree with the store's. + const kinds = value.segments.map((segment) => segment.kind); + if (new Set(kinds).size !== kinds.length) return false; + if (value.tools !== undefined) { + if (!Array.isArray(value.tools) || value.tools.length > PROMPT_COMPOSITION_MAX_TOOLS) { + return false; + } + if (!value.tools.every(isPromptCompositionTool)) return false; + } + if ( + value.remainingTools !== undefined && + !( + isRecord(value.remainingTools) && + hasExactShape(value.remainingTools, PROMPT_COMPOSITION_REMAINING_TOOLS_SHAPE) && + isNonNegativeInteger(value.remainingTools.count) && + isNonNegativeInteger(value.remainingTools.bytes) + ) + ) { + return false; + } + return value.unlabelledToolBytes === undefined || isNonNegativeInteger(value.unlabelledToolBytes); +} + +function isPromptCompositionSegment(value: unknown): value is PromptCompositionSegment { + return ( + isRecord(value) && + hasExactShape(value, PROMPT_COMPOSITION_SEGMENT_SHAPE) && + (PROMPT_COMPOSITION_SEGMENT_KINDS as readonly unknown[]).includes(value.kind) && + isNonNegativeInteger(value.bytes) + ); +} + +function isPromptCompositionTool(value: unknown): value is PromptCompositionTool { + return ( + isRecord(value) && + hasExactShape(value, PROMPT_COMPOSITION_TOOL_SHAPE) && + typeof value.name === 'string' && + value.name.length <= PREPARED_REQUEST_OBSERVATION_TEXT_MAX_LENGTH && + isNonNegativeInteger(value.bytes) + ); +} + function isPreparedRequestObservation(value: unknown): value is PreparedRequestObservation { if (!isRecord(value) || !hasExactShape(value, PREPARED_REQUEST_OBSERVATION_SHAPE)) return false; return ( @@ -425,6 +555,7 @@ export function decodeModelCallAttempt(value: unknown): ModelCallAttempt { isNonEmptyString(value.modelId) && isOptionalNonNegativeNumber(value.contextWindow) && isOptionalString(value.captureArtifactId) && + (value.promptComposition === undefined || isPromptComposition(value.promptComposition)) && (value.requestObservation === undefined || isPreparedRequestObservation(value.requestObservation)) && isFiniteNumber(value.startedAt) && diff --git a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts index d7394f75a0..b94f08d66a 100644 --- a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts @@ -2029,7 +2029,7 @@ test('production Host executes a canonical ai-sdk Session against a real provide const capturedRequestCount = mainRequests.length + compactRequests.length; const attempts = await waitForCanonicalAttempts(usageStores, session.id, capturedRequestCount); assert.equal(attempts.length, capturedRequestCount); - assert.ok(attempts.every((attempt) => attempt.requestObservation)); + assert.ok(attempts.every((attempt) => attempt.promptComposition)); const contextDiagnostics = await composition.handlers['context.diagnostics.query']( { sessionId: session.id }, connectionContext, diff --git a/packages/runtime/src/__tests__/latest-context-commit.test.ts b/packages/runtime/src/__tests__/latest-context-commit.test.ts index f3eb59d021..733cdea19f 100644 --- a/packages/runtime/src/__tests__/latest-context-commit.test.ts +++ b/packages/runtime/src/__tests__/latest-context-commit.test.ts @@ -38,7 +38,7 @@ import { MockLanguageModelV4, simulateReadableStream } from 'ai/test'; import type { LanguageModelV4StreamPart } from '@ai-sdk/provider'; import { decodeModelCallAttempt, - PREPARED_REQUEST_OBSERVATION_MAX_SEGMENTS, + PROMPT_COMPOSITION_MAX_TOOLS, type ModelCallAttempt, } from '@maka/core/model-call-attempt'; import { createSqliteAgentRunStore } from '@maka/storage/agent-run-store'; @@ -154,11 +154,10 @@ test('a real send seals its observation into SQLite and reconstructs it after re ) ).flat(); assert.equal(canonicalAttempts.length, 1); - const observation = canonicalAttempts[0]?.requestObservation; - assert.ok(observation); - assert.ok(observation.segments.length <= PREPARED_REQUEST_OBSERVATION_MAX_SEGMENTS); - assert.ok(observation.segments.length > 0); - assert.ok(observation.segments.every((segment) => segment.comparison === 'exact')); + const composition = canonicalAttempts[0]?.promptComposition; + assert.ok(composition); + assert.ok(composition.segments.length > 0); + assert.ok((composition.tools?.length ?? 0) <= PROMPT_COMPOSITION_MAX_TOOLS); let coldScans = 0; const cold = await readLatestContextDiagnostics( diff --git a/packages/runtime/src/__tests__/prompt-composition.test.ts b/packages/runtime/src/__tests__/prompt-composition.test.ts index ce0f14af0f..039f7fc61e 100644 --- a/packages/runtime/src/__tests__/prompt-composition.test.ts +++ b/packages/runtime/src/__tests__/prompt-composition.test.ts @@ -30,7 +30,6 @@ import { PROVIDER_REQUEST_ATTEMPT_EVENT_TYPE, } from '../prompt-composition.js'; import type { SizedRequestSegment } from '../prompt-composition.js'; -import { prepareRequestObservation } from '../request-shape.js'; function segment(overrides: Partial = {}): SizedRequestSegment { return { kind: 'message', bytes: 10, ...overrides }; @@ -113,66 +112,6 @@ describe('foldPromptComposition', () => { }); }); -describe('a real observation survives the whole chain into one fold', () => { - test('prepare -> canonical segments -> fold keeps the same breakdown', () => { - // Every other test here writes its own segments, so a field renamed on one - // side and not the other would pass all of them; and the decode side reads - // `label` and `bytes` off an untyped record, so a hand-written fixture - // agrees with itself by construction. This is the one test where the - // writer, the storage encoding, the reader and the fold all meet. - const material = prepareRequestObservation({ - prompt: [ - { role: 'system', content: 'you are a helpful assistant' }, - { role: 'user', content: 'hello' }, - ], - tools: [ - { name: 'Bash', description: 'Run a command', inputSchema: { type: 'object' } }, - { name: 'Read', inputSchema: { type: 'object' } }, - ], - providerOptions: { anthropic: { thinking: { type: 'enabled' } } }, - }); - - const composition = foldPromptComposition(material.segments); - - assert.deepEqual( - composition?.tools?.map((tool) => tool.name), - ['Bash', 'Read'], - 'the tool names survive capture, storage shape and fold', - ); - assert.equal(composition?.unlabelledToolBytes, undefined, 'both tools were named'); - assert.deepEqual( - composition?.segments.map((part) => part.kind), - ['system_instructions', 'tool_definitions', 'messages', 'other'], - ); - assert.equal( - composition?.segments.find((part) => part.kind === 'tool_definitions')?.bytes, - composition!.tools!.reduce((carry, tool) => carry + tool.bytes, 0), - 'the per-tool rows sum to the tool total above them', - ); - }); - - test('a single-tool bounded remainder still counts as one remaining tool', () => { - const material = prepareRequestObservation({ - prompt: [ - { role: 'system', content: 'system' }, - { role: 'user', content: 'hello' }, - { role: 'assistant', content: 'hi' }, - ], - tools: Array.from({ length: 253 }, (_, index) => ({ - name: `tool-${String(index).padStart(3, '0')}`, - inputSchema: { type: 'object' }, - })), - providerOptions: { anthropic: { thinking: { type: 'enabled' } } }, - }); - - assert.equal(material.segments.length, 256); - const composition = foldPromptComposition(material.segments); - assert.equal(composition?.tools?.length, 64); - assert.equal(composition?.remainingTools?.count, 189); - assert.equal(composition?.unlabelledToolBytes, undefined); - }); -}); - describe('readPromptCompositionEvent', () => { const event = (data: unknown) => ({ type: PROVIDER_REQUEST_ATTEMPT_EVENT_TYPE, data }); diff --git a/packages/runtime/src/__tests__/provider-request-telemetry.test.ts b/packages/runtime/src/__tests__/provider-request-telemetry.test.ts index c6c881a7db..9fe530f93f 100644 --- a/packages/runtime/src/__tests__/provider-request-telemetry.test.ts +++ b/packages/runtime/src/__tests__/provider-request-telemetry.test.ts @@ -352,8 +352,8 @@ describe('provider request tracker', () => { { step: 2, attempt: 1, status: 'completed' }, ], ); - // Both attempts observed the same request, so they share its identity. - assert.equal(attempts[0]?.requestObservation?.digest, attempts[1]?.requestObservation?.digest); + // Both attempts measured the same request, so they describe it the same way. + assert.deepEqual(attempts[0]?.promptComposition, attempts[1]?.promptComposition); assert.equal(attempts[1]?.cacheReadInputTokens, 4); assert.equal(attempts[1]?.cacheMissInputTokens, 6); }); @@ -400,10 +400,10 @@ describe('provider request tracker', () => { })), [{ status: 'completed', finishReason: 'stop', inputTokens: 7, outputTokens: 3 }], ); - assert.ok(attempts[0]?.requestObservation); + assert.ok(attempts[0]?.promptComposition); }); - test('derives a canonical opaque observation from a redacted request', async () => { + test('keeps a redacted request out of the canonical attempt', async () => { const attempts: ModelCallAttempt[] = []; const tracker = new telemetry.ProviderRequestTracker({ traceId: 'compaction-trace', @@ -458,7 +458,7 @@ describe('provider request tracker', () => { }); assert.equal(attempts.length, 1); - assert.equal(attempts[0]?.requestObservation?.segments[0]?.comparison, 'opaque'); + assert.ok(attempts[0]?.promptComposition); assert.doesNotMatch(JSON.stringify(attempts[0]), /cmp_secret|OPAQUE_ENCRYPTED_STATE/); }); @@ -903,9 +903,9 @@ describe('canonical model-call accounting', () => { assert.equal(attempt.costUsd, undefined); }); - test('carries the bounded request observation on the canonical attempt', async () => { - // The observation is the whole record of what was sent: nothing stores a - // copy of the request body for it to be joined against. + test('carries the folded prompt composition on the canonical attempt', async () => { + // The composition is the whole record of what the prompt was made of: + // nothing stores the parts it folds, or a copy of the request body. const recorded: ModelCallAttempt[] = []; const tracker = accountingTracker({ record: ({ attempt }) => { @@ -924,8 +924,7 @@ describe('canonical model-call accounting', () => { const attempt = decodeModelCallAttempt(recorded[0]); assert.equal(attempt.usageBasis, 'reported'); assert.equal(attempt.captureArtifactId, undefined, 'nothing writes a capture join any more'); - assert.match(attempt.requestObservation?.digest ?? '', /^sha256:[a-f0-9]{64}$/); - assert.ok((attempt.requestObservation?.segments.length ?? 0) > 0); + assert.ok((attempt.promptComposition?.segments.length ?? 0) > 0); }); test('an unresolvable price records unpriced rather than zero', async () => { diff --git a/packages/runtime/src/__tests__/request-shape.test.ts b/packages/runtime/src/__tests__/request-shape.test.ts index e102907374..6647293b59 100644 --- a/packages/runtime/src/__tests__/request-shape.test.ts +++ b/packages/runtime/src/__tests__/request-shape.test.ts @@ -20,9 +20,10 @@ import assert from 'node:assert/strict'; import { describe, test } from 'node:test'; +import { PROMPT_COMPOSITION_MAX_TOOLS } from '@maka/core/model-call-attempt'; import { canonicalizeToolSet, - prepareRequestObservation, + preparedPromptComposition, toolSchemaCharsForDiagnostics, } from '../request-shape.js'; import type { MakaTool } from '../tool-runtime.js'; @@ -61,131 +62,92 @@ describe('canonicalizeToolSet active allow-list', () => { }); }); -describe('prepared request observation', () => { - test('sizes and identifies the whole request, not just its named segments', () => { - const material = prepareRequestObservation({ - prompt: [{ role: 'user', content: 'hello' }], - maxOutputTokens: 1_024, +describe('prepared prompt composition', () => { + test('folds every semantic part into its bucket and names the tools', () => { + const composition = preparedPromptComposition({ + prompt: [ + { role: 'system', content: 'system' }, + { role: 'user', content: 'hello' }, + ], + tools: [{ name: 'Bash', inputSchema: { type: 'object' } }, { inputSchema: {} }], + providerOptions: { anthropic: { thinking: { type: 'enabled' } } }, }); - assert.match(material.digest, /^sha256:[a-f0-9]{64}$/); - // `maxOutputTokens` is part of the request but is not a segment, so the - // total has to exceed what the segments account for. - assert.ok( - material.bytes > material.segments.reduce((total, segment) => total + segment.bytes, 0), + assert.deepEqual( + composition?.segments.map((segment) => segment.kind), + ['system_instructions', 'tool_definitions', 'messages', 'other'], ); + assert.deepEqual( + composition?.tools?.map((tool) => tool.name), + ['Bash'], + ); + // The unnamed tool's schema is still counted; it just cannot be listed. + assert.ok((composition?.unlabelledToolBytes ?? 0) > 0); }); - test('serializes non-JSON values without collapsing their semantic identity', () => { - const observed = prepareRequestObservation({ - bigint: 42n, - missing: undefined, - createdAt: new Date('2026-08-31T00:00:00.000Z'), - headers: new Map([['x-observation', 'present']]), - }); - const plain = prepareRequestObservation({ - bigint: '42', - missing: '[undefined]', - createdAt: '2026-08-31T00:00:00.000Z', - headers: { 'x-observation': 'present' }, - }); - - assert.notEqual(observed.digest, plain.digest); - }); - - test('preserves the semantic identity of binary request content', () => { - const observe = (byte: number) => - prepareRequestObservation({ - prompt: [ - { - role: 'user', - content: [ - { - type: 'file', - data: { type: 'data', data: new Uint8Array([byte]) }, - mediaType: 'application/octet-stream', - }, - ], - }, - ], - }); - - const first = observe(1); - const second = observe(2); - assert.notEqual(first.digest, second.digest); - assert.notEqual(first.segments[0]?.digest, second.segments[0]?.digest); - assert.equal(first.segments[0]?.comparison, 'exact'); - }); - - test('marks redacted compaction content comparison-opaque', () => { - const material = prepareRequestObservation({ + test('sizes non-JSON values rather than dropping them', () => { + const composition = preparedPromptComposition({ prompt: [ { - role: 'assistant', + role: 'user', content: [ { - type: 'custom', - kind: 'openai.compaction', - providerOptions: { openai: { redacted: true } }, + type: 'file', + data: { type: 'data', data: new Uint8Array([1, 2, 3]) }, + mediaType: 'application/octet-stream', }, ], }, ], }); - assert.equal(material.segments[0]?.kind, 'message'); - assert.equal(material.segments[0]?.comparison, 'opaque'); + assert.ok((composition?.segments[0]?.bytes ?? 0) > 0); }); - test('bounds ordered segments without dropping their count or bytes', () => { + test('folds a long conversation into one row without losing its bytes', () => { const prompt = Array.from({ length: 1_000 }, (_, index) => ({ role: 'user', content: `message-${index}`, })); - const material = prepareRequestObservation({ prompt }); const expectedBytes = prompt.reduce( (total, message) => - total + prepareRequestObservation({ prompt: [message] }).segments[0]!.bytes, + total + (preparedPromptComposition({ prompt: [message] })?.segments[0]?.bytes ?? 0), 0, ); - assert.ok(material.segments.length <= 256); - assert.equal( - material.segments.reduce((total, segment) => total + segment.bytes, 0), - expectedBytes, - ); - assert.equal( - material.segments.reduce((total, segment) => total + (segment.representedSegments ?? 1), 0), - prompt.length, + const composition = preparedPromptComposition({ prompt }); + assert.deepEqual( + composition?.segments.map((segment) => segment.kind), + ['messages'], ); - assert.equal(material.segments.at(-1)?.comparison, 'opaque'); + assert.equal(composition?.segments[0]?.bytes, expectedBytes); }); - test('records semantic segments in provider-prefix order and labels only tools', () => { - const material = prepareRequestObservation({ - prompt: [ - { role: 'system', content: 'system' }, - { role: 'user', content: 'hello' }, - ], - tools: [{ name: 'Bash', inputSchema: { type: 'object' } }, { inputSchema: {} }], - providerOptions: { anthropic: { thinking: { type: 'enabled' } } }, + test('names the largest tools and carries the rest as a counted remainder', () => { + const composition = preparedPromptComposition({ + tools: Array.from({ length: PROMPT_COMPOSITION_MAX_TOOLS + 5 }, (_, index) => ({ + name: `tool-${String(index).padStart(3, '0')}`, + inputSchema: { type: 'object', padding: 'x'.repeat(index) }, + })), }); + assert.equal(composition?.tools?.length, PROMPT_COMPOSITION_MAX_TOOLS); + assert.equal(composition?.remainingTools?.count, 5); + assert.ok((composition?.remainingTools?.bytes ?? 0) > 0); + // Largest first, so what a reader could remove is at the top. + const bytes = composition?.tools?.map((tool) => tool.bytes) ?? []; assert.deepEqual( - material.segments.map(({ kind, index, cacheable, role, label }) => ({ - kind, - index, - cacheable, - ...(role ? { role } : {}), - ...(label ? { label } : {}), - })), - [ - { kind: 'tool_schema', index: 0, cacheable: true, label: 'Bash' }, - { kind: 'tool_schema', index: 1, cacheable: true }, - { kind: 'system_prompt', index: 0, cacheable: true }, - { kind: 'message', index: 0, cacheable: true, role: 'user' }, - { kind: 'provider_options', index: 0, cacheable: false }, - ], + bytes, + [...bytes].sort((left, right) => right - left), ); + // Every tool byte is still accounted for, named or not. + assert.equal( + bytes.reduce((total, size) => total + size, 0) + (composition?.remainingTools?.bytes ?? 0), + composition?.segments.find((segment) => segment.kind === 'tool_definitions')?.bytes, + ); + }); + + test('has nothing to say about an empty request', () => { + assert.equal(preparedPromptComposition({}), undefined); }); }); diff --git a/packages/runtime/src/context-diagnostics.ts b/packages/runtime/src/context-diagnostics.ts index 81416800d7..5fd98bc74c 100644 --- a/packages/runtime/src/context-diagnostics.ts +++ b/packages/runtime/src/context-diagnostics.ts @@ -22,7 +22,14 @@ import { type AgentRunEvent, type AgentRunStore, } from '@maka/core/agent-run'; -import { decodeModelCallAttempt, type ModelCallAttempt } from '@maka/core/model-call-attempt'; +import { + decodeModelCallAttempt, + type ModelCallAttempt, + type PromptComposition, + type PromptCompositionSegment, + type PromptCompositionSegmentKind, + type PromptCompositionTool, +} from '@maka/core/model-call-attempt'; import { foldPromptComposition, PROVIDER_REQUEST_ATTEMPT_EVENT_TYPE, @@ -41,30 +48,17 @@ import { export type ContextDiagnosticsUnavailableReason = 'no_completed_request' | 'trace_unavailable'; -export type ContextDiagnosticsSegmentKind = - | 'system_instructions' - | 'tool_definitions' - | 'messages' - | 'other'; - /** - * One part of the latest request, measured in bytes of serialized request. + * The composition vocabulary is the stored one. * - * Bytes only. `bytes / 4` is a rule of thumb over serialized JSON — wrong in a - * direction nobody here can correct for, badly so for an attachment's base64 — - * so the estimate is made where it is shown and labelled `≈` there. A figure - * rounded into this contract could no longer be labelled at all (#2323). + * These names are what a `ModelCallAttempt` durably carries, so serving them + * under a second set of diagnostic-only types would be two spellings of one + * fact, kept in step by hand. */ -export interface ContextDiagnosticsSegment { - kind: ContextDiagnosticsSegmentKind; - bytes: number; -} - -/** One tool's schema, sized on its own, so a reader knows which to remove. */ -export interface ContextDiagnosticsTool { - name: string; - bytes: number; -} +export type ContextDiagnosticsSegmentKind = PromptCompositionSegmentKind; +export type ContextDiagnosticsSegment = PromptCompositionSegment; +export type ContextDiagnosticsTool = PromptCompositionTool; +export type ContextDiagnosticsComposition = PromptComposition; export interface ContextDiagnosticsCompaction { kind: 'history'; @@ -99,16 +93,6 @@ export type ContextDiagnostics = compaction?: ContextDiagnosticsCompaction; }; -export interface ContextDiagnosticsComposition { - segments: ContextDiagnosticsSegment[]; - /** The largest named tool schemas, largest first; bounded at the fold. */ - tools?: ContextDiagnosticsTool[]; - /** Everything past the named rows, so the bytes still account for every tool. */ - remainingTools?: { count: number; bytes: number }; - /** Tool schemas the payload did not name, so their bytes are still counted. */ - unlabelledToolBytes?: number; -} - type ContextRunStore = Pick< AgentRunStore, 'readEvents' | 'readEventProjection' | 'readEventLedgerRevision' | 'repairEventProjection' @@ -247,7 +231,7 @@ async function rebuildContextFromLedger( // fallback for provider/model/status/timing/usage or for a different attempt. const composition = resolved.composition ?? - (anchor && !anchor.hasRequestObservation + (anchor && !anchor.composition ? exactHistoricalComposition(anchor, historicalAttempts) : undefined); const snapshot: LatestContextSnapshot = { @@ -338,7 +322,6 @@ function legacyProviderAnchor(event: AgentRunEvent): MeteringAnchor | undefined modelId, startedAt: typeof startedAt === 'number' ? startedAt : completedAt, completedAt, - hasRequestObservation: false, ...(typeof data.inputTokens === 'number' ? { inputTokens: data.inputTokens } : {}), ...(typeof data.contextWindow === 'number' ? { contextWindow: data.contextWindow } : {}), ...(composition ? { composition } : {}), @@ -377,7 +360,6 @@ interface MeteringAnchor { cacheReadInputTokens?: number; contextWindow?: number; composition?: ContextDiagnosticsComposition; - hasRequestObservation: boolean; } interface CheckpointCandidate { @@ -395,9 +377,14 @@ function meteringAnchor(event: AgentRunEvent): MeteringAnchor | undefined { return undefined; } if (attempt.callKind !== 'main' || attempt.status !== 'completed') return undefined; - const composition = attempt.requestObservation - ? foldPromptComposition(attempt.requestObservation.segments) - : undefined; + // Attempts recorded before the fold moved onto the record still carry their + // parts, and folding them here is the only way to say what those requests + // were made of. Current attempts arrive already folded. + const composition = + attempt.promptComposition ?? + (attempt.requestObservation + ? foldPromptComposition(attempt.requestObservation.segments) + : undefined); return { attemptId: attempt.attemptId, traceId: attempt.traceId, @@ -405,7 +392,6 @@ function meteringAnchor(event: AgentRunEvent): MeteringAnchor | undefined { modelId: attempt.modelId, startedAt: attempt.startedAt, completedAt: attempt.completedAt, - hasRequestObservation: attempt.requestObservation !== undefined, ...(attempt.inputTokens !== undefined ? { inputTokens: attempt.inputTokens } : {}), ...(attempt.cacheReadInputTokens !== undefined ? { cacheReadInputTokens: attempt.cacheReadInputTokens } diff --git a/packages/runtime/src/latest-context-snapshot.ts b/packages/runtime/src/latest-context-snapshot.ts index d2f53652d6..70d320ef64 100644 --- a/packages/runtime/src/latest-context-snapshot.ts +++ b/packages/runtime/src/latest-context-snapshot.ts @@ -24,11 +24,11 @@ import { } from '@maka/core/agent-run'; export { LATEST_CONTEXT_PROJECTION_TYPE }; +import type { PromptComposition } from '@maka/core/model-call-attempt'; import type { ContextDiagnosticsCompaction, ContextDiagnosticsComposition, } from './context-diagnostics.js'; -import { foldPromptComposition, type SizedRequestSegment } from './prompt-composition.js'; /** * One request's context, frozen by the transaction that committed it (#2323). @@ -78,10 +78,9 @@ export interface LatestContextSnapshot { */ export function latestContextProjectionInput( attempt: LatestContextFacts, - segments: readonly SizedRequestSegment[] | undefined, + composition: PromptComposition | undefined, compaction: ContextDiagnosticsCompaction | undefined, ): LatestContextProjectionInput { - const composition = segments ? foldPromptComposition(segments) : undefined; const snapshot: LatestContextSnapshot = { schemaVersion: LATEST_CONTEXT_SNAPSHOT_SCHEMA_VERSION, attemptId: attempt.attemptId, diff --git a/packages/runtime/src/prompt-composition.ts b/packages/runtime/src/prompt-composition.ts index c6863c913e..c2cacf1c8d 100644 --- a/packages/runtime/src/prompt-composition.ts +++ b/packages/runtime/src/prompt-composition.ts @@ -21,7 +21,10 @@ import type { ContextDiagnosticsComposition, ContextDiagnosticsSegment, } from './context-diagnostics.js'; -import type { PreparedRequestObservationSegmentKind } from '@maka/core/model-call-attempt'; +import { + PROMPT_COMPOSITION_MAX_TOOLS, + type PreparedRequestObservationSegmentKind, +} from '@maka/core/model-call-attempt'; /** * The three fields a fold needs, and no more. @@ -97,8 +100,8 @@ export function foldPromptComposition( // downstream only moves the cliff: the 257th tool would fail the whole query // instead of being summarised. What falls below the cut is carried as a // remainder, so the rows still account for every tool byte. - const tools = ranked.slice(0, MAX_TOOL_ROWS); - const remainder = ranked.slice(MAX_TOOL_ROWS); + const tools = ranked.slice(0, PROMPT_COMPOSITION_MAX_TOOLS); + const remainder = ranked.slice(PROMPT_COMPOSITION_MAX_TOOLS); const remainingToolCount = remainder.length + boundedToolCount; const remainingToolBytes = remainder.reduce((carry, tool) => carry + tool.bytes, 0) + boundedToolBytes; @@ -178,15 +181,6 @@ function isNonNegativeInteger(value: unknown): value is number { return Number.isSafeInteger(value) && (value as number) >= 0; } -/** - * How many tools the fold names individually. - * - * Generous enough that a normal registry is listed whole, small enough that a - * pathological one cannot make this record unbounded. The panel shows fewer - * still; this is the bound on what crosses a wire and sits in a projection. - */ -const MAX_TOOL_ROWS = 64; - const KIND_ORDER: readonly PreparedRequestObservationSegmentKind[] = [ 'system_prompt', 'tool_schema', diff --git a/packages/runtime/src/provider-request-telemetry.ts b/packages/runtime/src/provider-request-telemetry.ts index 1171f697e3..75d892c4fe 100644 --- a/packages/runtime/src/provider-request-telemetry.ts +++ b/packages/runtime/src/provider-request-telemetry.ts @@ -23,10 +23,10 @@ import { type ModelCallAttempt, type ModelCallKind, type ModelCallUsageBasis, - type PreparedRequestObservation, + type PromptComposition, } from '@maka/core/model-call-attempt'; import type { PricingConfig } from '@maka/core/usage-stats/types'; -import { prepareRequestObservation } from './request-shape.js'; +import { preparedPromptComposition } from './request-shape.js'; import { rawFinishReasonString } from './model-protocol.js'; import { providerFailureDiagnostic, @@ -325,10 +325,10 @@ export class ProviderRequestTracker { throwIfAbortedBeforeDispatch(input.abortSignal); this.input.accounting?.assertReady?.(); const step = this.step; - const observation = this.observe(input); + const composition = this.observe(input); throwIfAbortedBeforeDispatch(input.abortSignal); let sawOutput = false; - const attempt = this.beginAttempt(step, observation, input); + const attempt = this.beginAttempt(step, composition, input); let result: ProviderStreamResult; try { @@ -394,9 +394,9 @@ export class ProviderRequestTracker { throwIfAbortedBeforeDispatch(input.abortSignal); this.input.accounting?.assertReady?.(); const step = this.step; - const observation = this.observe(input); + const composition = this.observe(input); throwIfAbortedBeforeDispatch(input.abortSignal); - const attempt = this.beginAttempt(step, observation, input); + const attempt = this.beginAttempt(step, composition, input); try { const result = await input.doGenerate(); await attempt.finalize(input.abortSignal?.aborted ? 'aborted' : 'completed', { @@ -412,7 +412,7 @@ export class ProviderRequestTracker { private beginAttempt( step: number, - observation: PreparedRequestObservation, + composition: PromptComposition | undefined, input: Pick< TrackProviderStreamInput | TrackProviderGenerateInput, 'providerId' | 'modelId' | 'abortSignal' | 'historyCompactBoundary' | 'historyCompactRoute' @@ -484,7 +484,7 @@ export class ProviderRequestTracker { logicalCallId, usage, contextWindow, - requestObservation: observation, + promptComposition: composition, // Frozen when THIS request was prepared, so a checkpoint published // mid-flight by another turn cannot be sealed into a prompt built // before it existed. @@ -528,7 +528,7 @@ export class ProviderRequestTracker { logicalCallId: string; usage: ProviderRequestUsage | undefined; contextWindow: number | undefined; - requestObservation: PreparedRequestObservation; + promptComposition: PromptComposition | undefined; historyCompactBoundary: ContextDiagnosticsCompaction | undefined; historyCompactRoute: HistoryCompactRoute | undefined; }, @@ -565,7 +565,7 @@ export class ProviderRequestTracker { providerId: accounting.providerId ?? record.providerId, modelId: record.modelId, ...(context.contextWindow !== undefined ? { contextWindow: context.contextWindow } : {}), - requestObservation: context.requestObservation, + ...(context.promptComposition ? { promptComposition: context.promptComposition } : {}), startedAt: record.startedAt, completedAt: record.completedAt, latencyMs: record.latencyMs, @@ -596,7 +596,7 @@ export class ProviderRequestTracker { attempt.callKind === 'main' && attempt.status === 'completed' ? latestContextProjectionInput( attempt, - attempt.requestObservation?.segments, + attempt.promptComposition, context.historyCompactBoundary, ) : undefined; @@ -611,8 +611,8 @@ export class ProviderRequestTracker { private observe( input: TrackProviderStreamInput | TrackProviderGenerateInput, - ): PreparedRequestObservation { - return prepareRequestObservation(secretFreeParams(input.params)); + ): PromptComposition | undefined { + return preparedPromptComposition(secretFreeParams(input.params)); } } diff --git a/packages/runtime/src/request-shape.ts b/packages/runtime/src/request-shape.ts index 33ac26ac5c..e98d269c22 100644 --- a/packages/runtime/src/request-shape.ts +++ b/packages/runtime/src/request-shape.ts @@ -20,13 +20,10 @@ import { Buffer } from 'node:buffer'; import { createHash } from 'node:crypto'; import { - PREPARED_REQUEST_OBSERVATION_MAX_SEGMENTS, - PREPARED_REQUEST_OBSERVATION_SCHEMA_VERSION, PREPARED_REQUEST_OBSERVATION_TEXT_MAX_LENGTH, - type PreparedRequestObservation, - type PreparedRequestObservationSegment, - type PreparedRequestObservationSegmentKind, + type PromptComposition, } from '@maka/core/model-call-attempt'; +import { foldPromptComposition, type SizedRequestSegment } from './prompt-composition.js'; import { toJSONSchema } from 'zod'; import type { MakaTool } from './tool-runtime.js'; @@ -86,75 +83,42 @@ export function toolSchemaCharsForDiagnostics( * exact request evidence, but are not claimed to be a provider-cacheable prefix * segment. None of this is presented as the provider's final wire body. */ -export function prepareRequestObservation(payload: unknown): PreparedRequestObservation { - const normalizedPayload = normalizePreparedValue(payload); - // Serialized to size and identify the request, then dropped. Keeping the - // whole body was what filled the artifact store with re-serialized copies of - // the same conversation, one per step. - const serializedRequest = JSON.stringify(normalizedPayload.value); - const segments: PreparedRequestObservationSegment[] = []; +export function preparedPromptComposition(payload: unknown): PromptComposition | undefined { + const segments: SizedRequestSegment[] = []; const parts = semanticRequestParts(payload); - for (const [index, tool] of parts.tools.entries()) { - segments.push(preparedSegment('tool_schema', index, tool, true, undefined, toolLabel(tool))); - } + for (const tool of parts.tools) segments.push(sizedSegment('tool_schema', tool, toolLabel(tool))); if (parts.instructions !== undefined) { const instructions = Array.isArray(parts.instructions) ? parts.instructions : [parts.instructions]; - for (const [index, instruction] of instructions.entries()) { - segments.push(preparedSegment('system_prompt', index, instruction, true)); - } - } - for (const [index, message] of parts.messages.entries()) { - const role = - isObjectLike(message) && typeof message.role === 'string' ? message.role : undefined; - segments.push(preparedSegment('message', index, message, true, role)); + for (const instruction of instructions) + segments.push(sizedSegment('system_prompt', instruction)); } + for (const message of parts.messages) segments.push(sizedSegment('message', message)); if (parts.providerOptions !== undefined) { - segments.push(preparedSegment('provider_options', 0, parts.providerOptions, false)); + segments.push(sizedSegment('provider_options', parts.providerOptions)); } - return { - schemaVersion: PREPARED_REQUEST_OBSERVATION_SCHEMA_VERSION, - digest: hashSerialized(serializedRequest), - bytes: Buffer.byteLength(serializedRequest, 'utf8'), - segments: boundPreparedRequestSegments(segments), - }; + // Folded here rather than stored part by part. The fold is bounded by its own + // output — four kinds and a capped tool list — so the unbounded segment list + // never leaves this function and needs no cap of its own. + return foldPromptComposition(segments); } -const MAX_PREPARED_REQUEST_REMAINDERS = 4; - -function boundPreparedRequestSegments( - segments: readonly PreparedRequestObservationSegment[], -): PreparedRequestObservationSegment[] { - if (segments.length <= PREPARED_REQUEST_OBSERVATION_MAX_SEGMENTS) return [...segments]; - const kept = segments.slice( - 0, - PREPARED_REQUEST_OBSERVATION_MAX_SEGMENTS - MAX_PREPARED_REQUEST_REMAINDERS, - ); - const remainders: PreparedRequestObservationSegment[] = []; - for (const segment of segments.slice(kept.length)) { - const previous = remainders.at(-1); - if (previous?.kind === segment.kind) { - previous.bytes += segment.bytes; - previous.representedSegments = (previous.representedSegments ?? 1) + 1; - previous.digest = hashSerialized( - JSON.stringify(['prepared-segment-remainder', previous.digest, segment.digest]), - ); - continue; - } - remainders.push({ - kind: segment.kind, - index: segment.index, - cacheable: segment.cacheable, - comparison: 'opaque', - digest: hashSerialized(JSON.stringify(['prepared-segment-remainder', segment.digest])), - bytes: segment.bytes, - representedSegments: 1, - }); - } - return [...kept, ...remainders]; +function sizedSegment( + kind: SizedRequestSegment['kind'], + value: unknown, + label?: string, +): SizedRequestSegment { + const serialized = JSON.stringify(normalizePreparedValue(value).value); + return { + kind, + bytes: Buffer.byteLength(serialized, 'utf8'), + ...(label !== undefined + ? { label: label.slice(0, PREPARED_REQUEST_OBSERVATION_TEXT_MAX_LENGTH) } + : {}), + }; } function semanticRequestParts(payload: unknown): { @@ -203,32 +167,6 @@ function providerVisibleTools( return providerTools.filter((tool) => active.has(tool.name)); } -function preparedSegment( - kind: PreparedRequestObservationSegmentKind, - index: number, - value: unknown, - cacheable: boolean, - role?: string, - label?: string, -): PreparedRequestObservationSegment { - const normalized = normalizePreparedValue(value); - const serialized = JSON.stringify(normalized.value); - return { - kind, - index, - cacheable, - comparison: normalized.opaque || containsComparisonOpaqueRedaction(value) ? 'opaque' : 'exact', - digest: hashSerialized(serialized), - bytes: Buffer.byteLength(serialized, 'utf8'), - ...(role !== undefined - ? { role: role.slice(0, PREPARED_REQUEST_OBSERVATION_TEXT_MAX_LENGTH) } - : {}), - ...(label !== undefined - ? { label: label.slice(0, PREPARED_REQUEST_OBSERVATION_TEXT_MAX_LENGTH) } - : {}), - }; -} - /** * The tool's own name as the payload carries it. * @@ -245,10 +183,6 @@ export function stableHash(value: unknown): `sha256:${string}` { return `sha256:${createHash('sha256').update(stableStringify(value)).digest('hex')}`; } -function hashSerialized(serialized: string): `sha256:${string}` { - return `sha256:${createHash('sha256').update(serialized).digest('hex')}`; -} - export function toolCatalogHash(tools: readonly MakaTool[]): `sha256:${string}` { return stableHash( [...tools] @@ -418,25 +352,6 @@ function normalizePreparedValue(value: unknown): NormalizedPreparedValue { return visit(value, 0); } -function containsComparisonOpaqueRedaction(value: unknown, seen = new Set()): boolean { - if (!isObjectLike(value)) return false; - if (seen.has(value)) return false; - seen.add(value); - if (Array.isArray(value)) { - return value.some((entry) => containsComparisonOpaqueRedaction(entry, seen)); - } - if ( - value.type === 'custom' && - value.kind === 'openai.compaction' && - isPlainObject(value.providerOptions) && - isPlainObject(value.providerOptions.openai) && - value.providerOptions.openai.redacted === true - ) { - return true; - } - return Object.values(value).some((entry) => containsComparisonOpaqueRedaction(entry, seen)); -} - function toolShapeForDiagnostics(tool: MakaTool): unknown { return { name: tool.name, From 991c34b8e1e6352f44d8a1fd3f242b56b81687f1 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 13:46:12 +0800 Subject: [PATCH 3/8] refactor(runtime): drop the two names and the flag the retired observation left Diagnostics declared its own segment, tool and composition types over the ones a ModelCallAttempt durably carries. Folding them onto the record left those as aliases with no consumer outside the package, which is two spellings of one fact kept in step by hand. Prepared-value normalization also tracked whether a value could be compared exactly. That fed the retired observation's per-segment comparison mode; the one caller left takes the normalized value and sizes it, so the flag was accumulated through every branch and read nowhere. Refs #4082 Generated-by: Claude Code --- packages/runtime/src/context-diagnostics.ts | 18 +-- .../runtime/src/latest-context-snapshot.ts | 25 ++-- packages/runtime/src/prompt-composition.ts | 14 +- packages/runtime/src/request-shape.ts | 140 +++++------------- 4 files changed, 61 insertions(+), 136 deletions(-) diff --git a/packages/runtime/src/context-diagnostics.ts b/packages/runtime/src/context-diagnostics.ts index 5fd98bc74c..ef2b303e95 100644 --- a/packages/runtime/src/context-diagnostics.ts +++ b/packages/runtime/src/context-diagnostics.ts @@ -48,18 +48,6 @@ import { export type ContextDiagnosticsUnavailableReason = 'no_completed_request' | 'trace_unavailable'; -/** - * The composition vocabulary is the stored one. - * - * These names are what a `ModelCallAttempt` durably carries, so serving them - * under a second set of diagnostic-only types would be two spellings of one - * fact, kept in step by hand. - */ -export type ContextDiagnosticsSegmentKind = PromptCompositionSegmentKind; -export type ContextDiagnosticsSegment = PromptCompositionSegment; -export type ContextDiagnosticsTool = PromptCompositionTool; -export type ContextDiagnosticsComposition = PromptComposition; - export interface ContextDiagnosticsCompaction { kind: 'history'; phase: 'pre_turn' | 'mid_turn'; @@ -89,7 +77,7 @@ export type ContextDiagnostics = * quiet lie this separation exists to prevent, so readers never join an * independent capture stream (#2323). */ - composition?: ContextDiagnosticsComposition; + composition?: PromptComposition; compaction?: ContextDiagnosticsCompaction; }; @@ -359,7 +347,7 @@ interface MeteringAnchor { inputTokens?: number; cacheReadInputTokens?: number; contextWindow?: number; - composition?: ContextDiagnosticsComposition; + composition?: PromptComposition; } interface CheckpointCandidate { @@ -404,7 +392,7 @@ function meteringAnchor(event: AgentRunEvent): MeteringAnchor | undefined { function exactHistoricalComposition( anchor: MeteringAnchor, candidates: readonly LegacyProviderAnchor[], -): ContextDiagnosticsComposition | undefined { +): PromptComposition | undefined { const matches = candidates.filter( (candidate) => candidate.composition !== undefined && diff --git a/packages/runtime/src/latest-context-snapshot.ts b/packages/runtime/src/latest-context-snapshot.ts index 70d320ef64..0b4096e317 100644 --- a/packages/runtime/src/latest-context-snapshot.ts +++ b/packages/runtime/src/latest-context-snapshot.ts @@ -25,10 +25,7 @@ import { export { LATEST_CONTEXT_PROJECTION_TYPE }; import type { PromptComposition } from '@maka/core/model-call-attempt'; -import type { - ContextDiagnosticsCompaction, - ContextDiagnosticsComposition, -} from './context-diagnostics.js'; +import type { ContextDiagnosticsCompaction } from './context-diagnostics.js'; /** * One request's context, frozen by the transaction that committed it (#2323). @@ -64,7 +61,7 @@ export interface LatestContextSnapshot { * prepared-request observation — a request explains itself or says nothing, * never borrows another request's breakdown. */ - composition?: ContextDiagnosticsComposition; + composition?: PromptComposition; /** The boundary that applied when this request was built, if any. */ compaction?: ContextDiagnosticsCompaction; } @@ -141,7 +138,7 @@ export function readLatestContextSnapshot( !isOptionalCount(record.cacheReadInputTokens) || (record.contextWindow !== undefined && (!isCount(record.contextWindow) || record.contextWindow === 0)) || - (record.composition !== undefined && !isContextDiagnosticsComposition(record.composition)) || + (record.composition !== undefined && !isPromptComposition(record.composition)) || (record.compaction !== undefined && !isContextDiagnosticsCompaction(record.compaction)) ) { return undefined; @@ -149,7 +146,7 @@ export function readLatestContextSnapshot( return record as unknown as LatestContextSnapshot; } -function isContextDiagnosticsComposition(value: unknown): value is ContextDiagnosticsComposition { +function isPromptComposition(value: unknown): value is PromptComposition { const composition = shapedRecord( value, ['segments'], @@ -160,20 +157,20 @@ function isContextDiagnosticsComposition(value: unknown): value is ContextDiagno !Array.isArray(composition.segments) || composition.segments.length === 0 || composition.segments.length > 4 || - !composition.segments.every(isContextDiagnosticsSegment) || + !composition.segments.every(isPromptCompositionSegment) || (composition.tools !== undefined && (!Array.isArray(composition.tools) || composition.tools.length === 0 || composition.tools.length > 64 || - !composition.tools.every(isContextDiagnosticsTool))) || + !composition.tools.every(isPromptCompositionTool))) || (composition.remainingTools !== undefined && - !isContextDiagnosticsRemainder(composition.remainingTools)) || + !isPromptCompositionRemainder(composition.remainingTools)) || !isOptionalCount(composition.unlabelledToolBytes) || (composition.unlabelledToolBytes !== undefined && composition.unlabelledToolBytes === 0) ) { return false; } - const valid = composition as unknown as ContextDiagnosticsComposition; + const valid = composition as unknown as PromptComposition; const segmentOrder = ['system_instructions', 'tool_definitions', 'messages', 'other']; const order = valid.segments.map((segment) => segmentOrder.indexOf(segment.kind)); if (order.some((value, index) => index > 0 && value <= order[index - 1]!)) return false; @@ -201,7 +198,7 @@ function isContextDiagnosticsComposition(value: unknown): value is ContextDiagno : describedToolBytes === 0 && valid.remainingTools === undefined; } -function isContextDiagnosticsSegment(value: unknown): boolean { +function isPromptCompositionSegment(value: unknown): boolean { const segment = shapedRecord(value, ['kind', 'bytes'], []); return Boolean( segment && @@ -214,12 +211,12 @@ function isContextDiagnosticsSegment(value: unknown): boolean { ); } -function isContextDiagnosticsTool(value: unknown): boolean { +function isPromptCompositionTool(value: unknown): boolean { const tool = shapedRecord(value, ['name', 'bytes'], []); return Boolean(tool && isBoundedString(tool.name, 512) && isCount(tool.bytes) && tool.bytes > 0); } -function isContextDiagnosticsRemainder(value: unknown): boolean { +function isPromptCompositionRemainder(value: unknown): boolean { const remainder = shapedRecord(value, ['count', 'bytes'], []); return Boolean( remainder && diff --git a/packages/runtime/src/prompt-composition.ts b/packages/runtime/src/prompt-composition.ts index c2cacf1c8d..ae55133e2e 100644 --- a/packages/runtime/src/prompt-composition.ts +++ b/packages/runtime/src/prompt-composition.ts @@ -17,13 +17,11 @@ * under the License. */ -import type { - ContextDiagnosticsComposition, - ContextDiagnosticsSegment, -} from './context-diagnostics.js'; import { PROMPT_COMPOSITION_MAX_TOOLS, type PreparedRequestObservationSegmentKind, + type PromptComposition, + type PromptCompositionSegment, } from '@maka/core/model-call-attempt'; /** @@ -59,7 +57,7 @@ export interface SizedRequestSegment { */ export function foldPromptComposition( segments: readonly SizedRequestSegment[], -): ContextDiagnosticsComposition | undefined { +): PromptComposition | undefined { if (segments.length === 0) return undefined; const byKind = new Map(); @@ -83,7 +81,7 @@ export function foldPromptComposition( // A zero-byte kind is dropped rather than shown as `≈0`, the same way // `/context` folds it — a part nothing contributed to is not a part. - const folded: ContextDiagnosticsSegment[] = KIND_ORDER.flatMap((kind) => { + const folded: PromptCompositionSegment[] = KIND_ORDER.flatMap((kind) => { const bytes = byKind.get(kind) ?? 0; return bytes > 0 ? [{ kind: PART_KINDS[kind], bytes }] : []; }); @@ -130,7 +128,7 @@ export const PROVIDER_REQUEST_ATTEMPT_EVENT_TYPE = 'provider_request_attempt_rec export function readPromptCompositionEvent(event: { readonly type: string; readonly data?: unknown; -}): { attemptId: string; composition: ContextDiagnosticsComposition } | undefined { +}): { attemptId: string; composition: PromptComposition } | undefined { if (event.type !== PROVIDER_REQUEST_ATTEMPT_EVENT_TYPE) return undefined; const data = event.data; if (!isRecord(data)) return undefined; @@ -193,7 +191,7 @@ const KIND_ORDER: readonly PreparedRequestObservationSegmentKind[] = [ * four buckets already fold the same segments for `readLatestContextDiagnostics` * (#1580), and two names for one fact is how two surfaces start disagreeing. */ -const PART_KINDS: Record = +const PART_KINDS: Record = { system_prompt: 'system_instructions', tool_schema: 'tool_definitions', diff --git a/packages/runtime/src/request-shape.ts b/packages/runtime/src/request-shape.ts index e98d269c22..361f649932 100644 --- a/packages/runtime/src/request-shape.ts +++ b/packages/runtime/src/request-shape.ts @@ -111,7 +111,7 @@ function sizedSegment( value: unknown, label?: string, ): SizedRequestSegment { - const serialized = JSON.stringify(normalizePreparedValue(value).value); + const serialized = JSON.stringify(normalizePreparedValue(value)); return { kind, bytes: Buffer.byteLength(serialized, 'utf8'), @@ -195,29 +195,22 @@ export function stableStringify(value: unknown): string { return JSON.stringify(canonicalize(value)); } -interface NormalizedPreparedValue { - value: unknown; - opaque: boolean; -} - /** * Lossless JSON representation for the semantic values accepted by the model * seam. Every value is tagged, so a bigint cannot collide with a user string - * and an undefined property cannot disappear. Values that cannot be described - * exactly are retained as explicit opaque markers instead of pretending they - * were equal to another request. + * and an undefined property cannot disappear, and values that cannot be + * described exactly are kept as explicit markers rather than dropped — a size + * taken from this covers the whole payload. */ -function normalizePreparedValue(value: unknown): NormalizedPreparedValue { +function normalizePreparedValue(value: unknown): unknown { const tag = '__makaPreparedValue'; const ancestors = new Set(); - const visit = (current: unknown, depth: number): NormalizedPreparedValue => { + const visit = (current: unknown, depth: number): unknown => { if (current === null || typeof current === 'string' || typeof current === 'boolean') { - return { value: current, opaque: false }; + return current; } if (typeof current === 'number') { - if (Number.isFinite(current) && !Object.is(current, -0)) { - return { value: current, opaque: false }; - } + if (Number.isFinite(current) && !Object.is(current, -0)) return current; const encoded = Number.isNaN(current) ? 'NaN' : current === Infinity @@ -225,126 +218,75 @@ function normalizePreparedValue(value: unknown): NormalizedPreparedValue { : current === -Infinity ? '-Infinity' : '-0'; - return { value: { [tag]: 'number', value: encoded }, opaque: false }; - } - if (typeof current === 'bigint') { - return { value: { [tag]: 'bigint', value: current.toString() }, opaque: false }; - } - if (typeof current === 'undefined') { - return { value: { [tag]: 'undefined' }, opaque: false }; - } - if (typeof current === 'function' || typeof current === 'symbol') { - return { value: { [tag]: 'opaque', kind: typeof current }, opaque: true }; - } - if (typeof current !== 'object') { - return { value: { [tag]: 'opaque', kind: typeof current }, opaque: true }; - } - if (depth >= 64) { - return { value: { [tag]: 'opaque', kind: 'max-depth' }, opaque: true }; - } - if (ancestors.has(current)) { - return { value: { [tag]: 'opaque', kind: 'cycle' }, opaque: true }; + return { [tag]: 'number', value: encoded }; } + if (typeof current === 'bigint') return { [tag]: 'bigint', value: current.toString() }; + if (typeof current === 'undefined') return { [tag]: 'undefined' }; + if (typeof current !== 'object') return { [tag]: 'opaque', kind: typeof current }; + if (depth >= 64) return { [tag]: 'opaque', kind: 'max-depth' }; + if (ancestors.has(current)) return { [tag]: 'opaque', kind: 'cycle' }; ancestors.add(current); try { if (current instanceof ArrayBuffer) { return { - value: { - [tag]: 'binary', - kind: 'ArrayBuffer', - encoding: 'base64', - value: Buffer.from(current).toString('base64'), - }, - opaque: false, + [tag]: 'binary', + kind: 'ArrayBuffer', + encoding: 'base64', + value: Buffer.from(current).toString('base64'), }; } if (ArrayBuffer.isView(current)) { return { - value: { - [tag]: 'binary', - kind: current.constructor?.name ?? 'ArrayBufferView', - encoding: 'base64', - value: Buffer.from(current.buffer, current.byteOffset, current.byteLength).toString( - 'base64', - ), - }, - opaque: false, + [tag]: 'binary', + kind: current.constructor?.name ?? 'ArrayBufferView', + encoding: 'base64', + value: Buffer.from(current.buffer, current.byteOffset, current.byteLength).toString( + 'base64', + ), }; } if (current instanceof Date) { const timestamp = current.getTime(); return { - value: { - [tag]: 'date', - value: Number.isNaN(timestamp) ? 'invalid' : current.toISOString(), - }, - opaque: false, + [tag]: 'date', + value: Number.isNaN(timestamp) ? 'invalid' : current.toISOString(), }; } if (current instanceof Map) { - let opaque = false; - const entries = [...current.entries()].map(([key, entry]) => { - const normalizedKey = visit(key, depth + 1); - const normalizedEntry = visit(entry, depth + 1); - opaque ||= normalizedKey.opaque || normalizedEntry.opaque; - return [normalizedKey.value, normalizedEntry.value]; - }); - return { value: { [tag]: 'map', entries }, opaque }; + const entries = [...current.entries()].map(([key, entry]) => [ + visit(key, depth + 1), + visit(entry, depth + 1), + ]); + return { [tag]: 'map', entries }; } if (current instanceof Set) { - let opaque = false; - const entries = [...current].map((entry) => { - const normalized = visit(entry, depth + 1); - opaque ||= normalized.opaque; - return normalized.value; - }); - return { value: { [tag]: 'set', entries }, opaque }; + return { [tag]: 'set', entries: [...current].map((entry) => visit(entry, depth + 1)) }; } if (Array.isArray(current)) { - let opaque = false; - const entries = Array.from({ length: current.length }, (_, index) => { - if (!(index in current)) return { [tag]: 'array-hole' }; - const normalized = visit(current[index], depth + 1); - opaque ||= normalized.opaque; - return normalized.value; - }); - return { value: entries, opaque }; + return Array.from({ length: current.length }, (_, index) => + index in current ? visit(current[index], depth + 1) : { [tag]: 'array-hole' }, + ); } if (isPlainObject(current)) { - let opaque = false; const entries = Object.keys(current).map((key) => { - let normalized: NormalizedPreparedValue; try { - normalized = visit(current[key], depth + 1); + return [key, visit(current[key], depth + 1)]; } catch { - normalized = { - value: { [tag]: 'opaque', kind: 'unreadable-property' }, - opaque: true, - }; + return [key, { [tag]: 'opaque', kind: 'unreadable-property' }]; } - opaque ||= normalized.opaque; - return [key, normalized.value]; }); - if (Object.hasOwn(current, tag)) { - return { value: { [tag]: 'object', entries }, opaque }; - } - return { value: Object.fromEntries(entries), opaque }; + if (Object.hasOwn(current, tag)) return { [tag]: 'object', entries }; + return Object.fromEntries(entries); } const toJSON = (current as { toJSON?: unknown }).toJSON; if (typeof toJSON === 'function') { try { return visit(toJSON.call(current), depth + 1); } catch { - return { value: { [tag]: 'opaque', kind: 'toJSON-failed' }, opaque: true }; + return { [tag]: 'opaque', kind: 'toJSON-failed' }; } } - return { - value: { - [tag]: 'opaque', - kind: current.constructor?.name ?? 'non-plain-object', - }, - opaque: true, - }; + return { [tag]: 'opaque', kind: current.constructor?.name ?? 'non-plain-object' }; } finally { ancestors.delete(current); } From 23c6677c2432700e59e55cbb914f3c979c5fb04e Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 13:49:10 +0800 Subject: [PATCH 4/8] refactor(runtime): give the projection validator one list of the fold's buckets The snapshot validator spelled the four segment kinds twice and capped the tool list with a literal 64 beside the constant that sets it. Both now read from one list and one constant, so a change to the fold's buckets cannot leave the validator agreeing with a stale copy of itself. Also drops a sweep assertion that could not fail: the batch size is a module constant, so asserting it is a positive integer pinned nothing. Generated-by: Claude Code --- .../runtime/src/latest-context-snapshot.ts | 26 ++++++++++++------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/packages/runtime/src/latest-context-snapshot.ts b/packages/runtime/src/latest-context-snapshot.ts index 0b4096e317..b55cf90515 100644 --- a/packages/runtime/src/latest-context-snapshot.ts +++ b/packages/runtime/src/latest-context-snapshot.ts @@ -24,7 +24,11 @@ import { } from '@maka/core/agent-run'; export { LATEST_CONTEXT_PROJECTION_TYPE }; -import type { PromptComposition } from '@maka/core/model-call-attempt'; +import { + PROMPT_COMPOSITION_MAX_TOOLS, + type PromptComposition, + type PromptCompositionSegment, +} from '@maka/core/model-call-attempt'; import type { ContextDiagnosticsCompaction } from './context-diagnostics.js'; /** @@ -146,6 +150,14 @@ export function readLatestContextSnapshot( return record as unknown as LatestContextSnapshot; } +/** The fold's buckets, in the order a composition must list them. */ +const COMPOSITION_SEGMENT_ORDER: readonly PromptCompositionSegment['kind'][] = [ + 'system_instructions', + 'tool_definitions', + 'messages', + 'other', +]; + function isPromptComposition(value: unknown): value is PromptComposition { const composition = shapedRecord( value, @@ -156,12 +168,12 @@ function isPromptComposition(value: unknown): value is PromptComposition { !composition || !Array.isArray(composition.segments) || composition.segments.length === 0 || - composition.segments.length > 4 || + composition.segments.length > COMPOSITION_SEGMENT_ORDER.length || !composition.segments.every(isPromptCompositionSegment) || (composition.tools !== undefined && (!Array.isArray(composition.tools) || composition.tools.length === 0 || - composition.tools.length > 64 || + composition.tools.length > PROMPT_COMPOSITION_MAX_TOOLS || !composition.tools.every(isPromptCompositionTool))) || (composition.remainingTools !== undefined && !isPromptCompositionRemainder(composition.remainingTools)) || @@ -171,8 +183,7 @@ function isPromptComposition(value: unknown): value is PromptComposition { return false; } const valid = composition as unknown as PromptComposition; - const segmentOrder = ['system_instructions', 'tool_definitions', 'messages', 'other']; - const order = valid.segments.map((segment) => segmentOrder.indexOf(segment.kind)); + const order = valid.segments.map((segment) => COMPOSITION_SEGMENT_ORDER.indexOf(segment.kind)); if (order.some((value, index) => index > 0 && value <= order[index - 1]!)) return false; const toolDefinitions = valid.segments.find((segment) => segment.kind === 'tool_definitions'); @@ -202,10 +213,7 @@ function isPromptCompositionSegment(value: unknown): boolean { const segment = shapedRecord(value, ['kind', 'bytes'], []); return Boolean( segment && - (segment.kind === 'system_instructions' || - segment.kind === 'tool_definitions' || - segment.kind === 'messages' || - segment.kind === 'other') && + COMPOSITION_SEGMENT_ORDER.includes(segment.kind as PromptCompositionSegment['kind']) && isCount(segment.bytes) && segment.bytes > 0, ); From 91e4855b7dceda64128db8facf945e7973cc8907 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 13:49:51 +0800 Subject: [PATCH 5/8] refactor(runtime): call the fold where the request is dispatched `observe` was the seam the capture machinery hung on. With that gone it named nothing: two calls, one line, two call sites. Generated-by: Claude Code --- packages/runtime/src/provider-request-telemetry.ts | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/packages/runtime/src/provider-request-telemetry.ts b/packages/runtime/src/provider-request-telemetry.ts index 75d892c4fe..56b184a3f0 100644 --- a/packages/runtime/src/provider-request-telemetry.ts +++ b/packages/runtime/src/provider-request-telemetry.ts @@ -325,7 +325,7 @@ export class ProviderRequestTracker { throwIfAbortedBeforeDispatch(input.abortSignal); this.input.accounting?.assertReady?.(); const step = this.step; - const composition = this.observe(input); + const composition = preparedPromptComposition(secretFreeParams(input.params)); throwIfAbortedBeforeDispatch(input.abortSignal); let sawOutput = false; const attempt = this.beginAttempt(step, composition, input); @@ -394,7 +394,7 @@ export class ProviderRequestTracker { throwIfAbortedBeforeDispatch(input.abortSignal); this.input.accounting?.assertReady?.(); const step = this.step; - const composition = this.observe(input); + const composition = preparedPromptComposition(secretFreeParams(input.params)); throwIfAbortedBeforeDispatch(input.abortSignal); const attempt = this.beginAttempt(step, composition, input); try { @@ -608,12 +608,6 @@ export class ProviderRequestTracker { // itself. Settlement must not fail the turn the call already completed. } } - - private observe( - input: TrackProviderStreamInput | TrackProviderGenerateInput, - ): PromptComposition | undefined { - return preparedPromptComposition(secretFreeParams(input.params)); - } } function throwIfAbortedBeforeDispatch(signal: AbortSignal | undefined): void { From 8c355444cd8c418533170ef1d851b6473013c710 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 15:29:46 +0800 Subject: [PATCH 6/8] perf(storage): seal a session's artifact snapshot when it is asked for MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every load and every mutation rebuilt a map holding one sealed snapshot per session, and a snapshot's revision hashes every record in its session — so a store with 400 sessions sorted and hashed all 400 to answer a question about one. The map never earned that: each of the five readers reloads the whole store from the database first, so a kept snapshot never survived to be read. Sealing on the way out instead deletes the map, the two methods that maintained it, and the per-mutation bookkeeping that told them what changed. At 6,000 records across 400 sessions, same machine, same run: one listPage 13.45 to 11.54 ms, one create 38.32 to 37.42 ms. Refs #4037 Generated-by: Claude Code --- packages/storage/src/artifact-store.ts | 64 +++++++++++--------------- 1 file changed, 28 insertions(+), 36 deletions(-) diff --git a/packages/storage/src/artifact-store.ts b/packages/storage/src/artifact-store.ts index 2c50842ccb..4f82cbd3bf 100644 --- a/packages/storage/src/artifact-store.ts +++ b/packages/storage/src/artifact-store.ts @@ -85,10 +85,6 @@ const PURGE_INTENT_SCHEMA_VERSION = 1 as const; const MAX_PURGE_INTENT_BYTES = 64 * 1024 * 1024; const ARTIFACT_PURGE_RESOLVE_CONCURRENCY = 8; -const EMPTY_SESSION_SNAPSHOT: ArtifactSessionSnapshot = { - records: [], - revision: artifactListRevision([]), -}; interface ArtifactSessionSnapshot { readonly records: readonly ArtifactRecord[]; @@ -291,7 +287,6 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { private artifactRoot: string; private purgeIntentPath: string; private records: ArtifactRecord[] = []; - private sessionSnapshots = new Map(); private metadataReady = false; private recoveryRequired: boolean; private selfManagedRecoveryRequired: boolean; @@ -576,7 +571,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { } throw error; } - this.replaceRecords(nextRecords); + this.records = nextRecords; return { ...record }; } finally { if (!preserveStaging) { @@ -643,7 +638,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { record.id === canonical.id ? revived : record, ); await this.writeMetadataUnlocked({ upserts: [revived] }); - this.replaceRecords(nextRecords); + this.records = nextRecords; return { ...revived }; } @@ -693,7 +688,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { const { offset, limit } = options; return this.enqueue(async () => { await this.load(); - const snapshot = this.sessionSnapshots.get(sessionId) ?? EMPTY_SESSION_SNAPSHOT; + const snapshot = this.sessionSnapshot(sessionId); return { revision: snapshot.revision, records: snapshot.records.slice(offset, offset + limit).map((record) => ({ ...record })), @@ -707,7 +702,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { assertArtifactTurnKey(turnId); return this.enqueue(async () => { await this.load(); - const snapshot = this.sessionSnapshots.get(sessionId) ?? EMPTY_SESSION_SNAPSHOT; + const snapshot = this.sessionSnapshot(sessionId); return snapshot.records .filter((record) => record.turnId === turnId && record.status !== 'deleted') .map((record) => ({ ...record })); @@ -717,7 +712,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { async getInSession(sessionId: string, artifactId: string): Promise { return this.enqueue(async () => { await this.load(); - const snapshot = this.sessionSnapshots.get(sessionId) ?? EMPTY_SESSION_SNAPSHOT; + const snapshot = this.sessionSnapshot(sessionId); const record = snapshot.records.find((candidate) => candidate.id === artifactId); return { revision: snapshot.revision, @@ -855,7 +850,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { record.id === artifactId ? tombstone : record, ); await this.writeMetadataUnlocked({ upserts: [tombstone] }); - this.replaceRecords(nextRecords); + this.records = nextRecords; }); } @@ -865,7 +860,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { ): Promise { return this.enqueueMutation(async () => { await this.prepareMutationUnlocked({ kind: 'delete' }); - const snapshot = this.sessionSnapshots.get(sessionId) ?? EMPTY_SESSION_SNAPSHOT; + const snapshot = this.sessionSnapshot(sessionId); const existing = snapshot.records.find((record) => record.id === artifactId); if (!existing) return { kind: 'not_found' }; if (!canUserDeleteArtifact(existing)) return { kind: 'protected' }; @@ -877,7 +872,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { record.id === existing.id ? tombstone : record, ); await this.writeMetadataUnlocked({ upserts: [tombstone] }); - this.replaceRecords(nextRecords); + this.records = nextRecords; return { kind: 'deleted', record: { ...tombstone } }; }); } @@ -998,7 +993,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { } const nextRecords = this.records.filter((record) => !ids.has(record.id)); await this.writeMetadataUnlocked({ deleteIds: [...ids] }); - this.replaceRecords(nextRecords); + this.records = nextRecords; await this.removePurgeIntentUnlocked(); } @@ -1018,7 +1013,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { maxBytes: number, ): Promise { await this.load(); - const snapshot = this.sessionSnapshots.get(sessionId) ?? EMPTY_SESSION_SNAPSHOT; + const snapshot = this.sessionSnapshot(sessionId); const record = snapshot.records.find((candidate) => candidate.id === artifactId); if (!record) return { ok: false, reason: 'not_found' }; return this.prepareRecordRead(record, maxBytes, false); @@ -1041,7 +1036,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { private async load(): Promise { await this.metadataRepository.ready(); this.metadataReady = true; - this.replaceRecords(this.metadataRepository.readAll()); + this.records = this.metadataRepository.readAll(); } private async writeMetadataUnlocked(changes: ArtifactMetadataChanges): Promise { @@ -1094,7 +1089,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { private async reloadForMutationUnlocked(): Promise { await this.metadataRepository.ready(); this.metadataReady = true; - this.replaceRecords(this.metadataRepository.readAll()); + this.records = this.metadataRepository.readAll(); } private async hasCanonicalRecoveryResidueUnlocked(): Promise { @@ -1203,7 +1198,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { }; const nextRecords = [...this.records, record]; await this.writeMetadataUnlocked({ upserts: [record] }); - this.replaceRecords(nextRecords); + this.records = nextRecords; this.recoverableOrphans.delete(filesystemPathKey(candidate.relativePath)); return { ...record }; } @@ -1360,28 +1355,25 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { this.workspaceRoot = canonicalRoot; this.artifactRoot = join(canonicalRoot, 'artifacts'); this.purgeIntentPath = join(this.artifactRoot, ARTIFACT_PURGE_INTENT_FILE); - this.replaceRecords([]); + this.records = []; this.recoverableOrphans.clear(); if (this.recoveryMode === 'self_managed') this.selfManagedRecoveryRequired = true; } - private replaceRecords(records: ArtifactRecord[]): void { - const bySession = new Map(); - for (const record of records) { - const sessionRecords = bySession.get(record.sessionId); - if (sessionRecords) sessionRecords.push(record); - else bySession.set(record.sessionId, [record]); - } - const snapshots = new Map(); - for (const [sessionId, sessionRecords] of bySession) { - sessionRecords.sort(compareArtifactRecords); - snapshots.set(sessionId, { - records: sessionRecords, - revision: artifactListRevision(sessionRecords), - }); - } - this.records = records; - this.sessionSnapshots = snapshots; + /** + * Orders one session's records and stamps the revision readers compare on. + * + * Sealed on the way out rather than kept in a map. A revision hashes every + * record in its session, and every reader reloads the whole store from the + * database before it reads one, so a kept snapshot never survived to be read + * -- sealing all of them on load only charged each reader for the sessions it + * did not ask about. + */ + private sessionSnapshot(sessionId: string): ArtifactSessionSnapshot { + const records = this.records + .filter((record) => record.sessionId === sessionId) + .sort(compareArtifactRecords); + return { records, revision: artifactListRevision(records) }; } private async publishPurgeIntentUnlocked(artifactIds: readonly string[]): Promise { From f05902d868a9ed320a23f0c3059716feb3687e28 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 16:38:20 +0800 Subject: [PATCH 7/8] test(runtime-host): stop the graceful-shutdown Turn at a defined point The test stopped the Host as soon as `turn.start` returned, which only says the Turn was admitted -- which state the drain then found was left to how fast the machine was, and it asserts the drain records `cancelled`. It failed twice while this branch was verified with builds and benchmarks running beside it. It now waits for the question the scenario is about to ask before stopping, the same checkpoint its sibling test one screen below already uses. That is what makes the Turn active, which is what this test is about. Generated-by: Claude Code --- .../src/__tests__/execution-host-queue.test.ts | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts index 25052409c6..88d4da5e37 100644 --- a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts @@ -530,6 +530,11 @@ test('graceful Host shutdown stops and drains an active Turn before releasing ow await withExecutionRoot(async (fixture) => { const host = await fixture.startHost(); const client = await connectClient(fixture.root); + const subscription = await client.openSessionSubscription({ + sessionId: fixture.sessionId, + transcript: { kind: 'none' }, + }); + const probe = new SubscriptionProbe(subscription); const turnId = randomUUID(); const started = requireStartedTurn( await client.request('turn.start', { @@ -538,10 +543,16 @@ test('graceful Host shutdown stops and drains an active Turn before releasing ow content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, }), ); + // What this pins is the drain of an ACTIVE Turn, and `turn.start` + // returning only says the Turn was admitted. Waiting for the question it + // is about to ask is what makes it active, so stopping before that would + // leave which state the Host drains up to how fast the machine is. + await waitForPendingInteraction(subscription, probe, started.runId); const exit = await fixture.stopHost(host); assert.deepEqual(exit, { code: 0, signal: null }); await client.closed; + await probe.waitForFailure('connection_closed'); const successor = await fixture.startHost(); const observer = await connectClient(fixture.root); From 2f69ee4bb3790f9c23cbe02eff4cd73139817846 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 4 Sep 2026 17:35:25 +0800 Subject: [PATCH 8/8] refactor(core): let one list name the fold's buckets for both validators The decoder and the projection validator each spelled the same four segment kinds in the same order, so a change to what a composition is made of could leave one of them agreeing with a stale copy of the other. The list moves to the owner of the shape and says there that its order is contract, not presentation. The two validators stay separate on purpose: a record that fails the decoder loses its usage and cost with it, while a derived projection row that fails is rebuilt from the ledger -- so the projection can afford to check the fold's ordering and byte conservation, and the durable record cannot. Generated-by: Claude Code --- packages/core/src/model-call-attempt.ts | 9 ++++++++- .../runtime/src/latest-context-snapshot.ts | 19 +++++++------------ 2 files changed, 15 insertions(+), 13 deletions(-) diff --git a/packages/core/src/model-call-attempt.ts b/packages/core/src/model-call-attempt.ts index 26b8854af7..844218cdbd 100644 --- a/packages/core/src/model-call-attempt.ts +++ b/packages/core/src/model-call-attempt.ts @@ -346,7 +346,14 @@ const PROMPT_COMPOSITION_REMAINING_TOOLS_SHAPE = defineObjectShape<{ bytes: number; }>()(['count', 'bytes'], []); -const PROMPT_COMPOSITION_SEGMENT_KINDS: readonly PromptCompositionSegmentKind[] = [ +/** + * The fold's buckets, in the order a composition lists them. + * + * The order is part of the contract, not presentation: a reader comparing two + * compositions compares them position by position, and the projection + * validator rejects a record whose segments arrive out of this order. + */ +export const PROMPT_COMPOSITION_SEGMENT_KINDS: readonly PromptCompositionSegmentKind[] = [ 'system_instructions', 'tool_definitions', 'messages', diff --git a/packages/runtime/src/latest-context-snapshot.ts b/packages/runtime/src/latest-context-snapshot.ts index b55cf90515..14f65e4f3c 100644 --- a/packages/runtime/src/latest-context-snapshot.ts +++ b/packages/runtime/src/latest-context-snapshot.ts @@ -26,8 +26,9 @@ import { export { LATEST_CONTEXT_PROJECTION_TYPE }; import { PROMPT_COMPOSITION_MAX_TOOLS, + PROMPT_COMPOSITION_SEGMENT_KINDS, type PromptComposition, - type PromptCompositionSegment, + type PromptCompositionSegmentKind, } from '@maka/core/model-call-attempt'; import type { ContextDiagnosticsCompaction } from './context-diagnostics.js'; @@ -150,14 +151,6 @@ export function readLatestContextSnapshot( return record as unknown as LatestContextSnapshot; } -/** The fold's buckets, in the order a composition must list them. */ -const COMPOSITION_SEGMENT_ORDER: readonly PromptCompositionSegment['kind'][] = [ - 'system_instructions', - 'tool_definitions', - 'messages', - 'other', -]; - function isPromptComposition(value: unknown): value is PromptComposition { const composition = shapedRecord( value, @@ -168,7 +161,7 @@ function isPromptComposition(value: unknown): value is PromptComposition { !composition || !Array.isArray(composition.segments) || composition.segments.length === 0 || - composition.segments.length > COMPOSITION_SEGMENT_ORDER.length || + composition.segments.length > PROMPT_COMPOSITION_SEGMENT_KINDS.length || !composition.segments.every(isPromptCompositionSegment) || (composition.tools !== undefined && (!Array.isArray(composition.tools) || @@ -183,7 +176,9 @@ function isPromptComposition(value: unknown): value is PromptComposition { return false; } const valid = composition as unknown as PromptComposition; - const order = valid.segments.map((segment) => COMPOSITION_SEGMENT_ORDER.indexOf(segment.kind)); + const order = valid.segments.map((segment) => + PROMPT_COMPOSITION_SEGMENT_KINDS.indexOf(segment.kind), + ); if (order.some((value, index) => index > 0 && value <= order[index - 1]!)) return false; const toolDefinitions = valid.segments.find((segment) => segment.kind === 'tool_definitions'); @@ -213,7 +208,7 @@ function isPromptCompositionSegment(value: unknown): boolean { const segment = shapedRecord(value, ['kind', 'bytes'], []); return Boolean( segment && - COMPOSITION_SEGMENT_ORDER.includes(segment.kind as PromptCompositionSegment['kind']) && + PROMPT_COMPOSITION_SEGMENT_KINDS.includes(segment.kind as PromptCompositionSegmentKind) && isCount(segment.bytes) && segment.bytes > 0, );