fix(powerbi): compact retained failures and reuse rotated fixture tokens

Author: BillLeoutsakosvl346

apps/sim/executor/execution/block-executor.test.ts

@@ -133,7 +133,10 @@ describe('BlockExecutor', () => {
     expect(context.mcpBlockId).toBeUndefined()
   })
 
-  function createFailedToolExecution(output: Record<string, unknown>) {
+  function createFailedToolExecution(
+    output: Record<string, unknown>,
+    options: { piiRedaction?: boolean } = {}
+  ) {
     const block = createBlock()
     const workflow: SerializedWorkflow = {
       version: '1',
@@ -159,16 +162,126 @@ describe('BlockExecutor', () => {
       state
     )
     const ctx = createContext(state)
-    ctx.piiBlockOutputRedaction = {
-      enabled: true,
-      entityTypes: ['EMAIL_ADDRESS'],
-      language: 'en',
+    if (options.piiRedaction !== false) {
+      ctx.piiBlockOutputRedaction = {
+        enabled: true,
+        entityTypes: ['EMAIL_ADDRESS'],
+        language: 'en',
+      }
     }
     const node = createNode(block)
     node.outgoingEdges.set('error-edge', { sourceHandle: EDGE.ERROR, target: 'error-handler' })
     return { executor, block, state, ctx, node, failure, onBlockComplete }
   }
 
+  it('durably compacts inline failed tool rows before error-port state and completion output', async () => {
+    const rows = Array.from({ length: 3072 }, (_, id) => ({ id, value: 'p'.repeat(3072) }))
+    const persisted = new Map<string, Buffer>()
+    mockUploadFile.mockImplementation(async ({ customKey, file }) => {
+      persisted.set(customKey, file)
+      return { key: customKey }
+    })
+    mockDownloadFile.mockImplementation(async ({ key }) => {
+      const file = persisted.get(key)
+      if (!file) throw new Error('Stored partial-row chunk was not found')
+      return file
+    })
+    const { executor, block, state, ctx, node, failure, onBlockComplete } =
+      createFailedToolExecution(
+        { rows, rowCount: rows.length, incomplete: true },
+        { piiRedaction: false }
+      )
+    const cost = { input: 0.1, output: 0.2, total: 0.3 }
+    attachTrustedExecutionCost(failure, cost)
+
+    const output = await executor.execute(ctx, node, block)
+    await vi.waitFor(() => expect(onBlockComplete).toHaveBeenCalledOnce())
+
+    expect(isLargeArrayManifest(output.rows)).toBe(true)
+    expect(isLargeValueRef(output)).toBe(false)
+    expect(output).toMatchObject({
+      error: 'query incomplete',
+      rowCount: rows.length,
+      incomplete: true,
+      cost,
+    })
+    if (!isLargeArrayManifest(output.rows)) throw new Error('Expected compacted failed rows')
+    expect(output.rows.totalCount).toBe(rows.length)
+    expect(output.rows.chunks.every(({ ref }) => ref.executionId === ctx.executionId)).toBe(true)
+    clearLargeValueCacheForTests()
+    expect(
+      await readLargeArrayManifestSlice(output.rows, 0, rows.length, {
+        workspaceId: ctx.workspaceId,
+        workflowId: ctx.workflowId,
+        executionId: ctx.executionId,
+      })
+    ).toEqual(rows)
+    expect(state.getBlockOutput(block.id)).toEqual(output)
+    expect(ctx.blockLogs[0]?.output).toEqual(output)
+    expect(onBlockComplete.mock.calls[0]?.[3]?.output).toEqual(output)
+    expect(ctx.blockLogs[0]).toMatchObject({ success: false, errorHandled: true })
+  })
+
+  it('keeps aggregate failed envelopes addressable by error routing and named fields', async () => {
+    const rows = Array.from({ length: 1536 }, (_, id) => ({ id, value: 'p'.repeat(3072) }))
+    const metadata = { notes: 'm'.repeat(4.5 * 1024 * 1024) }
+    const { executor, block, state, ctx, node, failure, onBlockComplete } =
+      createFailedToolExecution(
+        { rows, metadata, rowCount: rows.length, incomplete: true },
+        { piiRedaction: false }
+      )
+    const cost = { input: 0.1, output: 0.2, total: 0.3 }
+    attachTrustedExecutionCost(failure, cost)
+
+    const output = await executor.execute(ctx, node, block)
+    await vi.waitFor(() => expect(onBlockComplete).toHaveBeenCalledOnce())
+
+    expect(isLargeValueRef(output)).toBe(false)
+    expect(output.error).toBe('query incomplete')
+    expect(output.rowCount).toBe(rows.length)
+    expect(output.incomplete).toBe(true)
+    expect(output.cost).toEqual(cost)
+    expect(JSON.stringify(output.rows) === JSON.stringify(rows)).toBe(true)
+    expect(JSON.stringify(output.metadata) === JSON.stringify(metadata)).toBe(true)
+    expect(state.getBlockOutput(block.id)).toEqual(output)
+    expect(ctx.blockLogs[0]?.output).toEqual(output)
+    expect(onBlockComplete.mock.calls[0]?.[3]?.output).toEqual(output)
+    expect(ctx.blockLogs[0]).toMatchObject({ success: false, errorHandled: true })
+  })
+
+  it('omits retained tool payloads when durable compaction fails while preserving trusted cost', async () => {
+    const rows = Array.from({ length: 3072 }, (_, id) => ({ id, value: 'p'.repeat(3072) }))
+    const unsafeFailure = 'storage rejected private-partial-row-payload'
+    mockUploadFile.mockRejectedValueOnce(new Error(unsafeFailure))
+    const { executor, block, state, ctx, node, failure, onBlockComplete } =
+      createFailedToolExecution(
+        { rows, rowCount: rows.length, incomplete: true },
+        { piiRedaction: false }
+      )
+    const cost = { input: 0.1, output: 0.2, total: 0.3 }
+    attachTrustedExecutionCost(failure, cost)
+
+    const output = await executor.execute(ctx, node, block)
+    await vi.waitFor(() => expect(onBlockComplete).toHaveBeenCalledOnce())
+
+    expect(Object.hasOwn(output, 'rows')).toBe(false)
+    expect(Object.hasOwn(output, 'rowCount')).toBe(false)
+    expect(Object.hasOwn(output, 'incomplete')).toBe(false)
+    const expected = {
+      error: 'Partial tool output could not be stored and was omitted.',
+      cost,
+    }
+    expect(output).toEqual(expected)
+    expect(state.getBlockOutput(block.id)).toEqual(expected)
+    expect(ctx.blockLogs[0]?.output).toEqual(expected)
+    expect(ctx.blockLogs[0]?.error).toBe(expected.error)
+    expect(onBlockComplete.mock.calls[0]?.[3]?.output).toEqual(expected)
+    expect(ctx.blockLogs[0]).toMatchObject({ success: false, errorHandled: true })
+    expect(JSON.stringify([output, ctx.blockLogs, onBlockComplete.mock.calls])).not.toContain(
+      unsafeFailure
+    )
+  })
+
   it('masks partial failed tool rows before error-port state and completion output', async () => {
     mockMaskBatch.mockImplementation(async (texts: string[]) =>
       texts.map((text) => text.replaceAll('alice@example.com', '<EMAIL_ADDRESS>'))

apps/sim/executor/execution/block-executor.ts

@@ -717,13 +717,33 @@ export class BlockExecutor {
         errorMessage = 'PII redaction failed. Partial tool output was omitted.'
       }
     }
-    endedAt = new Date().toISOString()
-    duration = performance.now() - startTime
-    const errorOutput: NormalizedBlockOutput = {
+    let errorOutput: NormalizedBlockOutput = {
       ...partialOutput,
       error: errorMessage,
       ...(trustedExecutionCost ? { cost: trustedExecutionCost } : {}),
     }
+    if (partialOutput && Object.keys(partialOutput).length > 0) {
+      try {
+        const compacted = await compactBlockOutput(errorOutput, {
+          workspaceId: ctx.workspaceId,
+          workflowId: ctx.workflowId,
+          executionId: ctx.executionId,
+          userId: ctx.userId,
+          preserveUserFileBase64: ctx.includeFileBase64 === true,
+          preserveRoot: true,
+          requireDurable: true,
+        })
+        errorOutput = compacted.output
+      } catch {
+        errorMessage = 'Partial tool output could not be stored and was omitted.'
+        errorOutput = {
+          error: errorMessage,
+          ...(trustedExecutionCost ? { cost: trustedExecutionCost } : {}),
+        }
+      }
+    }
+    endedAt = new Date().toISOString()
+    duration = performance.now() - startTime
 
     // Keep any answer text already drained before timeout/failure so logs match
     // what was projected to the client.

apps/sim/tools/powerbi/__fixtures__/provider-fixture.ts

@@ -97,7 +97,8 @@ export async function startPowerBIProviderFixture(http: typeof NodeHTTP) {
       entry.body = Object.fromEntries(parameters)
       entry.authorized =
         parameters.get('grant_type') === 'refresh_token' &&
-        parameters.get('refresh_token') === POWERBI_FIXTURE_REFRESH_TOKEN &&
+        (parameters.get('refresh_token') === POWERBI_FIXTURE_REFRESH_TOKEN ||
+          parameters.get('refresh_token') === POWERBI_FIXTURE_ROTATED_REFRESH_TOKEN) &&
         parameters.get('client_id') === 'powerbi-fixture-client' &&
         parameters.get('client_secret') === 'powerbi-fixture-secret'
       if (!entry.authorized) return send(400, { error: 'invalid_grant' })

apps/sim/tools/powerbi/powerbi.integration.ts

@@ -347,8 +347,8 @@ afterAll(async () => {
 })
 
 describe('Power BI with persisted delegated credentials and provider wire responses', () => {
-  it('refreshes an expired delegated token, stores rotation, and requests the Power BI audience', () =>
-    checked('expired delegated token rotation', async () => {
+  it('refreshes an expired delegated token, reuses its stored rotation, and requests the Power BI audience', () =>
+    checked('expired delegated token rotation and reuse', async () => {
       const run = await runWorkflow(['powerbi_list_datasets'], undefined, expiredCredentialId)
       expect(run.result).toMatchObject({ ok: true, status: 'completed' })
       expect(provider.requests).toHaveLength(2)
@@ -374,6 +374,45 @@ describe('Power BI with persisted delegated credentials and provider wire respon
         Date.now() + 89 * 24 * 60 * 60 * 1000
       )
       expect(run.state?.blockStates[run.blockIds[0]]?.output).toMatchObject({ datasetCount: 1 })
+
+      await db
+        .update(account)
+        .set({ accessTokenExpiresAt: new Date(Date.now() - 60_000) })
+        .where(eq(account.id, expiredAccountId))
+      const beforeRepeatedRefresh = provider.requests.length
+      const repeated = await runWorkflow(['powerbi_list_datasets'], undefined, expiredCredentialId)
+      expect(provider.requests.slice(beforeRepeatedRefresh)).toEqual([
+        {
+          method: 'POST',
+          path: '/common/oauth2/v2.0/token',
+          body: {
+            grant_type: 'refresh_token',
+            refresh_token: POWERBI_FIXTURE_ROTATED_REFRESH_TOKEN,
+            client_id: 'powerbi-fixture-client',
+            client_secret: 'powerbi-fixture-secret',
+            scope: tokenBody.scope,
+          },
+          authorized: true,
+          status: 200,
+        },
+        {
+          method: 'GET',
+          path: `/v1.0/myorg/groups/${POWERBI_FIXTURE_IDS.groupId}/datasets`,
+          body: null,
+          authorized: true,
+          status: 200,
+        },
+      ])
+      expect(repeated.result).toMatchObject({ ok: true, status: 'completed' })
+      expect(repeated.state?.blockStates[repeated.blockIds[0]]?.output).toMatchObject({
+        datasetCount: 1,
+      })
+      const [storedAgain] = await db.select().from(account).where(eq(account.id, expiredAccountId))
+      expect(storedAgain).toMatchObject({
+        accessToken: POWERBI_FIXTURE_TOKEN,
+        refreshToken: POWERBI_FIXTURE_ROTATED_REFRESH_TOKEN,
+      })
+      expect(storedAgain.accessTokenExpiresAt?.getTime()).toBeGreaterThan(Date.now())
     }))
 
   it('rejects missing resource identities instead of storing successful null identities', () =>