fix(powerbi): use standard workflow failure handling

Author: Bill Leoutsakos

apps/docs/content/docs/integrations/powerbi.mdx

@@ -15,7 +15,7 @@ import { BlockInfoCard } from "@/components/ui/block-info-card"
 
 Connect a home-tenant organizational account through the **Power BI** Microsoft connection. Existing Microsoft Graph connections cannot substitute for it. Self-hosted installations use the existing Microsoft client environment variables and register `/api/auth/oauth2/callback/microsoft-powerbi` as a Web redirect URI. This version supports delegated OAuth only, without Credential Groups or service principals.
 
-**Execute DAX Query** requires the tenant's **Dataset Execute Queries REST API** setting, workspace access, and Read and Build permissions on the semantic model. Submit one DAX query that returns one table. The API limits each query to 100,000 rows or 1,000,000 values, 15 MB of returned data, and 120 query requests per minute per user. Dynamic column names are preserved. An HTTP 200 response can contain an error with partial rows; check the block's error and incomplete outputs before using those rows. See Microsoft's [Execute Queries documentation](https://learn.microsoft.com/en-us/rest/api/power-bi/datasets/execute-queries-in-group).
+**Execute DAX Query** requires the tenant's **Dataset Execute Queries REST API** setting, workspace access, and Read and Build permissions on the semantic model. Submit one DAX query that returns one table. The API limits each query to 100,000 rows or 1,000,000 values, 15 MB of returned data, and 120 query requests per minute per user. Dynamic column names are preserved. An HTTP 200 response can contain an error with partial rows; any reported query error fails the action. Connect the block's error port to handle its standard `error` output. Failed workflow outputs omit query fields, including partial rows, `errors`, and `incomplete`; successful queries expose the declared outputs. Direct tool responses retain available partial rows and typed errors, but those fields are unavailable to downstream workflow blocks after failure. See Microsoft's [Execute Queries documentation](https://learn.microsoft.com/en-us/rest/api/power-bi/datasets/execute-queries-in-group).
 
 **Refresh Semantic Model** requests a standard refresh and returns acceptance immediately. Acceptance does not mean the refresh completed. Use **Get Refresh History** separately to inspect status; it requires write permission on the model. `Unknown` can mean a refresh is in progress or its completion state is unknown. Standard refresh is subject to capacity limits, including eight refresh requests per day on shared capacity. See Microsoft's [refresh documentation](https://learn.microsoft.com/en-us/rest/api/power-bi/datasets/refresh-dataset-in-group) and [refresh history documentation](https://learn.microsoft.com/en-us/rest/api/power-bi/datasets/get-refresh-history-in-group).
 
@@ -132,7 +132,7 @@ Get semantic model metadata from a Power BI workspace.
 
 ### Power BI Execute DAX Query
 
-Execute one DAX query against a semantic model and preserve rows alongside reported query errors.
+Execute one DAX query against a semantic model and detect reported query errors.
 
 #### Input
 
@@ -147,7 +147,7 @@ Execute one DAX query against a semantic model and preserve rows alongside repor
 
 | Parameter | Type | Description |
 | --------- | ---- | ----------- |
-| `rows` | array | Returned rows, preserving provider column keys and partial results |
+| `rows` | array | Returned query rows, preserving provider column keys |
 | `rowCount` | number | Number of rows returned |
 | `errors` | array | Errors reported at response, query, or table scope |
 |   ↳ `scope` | string | Error scope: response, query, or table |

apps/sim/blocks/blocks/powerbi.ts

@@ -385,8 +385,7 @@ export const PowerBIBlock: BlockConfig = {
     },
     rows: {
       type: 'json',
-      description:
-        'Query rows preserving DAX column names, including available partial rows on error',
+      description: 'Rows from a successful query, preserving DAX column names',
       condition: { field: 'operation', value: 'powerbi_execute_query' },
     },
     rowCount: {
@@ -447,7 +446,7 @@ export const PowerBIBlockMeta = {
       icon: PowerBIIcon,
       title: 'Power BI KPI briefing',
       prompt:
-        'Build a scheduled workflow that runs an approved DAX query for daily KPIs in a Power BI semantic model, summarizes complete results, and posts the briefing to Microsoft Teams. Route query errors to a separate notification instead of treating partial rows as a complete report.',
+        "Build a scheduled workflow that runs an approved DAX query for daily KPIs in a Power BI semantic model, summarizes successful results, and posts the briefing to Microsoft Teams. Connect the query block's error port to a separate failure notification.",
       modules: ['scheduled', 'agent', 'workflows'],
       category: 'operations',
       tags: ['analytics', 'reporting'],
@@ -518,7 +517,7 @@ export const PowerBIBlockMeta = {
       description:
         'Run an approved DAX query and turn complete semantic model results into a KPI summary.',
       content:
-        '# Summarize Semantic Model KPIs\n\n## Steps\n1. Select an accessible workspace and semantic model. Ask for known measure names or an approved query; these actions do not discover the model schema.\n2. Use Execute DAX Query with one EVALUATE query returning one table. Keep the result small and include nulls when blanks matter.\n3. Check errors and incomplete before summarizing. Keep partial rows separate from a complete KPI report.\n\n## Output\nKPI values, their model context, and any query errors.\n\n## Sources\n[Microsoft DAX queries](https://learn.microsoft.com/en-us/dax/dax-queries) · [Execute Queries API](https://learn.microsoft.com/en-us/rest/api/power-bi/datasets/execute-queries-in-group)',
+        "# Summarize Semantic Model KPIs\n\n## Steps\n1. Select an accessible workspace and semantic model. Ask for known measure names or an approved query; these actions do not discover the model schema.\n2. Use Execute DAX Query with one EVALUATE query returning one table. Keep the result small and include nulls when blanks matter.\n3. Summarize only successful query results. Connect the query block's error port to a failure notification using its error output; partial rows are unavailable on the workflow error path.\n\n## Output\nKPI values and their model context on success, or a separate query failure notification.\n\n## Sources\n[Microsoft DAX queries](https://learn.microsoft.com/en-us/dax/dax-queries) · [Execute Queries API](https://learn.microsoft.com/en-us/rest/api/power-bi/datasets/execute-queries-in-group)",
     },
     {
       name: 'inventory-workspace-reports',

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

@@ -5,11 +5,7 @@ import { uploadsMock } from '@sim/testing/mocks/uploads.mock'
 import { DrizzleQueryError } from 'drizzle-orm/errors'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { clearLargeValueCacheForTests } from '@/lib/execution/payloads/cache'
-import {
-  createLargeArrayManifest,
-  isLargeArrayManifest,
-  readLargeArrayManifestSlice,
-} from '@/lib/execution/payloads/large-array-manifest'
+import { createLargeArrayManifest } from '@/lib/execution/payloads/large-array-manifest'
 import { isLargeValueRef } from '@/lib/execution/payloads/large-value-ref'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
 import { validateBlockType } from '@/ee/access-control/utils/permission-check'
@@ -18,7 +14,7 @@ import type { DAGNode } from '@/executor/dag/builder'
 import { BlockExecutor } from '@/executor/execution/block-executor'
 import { ExecutionState } from '@/executor/execution/state'
 import type { BlockHandler, ExecutionContext } from '@/executor/types'
-import { attachToolFailureOutput, attachTrustedExecutionCost } from '@/executor/utils/errors'
+import { attachTrustedExecutionCost } from '@/executor/utils/errors'
 import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
 import { VariableResolver } from '@/executor/variables/resolver'
 import type { SerializedBlock, SerializedWorkflow } from '@/serializer/types'
@@ -133,256 +129,6 @@ describe('BlockExecutor', () => {
     expect(context.mcpBlockId).toBeUndefined()
   })
 
-  function createFailedToolExecution(
-    output: Record<string, unknown>,
-    options: { piiRedaction?: boolean } = {}
-  ) {
-    const block = createBlock()
-    const workflow: SerializedWorkflow = {
-      version: '1',
-      blocks: [block],
-      connections: [],
-      loops: {},
-      parallels: {},
-    }
-    const state = new ExecutionState()
-    const failure = new Error('query incomplete')
-    attachToolFailureOutput(failure, output)
-    const handler: BlockHandler = {
-      canHandle: () => true,
-      execute: async () => {
-        throw failure
-      },
-    }
-    const onBlockComplete = vi.fn(async () => {})
-    const executor = new BlockExecutor(
-      [handler],
-      new VariableResolver(workflow, {}, state),
-      { onBlockComplete },
-      state
-    )
-    const ctx = createContext(state)
-    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>'))
-    )
-    const { executor, block, state, ctx, node, onBlockComplete } = createFailedToolExecution({
-      rows: [{ email: 'alice@example.com', count: 7 }],
-      rowCount: 1,
-      incomplete: true,
-    })
-
-    const output = await executor.execute(ctx, node, block)
-    await vi.waitFor(() => expect(onBlockComplete).toHaveBeenCalledOnce())
-
-    const expected = {
-      rows: [{ email: '<EMAIL_ADDRESS>', count: 7 }],
-      rowCount: 1,
-      incomplete: true,
-      error: 'query incomplete',
-    }
-    expect(output).toEqual(expected)
-    expect(state.getBlockOutput(block.id)).toEqual(expected)
-    expect(ctx.blockLogs[0]?.output).toEqual(expected)
-    expect(onBlockComplete.mock.calls[0]?.[3]?.output).toEqual(expected)
-    expect(ctx.blockLogs[0]).toMatchObject({ success: false, errorHandled: true })
-  })
-
-  it('masks and re-stores partial failed tool manifests under the current execution', async () => {
-    const items = [{ email: 'alice@example.com', count: 7 }]
-    const manifest = await createLargeArrayManifest(items, {
-      workspaceId: 'workspace-1',
-      workflowId: 'workflow-1',
-      executionId: 'source-execution',
-    })
-    clearLargeValueCacheForTests()
-    mockDownloadFile.mockResolvedValue(Buffer.from(JSON.stringify(items)))
-    mockMaskBatch.mockImplementation(async (texts: string[]) =>
-      texts.map((text) => text.replaceAll('alice@example.com', '<EMAIL_ADDRESS>'))
-    )
-    const { executor, block, state, ctx, node, onBlockComplete } = createFailedToolExecution({
-      rows: manifest,
-      rowCount: 1,
-      incomplete: true,
-    })
-    ctx.largeValueExecutionIds = ['source-execution']
-
-    const output = await executor.execute(ctx, node, block)
-    await vi.waitFor(() => expect(onBlockComplete).toHaveBeenCalledOnce())
-
-    expect(output).toMatchObject({ rowCount: 1, incomplete: true, error: 'query incomplete' })
-    expect(output.rows).toMatchObject({
-      preview: [{ email: '<EMAIL_ADDRESS>', count: 7 }],
-      chunks: [{ ref: { executionId: 'execution-1' } }],
-    })
-    if (!isLargeArrayManifest(output.rows)) throw new Error('Expected a masked row manifest')
-    expect(
-      await readLargeArrayManifestSlice(output.rows, 0, 1, {
-        workspaceId: ctx.workspaceId,
-        workflowId: ctx.workflowId,
-        executionId: ctx.executionId,
-      })
-    ).toEqual([{ email: '<EMAIL_ADDRESS>', count: 7 }])
-    expect(state.getBlockOutput(block.id)).toEqual(output)
-    expect(ctx.blockLogs[0]?.output).toEqual(output)
-    expect(onBlockComplete.mock.calls[0]?.[3]?.output).toEqual(output)
-  })
-
-  it('omits failed tool payloads when masking fails while preserving trusted cost', async () => {
-    const unsafeFailure = 'mask service failed while processing alice@example.com'
-    mockMaskBatch.mockRejectedValueOnce(new Error(unsafeFailure))
-    const { executor, block, state, ctx, node, failure, onBlockComplete } =
-      createFailedToolExecution({
-        rows: [{ email: 'alice@example.com', count: 7 }],
-        rowCount: 1,
-        incomplete: true,
-      })
-    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())
-
-    const expected = {
-      error: 'PII redaction failed. Partial tool output 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 })
-    const surfaced = JSON.stringify([
-      output,
-      ctx.blockLogs,
-      onBlockComplete.mock.calls,
-      blockExecutorBaseLogger.error.mock.calls,
-    ])
-    expect(surfaced).not.toContain('alice@example.com')
-    expect(surfaced).not.toContain(unsafeFailure)
-  })
-
   it('redacts an authorized prior-execution manifest returned by a block under the current execution', async () => {
     const items = [{ email: 'alice@example.com', count: 7 }]
     const manifest = await createLargeArrayManifest(items, {