perf: bound what one model call costs to store

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> = {}): 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: {

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 },

packages/core/src/model-call-attempt.ts

@@ -109,12 +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. The full secret-free serialization stays
- * in the private request artifact referenced by `captureArtifactId` when that
- * sink is available.
+ * 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;
@@ -165,9 +208,18 @@ 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. */
+  /** 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;
@@ -226,6 +278,7 @@ const MODEL_CALL_ATTEMPT_SHAPE = defineObjectShape<ModelCallAttempt>()(
     'historyCompactRoute',
     'contextWindow',
     'captureArtifactId',
+    'promptComposition',
     'requestObservation',
     'timeToFirstTokenMs',
     'finishReason',
@@ -273,6 +326,48 @@ const PREPARED_REQUEST_SEGMENT_KINDS: readonly PreparedRequestObservationSegment
   'provider_options',
 ];
 
+const PROMPT_COMPOSITION_SHAPE = defineObjectShape<PromptComposition>()(
+  ['segments'],
+  ['tools', 'remainingTools', 'unlabelledToolBytes'],
+);
+
+const PROMPT_COMPOSITION_SEGMENT_SHAPE = defineObjectShape<PromptCompositionSegment>()(
+  ['kind', 'bytes'],
+  [],
+);
+
+const PROMPT_COMPOSITION_TOOL_SHAPE = defineObjectShape<PromptCompositionTool>()(
+  ['name', 'bytes'],
+  [],
+);
+
+const PROMPT_COMPOSITION_REMAINING_TOOLS_SHAPE = defineObjectShape<{
+  count: number;
+  bytes: number;
+}>()(['count', 'bytes'], []);
+
+/**
+ * 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',
+  '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;
 }
@@ -339,6 +434,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 (
@@ -419,6 +562,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) &&

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);

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,
@@ -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<ReturnType<typeof openInteractiveArtifactStoreForWrite>>,
-  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<void>((resolve) => setTimeout(resolve, 10));
-  }
-  throw new Error(`Hosted request artifacts did not reach ${expectedRequests}`);
-}
-
 async function waitForAutomaticMemoryRequestsToSettle(
   requests: readonly ProviderRequest[],
 ): Promise<void> {