mirror of
https://github.com/n8n-io/n8n.git
synced 2026-07-30 20:45:07 +02:00
feat(ai-builder): Verify approval-loop branches with scripted gate decisions (#34822)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
José Braulio González Valido
GitHub
parent
fcaadfa375
commit
00fe448a68
|
|
@ -86,3 +86,41 @@ describe('prepareVerificationRun — halted wait gates', () => {
|
|||
expect(result.prepared.verificationPinData?.['Email Approval']).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('prepareVerificationRun — gate scripts', () => {
|
||||
it('exposes the script that matches a halted gate', () => {
|
||||
const script = {
|
||||
nodeName: 'Email Approval',
|
||||
cutEdge: { source: 'Revise', target: 'Format' },
|
||||
decisions: [{ label: 'approve', items: [{ data: { approved: true } }] }],
|
||||
};
|
||||
const result = prepareVerificationRun(
|
||||
makeBuildOutcome({ nodeSimulationPlan: [gateVerdict], waitGateScripts: [script] }),
|
||||
undefined,
|
||||
);
|
||||
|
||||
expect(result.kind).toBe('ready');
|
||||
if (result.kind !== 'ready') return;
|
||||
expect(result.prepared.gateScript).toEqual(script);
|
||||
});
|
||||
|
||||
it('ignores scripts that do not match a halted gate', () => {
|
||||
const result = prepareVerificationRun(
|
||||
makeBuildOutcome({
|
||||
nodeSimulationPlan: [plainVerdict],
|
||||
waitGateScripts: [
|
||||
{
|
||||
nodeName: 'Some Other Node',
|
||||
cutEdge: { source: 'A', target: 'B' },
|
||||
decisions: [{ label: 'x', items: [{}] }],
|
||||
},
|
||||
],
|
||||
}),
|
||||
undefined,
|
||||
);
|
||||
|
||||
expect(result.kind).toBe('ready');
|
||||
if (result.kind !== 'ready') return;
|
||||
expect(result.prepared.gateScript).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -0,0 +1,164 @@
|
|||
import type {
|
||||
WaitGateScript,
|
||||
WorkflowBuildOutcome,
|
||||
} from '../../../../workflow-loop/workflow-loop-state';
|
||||
import type { PreparedVerificationRun } from '../prepare-run';
|
||||
import { runScriptedGateVerification, type ScriptedGateRunArgs } from '../scripted-gate-run';
|
||||
|
||||
function makeBuildOutcome(overrides: Partial<WorkflowBuildOutcome> = {}): WorkflowBuildOutcome {
|
||||
return {
|
||||
workItemId: 'wi_1',
|
||||
taskId: 'task_1',
|
||||
workflowId: 'wf_1',
|
||||
submitted: true,
|
||||
triggerType: 'manual_or_testable',
|
||||
needsUserInput: false,
|
||||
summary: 'Built',
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
const script: WaitGateScript = {
|
||||
nodeName: 'Gate',
|
||||
cutEdge: { source: 'Revise', target: 'Format' },
|
||||
decisions: [
|
||||
{ label: 'approve', items: [{ data: { approved: true } }] },
|
||||
{ label: 'decline', items: [{ data: { approved: false } }] },
|
||||
],
|
||||
};
|
||||
|
||||
const prepared: PreparedVerificationRun = {
|
||||
verificationPinData: { Gate: [], Generate: [{ text: 'draft' }] },
|
||||
simulatedNodes: [
|
||||
{ nodeName: 'Gate', reason: 'gate' },
|
||||
{ nodeName: 'Generate', reason: 'no credentials' },
|
||||
],
|
||||
haltedGateNames: ['Gate'],
|
||||
gateScript: script,
|
||||
};
|
||||
|
||||
const approvePassResult = {
|
||||
executionId: 'exec-approve',
|
||||
status: 'success',
|
||||
executedNodeNames: ['Trigger', 'Generate', 'Gate', 'Router', 'Publish'],
|
||||
lastNodeExecuted: 'Publish',
|
||||
data: { Publish: [{ ok: true }] },
|
||||
};
|
||||
|
||||
const declinePassResult = {
|
||||
executionId: 'exec-decline',
|
||||
status: 'success',
|
||||
executedNodeNames: ['Trigger', 'Generate', 'Gate', 'Router', 'Revise'],
|
||||
lastNodeExecuted: 'Revise',
|
||||
data: { Revise: [{ text: 'revised' }] },
|
||||
};
|
||||
|
||||
function makeArgs(runMock: ReturnType<typeof vi.fn>): ScriptedGateRunArgs {
|
||||
return {
|
||||
script,
|
||||
prepared,
|
||||
executionService: { run: runMock } as unknown as ScriptedGateRunArgs['executionService'],
|
||||
workflowId: 'wf_1',
|
||||
timeout: 30_000,
|
||||
buildOutcome: makeBuildOutcome({
|
||||
nodeSimulationPlan: [
|
||||
{
|
||||
nodeName: 'Gate',
|
||||
verdict: 'simulate',
|
||||
reason: 'gate',
|
||||
confidence: 'high',
|
||||
source: 'deterministic',
|
||||
haltBranch: true,
|
||||
},
|
||||
],
|
||||
}),
|
||||
stateBefore: undefined,
|
||||
runId: 'run-1',
|
||||
};
|
||||
}
|
||||
|
||||
describe('runScriptedGateVerification', () => {
|
||||
it('runs one pass per decision with the gate pin swapped and the loop edge cut', async () => {
|
||||
const run = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce(approvePassResult)
|
||||
.mockResolvedValueOnce(declinePassResult);
|
||||
|
||||
const { result, analysis } = await runScriptedGateVerification(makeArgs(run));
|
||||
|
||||
expect(run).toHaveBeenCalledTimes(2);
|
||||
type RunOptions = {
|
||||
verificationPinData?: Record<string, unknown[]>;
|
||||
omitConnections?: Array<{ source: string; target: string }>;
|
||||
};
|
||||
const optionsOfCall = (index: number): RunOptions =>
|
||||
(run.mock.calls[index] as unknown[])[2] as RunOptions;
|
||||
const firstOptions = optionsOfCall(0);
|
||||
const secondOptions = optionsOfCall(1);
|
||||
expect(firstOptions.verificationPinData?.Gate).toEqual([{ data: { approved: true } }]);
|
||||
expect(secondOptions.verificationPinData?.Gate).toEqual([{ data: { approved: false } }]);
|
||||
// Base pins for other simulated nodes ride along on every pass.
|
||||
expect(firstOptions.verificationPinData?.Generate).toEqual([{ text: 'draft' }]);
|
||||
expect(firstOptions.omitConnections).toEqual([{ source: 'Revise', target: 'Format' }]);
|
||||
|
||||
expect(analysis.success).toBe(true);
|
||||
expect(analysis.nodesExecuted).toEqual(expect.arrayContaining(['Publish', 'Revise']));
|
||||
expect(analysis.coverageNote).toContain('"approve" ended at "Publish"');
|
||||
expect(analysis.coverageNote).toContain('"decline" ended at "Revise"');
|
||||
expect(analysis.coverageNote).toContain('"Revise" → "Format"');
|
||||
expect(result.executedNodeNames).toEqual(expect.arrayContaining(['Publish', 'Revise']));
|
||||
});
|
||||
|
||||
it('fails the merged analysis when any pass fails', async () => {
|
||||
const run = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce(approvePassResult)
|
||||
.mockResolvedValueOnce({
|
||||
...declinePassResult,
|
||||
status: 'error',
|
||||
error: 'expression error in Revise',
|
||||
});
|
||||
|
||||
const { analysis } = await runScriptedGateVerification(makeArgs(run));
|
||||
|
||||
expect(analysis.success).toBe(false);
|
||||
expect(analysis.errorMessage).toContain('expression error');
|
||||
expect(analysis.remediation).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('runScriptedGateVerification — merge consistency', () => {
|
||||
it('normalizes the merged status when a pass fails on node errors despite engine success', async () => {
|
||||
const run = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce(approvePassResult)
|
||||
.mockResolvedValueOnce({
|
||||
...declinePassResult,
|
||||
status: 'success',
|
||||
nodeErrors: [{ nodeName: 'Revise', message: 'expression error' }],
|
||||
});
|
||||
|
||||
const { result, analysis } = await runScriptedGateVerification(makeArgs(run));
|
||||
|
||||
expect(analysis.success).toBe(false);
|
||||
expect(result.status).toBe('error');
|
||||
});
|
||||
|
||||
it('discloses simulated nodes reached only in an earlier pass', async () => {
|
||||
const run = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce(approvePassResult)
|
||||
.mockResolvedValueOnce(declinePassResult);
|
||||
|
||||
const args = makeArgs(run);
|
||||
args.prepared = {
|
||||
...prepared,
|
||||
simulatedNodes: [...prepared.simulatedNodes, { nodeName: 'Publish', reason: 'no creds' }],
|
||||
};
|
||||
|
||||
const { analysis } = await runScriptedGateVerification(args);
|
||||
|
||||
// Publish is reached only by the approve pass; the decline pass ran last.
|
||||
expect(analysis.simulationNote).toContain('Publish');
|
||||
});
|
||||
});
|
||||
|
|
@ -199,7 +199,7 @@ function classifyVerificationFailure(
|
|||
});
|
||||
}
|
||||
|
||||
function buildSimulationNote(
|
||||
export function buildSimulationNote(
|
||||
reachedSimulatedNodes: Array<{ nodeName: string; reason: string }>,
|
||||
planMissing: boolean,
|
||||
): string | undefined {
|
||||
|
|
|
|||
|
|
@ -1,12 +1,17 @@
|
|||
import type { VerifyBuiltWorkflowOutput, VerifyToolInput } from './types';
|
||||
import { createRemediation } from '../../../workflow-loop/remediation';
|
||||
import type { WorkflowBuildOutcome } from '../../../workflow-loop/workflow-loop-state';
|
||||
import type {
|
||||
WaitGateScript,
|
||||
WorkflowBuildOutcome,
|
||||
} from '../../../workflow-loop/workflow-loop-state';
|
||||
|
||||
export interface PreparedVerificationRun {
|
||||
verificationPinData: Record<string, unknown[]> | undefined;
|
||||
simulatedNodes: Array<{ nodeName: string; reason: string }>;
|
||||
/** Wait-gate nodes pinned with zero items — verification halts at these. */
|
||||
haltedGateNames: string[];
|
||||
/** When set, verify runs one scripted pass per decision instead of halting. */
|
||||
gateScript?: WaitGateScript;
|
||||
}
|
||||
|
||||
function getInvalidFixtureOverrideNodeNames(
|
||||
|
|
@ -61,10 +66,15 @@ function buildVerificationPinData(
|
|||
}
|
||||
}
|
||||
|
||||
const gateScript = (buildOutcome.waitGateScripts ?? []).find((script) =>
|
||||
haltedGateNames.includes(script.nodeName),
|
||||
);
|
||||
|
||||
return {
|
||||
verificationPinData: Object.keys(merged).length > 0 ? merged : undefined,
|
||||
simulatedNodes,
|
||||
haltedGateNames,
|
||||
...(gateScript ? { gateScript } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -83,9 +93,10 @@ export function prepareVerificationRun(
|
|||
if (haltedOverrideNodeNames.length > 0) {
|
||||
const guidance =
|
||||
`Node(s) ${haltedOverrideNodeNames.join(', ')} pause the workflow for a human decision and sit on a loop — ` +
|
||||
'verification always halts there and their output cannot be overridden: a canned response would ' +
|
||||
're-run the loop with the same answer forever. Treat reaching the gate as verified and tell the ' +
|
||||
'user the approval loop needs a manual end-to-end test.';
|
||||
'their output cannot be overridden: a constant canned response would re-run the loop with the same ' +
|
||||
'answer forever. Verification drives such gates with scripted decisions where possible, or halts at ' +
|
||||
'them otherwise. Treat the scripted/halted result as authoritative and tell the user the approval ' +
|
||||
'loop needs a manual end-to-end test for final confirmation.';
|
||||
const remediation = createRemediation({
|
||||
category: 'blocked',
|
||||
shouldEdit: false,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,154 @@
|
|||
/**
|
||||
* Scripted wait-gate verification: one execution per gate decision, each on an
|
||||
* ephemeral workflow copy with the script's loop edge removed so no decision
|
||||
* can re-enter the gate. Pass results merge into a single verification
|
||||
* analysis — success requires every pass to succeed, coverage is the union.
|
||||
*/
|
||||
|
||||
import {
|
||||
analyzeVerificationResult,
|
||||
buildSimulationNote,
|
||||
type VerificationAnalysis,
|
||||
} from './analyze-result';
|
||||
import type { PreparedVerificationRun } from './prepare-run';
|
||||
import type { ExecutionRunResult } from './types';
|
||||
import type { OrchestrationContext } from '../../../types';
|
||||
import type {
|
||||
WaitGateScript,
|
||||
WorkflowBuildOutcome,
|
||||
WorkflowLoopState,
|
||||
} from '../../../workflow-loop/workflow-loop-state';
|
||||
|
||||
type VerificationExecutionService = NonNullable<
|
||||
OrchestrationContext['domainContext']
|
||||
>['executionService'];
|
||||
|
||||
export interface ScriptedGateRunArgs {
|
||||
script: WaitGateScript;
|
||||
prepared: PreparedVerificationRun;
|
||||
executionService: VerificationExecutionService;
|
||||
workflowId: string;
|
||||
inputData?: Record<string, unknown>;
|
||||
timeout?: number;
|
||||
abortSignal?: AbortSignal;
|
||||
buildOutcome: WorkflowBuildOutcome;
|
||||
stateBefore: WorkflowLoopState | undefined;
|
||||
runId: string;
|
||||
}
|
||||
|
||||
interface DecisionPass {
|
||||
label: string;
|
||||
result: ExecutionRunResult;
|
||||
analysis: VerificationAnalysis;
|
||||
}
|
||||
|
||||
export async function runScriptedGateVerification(
|
||||
args: ScriptedGateRunArgs,
|
||||
): Promise<{ result: ExecutionRunResult; analysis: VerificationAnalysis }> {
|
||||
const { script, prepared, executionService, workflowId, buildOutcome, stateBefore, runId } = args;
|
||||
const basePins = prepared.verificationPinData ?? {};
|
||||
|
||||
const passes: DecisionPass[] = [];
|
||||
for (const decision of script.decisions) {
|
||||
const result = await executionService.run(workflowId, args.inputData, {
|
||||
timeout: args.timeout,
|
||||
verificationPinData: { ...basePins, [script.nodeName]: decision.items },
|
||||
omitConnections: [script.cutEdge],
|
||||
abortSignal: args.abortSignal,
|
||||
});
|
||||
const analysis = analyzeVerificationResult({
|
||||
result,
|
||||
buildOutcome,
|
||||
simulatedNodes: prepared.simulatedNodes,
|
||||
stateBefore,
|
||||
runId,
|
||||
});
|
||||
passes.push({ label: decision.label, result, analysis });
|
||||
}
|
||||
|
||||
return { result: mergeResults(passes), analysis: mergeAnalyses(script, prepared, passes) };
|
||||
}
|
||||
|
||||
function mergeResults(passes: DecisionPass[]): ExecutionRunResult {
|
||||
const last = passes[passes.length - 1];
|
||||
const failing = passes.find((pass) => !pass.analysis.success);
|
||||
const data: Record<string, unknown> = {};
|
||||
const executedNodeNames = new Set<string>();
|
||||
for (const pass of passes) {
|
||||
Object.assign(data, pass.result.data ?? {});
|
||||
for (const name of pass.analysis.reachedNames) executedNodeNames.add(name);
|
||||
}
|
||||
|
||||
// A pass can fail on node errors while its engine status is still
|
||||
// 'success' — normalize so the merged status never contradicts `success`.
|
||||
const failingStatus = failing
|
||||
? failing.result.status === 'success'
|
||||
? 'error'
|
||||
: failing.result.status
|
||||
: undefined;
|
||||
|
||||
return {
|
||||
...last.result,
|
||||
status: failingStatus ?? last.result.status,
|
||||
data: Object.keys(data).length > 0 ? data : undefined,
|
||||
executedNodeNames: [...executedNodeNames],
|
||||
error: failing?.result.error ?? failing?.analysis.errorMessage,
|
||||
};
|
||||
}
|
||||
|
||||
function mergeAnalyses(
|
||||
script: WaitGateScript,
|
||||
prepared: PreparedVerificationRun,
|
||||
passes: DecisionPass[],
|
||||
): VerificationAnalysis {
|
||||
const reachedNames = new Set<string>();
|
||||
for (const pass of passes) {
|
||||
for (const name of pass.analysis.reachedNames) reachedNames.add(name);
|
||||
}
|
||||
const failing = passes.find((pass) => !pass.analysis.success);
|
||||
const nodesNotReached = [
|
||||
...new Set(passes.flatMap((pass) => pass.analysis.nodesNotReached)),
|
||||
].filter((name) => !reachedNames.has(name));
|
||||
const nodeErrors = passes.flatMap((pass) =>
|
||||
pass.analysis.nodeErrors.map((nodeError) => ({
|
||||
...nodeError,
|
||||
message: `[decision: ${pass.label}] ${nodeError.message ?? 'node execution failed'}`,
|
||||
})),
|
||||
);
|
||||
|
||||
const endings = passes
|
||||
.map(
|
||||
(pass) =>
|
||||
`"${pass.label}" ended at ${pass.result.lastNodeExecuted ? `"${pass.result.lastNodeExecuted}"` : 'an unknown node'}`,
|
||||
)
|
||||
.join('; ');
|
||||
const coverageNote =
|
||||
`Scripted gate verification: ran ${String(passes.length)} decision pass(es) through ` +
|
||||
`"${script.nodeName}" with loop edge "${script.cutEdge.source}" → "${script.cutEdge.target}" ` +
|
||||
`removed so each pass is loop-safe (${endings}). The gate's send/wait behaviour itself was ` +
|
||||
'simulated — recommend a live end-to-end test for final confirmation.' +
|
||||
(nodesNotReached.length > 0
|
||||
? ` ${String(nodesNotReached.length)} planned node(s) not reached by any pass: ` +
|
||||
`${nodesNotReached.join(', ')}. If a decision should route to them, inspect the ` +
|
||||
'branch wiring before editing.'
|
||||
: '');
|
||||
|
||||
// Rebuild the note from the union — a node simulated only in an earlier
|
||||
// pass must still be disclosed.
|
||||
const reachedSimulatedNodes = prepared.simulatedNodes.filter((node) =>
|
||||
reachedNames.has(node.nodeName),
|
||||
);
|
||||
|
||||
return {
|
||||
success: passes.every((pass) => pass.analysis.success),
|
||||
reachedNames,
|
||||
reachedSimulatedNodes,
|
||||
nodesNotReached,
|
||||
remediation: failing?.analysis.remediation,
|
||||
nodesExecuted: [...reachedNames],
|
||||
simulationNote: buildSimulationNote(reachedSimulatedNodes, false),
|
||||
coverageNote,
|
||||
errorMessage: failing?.analysis.errorMessage,
|
||||
nodeErrors,
|
||||
};
|
||||
}
|
||||
|
|
@ -18,6 +18,7 @@ import {
|
|||
import { prepareVerificationRun } from './verification/prepare-run';
|
||||
import { reconcileStaleCredentialPlan } from './verification/reconcile-plan';
|
||||
import { resolveVerificationTarget } from './verification/resolve-target';
|
||||
import { runScriptedGateVerification } from './verification/scripted-gate-run';
|
||||
import { executionNodeErrorSchema } from '../../workflow-loop/workflow-loop-state';
|
||||
|
||||
const DEFAULT_NODE_PREVIEW_CHARS = 600;
|
||||
|
|
@ -157,24 +158,43 @@ export function createVerifyBuiltWorkflowTool(context: OrchestrationContext) {
|
|||
}
|
||||
const { prepared } = preparedResult;
|
||||
|
||||
const result = await target.domainContext.executionService.run(
|
||||
workflowId,
|
||||
resolvedInput.inputData,
|
||||
{
|
||||
timeout: resolvedInput.timeout,
|
||||
verificationPinData: prepared.verificationPinData,
|
||||
abortSignal: context.abortSignal,
|
||||
},
|
||||
);
|
||||
|
||||
const analysis = analyzeVerificationResult({
|
||||
result,
|
||||
buildOutcome,
|
||||
simulatedNodes: prepared.simulatedNodes,
|
||||
haltedGateNames: prepared.haltedGateNames,
|
||||
stateBefore: target.stateBefore,
|
||||
runId: context.runId,
|
||||
});
|
||||
// A scripted gate replaces the halt with one loop-safe pass per decision;
|
||||
// otherwise run the single standard pass (halted gates pin zero items).
|
||||
const { result, analysis } = prepared.gateScript
|
||||
? await runScriptedGateVerification({
|
||||
script: prepared.gateScript,
|
||||
prepared,
|
||||
executionService: target.domainContext.executionService,
|
||||
workflowId,
|
||||
inputData: resolvedInput.inputData,
|
||||
timeout: resolvedInput.timeout,
|
||||
abortSignal: context.abortSignal,
|
||||
buildOutcome,
|
||||
stateBefore: target.stateBefore,
|
||||
runId: context.runId,
|
||||
})
|
||||
: await (async () => {
|
||||
const runResult = await target.domainContext.executionService.run(
|
||||
workflowId,
|
||||
resolvedInput.inputData,
|
||||
{
|
||||
timeout: resolvedInput.timeout,
|
||||
verificationPinData: prepared.verificationPinData,
|
||||
abortSignal: context.abortSignal,
|
||||
},
|
||||
);
|
||||
return {
|
||||
result: runResult,
|
||||
analysis: analyzeVerificationResult({
|
||||
result: runResult,
|
||||
buildOutcome,
|
||||
simulatedNodes: prepared.simulatedNodes,
|
||||
haltedGateNames: prepared.haltedGateNames,
|
||||
stateBefore: target.stateBefore,
|
||||
runId: context.runId,
|
||||
}),
|
||||
};
|
||||
})();
|
||||
await persistVerificationOutcome({
|
||||
input: resolvedInput,
|
||||
context,
|
||||
|
|
|
|||
|
|
@ -428,3 +428,70 @@ describe('planVerificationSimulation — wait-gate halt verdicts', () => {
|
|||
).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('planVerificationSimulation — wait-gate scripts', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
mockGenerateFixtures.mockResolvedValue({});
|
||||
});
|
||||
|
||||
it('derives a decision script for a halted send-and-wait gate', async () => {
|
||||
mockClassify.mockResolvedValue([
|
||||
{
|
||||
nodeName: 'Email Approval',
|
||||
verdict: 'simulate',
|
||||
reason: 'Credentials are not configured for this node',
|
||||
confidence: 'high',
|
||||
source: 'deterministic',
|
||||
},
|
||||
]);
|
||||
|
||||
const workflow = {
|
||||
name: 'approval loop',
|
||||
nodes: [
|
||||
{
|
||||
id: 'id-0',
|
||||
name: 'Every day',
|
||||
type: 'n8n-nodes-base.scheduleTrigger',
|
||||
typeVersion: 1,
|
||||
position: [0, 0],
|
||||
parameters: {},
|
||||
},
|
||||
{
|
||||
id: 'id-1',
|
||||
name: 'Email Approval',
|
||||
type: 'n8n-nodes-base.gmail',
|
||||
typeVersion: 2,
|
||||
position: [100, 0],
|
||||
parameters: { operation: 'sendAndWait', responseType: 'approval' },
|
||||
},
|
||||
{
|
||||
id: 'id-2',
|
||||
name: 'Revise',
|
||||
type: 'n8n-nodes-base.set',
|
||||
typeVersion: 1,
|
||||
position: [200, 0],
|
||||
parameters: {},
|
||||
},
|
||||
],
|
||||
connections: {
|
||||
'Every day': { main: [[{ node: 'Email Approval', type: 'main', index: 0 }]] },
|
||||
'Email Approval': { main: [[{ node: 'Revise', type: 'main', index: 0 }]] },
|
||||
Revise: { main: [[{ node: 'Email Approval', type: 'main', index: 0 }]] },
|
||||
},
|
||||
} as unknown as WorkflowJSON;
|
||||
|
||||
const { nodeSimulationPlan, waitGateScripts } = await planVerificationSimulation({
|
||||
workflow,
|
||||
workflowId: 'wf-1',
|
||||
});
|
||||
|
||||
expect(nodeSimulationPlan?.find((v) => v.nodeName === 'Email Approval')?.haltBranch).toBe(true);
|
||||
expect(waitGateScripts).toHaveLength(1);
|
||||
expect(waitGateScripts?.[0]).toMatchObject({
|
||||
nodeName: 'Email Approval',
|
||||
cutEdge: { source: 'Revise', target: 'Email Approval' },
|
||||
});
|
||||
expect(waitGateScripts?.[0].decisions.map((d) => d.label)).toEqual(['approve', 'decline']);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -0,0 +1,238 @@
|
|||
import type { WorkflowJSON } from '@n8n/workflow-sdk';
|
||||
|
||||
import type { NodeSimulationVerdict } from '../../../workflow-loop/workflow-loop-state';
|
||||
import { deriveWaitGateScripts, findCycleCutEdge } from '../wait-gate-script';
|
||||
|
||||
const gateVerdict = (nodeName: string): NodeSimulationVerdict => ({
|
||||
nodeName,
|
||||
verdict: 'simulate',
|
||||
reason: 'Send-and-wait gate on a loop',
|
||||
confidence: 'high',
|
||||
source: 'deterministic',
|
||||
haltBranch: true,
|
||||
});
|
||||
|
||||
const main = (...targets: string[]) => ({
|
||||
main: [targets.map((target) => ({ node: target, type: 'main', index: 0 }))],
|
||||
});
|
||||
|
||||
/** trigger → Generate → Format → Gate → Router → (Publish | Revise → Format). */
|
||||
function approvalLoopWorkflow(gateParameters: Record<string, unknown>): WorkflowJSON {
|
||||
return {
|
||||
name: 'approval loop',
|
||||
nodes: [
|
||||
{ name: 'Every day', type: 'n8n-nodes-base.scheduleTrigger', parameters: {} },
|
||||
{ name: 'Generate', type: '@n8n/n8n-nodes-langchain.chainLlm', parameters: {} },
|
||||
{ name: 'Format', type: 'n8n-nodes-base.set', parameters: {} },
|
||||
{ name: 'Gate', type: 'n8n-nodes-base.gmail', parameters: gateParameters },
|
||||
{ name: 'Router', type: 'n8n-nodes-base.if', parameters: {} },
|
||||
{ name: 'Publish', type: 'n8n-nodes-base.linkedIn', parameters: {} },
|
||||
{ name: 'Revise', type: '@n8n/n8n-nodes-langchain.chainLlm', parameters: {} },
|
||||
].map((node, index) => ({
|
||||
id: `id-${index}`,
|
||||
typeVersion: 1,
|
||||
position: [index * 100, 0],
|
||||
...node,
|
||||
})),
|
||||
connections: {
|
||||
'Every day': main('Generate'),
|
||||
Generate: main('Format'),
|
||||
Format: main('Gate'),
|
||||
Gate: main('Router'),
|
||||
Router: {
|
||||
main: [
|
||||
[{ node: 'Publish', type: 'main', index: 0 }],
|
||||
[{ node: 'Revise', type: 'main', index: 0 }],
|
||||
],
|
||||
},
|
||||
Revise: main('Format'),
|
||||
},
|
||||
} as unknown as WorkflowJSON;
|
||||
}
|
||||
|
||||
describe('findCycleCutEdge', () => {
|
||||
it('picks the loop-back edge, never one that costs downstream coverage', () => {
|
||||
const workflow = approvalLoopWorkflow({ operation: 'sendAndWait' });
|
||||
|
||||
// Gate→Router also breaks the cycle but would disconnect the router,
|
||||
// publish and revise branches — only Revise→Format preserves coverage.
|
||||
expect(findCycleCutEdge(workflow, 'Gate')).toEqual({ source: 'Revise', target: 'Format' });
|
||||
});
|
||||
|
||||
it('returns undefined when the gate is not on a cycle or no trigger exists', () => {
|
||||
const acyclic = approvalLoopWorkflow({ operation: 'sendAndWait' });
|
||||
(acyclic.connections as Record<string, unknown>).Revise = { main: [[]] };
|
||||
expect(findCycleCutEdge(acyclic, 'Gate')).toBeUndefined();
|
||||
|
||||
const triggerless = approvalLoopWorkflow({ operation: 'sendAndWait' });
|
||||
triggerless.nodes = triggerless.nodes.filter((node) => node.name !== 'Every day');
|
||||
expect(findCycleCutEdge(triggerless, 'Gate')).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('deriveWaitGateScripts', () => {
|
||||
it('derives approve/decline passes for the approval response type', () => {
|
||||
const workflow = approvalLoopWorkflow({ operation: 'sendAndWait', responseType: 'approval' });
|
||||
|
||||
const scripts = deriveWaitGateScripts(workflow, [gateVerdict('Gate')]);
|
||||
|
||||
expect(scripts).toHaveLength(1);
|
||||
expect(scripts[0].nodeName).toBe('Gate');
|
||||
expect(scripts[0].cutEdge).toEqual({ source: 'Revise', target: 'Format' });
|
||||
expect(scripts[0].decisions.map((d) => d.label)).toEqual(['approve', 'decline']);
|
||||
expect(scripts[0].decisions[0].items[0]).toMatchObject({ data: { approved: true } });
|
||||
expect(scripts[0].decisions[1].items[0]).toMatchObject({ data: { approved: false } });
|
||||
});
|
||||
|
||||
it('defaults to the approval shape when responseType is omitted', () => {
|
||||
const workflow = approvalLoopWorkflow({ operation: 'sendAndWait' });
|
||||
|
||||
const scripts = deriveWaitGateScripts(workflow, [gateVerdict('Gate')]);
|
||||
|
||||
expect(scripts[0]?.decisions[0].items[0]).toMatchObject({ data: { approved: true } });
|
||||
});
|
||||
|
||||
it('enumerates the first single-select dropdown for custom forms', () => {
|
||||
const workflow = approvalLoopWorkflow({
|
||||
operation: 'sendAndWait',
|
||||
responseType: 'customForm',
|
||||
formFields: {
|
||||
values: [
|
||||
{
|
||||
fieldLabel: 'Decision',
|
||||
fieldType: 'dropdown',
|
||||
fieldOptions: { values: [{ option: 'Approve' }, { option: 'Request changes' }] },
|
||||
},
|
||||
{ fieldLabel: 'Requested changes', fieldType: 'textarea' },
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
const scripts = deriveWaitGateScripts(workflow, [gateVerdict('Gate')]);
|
||||
|
||||
expect(scripts[0]?.decisions.map((d) => d.label)).toEqual(['Approve', 'Request changes']);
|
||||
expect(scripts[0]?.decisions[1].items[0]).toMatchObject({
|
||||
data: { Decision: 'Request changes' },
|
||||
});
|
||||
});
|
||||
|
||||
it('derives canned approve/revise texts for free-text gates', () => {
|
||||
const workflow = approvalLoopWorkflow({ operation: 'sendAndWait', responseType: 'freeText' });
|
||||
|
||||
const scripts = deriveWaitGateScripts(workflow, [gateVerdict('Gate')]);
|
||||
|
||||
expect(scripts[0]?.decisions).toHaveLength(2);
|
||||
const firstItem = scripts[0]?.decisions[0].items[0] as { data?: { text?: string } };
|
||||
expect(firstItem.data?.text).toContain('Approved');
|
||||
});
|
||||
|
||||
it('returns [] for underivable shapes', () => {
|
||||
const noDropdown = approvalLoopWorkflow({
|
||||
operation: 'sendAndWait',
|
||||
responseType: 'customForm',
|
||||
formFields: { values: [{ fieldLabel: 'Decision', fieldType: 'text' }] },
|
||||
});
|
||||
expect(deriveWaitGateScripts(noDropdown, [gateVerdict('Gate')])).toEqual([]);
|
||||
|
||||
// Multiple halted gates — v1 scripts exactly one.
|
||||
const workflow = approvalLoopWorkflow({ operation: 'sendAndWait' });
|
||||
expect(deriveWaitGateScripts(workflow, [gateVerdict('Gate'), gateVerdict('Other')])).toEqual(
|
||||
[],
|
||||
);
|
||||
|
||||
// Not a send-and-wait node (Wait node keeps the halt in v1).
|
||||
const waitNode = approvalLoopWorkflow({});
|
||||
waitNode.nodes = waitNode.nodes.map((node) =>
|
||||
node.name === 'Gate' ? { ...node, type: 'n8n-nodes-base.wait', parameters: {} } : node,
|
||||
);
|
||||
expect(deriveWaitGateScripts(waitNode, [gateVerdict('Gate')])).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('deriveWaitGateScripts — field identifiers and JSON forms', () => {
|
||||
const formFields = {
|
||||
values: [
|
||||
{
|
||||
fieldLabel: 'Decision',
|
||||
fieldName: 'decision_key',
|
||||
fieldType: 'dropdown',
|
||||
fieldOptions: { values: [{ option: 'Approve' }, { option: 'Reject' }] },
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
it('keys fixtures by fieldName on v2.4+ gates, mirroring getFieldIdentifier', () => {
|
||||
const workflow = approvalLoopWorkflow({
|
||||
operation: 'sendAndWait',
|
||||
responseType: 'customForm',
|
||||
formFields,
|
||||
});
|
||||
workflow.nodes = workflow.nodes.map((node) =>
|
||||
node.name === 'Gate' ? { ...node, typeVersion: 2.4 } : node,
|
||||
);
|
||||
|
||||
const scripts = deriveWaitGateScripts(workflow, [gateVerdict('Gate')]);
|
||||
|
||||
expect(scripts[0]?.decisions[0].items[0]).toMatchObject({
|
||||
data: { decision_key: 'Approve' },
|
||||
});
|
||||
});
|
||||
|
||||
it('keys fixtures by label on pre-2.4 gates even when fieldName is set', () => {
|
||||
const workflow = approvalLoopWorkflow({
|
||||
operation: 'sendAndWait',
|
||||
responseType: 'customForm',
|
||||
formFields,
|
||||
});
|
||||
|
||||
const scripts = deriveWaitGateScripts(workflow, [gateVerdict('Gate')]);
|
||||
|
||||
expect(scripts[0]?.decisions[0].items[0]).toMatchObject({ data: { Decision: 'Approve' } });
|
||||
});
|
||||
|
||||
it('parses static JSON-defined forms and bails on expressions', () => {
|
||||
const staticJson = approvalLoopWorkflow({
|
||||
operation: 'sendAndWait',
|
||||
responseType: 'customForm',
|
||||
defineForm: 'json',
|
||||
jsonOutput: JSON.stringify([
|
||||
{
|
||||
fieldLabel: 'Decision',
|
||||
fieldType: 'dropdown',
|
||||
fieldOptions: { values: [{ option: 'Ship it' }, { option: 'Revise' }] },
|
||||
},
|
||||
]),
|
||||
});
|
||||
expect(
|
||||
deriveWaitGateScripts(staticJson, [gateVerdict('Gate')])[0]?.decisions.map((d) => d.label),
|
||||
).toEqual(['Ship it', 'Revise']);
|
||||
|
||||
const dynamicJson = approvalLoopWorkflow({
|
||||
operation: 'sendAndWait',
|
||||
responseType: 'customForm',
|
||||
defineForm: 'json',
|
||||
jsonOutput: '{{ $json.fields }}',
|
||||
});
|
||||
expect(deriveWaitGateScripts(dynamicJson, [gateVerdict('Gate')])).toEqual([]);
|
||||
});
|
||||
|
||||
it('accepts a radio field as the decision driver', () => {
|
||||
const workflow = approvalLoopWorkflow({
|
||||
operation: 'sendAndWait',
|
||||
responseType: 'customForm',
|
||||
formFields: {
|
||||
values: [
|
||||
{
|
||||
fieldLabel: 'Verdict',
|
||||
fieldType: 'radio',
|
||||
fieldOptions: { values: [{ option: 'Yes' }, { option: 'No' }] },
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
expect(
|
||||
deriveWaitGateScripts(workflow, [gateVerdict('Gate')])[0]?.decisions.map((d) => d.label),
|
||||
).toEqual(['Yes', 'No']);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,47 @@
|
|||
import type { InstanceAiContext } from '../../../types';
|
||||
import { trackWaitGateVerificationPlan } from '../workflow-build-telemetry';
|
||||
|
||||
function makeContext(trackTelemetry = vi.fn()): InstanceAiContext {
|
||||
return {
|
||||
trackTelemetry,
|
||||
threadId: 'thread-1',
|
||||
runId: 'run-1',
|
||||
} as unknown as InstanceAiContext;
|
||||
}
|
||||
|
||||
describe('trackWaitGateVerificationPlan', () => {
|
||||
it('emits gate counts and the multi-gate flag', () => {
|
||||
const trackTelemetry = vi.fn();
|
||||
|
||||
trackWaitGateVerificationPlan(makeContext(trackTelemetry), {
|
||||
haltedGateCount: 2,
|
||||
scriptedGateCount: 0,
|
||||
savedWorkflowId: 'wf-1',
|
||||
});
|
||||
|
||||
expect(trackTelemetry).toHaveBeenCalledWith(
|
||||
'instance_ai_wait_gate_verification_plan',
|
||||
expect.objectContaining({
|
||||
halted_gate_count: 2,
|
||||
scripted_gate_count: 0,
|
||||
multi_gate: true,
|
||||
scripted: false,
|
||||
workflow_id: 'wf-1',
|
||||
thread_id: 'thread-1',
|
||||
run_id: 'run-1',
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('stays silent for builds without halted gates', () => {
|
||||
const trackTelemetry = vi.fn();
|
||||
|
||||
trackWaitGateVerificationPlan(makeContext(trackTelemetry), {
|
||||
haltedGateCount: 0,
|
||||
scriptedGateCount: 0,
|
||||
savedWorkflowId: 'wf-1',
|
||||
});
|
||||
|
||||
expect(trackTelemetry).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
|
@ -37,7 +37,10 @@ import {
|
|||
reportWorkflowBuildOutcome,
|
||||
} from './workflow-build-reporting';
|
||||
import { withDeterministicRouting } from './workflow-build-routing';
|
||||
import { trackWorkflowSourceBuild } from './workflow-build-telemetry';
|
||||
import {
|
||||
trackWaitGateVerificationPlan,
|
||||
trackWorkflowSourceBuild,
|
||||
} from './workflow-build-telemetry';
|
||||
import {
|
||||
bindSourceFileToExistingWorkflow,
|
||||
getWorkflowSourceFileBinding,
|
||||
|
|
@ -757,14 +760,21 @@ export function createBuildWorkflowTool(context: InstanceAiContext) {
|
|||
) => {
|
||||
const setupRequests = await analyzeWorkflow(context, saved.id);
|
||||
const workflowNeedsSetup = setupRequests.some((request) => request.needsAction);
|
||||
const { nodeSimulationPlan, simulationFixtures } = await planVerificationSimulation({
|
||||
workflow: json,
|
||||
mockedNodeNames: mockResult.mockedNodeNames,
|
||||
declaredOutputFixtures: compiled.declaredOutputFixtures,
|
||||
workflowId: saved.id,
|
||||
outputSchemaLookup: context.outputSchemaLookup,
|
||||
fallbackModelConfig: context.modelId,
|
||||
logger: context.logger,
|
||||
const { nodeSimulationPlan, simulationFixtures, waitGateScripts } =
|
||||
await planVerificationSimulation({
|
||||
workflow: json,
|
||||
mockedNodeNames: mockResult.mockedNodeNames,
|
||||
declaredOutputFixtures: compiled.declaredOutputFixtures,
|
||||
workflowId: saved.id,
|
||||
outputSchemaLookup: context.outputSchemaLookup,
|
||||
fallbackModelConfig: context.modelId,
|
||||
logger: context.logger,
|
||||
});
|
||||
trackWaitGateVerificationPlan(context, {
|
||||
haltedGateCount: (nodeSimulationPlan ?? []).filter((verdict) => verdict.haltBranch)
|
||||
.length,
|
||||
scriptedGateCount: waitGateScripts?.length ?? 0,
|
||||
savedWorkflowId: saved.id,
|
||||
});
|
||||
const runId = buildContext?.runId ?? context.runId;
|
||||
const workflowName = json.name || 'workflow';
|
||||
|
|
@ -831,6 +841,7 @@ export function createBuildWorkflowTool(context: InstanceAiContext) {
|
|||
workflowNeedsSetup,
|
||||
nodeSimulationPlan,
|
||||
simulationFixtures,
|
||||
waitGateScripts,
|
||||
supportingWorkflowIds:
|
||||
referencedWorkflowIds.length > 0 ? referencedWorkflowIds : undefined,
|
||||
hasUnresolvedPlaceholders: hasPlaceholders || undefined,
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ import {
|
|||
generateSimulationFixtures,
|
||||
type SimulationFixtures,
|
||||
} from './generate-simulation-fixtures.service';
|
||||
import { deriveWaitGateScripts } from './wait-gate-script';
|
||||
import {
|
||||
isMockableTriggerNodeType,
|
||||
isTriggerNodeType,
|
||||
|
|
@ -28,7 +29,10 @@ import {
|
|||
} from './workflow-json-utils';
|
||||
import type { Logger } from '../../logger';
|
||||
import type { ModelConfig } from '../../types';
|
||||
import type { NodeSimulationVerdict } from '../../workflow-loop/workflow-loop-state';
|
||||
import type {
|
||||
NodeSimulationVerdict,
|
||||
WaitGateScript,
|
||||
} from '../../workflow-loop/workflow-loop-state';
|
||||
|
||||
export interface PlanVerificationSimulationInput {
|
||||
workflow: WorkflowJSON;
|
||||
|
|
@ -50,6 +54,8 @@ export interface PlanVerificationSimulationInput {
|
|||
|
||||
export interface VerificationSimulationPlan {
|
||||
nodeSimulationPlan?: NodeSimulationVerdict[];
|
||||
/** Scripted decisions for wait gates on loops; absent when none derivable. */
|
||||
waitGateScripts?: WaitGateScript[];
|
||||
simulationFixtures?: SimulationFixtures;
|
||||
}
|
||||
|
||||
|
|
@ -299,6 +305,7 @@ export async function planVerificationSimulation({
|
|||
}: PlanVerificationSimulationInput): Promise<VerificationSimulationPlan> {
|
||||
let nodeSimulationPlan: NodeSimulationVerdict[] | undefined;
|
||||
let simulationFixtures: SimulationFixtures | undefined;
|
||||
let waitGateScripts: WaitGateScript[] | undefined;
|
||||
const declaredFixtures = nonEmptyDeclaredFixtures(declaredOutputFixtures);
|
||||
try {
|
||||
nodeSimulationPlan = await classifyNodesForSimulation({
|
||||
|
|
@ -314,6 +321,8 @@ export async function planVerificationSimulation({
|
|||
new Set(mockedNodeNames ?? []),
|
||||
);
|
||||
nodeSimulationPlan = withWaitGateHaltVerdicts(nodeSimulationPlan, workflow);
|
||||
const derivedScripts = deriveWaitGateScripts(workflow, nodeSimulationPlan);
|
||||
waitGateScripts = derivedScripts.length > 0 ? derivedScripts : undefined;
|
||||
if (nodeSimulationPlan.length > 0) {
|
||||
const planNeedingGeneratedFixtures = nodeSimulationPlan.filter(
|
||||
(verdict) =>
|
||||
|
|
@ -365,5 +374,5 @@ export async function planVerificationSimulation({
|
|||
simulationFixtures = declaredFixtures;
|
||||
}
|
||||
}
|
||||
return { nodeSimulationPlan, simulationFixtures };
|
||||
return { nodeSimulationPlan, simulationFixtures, waitGateScripts };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,277 @@
|
|||
/**
|
||||
* Wait-gate decision scripts for verification.
|
||||
*
|
||||
* A send-and-wait gate on a loop cannot be verified with a single pinned
|
||||
* response (the constant decision re-runs the loop forever). Instead, verify
|
||||
* runs one pass per scripted decision — approve-ish and reject-ish — on an
|
||||
* ephemeral workflow copy with one loop edge removed, so every pass is
|
||||
* acyclic regardless of how the decision routes. This module derives that
|
||||
* script (cut edge + decision fixtures) from the workflow at build time.
|
||||
*
|
||||
* V1 limits: send-and-wait operations only (Wait/Form keep the halt), exactly
|
||||
* one scripted gate per workflow, and custom forms need a dropdown field to
|
||||
* enumerate decisions. Anything underivable returns [] and the gate falls
|
||||
* back to its `haltBranch` behaviour.
|
||||
*/
|
||||
|
||||
import { isRecord } from '@n8n/utils/is-record';
|
||||
import type { WorkflowJSON } from '@n8n/workflow-sdk';
|
||||
|
||||
import { isTriggerNodeType } from './workflow-json-utils';
|
||||
import type {
|
||||
NodeSimulationVerdict,
|
||||
WaitGateScript,
|
||||
} from '../../workflow-loop/workflow-loop-state';
|
||||
|
||||
type WorkflowNode = WorkflowJSON['nodes'][number];
|
||||
type Edge = { source: string; target: string };
|
||||
|
||||
const SEND_AND_WAIT_OPERATION = 'sendAndWait';
|
||||
|
||||
const APPROVE_TEXT = 'Approved — looks good, go ahead.';
|
||||
const REVISE_TEXT = 'Please revise: tighten the copy and add one concrete example.';
|
||||
|
||||
function isSendAndWaitNode(node: WorkflowNode): boolean {
|
||||
const parameters = isRecord(node.parameters) ? node.parameters : {};
|
||||
return parameters.operation === SEND_AND_WAIT_OPERATION;
|
||||
}
|
||||
|
||||
/** All `main` edges of the workflow, in stable declaration order. */
|
||||
function mainEdges(json: WorkflowJSON): Edge[] {
|
||||
const edges: Edge[] = [];
|
||||
for (const [source, byType] of Object.entries(json.connections ?? {})) {
|
||||
if (!isRecord(byType) || !Array.isArray(byType.main)) continue;
|
||||
for (const group of byType.main) {
|
||||
if (!Array.isArray(group)) continue;
|
||||
for (const connection of group) {
|
||||
if (isRecord(connection) && typeof connection.node === 'string') {
|
||||
edges.push({ source, target: connection.node });
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return edges;
|
||||
}
|
||||
|
||||
function buildAdjacency(edges: Edge[], skip?: Edge): Map<string, string[]> {
|
||||
const adjacency = new Map<string, string[]>();
|
||||
for (const edge of edges) {
|
||||
if (skip && edge.source === skip.source && edge.target === skip.target) continue;
|
||||
const targets = adjacency.get(edge.source);
|
||||
if (targets) targets.push(edge.target);
|
||||
else adjacency.set(edge.source, [edge.target]);
|
||||
}
|
||||
return adjacency;
|
||||
}
|
||||
|
||||
function reaches(edges: Edge[], from: string, to: string, skip?: Edge): boolean {
|
||||
const adjacency = buildAdjacency(edges, skip);
|
||||
// Seed with successors so `from === to` requires traversing at least one edge.
|
||||
const queue = [...(adjacency.get(from) ?? [])];
|
||||
const visited = new Set<string>();
|
||||
while (queue.length > 0) {
|
||||
const current = queue.pop();
|
||||
if (current === undefined) break;
|
||||
if (current === to) return true;
|
||||
if (visited.has(current)) continue;
|
||||
visited.add(current);
|
||||
queue.push(...(adjacency.get(current) ?? []));
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
function reachableFrom(edges: Edge[], sources: string[], skip?: Edge): Set<string> {
|
||||
const adjacency = buildAdjacency(edges, skip);
|
||||
const visited = new Set<string>(sources);
|
||||
const queue = [...sources];
|
||||
while (queue.length > 0) {
|
||||
const current = queue.pop();
|
||||
if (current === undefined) break;
|
||||
for (const next of adjacency.get(current) ?? []) {
|
||||
if (visited.has(next)) continue;
|
||||
visited.add(next);
|
||||
queue.push(next);
|
||||
}
|
||||
}
|
||||
return visited;
|
||||
}
|
||||
|
||||
/**
|
||||
* Find a `main` edge whose removal breaks every cycle through `gateName`
|
||||
* without shrinking the set of nodes reachable from the triggers — i.e. a
|
||||
* true loop-back edge whose cut costs no coverage. Returns undefined when no
|
||||
* such edge exists (e.g. the loop-back edge is also the only inbound path).
|
||||
*/
|
||||
export function findCycleCutEdge(json: WorkflowJSON, gateName: string): Edge | undefined {
|
||||
const edges = mainEdges(json);
|
||||
const triggers = (json.nodes ?? [])
|
||||
.filter(
|
||||
(node): node is WorkflowNode & { name: string } =>
|
||||
Boolean(node.name) && !node.disabled && isTriggerNodeType(node.type),
|
||||
)
|
||||
.map((node) => node.name);
|
||||
if (triggers.length === 0) return undefined;
|
||||
if (!reaches(edges, gateName, gateName)) return undefined;
|
||||
|
||||
const baseline = reachableFrom(edges, triggers);
|
||||
for (const edge of edges) {
|
||||
if (reaches(edges, gateName, gateName, edge)) continue;
|
||||
const afterCut = reachableFrom(edges, triggers, edge);
|
||||
const preservesCoverage = [...baseline].every((node) => afterCut.has(node));
|
||||
if (preservesCoverage) return edge;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
type Decision = WaitGateScript['decisions'][number];
|
||||
|
||||
interface FormField {
|
||||
fieldLabel?: string;
|
||||
fieldName?: string;
|
||||
fieldType?: string;
|
||||
options: string[];
|
||||
multiselect: boolean;
|
||||
}
|
||||
|
||||
function toFormField(value: unknown): FormField | undefined {
|
||||
if (!isRecord(value)) return undefined;
|
||||
const fieldLabel = typeof value.fieldLabel === 'string' ? value.fieldLabel : undefined;
|
||||
const fieldName = typeof value.fieldName === 'string' ? value.fieldName : undefined;
|
||||
if (!fieldLabel && !fieldName) return undefined;
|
||||
const fieldOptions = isRecord(value.fieldOptions) ? value.fieldOptions : {};
|
||||
const optionValues = Array.isArray(fieldOptions.values) ? fieldOptions.values : [];
|
||||
const options = optionValues
|
||||
.map((option) => (isRecord(option) ? option.option : undefined))
|
||||
.filter((option): option is string => typeof option === 'string');
|
||||
return {
|
||||
fieldLabel,
|
||||
fieldName,
|
||||
fieldType: typeof value.fieldType === 'string' ? value.fieldType : undefined,
|
||||
options,
|
||||
multiselect: value.multiselectDropdown === true || value.multiselect === true,
|
||||
};
|
||||
}
|
||||
|
||||
/** UI-defined and static-JSON forms parse; JSON with expressions is dynamic → undefined. */
|
||||
function parseFormFields(parameters: Record<string, unknown>): FormField[] | undefined {
|
||||
let values: unknown[];
|
||||
if (parameters.defineForm === 'json') {
|
||||
const jsonOutput = parameters.jsonOutput;
|
||||
if (typeof jsonOutput !== 'string' || jsonOutput.includes('{{')) return undefined;
|
||||
try {
|
||||
const parsed: unknown = JSON.parse(jsonOutput);
|
||||
if (!Array.isArray(parsed)) return undefined;
|
||||
values = parsed;
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
} else {
|
||||
const formFields = isRecord(parameters.formFields) ? parameters.formFields : {};
|
||||
values = Array.isArray(formFields.values) ? formFields.values : [];
|
||||
}
|
||||
return values.map(toFormField).filter((field): field is FormField => field !== undefined);
|
||||
}
|
||||
|
||||
/** Mirrors nodes-base `getFieldIdentifier`: v2.4+ prefers `fieldName`, else the label. */
|
||||
function fieldKey(field: FormField, typeVersion: number): string | undefined {
|
||||
if (typeVersion >= 2.4 && field.fieldName) return field.fieldName;
|
||||
return field.fieldLabel ?? field.fieldName;
|
||||
}
|
||||
|
||||
function cannedFieldValue(field: FormField, passLabel: string): unknown {
|
||||
switch (field.fieldType) {
|
||||
case 'dropdown':
|
||||
return field.multiselect ? field.options.slice(0, 1) : (field.options[0] ?? '');
|
||||
case 'radio':
|
||||
return field.options[0] ?? '';
|
||||
case 'number':
|
||||
return 1;
|
||||
case 'date':
|
||||
return '2026-01-01';
|
||||
case 'email':
|
||||
return 'reviewer@example.com';
|
||||
case 'checkbox':
|
||||
return false;
|
||||
default:
|
||||
return passLabel;
|
||||
}
|
||||
}
|
||||
|
||||
/** Response fixtures per decision, mirroring the shared send-and-wait output shapes. */
|
||||
function deriveDecisions(node: WorkflowNode): Decision[] | undefined {
|
||||
const parameters = isRecord(node.parameters) ? node.parameters : {};
|
||||
const responseType =
|
||||
typeof parameters.responseType === 'string' ? parameters.responseType : 'approval';
|
||||
const respondedAt = new Date().toISOString();
|
||||
|
||||
if (responseType === 'approval') {
|
||||
return [
|
||||
{ label: 'approve', items: [{ data: { approved: true, respondedAt } }] },
|
||||
{ label: 'decline', items: [{ data: { approved: false, respondedAt } }] },
|
||||
];
|
||||
}
|
||||
|
||||
if (responseType === 'freeText') {
|
||||
return [
|
||||
{ label: 'approve', items: [{ data: { text: APPROVE_TEXT, respondedAt } }] },
|
||||
{ label: 'request changes', items: [{ data: { text: REVISE_TEXT, respondedAt } }] },
|
||||
];
|
||||
}
|
||||
|
||||
if (responseType === 'customForm') {
|
||||
const fields = parseFormFields(parameters);
|
||||
if (!fields || fields.length === 0) return undefined;
|
||||
const typeVersion = typeof node.typeVersion === 'number' ? node.typeVersion : 0;
|
||||
// The first single-select dropdown/radio with 2+ options drives the
|
||||
// decision; without one the routing condition is free text we cannot
|
||||
// enumerate.
|
||||
const driver = fields.find(
|
||||
(field) =>
|
||||
(field.fieldType === 'dropdown' || field.fieldType === 'radio') &&
|
||||
!field.multiselect &&
|
||||
field.options.length >= 2 &&
|
||||
fieldKey(field, typeVersion) !== undefined,
|
||||
);
|
||||
if (!driver) return undefined;
|
||||
|
||||
const buildPass = (driverValue: string): Decision => {
|
||||
const data: Record<string, unknown> = {};
|
||||
for (const field of fields) {
|
||||
if (field.fieldType === 'html') continue;
|
||||
const key = fieldKey(field, typeVersion);
|
||||
if (!key) continue;
|
||||
data[key] = field === driver ? driverValue : cannedFieldValue(field, driverValue);
|
||||
}
|
||||
data.respondedAt = respondedAt;
|
||||
return { label: driverValue, items: [{ data }] };
|
||||
};
|
||||
return [buildPass(driver.options[0]), buildPass(driver.options[1])];
|
||||
}
|
||||
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Derive scripted-decision verification for the workflow's halted wait gates.
|
||||
* V1 scripts exactly one send-and-wait gate; every underivable shape returns
|
||||
* [] so the gate keeps its halt fallback.
|
||||
*/
|
||||
export function deriveWaitGateScripts(
|
||||
workflow: WorkflowJSON,
|
||||
plan: NodeSimulationVerdict[],
|
||||
): WaitGateScript[] {
|
||||
const halted = plan.filter((verdict) => verdict.verdict === 'simulate' && verdict.haltBranch);
|
||||
if (halted.length !== 1) return [];
|
||||
|
||||
const gateName = halted[0].nodeName;
|
||||
const node = (workflow.nodes ?? []).find((candidate) => candidate.name === gateName);
|
||||
if (!node || !isSendAndWaitNode(node)) return [];
|
||||
|
||||
const cutEdge = findCycleCutEdge(workflow, gateName);
|
||||
if (!cutEdge) return [];
|
||||
|
||||
const decisions = deriveDecisions(node);
|
||||
if (!decisions) return [];
|
||||
|
||||
return [{ nodeName: gateName, cutEdge, decisions }];
|
||||
}
|
||||
|
|
@ -57,3 +57,29 @@ export function trackWorkflowSourceBuild(
|
|||
: {}),
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Emitted once per successful build whose plan halts wait gates, so we learn
|
||||
* how often scripted verification applies and whether multi-gate loop shapes
|
||||
* (currently always halt-fallback) occur in the wild.
|
||||
*/
|
||||
export function trackWaitGateVerificationPlan(
|
||||
context: InstanceAiContext,
|
||||
input: {
|
||||
haltedGateCount: number;
|
||||
scriptedGateCount: number;
|
||||
savedWorkflowId: string;
|
||||
},
|
||||
): void {
|
||||
if (input.haltedGateCount === 0) return;
|
||||
const buildContext = context.workflowBuildContext;
|
||||
context.trackTelemetry?.('instance_ai_wait_gate_verification_plan', {
|
||||
halted_gate_count: input.haltedGateCount,
|
||||
scripted_gate_count: input.scriptedGateCount,
|
||||
multi_gate: input.haltedGateCount >= 2,
|
||||
scripted: input.scriptedGateCount > 0,
|
||||
workflow_id: input.savedWorkflowId,
|
||||
thread_id: context.threadId ?? buildContext?.threadId ?? 'unknown',
|
||||
run_id: buildContext?.runId ?? context.runId ?? 'unknown',
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -387,6 +387,12 @@ export interface InstanceAiExecutionService {
|
|||
verificationPinData?: Record<string, unknown[]>;
|
||||
/** When set, execute this specific trigger node instead of auto-detecting. */
|
||||
triggerNodeName?: string;
|
||||
/**
|
||||
* Connections removed from this run's ephemeral workflow copy (the saved
|
||||
* workflow is untouched). Used to sever a loop edge so scripted wait-gate
|
||||
* verification passes are acyclic.
|
||||
*/
|
||||
omitConnections?: Array<{ source: string; target: string }>;
|
||||
abortSignal?: AbortSignal;
|
||||
},
|
||||
): Promise<ExecutionResult>;
|
||||
|
|
|
|||
|
|
@ -252,6 +252,25 @@ export const nodeSimulationVerdictSchema = z.object({
|
|||
|
||||
export type NodeSimulationVerdict = z.infer<typeof nodeSimulationVerdictSchema>;
|
||||
|
||||
/**
|
||||
* Scripted verification for a wait gate on a loop: verify runs one pass per
|
||||
* decision with `cutEdge` removed from the run's connections, so every pass
|
||||
* is acyclic no matter which way the decision routes. The gate keeps its
|
||||
* `haltBranch` verdict as the fallback for older readers of this outcome.
|
||||
*/
|
||||
export const waitGateScriptSchema = z.object({
|
||||
nodeName: z.string(),
|
||||
/** Loop edge (main connection) removed for the scripted passes. */
|
||||
cutEdge: z.object({ source: z.string(), target: z.string() }),
|
||||
/** Response fixtures, one verification pass per entry. */
|
||||
decisions: z
|
||||
.array(z.object({ label: z.string(), items: z.array(z.record(z.unknown())) }))
|
||||
.min(1)
|
||||
.max(3),
|
||||
});
|
||||
|
||||
export type WaitGateScript = z.infer<typeof waitGateScriptSchema>;
|
||||
|
||||
export const workflowBuildOutcomeSchema = z.object({
|
||||
workItemId: z.string(),
|
||||
runId: z.string().optional(),
|
||||
|
|
@ -299,6 +318,8 @@ export const workflowBuildOutcomeSchema = z.object({
|
|||
* this build, never persisted to the workflow.
|
||||
*/
|
||||
nodeSimulationPlan: z.array(nodeSimulationVerdictSchema).optional(),
|
||||
/** Scripted decisions for wait gates on loops — see `waitGateScriptSchema`. */
|
||||
waitGateScripts: z.array(waitGateScriptSchema).optional(),
|
||||
/**
|
||||
* LLM-generated mock output for `simulate`-verdict nodes, keyed by node
|
||||
* name. Items are plain objects (no `{json}` envelope) — the shape
|
||||
|
|
|
|||
|
|
@ -2998,6 +2998,32 @@ describe('createExecutionAdapter run()', () => {
|
|||
expect(secondRun.workflowData.settings?.executionTimeout).toBe(600);
|
||||
});
|
||||
|
||||
it('omits the listed connections on the ephemeral run copy only', async () => {
|
||||
const workflow = {
|
||||
id: 'wf-1',
|
||||
nodes: [],
|
||||
connections: {
|
||||
Revise: { main: [[{ node: 'Format', type: 'main', index: 0 }]] },
|
||||
Format: { main: [[{ node: 'Gate', type: 'main', index: 0 }]] },
|
||||
},
|
||||
};
|
||||
const { adapter, mockWorkflowRunner } = createRunAdapterForTests(workflow);
|
||||
|
||||
await adapter.run('wf-1', undefined, {
|
||||
omitConnections: [{ source: 'Revise', target: 'Format' }],
|
||||
});
|
||||
|
||||
const runData = mockWorkflowRunner.run.mock.calls[0][0];
|
||||
expect(runData.workflowData.connections.Revise.main).toEqual([[]]);
|
||||
expect(runData.workflowData.connections.Format.main).toEqual([
|
||||
[{ node: 'Gate', type: 'main', index: 0 }],
|
||||
]);
|
||||
// The saved workflow object is untouched.
|
||||
expect(workflow.connections.Revise.main).toEqual([
|
||||
[{ node: 'Format', type: 'main', index: 0 }],
|
||||
]);
|
||||
});
|
||||
|
||||
it('attaches Instance AI execution telemetry metadata to workflow runs', async () => {
|
||||
const { adapter, mockWorkflowRunner } = createRunAdapterForTests({
|
||||
id: 'wf-1',
|
||||
|
|
|
|||
|
|
@ -1138,6 +1138,12 @@ export class InstanceAiAdapterService {
|
|||
|
||||
const timeoutMs = Math.min(options?.timeout ?? DEFAULT_TIMEOUT_MS, MAX_TIMEOUT_MS);
|
||||
|
||||
// Sever the listed edges on this run's ephemeral copy only — used by
|
||||
// scripted wait-gate verification to keep each pass acyclic.
|
||||
const connections = options?.omitConnections?.length
|
||||
? omitWorkflowConnections(workflow.connections, options.omitConnections)
|
||||
: workflow.connections;
|
||||
|
||||
// Force-save AI-initiated executions so that follow-up
|
||||
// `executions(list/get/debug)` calls can read the result, regardless of
|
||||
// instance-wide or per-workflow save settings. Manual mode is gated by
|
||||
|
|
@ -1149,6 +1155,7 @@ export class InstanceAiAdapterService {
|
|||
: ('manual' as WorkflowExecuteMode),
|
||||
workflowData: {
|
||||
...workflow,
|
||||
connections,
|
||||
settings: {
|
||||
...workflow.settings,
|
||||
saveManualExecutions: true,
|
||||
|
|
@ -3461,6 +3468,25 @@ function findTriggerNode(nodes: INode[]): INode | undefined {
|
|||
return nodes.find((n) => isTriggerNodeType(n.type));
|
||||
}
|
||||
|
||||
/** Copy of `connections` minus every listed source→target edge (any type). */
|
||||
function omitWorkflowConnections(
|
||||
connections: IConnections,
|
||||
omit: Array<{ source: string; target: string }>,
|
||||
): IConnections {
|
||||
const omitted = new Set(omit.map((edge) => `${edge.source}→${edge.target}`));
|
||||
const result: IConnections = {};
|
||||
for (const [source, byType] of Object.entries(connections)) {
|
||||
const nextByType: IConnections[string] = {};
|
||||
for (const [type, groups] of Object.entries(byType)) {
|
||||
nextByType[type] = (groups ?? []).map((group) =>
|
||||
group ? group.filter((connection) => !omitted.has(`${source}→${connection.node}`)) : group,
|
||||
);
|
||||
}
|
||||
result[source] = nextByType;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/** Get the execution mode based on the trigger node type. */
|
||||
function getExecutionModeForTrigger(node: INode): WorkflowExecuteMode {
|
||||
switch (node.type) {
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user