fix(tools): bind schema-enrichment reads to executor delegations (#6593)

* fix(tools): bind schema-enrichment reads to executor delegations

Two schema-enrichment callers still sent the deprecated legacy internal JWT
to routes that moved onto delegation-only auth, so both 401'd and swallowed
the failure:

- tools/params.ts fetched a child workflow's input fields with an unsubjected
  buildAuthHeaders(), leaving the Agent block's workflow_executor inputMapping
  untyped so the model guessed the child's field names.
- tools/schema-enrichers.ts fetched KB tag definitions the same way, which
  dropped the tags/tagFilters parameter from the knowledge tools entirely.

Extract the executionId binding rule into executionScopeForTarget so the three
enrichment call sites share one definition.

* improvement(tools): surface tag-definition read failures at error level

Enrichment degrades silently by design, so the log line is the only signal a
credential break leaves. Drain the body on the failure path too.
This commit is contained in:
Waleed
2026-08-11 21:22:16 -07:00
committed by GitHub
parent a1a05c6ece
commit a61cbf7287
7 changed files with 245 additions and 22 deletions
@@ -0,0 +1,27 @@
/**
* @vitest-environment node
*/
import { describe, expect, it } from 'vitest'
import { executionScopeForTarget } from '@/executor/utils/delegation'
describe('executionScopeForTarget', () => {
it('binds the execution when the target is the running workflow', () => {
expect(
executionScopeForTarget({ workflowId: 'workflow-1', executionId: 'run-1' }, 'workflow-1')
).toEqual({ executionId: 'run-1' })
})
it('omits the execution for a child workflow, which binds on its own id', () => {
expect(
executionScopeForTarget({ workflowId: 'parent', executionId: 'run-1' }, 'child')
).toEqual({})
})
it('omits the execution outside an active run', () => {
expect(executionScopeForTarget({ workflowId: 'workflow-1' }, 'workflow-1')).toEqual({})
})
it('omits the execution when the context has no workflow to compare', () => {
expect(executionScopeForTarget({ executionId: 'run-1' }, 'workflow-1')).toEqual({})
})
})
+21
View File
@@ -0,0 +1,21 @@
import type { GenerateInternalDelegationTokenInput } from '@/lib/auth/internal'
/**
* Binds the running execution to a delegation only when it targets the workflow that
* is actually running.
*
* A child workflow is a separate resource and binds on its own id, so forwarding the
* parent's `executionId` would assert a run that does not cover the target and the
* delegation would fail to bind. Callers spread the result into their delegation input.
*
* Kept free of runtime imports so client-reachable modules can read it without pulling
* in the executor graph.
*/
export function executionScopeForTarget(
context: { workflowId?: string; executionId?: string },
targetWorkflowId: string
): Pick<GenerateInternalDelegationTokenInput, 'executionId'> {
return context.workflowId === targetWorkflowId && context.executionId
? { executionId: context.executionId }
: {}
}
+2 -3
View File
@@ -90,13 +90,12 @@ async function fetchWorkflowMetadata(
throw new Error('Workflow metadata enrichment requires a trusted execution subject')
}
const { buildAPIUrl, buildExecutorDelegationHeaders } = await import('@/executor/utils/http')
const { executionScopeForTarget } = await import('@/executor/utils/delegation')
const headers = await buildExecutorDelegationHeaders({
subjectUserId: executionContext.userId,
workflowId,
...(executionContext.workflowId === workflowId && executionContext.executionId
? { executionId: executionContext.executionId }
: {}),
...executionScopeForTarget(executionContext, workflowId),
})
const url = buildAPIUrl(`/api/workflows/${workflowId}`)
+110 -1
View File
@@ -1,4 +1,4 @@
import { afterAll, describe, expect, it, vi } from 'vitest'
import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
import { mergeToolParameters } from '@/tools/merge-params'
import * as toolMetadata from '@/tools/metadata'
import {
@@ -18,6 +18,17 @@ import {
} from '@/tools/params'
import type { HttpMethod, ParameterVisibility } from '@/tools/types'
const { mockBuildExecutorDelegationHeaders } = vi.hoisted(() => ({
mockBuildExecutorDelegationHeaders: vi
.fn()
.mockResolvedValue({ Authorization: 'Bearer delegation-token' }),
}))
vi.mock('@/executor/utils/http', () => ({
buildExecutorDelegationHeaders: mockBuildExecutorDelegationHeaders,
buildAPIUrl: (path: string) => new URL(path, 'http://localhost:3000'),
}))
const mockToolConfig = {
id: 'test_tool',
name: 'Test Tool',
@@ -648,6 +659,104 @@ describe('Tool Parameters Utils', () => {
})
})
describe('createLLMToolSchema - child workflow input enrichment', () => {
const childWorkflowPayload = {
data: {
state: {
blocks: {
'block-1': {
type: 'starter',
subBlocks: {
inputFormat: {
value: [
{ name: 'email', type: 'string', description: 'Recipient address' },
{ name: 'attempts', type: 'number' },
],
},
},
},
},
},
},
}
const mockFetch = vi.fn()
beforeEach(() => {
mockBuildExecutorDelegationHeaders.mockClear()
mockFetch.mockReset()
mockFetch.mockResolvedValue(
new Response(JSON.stringify(childWorkflowPayload), {
status: 200,
headers: { 'Content-Type': 'application/json' },
})
)
// The suite runs with `unstubGlobals`, which restores globals between tests.
vi.stubGlobal('fetch', mockFetch)
})
it('binds the delegation to the execution subject and the target workflow', async () => {
const { schema } = await createLLMToolSchema(
mockWorkflowExecutorConfig,
{ workflowId: 'child-workflow' },
{
userId: 'user-1',
workflowId: 'parent-workflow',
executionId: 'execution-1',
workspaceId: 'workspace-1',
}
)
expect(mockBuildExecutorDelegationHeaders).toHaveBeenCalledWith({
subjectUserId: 'user-1',
workflowId: 'child-workflow',
})
expect(schema.properties.inputMapping.properties).toEqual({
email: { type: 'string', description: 'Recipient address' },
attempts: { type: 'number', description: 'Input field: attempts' },
})
expect(schema.properties.inputMapping.required).toEqual(['email', 'attempts'])
})
it('carries the executionId when the target is the running workflow', async () => {
await createLLMToolSchema(
mockWorkflowExecutorConfig,
{ workflowId: 'parent-workflow' },
{ userId: 'user-1', workflowId: 'parent-workflow', executionId: 'execution-1' }
)
expect(mockBuildExecutorDelegationHeaders).toHaveBeenCalledWith({
subjectUserId: 'user-1',
workflowId: 'parent-workflow',
executionId: 'execution-1',
})
})
it('leaves inputMapping untyped and issues no request without an execution subject', async () => {
const { schema } = await createLLMToolSchema(
mockWorkflowExecutorConfig,
{ workflowId: 'child-workflow' },
{ workflowId: 'parent-workflow', executionId: 'execution-1' }
)
expect(mockBuildExecutorDelegationHeaders).not.toHaveBeenCalled()
expect(mockFetch).not.toHaveBeenCalled()
expect(schema.properties.inputMapping.properties).toBeUndefined()
})
it('leaves inputMapping untyped when the workflow read is rejected', async () => {
mockFetch.mockResolvedValue(new Response('Unauthorized', { status: 401 }))
const { schema } = await createLLMToolSchema(
mockWorkflowExecutorConfig,
{ workflowId: 'child-workflow' },
{ userId: 'user-1', workflowId: 'parent-workflow' }
)
expect(schema.properties.inputMapping.properties).toBeUndefined()
})
})
describe('mergeToolParameters - inputMapping deep merge', () => {
it.concurrent('should deep merge inputMapping when user provides empty object', () => {
const userProvided = {
+21 -7
View File
@@ -699,7 +699,7 @@ export async function createLLMToolSchema(
if (isWorkflowInputMapping) {
const workflowId = userProvidedParams.workflowId as string
if (workflowId) {
await applyDynamicSchemaForWorkflow(propertySchema, workflowId)
await applyDynamicSchemaForWorkflow(propertySchema, workflowId, enrichmentContext)
}
}
@@ -742,10 +742,11 @@ export async function createLLMToolSchema(
*/
async function applyDynamicSchemaForWorkflow(
propertySchema: SchemaProperty,
workflowId: string
workflowId: string,
context: WorkflowToolExecutionContext
): Promise<void> {
try {
const workflowInputFields = await fetchWorkflowInputFields(workflowId)
const workflowInputFields = await fetchWorkflowInputFields(workflowId, context)
if (workflowInputFields && workflowInputFields.length > 0) {
propertySchema.type = 'object'
@@ -771,19 +772,32 @@ async function applyDynamicSchemaForWorkflow(
/**
* Fetches workflow input fields from the API.
*
* The workflow read route accepts only scoped executor delegations, so the call is
* bound to the acting execution subject.
*/
async function fetchWorkflowInputFields(
workflowId: string
workflowId: string,
context: WorkflowToolExecutionContext
): Promise<Array<{ name: string; type: string; description?: string }>> {
try {
const { buildAuthHeaders, buildAPIUrl } = await import('@/executor/utils/http')
if (!context.userId) {
throw new Error('Workflow input enrichment requires a trusted execution subject')
}
const { buildAPIUrl, buildExecutorDelegationHeaders } = await import('@/executor/utils/http')
const { executionScopeForTarget } = await import('@/executor/utils/delegation')
const headers = await buildAuthHeaders()
const headers = await buildExecutorDelegationHeaders({
subjectUserId: context.userId,
workflowId,
...executionScopeForTarget(context, workflowId),
})
const url = buildAPIUrl(`/api/workflows/${workflowId}`)
const response = await fetch(url.toString(), { headers })
if (!response.ok) {
throw new Error('Failed to fetch workflow')
await response.text().catch(() => {})
throw new Error(`Failed to fetch workflow (${response.status})`)
}
const { data } = await response.json()
+45 -8
View File
@@ -3,7 +3,12 @@
*/
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
const { mockBuildAPIUrl, mockBuildAuthHeaders, mockExtractAPIErrorMessage } = vi.hoisted(() => ({
const {
mockBuildAPIUrl,
mockBuildAuthHeaders,
mockBuildExecutorDelegationHeaders,
mockExtractAPIErrorMessage,
} = vi.hoisted(() => ({
mockBuildAPIUrl: vi.fn((path: string, params?: Record<string, string>) => {
const url = new URL(path, 'http://localhost:3000')
for (const [key, value] of Object.entries(params ?? {})) {
@@ -12,12 +17,14 @@ const { mockBuildAPIUrl, mockBuildAuthHeaders, mockExtractAPIErrorMessage } = vi
return url
}),
mockBuildAuthHeaders: vi.fn(),
mockBuildExecutorDelegationHeaders: vi.fn(),
mockExtractAPIErrorMessage: vi.fn(),
}))
vi.mock('@/executor/utils/http', () => ({
buildAPIUrl: mockBuildAPIUrl,
buildAuthHeaders: mockBuildAuthHeaders,
buildExecutorDelegationHeaders: mockBuildExecutorDelegationHeaders,
extractAPIErrorMessage: mockExtractAPIErrorMessage,
}))
@@ -106,14 +113,16 @@ describe('enrichTableToolSchema', () => {
describe('enrichKBTagsSchema', () => {
beforeEach(() => {
vi.clearAllMocks()
mockBuildAuthHeaders.mockResolvedValue({ Authorization: 'Bearer internal-token' })
mockBuildExecutorDelegationHeaders.mockResolvedValue({
Authorization: 'Bearer delegation-token',
})
})
afterEach(() => {
vi.unstubAllGlobals()
})
it('fetches tag definitions as the acting user so the route can authorize them', async () => {
it('binds the tag-definition read to the acting subject and workflow execution', async () => {
const mockFetch = vi.fn().mockResolvedValue(
new Response(
JSON.stringify({
@@ -125,18 +134,46 @@ describe('enrichKBTagsSchema', () => {
)
vi.stubGlobal('fetch', mockFetch)
const result = await enrichKBTagsSchema('kb-1', { userId: 'user-1' })
const result = await enrichKBTagsSchema('kb-1', {
userId: 'user-1',
workflowId: 'workflow-1',
executionId: 'execution-1',
})
expect(mockBuildAuthHeaders).toHaveBeenCalledWith('user-1')
expect(mockBuildExecutorDelegationHeaders).toHaveBeenCalledWith({
subjectUserId: 'user-1',
workflowId: 'workflow-1',
executionId: 'execution-1',
})
expect(result?.properties).toEqual({ Client: { type: 'string', description: 'text tag' } })
})
it('skips enrichment without an acting user rather than issuing an unauthorized request', async () => {
it('omits the executionId outside an active run', async () => {
const mockFetch = vi.fn().mockResolvedValue(
new Response(JSON.stringify({ success: true, data: [] }), {
status: 200,
headers: { 'Content-Type': 'application/json' },
})
)
vi.stubGlobal('fetch', mockFetch)
await enrichKBTagsSchema('kb-1', { userId: 'user-1', workflowId: 'workflow-1' })
expect(mockBuildExecutorDelegationHeaders).toHaveBeenCalledWith({
subjectUserId: 'user-1',
workflowId: 'workflow-1',
})
})
it.each([
['no acting user', { workflowId: 'workflow-1' }],
['no acting workflow to bind the delegation on', { userId: 'user-1' }],
])('skips enrichment with %s rather than issuing an unauthorized request', async (_, context) => {
const mockFetch = vi.fn()
vi.stubGlobal('fetch', mockFetch)
await expect(enrichKBTagsSchema('kb-1', {})).resolves.toBeNull()
await expect(enrichKBTagsSchema('kb-1', context)).resolves.toBeNull()
expect(mockFetch).not.toHaveBeenCalled()
expect(mockBuildAuthHeaders).not.toHaveBeenCalled()
expect(mockBuildExecutorDelegationHeaders).not.toHaveBeenCalled()
})
})
+19 -3
View File
@@ -116,6 +116,10 @@ function mapFieldTypeToSchemaType(fieldType: string): string {
/**
* Fetches tag definitions from a knowledge base as the acting user, whose id the
* route requires to authorize the read.
*
* The tag-definition route accepts only scoped executor delegations. The delegation
* binds on the running workflow — that is what resolves the workspace the knowledge
* base must belong to — so both the subject and the workflow are required.
*/
async function fetchTagDefinitions(
knowledgeBaseId: string,
@@ -125,18 +129,30 @@ async function fetchTagDefinitions(
logger.warn(`Skipping tag definition enrichment for KB ${knowledgeBaseId}: no acting user`)
return []
}
if (!context.workflowId) {
logger.warn(`Skipping tag definition enrichment for KB ${knowledgeBaseId}: no acting workflow`)
return []
}
try {
const { buildAuthHeaders, buildAPIUrl } = await import('@/executor/utils/http')
const { buildAPIUrl, buildExecutorDelegationHeaders } = await import('@/executor/utils/http')
const { executionScopeForTarget } = await import('@/executor/utils/delegation')
const headers = await buildAuthHeaders(context.userId)
const headers = await buildExecutorDelegationHeaders({
subjectUserId: context.userId,
workflowId: context.workflowId,
...executionScopeForTarget(context, context.workflowId),
})
const url = buildAPIUrl(`/api/knowledge/${knowledgeBaseId}/tag-definitions`)
logger.info(`Fetching tag definitions for KB ${knowledgeBaseId} from ${url.toString()}`)
const response = await fetch(url.toString(), { headers })
if (!response.ok) {
logger.warn(`Failed to fetch tag definitions for KB ${knowledgeBaseId}: ${response.status}`)
await response.text().catch(() => {})
// Error, not warn: enrichment degrades silently, so a credential break is only
// ever visible here. A 401 means the delegation stopped satisfying the route.
logger.error(`Failed to fetch tag definitions for KB ${knowledgeBaseId}: ${response.status}`)
return []
}