diff --git a/agent-service/src/agent/tools/workflow-crud-tools.spec.ts b/agent-service/src/agent/tools/workflow-crud-tools.spec.ts index c1e2b6b93d6..44c6953de10 100644 --- a/agent-service/src/agent/tools/workflow-crud-tools.spec.ts +++ b/agent-service/src/agent/tools/workflow-crud-tools.spec.ts @@ -53,6 +53,20 @@ const OPERATOR_METADATA: OperatorMetadata = { dynamicOutputPorts: false, }, }, + { + operatorType: "TestJoinOp", + jsonSchema: TEST_OP_SCHEMA, + operatorVersion: "1.0", + additionalMetadata: { + userFriendlyName: "Test Join Operator", + operatorGroupName: "Test", + operatorDescription: "A test operator with two input ports", + inputPorts: [{ displayName: "in-0" }, { displayName: "in-1" }], + outputPorts: [{ displayName: "out-0" }], + dynamicInputPorts: false, + dynamicOutputPorts: false, + }, + }, ], groups: [], }; @@ -65,7 +79,10 @@ function buildMetadataStore(): WorkflowSystemMetadata { const metadataStore = buildMetadataStore(); const context = { metadataStore }; -const operatorSchemas = new Map([["TestOp", TEST_OP_SCHEMA]]); +const operatorSchemas = new Map([ + ["TestOp", TEST_OP_SCHEMA], + ["TestJoinOp", TEST_OP_SCHEMA], +]); // Invoke a tool's execute() with the minimal ToolExecutionOptions the SDK requires. async function runTool(agentTool: any, args: Record): Promise { @@ -224,9 +241,10 @@ describe("createModifyOperatorTool", () => { expect(result).toContain("[ERROR]"); expect(result).toContain("non-negative integer"); - // Existing incoming links are dropped before the port indices are validated, - // so the failed modify leaves the operator disconnected. - expect(state.getAllLinks()).toHaveLength(0); + // The port indices are validated before anything changes, so the failed + // modify keeps the existing link and doesn't rename the operator. + expect(state.getAllLinks().map(link => link.linkID)).toEqual(["l0"]); + expect(state.getOperator("op1")!.customDisplayName).toBeUndefined(); }); test("returns an error when an input port index is negative", async () => { @@ -261,6 +279,23 @@ describe("createModifyOperatorTool", () => { expect(state.getAllLinks()).toHaveLength(0); }); + test.each(["1abc", "1.5"])("returns an error when an input port index is not all digits: %p", async portKey => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + state.addOperator(makeOperator("op1", 2, 1)); + + const result = await runTool(createModifyOperatorTool(state), { + operatorId: "op1", + inputOperatorIds: { [portKey]: ["op0"] }, + summary: "bad port", + }); + + expect(result).toContain("[ERROR]"); + expect(result).toContain("non-negative integer"); + // parseInt alone would read the key as port 1, which exists here. + expect(state.getAllLinks()).toHaveLength(0); + }); + test("returns an error when a referenced source operator does not exist", async () => { const state = new WorkflowState(); state.addOperator(makeOperator("op1")); @@ -276,6 +311,109 @@ describe("createModifyOperatorTool", () => { expect(state.getAllLinks()).toHaveLength(0); }); + test("returns an error and keeps the links when the operator names itself as a source", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + state.addOperator(makeOperator("op1")); + state.addLink({ + linkID: "l0", + source: { operatorID: "op0", portID: "output-0" }, + target: { operatorID: "op1", portID: "input-0" }, + }); + const before = structuredClone(state.getWorkflowContent()); + + const result = await runTool(createModifyOperatorTool(state), { + operatorId: "op1", + inputOperatorIds: { "0": ["op1"] }, + summary: "self", + }); + + expect(result).toContain("[ERROR]"); + expect(result).toContain('Operator "op1" cannot be its own input source'); + expect(state.getWorkflowContent()).toEqual(before); + }); + + test("leaves properties, display name and links unchanged when a new source does not exist", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + state.addOperator({ + ...makeOperator("op1"), + operatorProperties: { title: "original" }, + customDisplayName: "original", + }); + state.addLink({ + linkID: "l0", + source: { operatorID: "op0", portID: "output-0" }, + target: { operatorID: "op1", portID: "input-0" }, + }); + const before = structuredClone(state.getWorkflowContent()); + + const result = await runTool(createModifyOperatorTool(state, context), { + operatorId: "op1", + properties: { title: "changed" }, + inputOperatorIds: { "0": ["ghost"] }, + summary: "changed", + }); + + expect(result).toContain("[ERROR]"); + expect(result).toContain('Source operator "ghost" not found'); + expect(state.getWorkflowContent()).toEqual(before); + }); + + test("leaves the existing links unchanged when a later port names a missing source", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + state.addOperator(makeOperator("op1", 0, 1)); + state.addOperator(makeOperator("op2", 2, 1)); + state.addLink({ + linkID: "l0", + source: { operatorID: "op0", portID: "output-0" }, + target: { operatorID: "op2", portID: "input-0" }, + }); + state.addLink({ + linkID: "l1", + source: { operatorID: "op1", portID: "output-0" }, + target: { operatorID: "op2", portID: "input-1" }, + }); + const before = structuredClone(state.getWorkflowContent()); + + // Port 0 is valid and would be relinked first; port 1 names a source that does not exist. + const result = await runTool(createModifyOperatorTool(state), { + operatorId: "op2", + inputOperatorIds: { "0": ["op1"], "1": ["ghost"] }, + summary: "re-link", + }); + + expect(result).toContain("[ERROR]"); + expect(result).toContain('Source operator "ghost" not found'); + expect(state.getWorkflowContent()).toEqual(before); + }); + + test("emits no change events when the call fails", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + state.addOperator(makeOperator("op1")); + state.addLink({ + linkID: "l0", + source: { operatorID: "op0", portID: "output-0" }, + target: { operatorID: "op1", portID: "input-0" }, + }); + const events: unknown[] = []; + const subscription = state.getWorkflowChangedStream().subscribe(event => events.push(event)); + + const result = await runTool(createModifyOperatorTool(state, context), { + operatorId: "op1", + properties: { title: "changed" }, + inputOperatorIds: { "0": ["ghost"] }, + summary: "changed", + }); + subscription.unsubscribe(); + + expect(result).toContain("[ERROR]"); + // Subscribers such as auto-persist see nothing from a failed call. + expect(events).toHaveLength(0); + }); + test("wraps an unexpected state failure with the operator id", async () => { const state = new WorkflowState(); state.addOperator(makeOperator("op1")); @@ -434,6 +572,24 @@ describe("createAddOperatorTool", () => { expect(result).toContain("non-negative integer"); }); + test.each(["1abc", "1.5"])("returns an error when an input port index is not all digits: %p", async portKey => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + const before = structuredClone(state.getWorkflowContent()); + + const result = await runTool(createAddOperatorTool(state, operatorSchemas, context), { + operatorId: "op1", + operatorType: "TestJoinOp", + properties: {}, + inputOperatorIds: { [portKey]: ["op0"] }, + summary: "x", + }); + + expect(result).toContain("[ERROR]"); + expect(result).toContain("non-negative integer"); + expect(state.getWorkflowContent()).toEqual(before); + }); + test("returns an error when a referenced source operator does not exist", async () => { const state = new WorkflowState(); @@ -450,6 +606,80 @@ describe("createAddOperatorTool", () => { expect(result).toContain("not found"); }); + test("returns an error when the operator names itself as a source", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + const before = structuredClone(state.getWorkflowContent()); + + const result = await runTool(createAddOperatorTool(state, operatorSchemas, context), { + operatorId: "op1", + operatorType: "TestOp", + properties: {}, + inputOperatorIds: { "0": ["op1"] }, + summary: "x", + }); + + expect(result).toContain("[ERROR]"); + expect(result).toContain('Operator "op1" cannot be its own input source'); + expect(state.getWorkflowContent()).toEqual(before); + }); + + test("leaves the workflow unchanged when an input port index is out of range", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + const before = structuredClone(state.getWorkflowContent()); + + const result = await runTool(createAddOperatorTool(state, operatorSchemas, context), { + operatorId: "op1", + operatorType: "TestOp", + properties: {}, + inputOperatorIds: { "5": ["op0"] }, + summary: "x", + }); + + expect(result).toContain("[ERROR]"); + expect(result).toContain("Input port index 5 out of range"); + expect(state.getWorkflowContent()).toEqual(before); + }); + + test("leaves the workflow unchanged when a later port names a missing source", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + const before = structuredClone(state.getWorkflowContent()); + + // Port 0 is valid and would be linked first; port 1 names a source that does not exist. + const result = await runTool(createAddOperatorTool(state, operatorSchemas, context), { + operatorId: "op1", + operatorType: "TestJoinOp", + properties: {}, + inputOperatorIds: { "0": ["op0"], "1": ["ghost"] }, + summary: "x", + }); + + expect(result).toContain("[ERROR]"); + expect(result).toContain('Source operator "ghost" not found'); + expect(state.getWorkflowContent()).toEqual(before); + }); + + test("emits no change events when the call fails", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + const events: unknown[] = []; + const subscription = state.getWorkflowChangedStream().subscribe(event => events.push(event)); + + const result = await runTool(createAddOperatorTool(state, operatorSchemas, context), { + operatorId: "op1", + operatorType: "TestJoinOp", + properties: {}, + inputOperatorIds: { "0": ["op0"], "1": ["ghost"] }, + summary: "x", + }); + subscription.unsubscribe(); + + expect(result).toContain("[ERROR]"); + expect(events).toHaveLength(0); + }); + test("adds a source operator with no input links", async () => { const state = new WorkflowState(); @@ -486,6 +716,53 @@ describe("createAddOperatorTool", () => { expect(links[0].target.operatorID).toBe("op2"); }); + test("links each input port of a multi-port operator", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + state.addOperator(makeOperator("op1", 0, 1)); + + const result = await runTool(createAddOperatorTool(state, operatorSchemas, context), { + operatorId: "op2", + operatorType: "TestJoinOp", + properties: {}, + inputOperatorIds: { "0": ["op0"], "1": ["op1"] }, + summary: "join", + }); + + expect(result).toBe( + "Added operator op2, input ports: 2, output ports: 1, created links: [op0 --> op2, op1 --> op2]" + ); + expect(state.getAllLinks().map(link => [link.source.operatorID, link.target.portID])).toEqual([ + ["op0", "input-0"], + ["op1", "input-1"], + ]); + }); + + test("accepts the same operator id again after a failed add", async () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op0", 0, 1)); + const addTool = createAddOperatorTool(state, operatorSchemas, context); + const failed = await runTool(addTool, { + operatorId: "op1", + operatorType: "TestOp", + properties: {}, + inputOperatorIds: { "0": ["ghost"] }, + summary: "x", + }); + expect(failed).toContain("[ERROR]"); + + const result = await runTool(addTool, { + operatorId: "op1", + operatorType: "TestOp", + properties: {}, + inputOperatorIds: { "0": ["op0"] }, + summary: "x", + }); + + expect(result).toContain("Added operator op1"); + expect(result).toContain("created links: [op0 --> op1]"); + }); + test("reports an unexpected state failure without adding the operator", async () => { const state = new WorkflowState(); // Force the state layer to blow up so the tool's catch block is exercised. diff --git a/agent-service/src/agent/tools/workflow-crud-tools.ts b/agent-service/src/agent/tools/workflow-crud-tools.ts index 54a0b29db99..1933b086aa1 100644 --- a/agent-service/src/agent/tools/workflow-crud-tools.ts +++ b/agent-service/src/agent/tools/workflow-crud-tools.ts @@ -22,7 +22,7 @@ import { tool } from "ai"; import { WorkflowState } from "../workflow-state"; import { autoLayoutWorkflow } from "../util/auto-layout"; import { WorkflowUtilService } from "../util/workflow-utils"; -import type { OperatorLink } from "../../types/workflow"; +import type { OperatorLink, PortDescription } from "../../types/workflow"; import { createToolResult, createErrorResult, @@ -57,6 +57,50 @@ function formatInputArgs(args: Record): string { return `Input: ${JSON.stringify(compact)}`; } +/** + * Checks every port index and source operator in inputOperatorIds and returns the links to create, + * or the first error. It doesn't change the workflow, so callers must run it before any mutation. + */ +function resolveInputLinks( + workflowState: WorkflowState, + operatorId: string, + inputPorts: PortDescription[], + inputOperatorIds: Record | undefined, + inputInfo: string +): { links: Omit[] } | { error: string } { + const links: Omit[] = []; + for (const [portIndexStr, sourceOpIds] of Object.entries(inputOperatorIds ?? {})) { + if (!/^\d+$/.test(portIndexStr)) { + return { error: `Invalid input port index: "${portIndexStr}". Must be a non-negative integer. ${inputInfo}` }; + } + const targetPortIdx = parseInt(portIndexStr, 10); + if (targetPortIdx >= inputPorts.length) { + return { + error: `Input port index ${targetPortIdx} out of range. Operator "${operatorId}" has ${inputPorts.length} input port(s). ${inputInfo}`, + }; + } + const targetPortId = inputPorts[targetPortIdx].portID; + + for (const sourceOpId of sourceOpIds) { + if (sourceOpId === operatorId) { + return { error: `Operator "${operatorId}" cannot be its own input source. ${inputInfo}` }; + } + const sourceOp = workflowState.getOperator(sourceOpId); + if (!sourceOp) { + return { + error: `Source operator "${sourceOpId}" not found. Make sure it exists before referencing it in inputOperatorIds. ${inputInfo}`, + }; + } + const sourcePortId = sourceOp.outputPorts.length > 0 ? sourceOp.outputPorts[0].portID : "output-0"; + links.push({ + source: { operatorID: sourceOpId, portID: sourcePortId }, + target: { operatorID: operatorId, portID: targetPortId }, + }); + } + } + return { links }; +} + export function createAddOperatorTool( workflowState: WorkflowState, operatorSchemas: Map, @@ -143,44 +187,21 @@ Examples: operatorProperties: { ...operator.operatorProperties, ...args.properties }, }; + const resolved = resolveInputLinks( + workflowState, + args.operatorId, + operator.inputPorts, + args.inputOperatorIds, + inputInfo + ); + if ("error" in resolved) return createErrorResult(resolved.error); + workflowState.addOperator(operator); const createdLinkPairs: { source: string; target: string }[] = []; - if (args.inputOperatorIds) { - const addedOperator = workflowState.getOperator(operator.operatorID)!; - for (const [portIndexStr, sourceOpIds] of Object.entries(args.inputOperatorIds)) { - const targetPortIdx = parseInt(portIndexStr, 10); - if (isNaN(targetPortIdx) || targetPortIdx < 0) { - return createErrorResult( - `Invalid input port index: "${portIndexStr}". Must be a non-negative integer. ${inputInfo}` - ); - } - if (targetPortIdx >= addedOperator.inputPorts.length) { - return createErrorResult( - `Input port index ${targetPortIdx} out of range. Operator "${args.operatorId}" has ${addedOperator.inputPorts.length} input port(s). ${inputInfo}` - ); - } - const targetPortId = addedOperator.inputPorts[targetPortIdx].portID; - - for (const sourceOpId of sourceOpIds) { - const sourceOp = workflowState.getOperator(sourceOpId); - if (!sourceOp) { - return createErrorResult( - `Source operator "${sourceOpId}" not found. Make sure it exists before referencing it in inputOperatorIds. ${inputInfo}` - ); - } - const sourcePortId = sourceOp.outputPorts.length > 0 ? sourceOp.outputPorts[0].portID : "output-0"; - - const linkId = workflowState.generateLinkId(); - const link: OperatorLink = { - linkID: linkId, - source: { operatorID: sourceOpId, portID: sourcePortId }, - target: { operatorID: args.operatorId, portID: targetPortId }, - }; - workflowState.addLink(link); - createdLinkPairs.push({ source: sourceOpId, target: args.operatorId }); - } - } + for (const link of resolved.links) { + workflowState.addLink({ linkID: workflowState.generateLinkId(), ...link }); + createdLinkPairs.push({ source: link.source.operatorID, target: args.operatorId }); } autoLayoutWorkflow(workflowState); @@ -254,6 +275,15 @@ Examples: } } + const resolved = resolveInputLinks( + workflowState, + args.operatorId, + operator.inputPorts, + args.inputOperatorIds, + inputInfo + ); + if ("error" in resolved) return createErrorResult(resolved.error); + const createdLinkPairs: { source: string; target: string }[] = []; const deletedLinkPairs: { source: string; target: string }[] = []; @@ -274,38 +304,9 @@ Examples: workflowState.deleteLink(link.linkID); } - for (const [portIndexStr, sourceOpIds] of Object.entries(args.inputOperatorIds)) { - const targetPortIdx = parseInt(portIndexStr, 10); - if (isNaN(targetPortIdx) || targetPortIdx < 0) { - return createErrorResult( - `Invalid input port index: "${portIndexStr}". Must be a non-negative integer. ${inputInfo}` - ); - } - if (targetPortIdx >= operator.inputPorts.length) { - return createErrorResult( - `Input port index ${targetPortIdx} out of range. Operator "${args.operatorId}" has ${operator.inputPorts.length} input port(s). ${inputInfo}` - ); - } - const targetPortId = operator.inputPorts[targetPortIdx].portID; - - for (const sourceOpId of sourceOpIds) { - const sourceOp = workflowState.getOperator(sourceOpId); - if (!sourceOp) { - return createErrorResult( - `Source operator "${sourceOpId}" not found. Make sure it exists before referencing it in inputOperatorIds. ${inputInfo}` - ); - } - const sourcePortId = sourceOp.outputPorts.length > 0 ? sourceOp.outputPorts[0].portID : "output-0"; - - const linkId = workflowState.generateLinkId(); - const link: OperatorLink = { - linkID: linkId, - source: { operatorID: sourceOpId, portID: sourcePortId }, - target: { operatorID: args.operatorId, portID: targetPortId }, - }; - workflowState.addLink(link); - createdLinkPairs.push({ source: sourceOpId, target: args.operatorId }); - } + for (const link of resolved.links) { + workflowState.addLink({ linkID: workflowState.generateLinkId(), ...link }); + createdLinkPairs.push({ source: link.source.operatorID, target: args.operatorId }); } autoLayoutWorkflow(workflowState);