fix(openai): tier parallel tool results independently

Author: chioarub

src/services/api/compressToolHistory.ts

@@ -246,18 +246,19 @@ function indexToolUses(messages: AnyMessage[]): Map<string, ToolUseBlock> {
   return map
 }
 
-function indexToolResultMessages(messages: AnyMessage[]): number[] {
-  const indices: number[] = []
+function indexToolResults(
+  messages: AnyMessage[],
+): Array<{ messageIndex: number; blockIndex: number }> {
+  const indices: Array<{ messageIndex: number; blockIndex: number }> = []
   for (let i = 0; i < messages.length; i++) {
     const inner = getInner(messages[i])
     const role = inner.role ?? messages[i].role
     const content = inner.content
-    if (
-      role === 'user' &&
-      Array.isArray(content) &&
-      content.some((b: { type?: string }) => b?.type === 'tool_result')
-    ) {
-      indices.push(i)
+    if (role !== 'user' || !Array.isArray(content)) continue
+    for (let blockIndex = 0; blockIndex < content.length; blockIndex++) {
+      if ((content[blockIndex] as { type?: string })?.type === 'tool_result') {
+        indices.push({ messageIndex: i, blockIndex })
+      }
     }
   }
   return indices
@@ -313,36 +314,45 @@ export function compressToolHistory<T extends AnyMessage>(
     options.effectiveContextWindowSize ?? getEffectiveContextWindowSize(model),
   )
 
-  const toolResultIndices = indexToolResultMessages(messages)
-  const total = toolResultIndices.length
+  const toolResults = indexToolResults(messages)
+  const total = toolResults.length
   // If every tool-result fits in the recent tier, no boundary crosses; return
   // the same reference for the same copy-elision reason.
   if (total <= tiers.recent) return messages
 
-  // O(1) lookup: messageIndex → tool-result position (0 = oldest). Replaces
-  // the naive Array.indexOf(i) that was O(n²) across the .map below.
-  const positionByIndex = new Map<number, number>()
-  for (let pos = 0; pos < toolResultIndices.length; pos++) {
-    positionByIndex.set(toolResultIndices[pos], pos)
+  // O(1) lookup within each carrier message: blockIndex → tool-result position
+  // (0 = oldest). Parallel results share a message but receive distinct tiers.
+  const positionsByMessage = new Map<number, Map<number, number>>()
+  for (let pos = 0; pos < toolResults.length; pos++) {
+    const { messageIndex, blockIndex } = toolResults[pos]
+    let positions = positionsByMessage.get(messageIndex)
+    if (!positions) {
+      positions = new Map<number, number>()
+      positionsByMessage.set(messageIndex, positions)
+    }
+    positions.set(blockIndex, pos)
   }
 
   const toolUsesById = indexToolUses(messages)
 
   return messages.map((msg, i) => {
-    const pos = positionByIndex.get(i)
-    if (pos === undefined) return msg
-
-    const fromEnd = total - 1 - pos
-    if (fromEnd < tiers.recent) return msg
+    const positions = positionsByMessage.get(i)
+    if (!positions) return msg
+    const firstPos = positions.values().next().value
+    if (firstPos === undefined || total - 1 - firstPos < tiers.recent) return msg
 
-    const inMidWindow = fromEnd < tiers.recent + tiers.mid
     const content = getInner(msg).content as unknown[]
-    const newContent = content.map(block => {
+    const newContent = content.map((block, blockIndex) => {
+      const pos = positions.get(blockIndex)
+      if (pos === undefined) return block
+      const fromEnd = total - 1 - pos
+      if (fromEnd < tiers.recent) return block
+
       const b = block as { type?: string }
       if (b?.type !== 'tool_result') return block
       const tr = block as ToolResultBlock
       if (!shouldCompressBlock(tr, toolUsesById)) return block
-      return inMidWindow
+      return fromEnd < tiers.recent + tiers.mid
         ? truncateBlock(tr, MID_MAX_CHARS)
         : buildStub(tr, toolUsesById)
     })

src/services/api/openaiShim.compression.test.ts

@@ -114,6 +114,37 @@ function buildLongConversation(numExchanges: number, resultLength = 5_000) {
   return out
 }
 
+function buildParallelToolConversation(
+  numBatches: number,
+  batchSize: number,
+  resultLength = 5_000,
+) {
+  const out: Array<{ role: string; content: unknown }> = [
+    { role: 'user', content: 'start the parallel work' },
+  ]
+  for (let batch = 0; batch < numBatches; batch++) {
+    const firstId = batch * batchSize
+    out.push({
+      role: 'assistant',
+      content: Array.from({ length: batchSize }, (_, offset) => ({
+        type: 'tool_use',
+        id: `toolu_${firstId + offset}`,
+        name: 'Read',
+        input: { file_path: `/path/to/file${firstId + offset}.ts` },
+      })),
+    })
+    out.push({
+      role: 'user',
+      content: Array.from({ length: batchSize }, (_, offset) => ({
+        type: 'tool_result',
+        tool_use_id: `toolu_${firstId + offset}`,
+        content: bigText(resultLength),
+      })),
+    })
+  }
+  return out
+}
+
 function makeFakeResponse(): Response {
   return new Response(
     JSON.stringify({
@@ -537,12 +568,12 @@ test('Responses compression preserves structured history and materially reduces
   expect(compressedByCallId.get('toolu_0')).toBe(
     '[Old tool result content cleared]',
   )
-  for (let i = 1; i <= 14; i++) {
+  for (let i = 1; i <= 15; i++) {
     expect(compressedByCallId.get(`toolu_${i}`)).toMatch(
       /^\[Read args=.*chars omitted\]$/,
     )
   }
-  for (let i = 15; i <= 24; i++) {
+  for (let i = 16; i <= 24; i++) {
     expect(compressedByCallId.get(`toolu_${i}`)).toContain('[…truncated')
   }
   expect(compressedByCallId.get('toolu_16_extra')).toContain('[…truncated')
@@ -571,6 +602,38 @@ test('Responses compression preserves structured history and materially reduces
   expect(compressedSize).toBeLessThan(uncompressedSize * 0.5)
 })
 
+test('Responses compression assigns tiers per parallel tool result', async () => {
+  mockState.enabled = true
+  mockState.effectiveWindow = 100_000
+  const messages = buildParallelToolConversation(5, 6)
+
+  const body = await captureResponsesRequestBody(messages, MID_TIER_MODEL)
+  const functionCalls = getResponsesFunctionCalls(body)
+  const functionOutputs = getResponsesFunctionOutputs(body)
+  const outputs = new Map(
+    functionOutputs.map(item => [item.call_id, item.output ?? '']),
+  )
+
+  expect(functionCalls).toHaveLength(30)
+  expect(functionOutputs).toHaveLength(30)
+  expect(functionCalls.map(item => item.call_id)).toEqual(
+    functionOutputs.map(item => item.call_id),
+  )
+  for (let i = 0; i < 15; i++) {
+    expect(outputs.get(`toolu_${i}`)).toMatch(
+      /^\[Read args=.*5000 chars omitted\]$/,
+    )
+  }
+  for (let i = 15; i < 25; i++) {
+    expect(outputs.get(`toolu_${i}`)).toContain(
+      '[…truncated 3000 chars from tool history]',
+    )
+  }
+  for (let i = 25; i < 30; i++) {
+    expect(outputs.get(`toolu_${i}`)).toHaveLength(5_000)
+  }
+})
+
 test('Responses compression preserves structured parts while compressing old and mid text', async () => {
   mockState.enabled = true
   mockState.effectiveWindow = 100_000
@@ -767,7 +830,7 @@ test('non-chat transports do not invoke the Chat message converter', () => {
   expect(selectMessages('chat_completions', () => chatMessages)).toBe(chatMessages)
 })
 
-test('Anthropic and Gemini transports do not invoke tool history compression', () => {
+test('only OpenAI-compatible transports invoke tool history compression', () => {
   const selectMessages = __test.getCompressedMessagesForTransport
   const rawMessages = [{ role: 'user', content: 'hello' }]
   const compressedMessages = [{ role: 'user', content: 'compressed' }]
@@ -779,6 +842,7 @@ test('Anthropic and Gemini transports do not invoke tool history compression', (
 
   expect(selectMessages('anthropic_messages', rawMessages, compress)).toBe(rawMessages)
   expect(selectMessages('gemini', rawMessages, compress)).toBe(rawMessages)
+  expect(selectMessages('future_transport', rawMessages, compress)).toBe(rawMessages)
   expect(compressionCalls).toBe(0)
 
   expect(selectMessages('responses', rawMessages, compress)).toBe(compressedMessages)

src/services/api/openaiShim.ts

@@ -1501,9 +1501,11 @@ function getCompressedMessagesForTransport<T>(
   rawMessages: T,
   compress: () => T,
 ): T {
-  return transport === 'anthropic_messages' || transport === 'gemini'
-    ? rawMessages
-    : compress()
+  return transport === 'chat_completions' ||
+    transport === 'responses' ||
+    transport === 'responses_compat'
+    ? compress()
+    : rawMessages
 }
 
 /**