Skip to content

fix(agent-service): check operator edits before changing the workflow #8981

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ShawnGarello wants to merge 4 commits into
base: main
Choose a base branch
from
Open

fix(agent-service): check operator edits before changing the workflow #8981

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2e11409
test(agent-service): reproduce half-applied operator edits
ShawnGarello Oct 14, 2026
5149ef3
fix(agent-service): check operator edits before changing the workflow
ShawnGarello Oct 14, 2026
bee292e
fix(agent-service): reject self-links and non-digit port keys
ShawnGarello Oct 20, 2026
18fbc87
Merge branch 'main' into fix/agent-tool-partial-edits
ShawnGarello Oct 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
285 changes: 281 additions & 4 deletions agent-service/src/agent/tools/workflow-crud-tools.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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: [],
};
Expand All @@ -65,7 +79,10 @@ function buildMetadataStore(): WorkflowSystemMetadata {

const metadataStore = buildMetadataStore();
const context = { metadataStore };
const operatorSchemas = new Map<string, any>([["TestOp", TEST_OP_SCHEMA]]);
const operatorSchemas = new Map<string, any>([
["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<string, any>): Promise<string> {
Expand Down Expand Up @@ -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 () => {
Expand Down Expand Up @@ -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"));
Expand All @@ -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"));
Expand Down Expand Up @@ -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();

Expand All @@ -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();

Expand Down Expand Up @@ -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.
Expand Down
Loading