From 2e11409433cab2917e6ba724cd06c43101622778 Mon Sep 17 00:00:00 2001 From: Shawn Garello Date: Wed, 14 Oct 2026 00:55:29 -0700 Subject: [PATCH 1/3] test(agent-service): reproduce half-applied operator edits addOperator and modifyOperator change the workflow before they check the input ports and source operators, so a call that returns an error can leave a new operator, a partial link, or changed properties with the operator's incoming links removed. Add three tests that assert the workflow is unchanged after such an error. They fail on main and are meant to pass once the tools check everything before changing the workflow. Generated-by: Claude Code (Opus 5.5) --- .../agent/tools/workflow-crud-tools.spec.ts | 83 ++++++++++++++++++- 1 file changed, 82 insertions(+), 1 deletion(-) 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..747e6a73a1b 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 { @@ -276,6 +293,33 @@ describe("createModifyOperatorTool", () => { expect(state.getAllLinks()).toHaveLength(0); }); + 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("wraps an unexpected state failure with the operator id", async () => { const state = new WorkflowState(); state.addOperator(makeOperator("op1")); @@ -450,6 +494,43 @@ describe("createAddOperatorTool", () => { expect(result).toContain("not found"); }); + 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("adds a source operator with no input links", async () => { const state = new WorkflowState(); From 5149ef31992bc64fb82d3e48f54e1069af1445ed Mon Sep 17 00:00:00 2001 From: Shawn Garello Date: Wed, 14 Oct 2026 08:56:50 -0700 Subject: [PATCH 2/3] fix(agent-service): check operator edits before changing the workflow addOperator and modifyOperator now check every input port index and source operator before they add the operator, apply properties or delete links, so a call that returns an error leaves the workflow as it was. Closes #8979 Generated-by: Claude Code (Opus 5.5) --- .../agent/tools/workflow-crud-tools.spec.ts | 7 +- .../src/agent/tools/workflow-crud-tools.ts | 134 +++++++++--------- 2 files changed, 70 insertions(+), 71 deletions(-) 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 747e6a73a1b..7f361b0d70f 100644 --- a/agent-service/src/agent/tools/workflow-crud-tools.spec.ts +++ b/agent-service/src/agent/tools/workflow-crud-tools.spec.ts @@ -241,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 () => { diff --git a/agent-service/src/agent/tools/workflow-crud-tools.ts b/agent-service/src/agent/tools/workflow-crud-tools.ts index 54a0b29db99..b17b09fa9a8 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,47 @@ 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 ?? {})) { + const targetPortIdx = parseInt(portIndexStr, 10); + if (isNaN(targetPortIdx) || targetPortIdx < 0) { + return { error: `Invalid input port index: "${portIndexStr}". Must be a non-negative integer. ${inputInfo}` }; + } + 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) { + 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 +184,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 +272,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 +301,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); From bee292e1ae82a4bbec21b2ad744dac8039e321b6 Mon Sep 17 00:00:00 2001 From: Shawn Garello Date: Mon, 19 Oct 2026 18:36:45 -0700 Subject: [PATCH 3/3] fix(agent-service): reject self-links and non-digit port keys Applies the review suggestions on #8981. resolveInputLinks now rejects an operator that names itself as a source (modifyOperator used to store a self-loop) and port keys that are not all digits (parseInt read "1abc" as port 1). Adds tests for both, plus a failed modify on a later port, no change events after a failed call, a two-port add and a retry after a failed add. Generated-by: Claude Code (Opus 5.5) --- .../agent/tools/workflow-crud-tools.spec.ts | 195 ++++++++++++++++++ .../src/agent/tools/workflow-crud-tools.ts | 7 +- 2 files changed, 200 insertions(+), 2 deletions(-) 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 7f361b0d70f..44c6953de10 100644 --- a/agent-service/src/agent/tools/workflow-crud-tools.spec.ts +++ b/agent-service/src/agent/tools/workflow-crud-tools.spec.ts @@ -279,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")); @@ -294,6 +311,28 @@ 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)); @@ -321,6 +360,60 @@ describe("createModifyOperatorTool", () => { 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")); @@ -479,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(); @@ -495,6 +606,24 @@ 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)); @@ -532,6 +661,25 @@ describe("createAddOperatorTool", () => { 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(); @@ -568,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 b17b09fa9a8..1933b086aa1 100644 --- a/agent-service/src/agent/tools/workflow-crud-tools.ts +++ b/agent-service/src/agent/tools/workflow-crud-tools.ts @@ -70,10 +70,10 @@ function resolveInputLinks( ): { links: Omit[] } | { error: string } { const links: Omit[] = []; for (const [portIndexStr, sourceOpIds] of Object.entries(inputOperatorIds ?? {})) { - const targetPortIdx = parseInt(portIndexStr, 10); - if (isNaN(targetPortIdx) || targetPortIdx < 0) { + 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}`, @@ -82,6 +82,9 @@ function resolveInputLinks( 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 {