fix(powerbi): harden partial results and selector hydration

Author: BillLeoutsakosvl346

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

@@ -209,7 +209,7 @@ Sim requests these scopes when someone connects a Power BI account. On a self-ho
 | ----- | ----------- |
 | `https://analysis.windows.net/powerbi/api/Workspace.Read.All` | View Power BI workspaces |
 | `https://analysis.windows.net/powerbi/api/Report.Read.All` | View Power BI reports |
-| `https://analysis.windows.net/powerbi/api/Dataset.ReadWrite.All` | Read and query Power BI semantic models, request refreshes, and view refresh history |
+| `https://analysis.windows.net/powerbi/api/Dataset.ReadWrite.All` | Read, create, update, refresh, and delete Power BI semantic models within your account’s permissions |
 | `openid` | Standard authentication |
 | `profile` | Access profile information |
 | `email` | Access email address |

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

@@ -5,7 +5,11 @@ 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 } from '@/lib/execution/payloads/large-array-manifest'
+import {
+  createLargeArrayManifest,
+  isLargeArrayManifest,
+  readLargeArrayManifestSlice,
+} 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'
@@ -14,7 +18,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 { attachTrustedExecutionCost } from '@/executor/utils/errors'
+import { attachToolFailureOutput, 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'
@@ -129,6 +133,143 @@ describe('BlockExecutor', () => {
     expect(context.mcpBlockId).toBeUndefined()
   })
 
+  function createFailedToolExecution(output: Record<string, unknown>) {
+    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)
+    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('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, {

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

@@ -350,35 +350,7 @@ export class BlockExecutor {
         })) as NormalizedBlockOutput
       }
 
-      if (blockCtx.piiBlockOutputRedaction?.enabled) {
-        // In-flight redaction before the log/state split below, so both the
-        // downstream state copy and the persisted log copy are masked.
-        // `onFailure: 'throw'` aborts the run rather than feeding corrupted/leaked
-        // data downstream.
-        const redactionOptions = {
-          entityTypes: blockCtx.piiBlockOutputRedaction.entityTypes,
-          language: blockCtx.piiBlockOutputRedaction.language,
-          customPatterns: blockCtx.piiBlockOutputRedaction.customPatterns,
-          onFailure: 'throw' as const,
-        }
-        // Tools like the function executor offload large outputs to large-value
-        // refs BEFORE they reach here, and the string walk treats a ref as opaque.
-        // So hydrate → mask → re-store any refs first, then mask inline strings —
-        // otherwise PII inside an offloaded output is never redacted.
-        normalizedOutput = await redactLargeValueRefsInValue(normalizedOutput, {
-          ...redactionOptions,
-          store: {
-            workspaceId: blockCtx.workspaceId,
-            workflowId: blockCtx.workflowId,
-            executionId: blockCtx.executionId,
-            largeValueExecutionIds: blockCtx.largeValueExecutionIds,
-            largeValueKeys: blockCtx.largeValueKeys,
-            allowLargeValueWorkflowScope: blockCtx.allowLargeValueWorkflowScope,
-            userId: blockCtx.userId,
-          },
-        })
-        normalizedOutput = await redactObjectStrings(normalizedOutput, redactionOptions)
-      }
+      normalizedOutput = await this.redactBlockOutput(normalizedOutput, blockCtx)
 
       const compacted = await compactBlockOutput(normalizedOutput, {
         workspaceId: blockCtx.workspaceId,
@@ -622,6 +594,33 @@ export class BlockExecutor {
     }
   }
 
+  private async redactBlockOutput(
+    output: NormalizedBlockOutput,
+    ctx: ExecutionContext
+  ): Promise<NormalizedBlockOutput> {
+    if (!ctx.piiBlockOutputRedaction?.enabled) return output
+    const options = {
+      entityTypes: ctx.piiBlockOutputRedaction.entityTypes,
+      language: ctx.piiBlockOutputRedaction.language,
+      customPatterns: ctx.piiBlockOutputRedaction.customPatterns,
+      onFailure: 'throw' as const,
+    }
+    // Offloaded content must be hydrated, masked, and re-stored before inline strings.
+    const redacted = await redactLargeValueRefsInValue(output, {
+      ...options,
+      store: {
+        workspaceId: ctx.workspaceId,
+        workflowId: ctx.workflowId,
+        executionId: ctx.executionId,
+        largeValueExecutionIds: ctx.largeValueExecutionIds,
+        largeValueKeys: ctx.largeValueKeys,
+        allowLargeValueWorkflowScope: ctx.allowLargeValueWorkflowScope,
+        userId: ctx.userId,
+      },
+    })
+    return redactObjectStrings(redacted, options)
+  }
+
   private async handleBlockError(
     error: unknown,
     ctx: ExecutionContext,
@@ -637,10 +636,10 @@ export class BlockExecutor {
     streamingPartialOutput?: Record<string, any>,
     completedHandlerCost?: TrustedExecutionCost
   ): Promise<NormalizedBlockOutput> {
-    const endedAt = new Date().toISOString()
-    const duration = performance.now() - startTime
+    let endedAt = new Date().toISOString()
+    let duration = performance.now() - startTime
     const isDatabaseError = error instanceof DrizzleQueryError
-    const errorMessage = isDatabaseError ? INTERNAL_DATABASE_ERROR_MESSAGE : normalizeError(error)
+    let errorMessage = isDatabaseError ? INTERNAL_DATABASE_ERROR_MESSAGE : normalizeError(error)
     const hasLogInputs =
       inputsForLog && typeof inputsForLog === 'object' && Object.keys(inputsForLog).length > 0
     const input = hasLogInputs
@@ -709,8 +708,19 @@ export class BlockExecutor {
     }
 
     const trustedExecutionCost = readTrustedExecutionCost(error) ?? completedHandlerCost
+    let partialOutput = readToolFailureOutput(error)
+    if (partialOutput) {
+      try {
+        partialOutput = await this.redactBlockOutput(partialOutput, ctx)
+      } catch {
+        partialOutput = undefined
+        errorMessage = 'PII redaction failed. Partial tool output was omitted.'
+      }
+    }
+    endedAt = new Date().toISOString()
+    duration = performance.now() - startTime
     const errorOutput: NormalizedBlockOutput = {
-      ...readToolFailureOutput(error),
+      ...partialOutput,
       error: errorMessage,
       ...(trustedExecutionCost ? { cost: trustedExecutionCost } : {}),
     }

apps/sim/executor/handlers/generic/generic-handler.ts

@@ -10,6 +10,7 @@ import { prepareResolvedSecretProjectedInputs } from '@/executor/utils/resolved-
 import type { ResolvedSecretInputPath } from '@/executor/utils/resolved-secret-trace-registry'
 import type { SerializedBlock } from '@/serializer/types'
 import { executeTool } from '@/tools'
+import { isToolDiagnosticFailure } from '@/tools/errors'
 import { isInternalToolConfig, type ToolConfig } from '@/tools/types'
 import { getTool } from '@/tools/utils'
 
@@ -348,7 +349,7 @@ export class GenericBlockHandler implements BlockHandler {
         const error = new Error(errorMessage)
 
         const declaredOutput: Record<string, unknown> = {}
-        if (isPlainRecord(result.output)) {
+        if (!isToolDiagnosticFailure(result) && isPlainRecord(result.output)) {
           for (const key of Object.keys(tool?.outputs ?? {})) {
             if (key !== 'cost' && key !== 'error' && Object.hasOwn(result.output, key)) {
               declaredOutput[key] = result.output[key]

apps/sim/lib/oauth/utils.ts

@@ -15,7 +15,7 @@ export const SCOPE_DESCRIPTIONS: Record<string, string> = {
   'https://analysis.windows.net/powerbi/api/Workspace.Read.All': 'View Power BI workspaces',
   'https://analysis.windows.net/powerbi/api/Report.Read.All': 'View Power BI reports',
   'https://analysis.windows.net/powerbi/api/Dataset.ReadWrite.All':
-    'Read and query Power BI semantic models, request refreshes, and view refresh history',
+    'Read, create, update, refresh, and delete Power BI semantic models within your account’s permissions',
   'users.profile:write': 'Update Slack user profiles',
   'users.profile:read': 'View Slack user profiles',
   'channels:join': 'Join public Slack channels',

apps/sim/lib/selectors/manifest.ts

@@ -380,7 +380,7 @@ export const selectorManifest = {
     search: true,
     detail: true,
   }),
-  'powerbi.workspaces': providerSelector([], { listMode: 'paginated' }),
+  'powerbi.workspaces': providerSelector([], { listMode: 'paginated', detail: true }),
   'powerbi.datasets': providerSelector(['groupId'], {
     readiness: { all: ['oauthCredential', 'groupId'] },
     detail: true,

apps/sim/lib/selectors/server/providers/powerbi.test.ts

@@ -161,30 +161,47 @@ describe('Power BI selector provider boundary', () => {
     }
   )
 
-  it.each(['datasets', 'reports'] as const)(
-    'resolves a saved %s selection without loading the full resource list',
+  it.each(['workspaces', 'datasets', 'reports'] as const)(
+    'resolves a saved %s selection through one direct encoded provider request',
     async (kind) => {
-      fetchMock.mockResolvedValueOnce(jsonResponse({ id: 'resource-id', name: 'Saved resource' }))
+      fetchMock.mockResolvedValueOnce(
+        jsonResponse({
+          id: 'resource#id',
+          name: 'Saved resource',
+          rawProviderSecret: 'must-not-escape',
+        })
+      )
       const key = `powerbi.${kind}` as const
       expect(
         await powerBISelectorAttachments[key].execute(
           args({ selectorKey: key, request: { kind: 'detail', id: ' resource#id ' } })
         )
-      ).toEqual({ kind: 'detail', item: { id: 'resource-id', label: 'Saved resource' } })
-      expect(new URL(String(fetchMock.mock.calls[0]?.[0])).pathname).toBe(
-        `/v1.0/myorg/groups/workspace-one/${kind}/resource%23id`
+      ).toEqual({ kind: 'detail', item: { id: 'resource#id', label: 'Saved resource' } })
+      expect(fetchMock).toHaveBeenCalledTimes(1)
+      const url = new URL(String(fetchMock.mock.calls[0]?.[0]))
+      expect(url.origin).toBe('https://api.powerbi.com')
+      expect(url.pathname).toBe(
+        kind === 'workspaces'
+          ? '/v1.0/myorg/groups/resource%23id'
+          : `/v1.0/myorg/groups/workspace-one/${kind}/resource%23id`
       )
+      expect(url.search).toBe('')
+      expect(url.hash).toBe('')
     }
   )
 
-  it('returns no detail for a deleted semantic model', async () => {
-    fetchMock.mockResolvedValueOnce(new Response(null, { status: 404 }))
-    expect(
-      await powerBISelectorAttachments['powerbi.datasets'].execute(
-        args({ selectorKey: 'powerbi.datasets', request: { kind: 'detail', id: 'removed-model' } })
-      )
-    ).toEqual({ kind: 'detail', item: null })
-  })
+  it.each(['workspaces', 'datasets'] as const)(
+    'returns no detail for a deleted %s selection',
+    async (kind) => {
+      fetchMock.mockResolvedValueOnce(new Response(null, { status: 404 }))
+      const key = `powerbi.${kind}` as const
+      expect(
+        await powerBISelectorAttachments[key].execute(
+          args({ selectorKey: key, request: { kind: 'detail', id: 'removed-resource' } })
+        )
+      ).toEqual({ kind: 'detail', item: null })
+    }
+  )
 
   it.each([
     {},

apps/sim/tools/errors.ts

@@ -1,4 +1,17 @@
 import { HttpError } from '@/lib/core/utils/http-error'
+import type { ToolResponse } from '@/tools/types'
+
+const diagnosticFailures = new WeakSet<ToolResponse>()
+
+/** Catch-generated HTTP diagnostics are not declared resource outputs. */
+export function markToolDiagnosticFailure(response: ToolResponse): ToolResponse {
+  diagnosticFailures.add(response)
+  return response
+}
+
+export function isToolDiagnosticFailure(response: ToolResponse): boolean {
+  return diagnosticFailures.has(response)
+}
 
 /**
  * Hosted-key acquisition blocked by the workspace's own rate bucket. Carries a