From bff5def04e9221c4226fa6a8f96a08953c8d9dae Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Sun, 9 Aug 2026 17:04:33 -0700 Subject: [PATCH 1/2] test(agent-service): drive sendMessage with a stand-in language model sendMessage was the largest untested region in the service: 238 of the 292 uncovered lines in texera-agent.ts, and nothing exercised it because the existing spec stops at the model boundary. Nothing there actually needs the network - ai/test ships a MockLanguageModelV4 that satisfies the same LanguageModel type the constructor takes. Adds 25 tests across two blocks. The first drives the ReAct loop itself: branch bookkeeping, per-step and summed usage, tool projection, the maxSteps cap, turn chaining, and every failure path - a thrown model, a non-Error throw, cancellation, and stop() mid-run. The second covers delegate mode through fetch: the one-time backend refresh, auto-execution after a tool call and where its result is keyed, the guards that suppress it, and the debounced auto-persist. texera-agent.ts goes from 51.01% to 99.82% lines. The one line left is getStepsById, which has no call site. No production file is touched. --- agent-service/src/agent/texera-agent.spec.ts | 794 ++++++++++++++++++- 1 file changed, 793 insertions(+), 1 deletion(-) diff --git a/agent-service/src/agent/texera-agent.spec.ts b/agent-service/src/agent/texera-agent.spec.ts index 27aa37b8139..93c9500c401 100644 --- a/agent-service/src/agent/texera-agent.spec.ts +++ b/agent-service/src/agent/texera-agent.spec.ts @@ -17,9 +17,11 @@ * under the License. */ -import { beforeEach, describe, expect, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; import { TexeraAgent } from "./texera-agent"; import { AgentState, INITIAL_STEP_ID, type ReActStep } from "../types/agent"; +import { MockLanguageModelV4 } from "ai/test"; +import { WorkflowSystemMetadata } from "./util/workflow-system-metadata"; /** * These tests cover the agent's bookkeeping — the ReAct step tree, settings, and client set — which @@ -297,3 +299,793 @@ describe("TexeraAgent", () => { }); }); }); + +/** + * LanguageModelV4FinishReason is an object, not a string: a bare `finishReason: "stop"` runs fine + * under `bun test` but fails `tsc --noEmit`, so the mocks below build it through here. + */ +const finish = (unified: "stop" | "tool-calls") => ({ unified, raw: undefined }); + +/** + * LanguageModelV4Usage is nested. A flat `{ inputTokens: 11 }` is silently discarded, and every + * usage assertion then passes against a gutted mapping — so usage is always built through here. + * `totalTokens` is deliberately absent: the SDK derives it. + */ +const usage = (i: number, o: number) => ({ + inputTokens: { total: i, noCache: i, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: o, text: o, reasoning: 0 }, +}); + +const textModel = (text: string, i = 11, o = 7) => + new MockLanguageModelV4({ + doGenerate: async () => ({ + content: [{ type: "text" as const, text }], + // `as const` on both: without it these widen to `string` and the mock no longer satisfies + // LanguageModelV4GenerateResult, which `bun test` accepts but `tsc --noEmit` rejects. + finishReason: finish("stop"), + usage: usage(i, o), + warnings: [], + }), + }); + +/** Sibling of makeAgent() that takes a stand-in model, so sendMessage runs with no network. */ +function makeAgentWith(model: any): TexeraAgent { + return new TexeraAgent({ model, modelType: "test-model", agentId: "agent-1", systemPrompt: "SYS-XYZ" }); +} + +/** A source operator. `inputPorts: []` matters — a non-empty one fails validateOperatorConnection + * and masks the delegate-mode assertions behind a validation error. */ +const srcOp = (id: string, props: Record = {}) => ({ + operatorID: id, + operatorType: "CSVFileScan", + operatorVersion: "1.0", + operatorProperties: props, + inputPorts: [], + outputPorts: [{ portID: "output-0" }], + showAdvanced: false, +}); + +/** + * A tripwire rather than a stub: with an empty workflow and no delegate config, sendMessage makes + * no network calls at all, and the tests below assert that. Tests that do need I/O install their + * own implementation over it. + */ +let fetchSpy: any; +let urls: string[]; + +beforeEach(() => { + urls = []; + fetchSpy = spyOn(globalThis, "fetch").mockImplementation((async (u: any) => { + urls.push(String(u)); + throw new Error("unexpected fetch"); + }) as any); +}); +afterEach(() => fetchSpy.mockRestore()); + +/** + * sendMessage was the largest untested region in the file. Everything it needs is in-process: + * `ai/test` supplies a model that never reaches the network, so the ReAct loop, its usage + * accounting, its branch bookkeeping and its failure paths can all be driven directly. + * + * One shape to know when adding tests here: a tool call's `input` must be a JSON *string*, e.g. + * `{ type: "tool-call", toolCallId: "c1", toolName: "modifyOperator", input: JSON.stringify(...) }`. + */ +describe("sendMessage", () => { + test("records the turn as a linear two-step branch", async () => { + const model = textModel("hello there"); + const agent = makeAgentWith(model); + const res = await agent.sendMessage("12345678", "feedback"); + const steps = agent.getAllSteps(); + expect(res.response).toBe("hello there"); + expect(res.stopped).toBe(false); + expect(res.error).toBeUndefined(); + expect(res.messages).toEqual([{ role: "assistant", content: [{ type: "text", text: "hello there" }] }]); + expect(res.usage).toEqual({ inputTokens: 11, outputTokens: 7, totalTokens: 18 }); + expect(steps.map(s => [s.role, s.stepId, s.content, s.isBegin, s.isEnd])).toEqual([ + ["user", 0, "12345678", true, true], + ["agent", 1, "hello there", true, true], + ]); + expect(steps[0].usage).toEqual({ inputTokens: 2, outputTokens: 0, totalTokens: 2 }); + expect(steps[0].messageSource).toBe("feedback"); + expect(steps[0].parentId).toBe(INITIAL_STEP_ID); + expect(steps[1].parentId).toBe(steps[0].id); + expect(agent.getHead()).toBe(steps[1].id); + expect(agent.getAncestorPath()).toEqual([INITIAL_STEP_ID, steps[0].id, steps[1].id]); + expect(steps[0].id).toMatch(/^step-agent-1-1-\d+$/); + expect(steps[1].id).toMatch(/^step-agent-1-2-\d+$/); + expect(steps[0].messageId).toMatch(/^msg-agent-1-1-\d+$/); + expect(agent.getState()).toBe(AgentState.AVAILABLE); + expect((agent as any).abortController).toBeNull(); + expect(urls).toEqual([]); + }); + + test("pinned call options + assembled context replaces the raw message", async () => { + const model = textModel("ok"); + const agent = makeAgentWith(model); + await agent.sendMessage("raw-user-text"); + const call = (model as any).doGenerateCalls[0]; + expect(call.temperature).toBe(0.2); + expect(call.providerOptions).toEqual({ + openai: { parallelToolCalls: false }, + anthropic: { disableParallelToolUse: true }, + mistral: { parallelToolCalls: false }, + }); + expect(call.abortSignal).toBeInstanceOf(AbortSignal); + expect(call.abortSignal.aborted).toBe(false); + expect(call.tools.map((t: any) => t.name)).toEqual(["deleteOperator", "addOperator", "modifyOperator"]); + expect(call.prompt[0]).toEqual({ role: "system", content: "SYS-XYZ" }); + const txt = call.prompt[1].content[0].text; + expect(txt).toContain("# Ongoing Task"); + expect(txt).toContain("raw-user-text"); + expect(agent.getAllSteps()[1].inputMessages).toEqual([{ role: "user", content: txt }]); + }); + + test("two-step tool run: per-step + summed usage, tool projection, rolling snapshots", async () => { + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + if (n === 1) + return { + content: [ + { + type: "tool-call", + toolCallId: "c1", + toolName: "deleteOperator", + input: JSON.stringify({ operatorId: "ghost" }), + }, + ], + finishReason: finish("tool-calls"), + usage: usage(100, 10), + warnings: [], + } as any; + return { + content: [{ type: "text", text: "done" }], + finishReason: finish("stop"), + usage: usage(200, 20), + warnings: [], + } as any; + }, + }); + const agent = makeAgentWith(model); + const res = await agent.sendMessage("delete it"); + const steps = agent.getAllSteps(); + expect(res.usage).toEqual({ inputTokens: 300, outputTokens: 30, totalTokens: 330 }); + expect(res.response).toBe("done"); + expect(steps[1].usage).toEqual({ inputTokens: 100, outputTokens: 10, totalTokens: 110 }); + expect(steps[2].usage).toEqual({ inputTokens: 200, outputTokens: 20, totalTokens: 220 }); + expect(steps[1].toolCalls).toEqual([ + { toolName: "deleteOperator", toolCallId: "c1", input: { operatorId: "ghost" } }, + ]); + expect(steps[1].toolResults).toEqual([ + { toolCallId: "c1", output: "[ERROR] Operator ghost not found", isError: false }, + ]); + expect(steps[1].content).toBe(""); + expect(steps.map(s => [s.stepId, s.isBegin, s.isEnd])).toEqual([ + [0, true, true], + [1, true, false], + [2, false, true], + ]); + expect(steps[2].parentId).toBe(steps[1].id); + expect(urls).toEqual([]); + }); + + test("rolling before/after snapshots across a mutating step", async () => { + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + if (n === 1) + return { + content: [ + { + type: "tool-call", + toolCallId: "c1", + toolName: "deleteOperator", + input: JSON.stringify({ operatorId: "op1" }), + }, + ], + finishReason: finish("tool-calls"), + usage: usage(1, 1), + warnings: [], + } as any; + return { + content: [{ type: "text", text: "x" }], + finishReason: finish("stop"), + usage: usage(1, 1), + warnings: [], + } as any; + }, + }); + const agent = makeAgentWith(model); + agent.getWorkflowState().addOperator(srcOp("op1") as any); + fetchSpy.mockImplementation((async (u: any) => { + urls.push(String(u)); + return { ok: true, json: async () => ({ operatorOutputSchemas: {}, operatorErrors: {} }) } as any; + }) as any); + await agent.sendMessage("del"); + const counts = agent + .getAllSteps() + .map(s => [s.role, s.beforeWorkflowContent?.operators.length, s.afterWorkflowContent?.operators.length]); + expect(counts).toEqual([ + ["user", 1, 1], + ["agent", 1, 0], + ["agent", 0, 0], + ]); + }); + + test("caps the loop at maxSteps", async () => { + // The cap is the only thing standing between a looping model and an unbounded run. + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + if (n > 20) throw new Error("runaway"); + return { + content: [ + { + type: "tool-call", + toolCallId: "c" + n, + toolName: "deleteOperator", + input: JSON.stringify({ operatorId: "ghost" }), + }, + ], + finishReason: finish("tool-calls"), + usage: usage(1, 1), + warnings: [], + } as any; + }, + }); + const agent = makeAgentWith(model); + agent.updateSettings({ maxSteps: 2 }); + await agent.sendMessage("go"); + expect(n).toBe(2); + expect(agent.getAllSteps().map(s => s.stepId)).toEqual([0, 1, 2]); + }); + + test("second turn chains on and reads turn 1 as completed", async () => { + const model = textModel("a", 1, 1); + const agent = makeAgentWith(model); + await agent.sendMessage("one"); + const headAfter1 = agent.getHead(); + await agent.sendMessage("two"); + const steps = agent.getAllSteps(); + expect(steps.map(s => s.stepId)).toEqual([0, 1, 0, 1]); + expect(steps[2].parentId).toBe(headAfter1); + expect(steps[0].messageId).not.toBe(steps[2].messageId); + expect(steps[2].messageId).toMatch(/^msg-agent-1-2-\d+$/); + expect(steps[3].id).toMatch(/^step-agent-1-4-\d+$/); + const txt = (model as any).doGenerateCalls[1].prompt[1].content[0].text; + expect(txt).toContain("# Completed Tasks"); + expect(txt).toContain("## Task (completed)"); + expect(txt).toContain("# Ongoing Task"); + }); + + test("abandoned branch is invisible to the model", async () => { + // Branches are how a retried turn discards its predecessor; if the abandoned one still reached + // the prompt the model would answer against history the user already rejected. + const model = textModel("a", 1, 1); + const agent = makeAgentWith(model); + await agent.sendMessage("first"); + (agent as any).head = agent.getAllSteps()[0].id; + await agent.sendMessage("second"); + const txt = (model as any).doGenerateCalls[1].prompt[1].content[0].text; + expect(txt).not.toContain("### Turn 1"); + expect(agent.getAllSteps().length).toBe(4); + expect(agent.getVisibleReActSteps().length).toBe(3); + }); + + test("compiles the DAG and feeds schemas + cached results into the prompt", async () => { + const model = textModel("ok", 1, 1); + const agent = makeAgentWith(model); + agent.getWorkflowState().addOperator(srcOp("op-1", { fileName: "f.csv" }) as any); + agent.getWorkflowResultState().set("op-1", INITIAL_STEP_ID, { + state: "COMPLETED", + inputTuples: 0, + outputTuples: 2, + resultMode: "SET_SNAPSHOT", + result: [{ a: 1 }, { a: 2 }], + } as any); + fetchSpy.mockImplementation((async (u: any) => { + urls.push(String(u)); + return { + ok: true, + json: async () => ({ + operatorOutputSchemas: { "op-1": { "0_0": [{ attributeName: "a", attributeType: "integer" }] } }, + operatorErrors: {}, + }), + } as any; + }) as any); + const res = await agent.sendMessage("look"); + const txt = (model as any).doGenerateCalls[0].prompt[1].content[0].text; + expect(res.error).toBeUndefined(); + expect(urls).toEqual(["http://localhost:9090/api/compile"]); + expect(txt).toContain("Output Schema: [a: integer]"); + expect(txt).toContain("(CSVFileScan, executed)"); + expect(txt).toContain("Result:\n Executed operator op-1\n Output table shape: (2, 1)"); + expect(txt).toContain("Properties:\n fileName: f.csv"); + }); + + test("swallows a plan-build failure and still answers", async () => { + const model = textModel("still-here", 1, 1); + const agent = makeAgentWith(model); + const ws = agent.getWorkflowState(); + ws.addOperator({ ...srcOp("src"), outputPorts: undefined } as any); + ws.addOperator({ ...srcOp("dst"), inputPorts: [{ portID: "input-0" }] } as any); + ws.addLink({ + linkID: "l1", + source: { operatorID: "src", portID: "output-0" }, + target: { operatorID: "dst", portID: "input-0" }, + } as any); + const res = await agent.sendMessage("hi"); + expect(res.response).toBe("still-here"); + expect(res.error).toBeUndefined(); + }); + + test("reports a model failure as an error step", async () => { + const model = new MockLanguageModelV4({ + doGenerate: async () => { + throw new Error("model exploded"); + }, + }); + const agent = makeAgentWith(model); + const res = await agent.sendMessage("hi"); + const steps = agent.getAllSteps(); + expect(res).toEqual({ + response: "", + messages: [], + usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 }, + stopped: false, + error: "model exploded", + }); + expect(steps.map(s => [s.role, s.stepId, s.content, s.isBegin, s.isEnd])).toEqual([ + ["user", 0, "hi", true, true], + ["agent", 1, "Error: model exploded", false, true], + ]); + expect(agent.getHead()).toBe(steps[1].id); + expect(agent.getState()).toBe(AgentState.AVAILABLE); + expect((agent as any).abortController).toBeNull(); + expect((agent as any).currentMessageId).toBeUndefined(); + }); + + test("a non-Error throw is stringified", async () => { + const model = new MockLanguageModelV4({ + doGenerate: async () => { + throw "just-a-string"; + }, + }); + const agent = makeAgentWith(model); + const res = await agent.sendMessage("hi"); + expect(res.error).toBe("just-a-string"); + expect(agent.getAllSteps()[1].content).toBe("Error: just-a-string"); + }); + + test("a failed turn stays on the branch", async () => { + const model = new MockLanguageModelV4({ + doGenerate: async () => { + throw new Error("nope"); + }, + }); + const agent = makeAgentWith(model); + await agent.sendMessage("one"); + const headAfter1 = agent.getHead(); + await agent.sendMessage("two"); + expect(agent.getAllSteps()[2].parentId).toBe(headAfter1); + expect(agent.getVisibleReActSteps().length).toBe(4); + }); + + test("reports a cancelled run as stopped and swallows the real error", async () => { + let agent!: TexeraAgent; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + agent.stop(); + throw new Error("real-provider-failure"); + }, + }); + agent = makeAgentWith(model); + const res = await agent.sendMessage("hi"); + const steps = agent.getAllSteps(); + expect(res).toEqual({ + response: "", + messages: [], + usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 }, + stopped: true, + }); + expect(res.error).toBeUndefined(); + expect(steps.map(s => [s.role, s.stepId, s.content, s.isBegin, s.isEnd])).toEqual([ + ["user", 0, "hi", true, true], + ["agent", 1, "Generation stopped by user.", false, true], + ]); + expect(JSON.stringify(steps)).not.toContain("real-provider-failure"); + expect(agent.getHead()).toBe(steps[1].id); + expect(agent.getState()).toBe(AgentState.AVAILABLE); + }); + + test("an AbortError-named provider error reads as a user stop", async () => { + // Providers signal cancellation by name rather than by type, so the name is what has to be read. + const model = new MockLanguageModelV4({ + doGenerate: async () => { + throw Object.assign(new Error("boom"), { name: "AbortError" }); + }, + }); + const agent = makeAgentWith(model); + const res = await agent.sendMessage("hi"); + expect(res.stopped).toBe(true); + expect(res.error).toBeUndefined(); + expect(agent.getAllSteps()[1].content).toBe("Generation stopped by user."); + }); + + test("stop() mid-run prevents the next model call and keeps the partial step", async () => { + // Stopping has to be observable at the next step boundary, and the work already done has to + // survive - otherwise pressing stop silently discards the turn. + let agent!: TexeraAgent; + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + agent.stop(); + return { + content: [ + { + type: "tool-call", + toolCallId: "c" + n, + toolName: "deleteOperator", + input: JSON.stringify({ operatorId: "ghost" }), + }, + ], + finishReason: finish("tool-calls"), + usage: usage(5, 5), + warnings: [], + } as any; + }, + }); + agent = makeAgentWith(model); + const res = await agent.sendMessage("go"); + expect(n).toBe(1); + expect(res.stopped).toBe(true); + expect(res.usage.totalTokens).toBe(0); + expect(res.messages).toEqual([]); + expect(agent.getAllSteps().map(s => [s.role, s.stepId, s.isEnd])).toEqual([ + ["user", 0, true], + ["agent", 1, false], + ["agent", 2, true], + ]); + expect(agent.getHead()).toBe(agent.getAllSteps()[2].id); + }); + + test("mid-run state is GENERATING", async () => { + let release!: () => void; + const gate = new Promise(r => (release = r)); + const observed: string[] = []; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + await gate; + return { + content: [{ type: "text", text: "x" }], + finishReason: finish("stop"), + usage: usage(1, 1), + warnings: [], + } as any; + }, + }); + const agent = makeAgentWith(model); + const p = agent.sendMessage("hi"); + await new Promise(r => setTimeout(r, 5)); + observed.push(agent.getState()); + release(); + await p; + expect(observed).toEqual([AgentState.GENERATING]); + expect(agent.getState()).toBe(AgentState.AVAILABLE); + }); +}); + +/** + * With a delegate config the agent talks to the backend, so these drive it through `fetch`. + * + * Two traps. Setting a delegate config makes the first turn refresh from the backend, and that + * refresh replaces the whole workflow — so operators have to arrive through `dispatch`'s stub + * rather than being seeded on the agent. And every test here ends with `agent.destroy()`, + * because the auto-persist debounce would otherwise fire after the fetch spy is restored and + * issue a real request. + */ +describe("delegate mode", () => { + const wfBody = (ops: any[]) => ({ + wid: 7, + name: "w", + content: { + operators: ops, + links: [], + operatorPositions: {}, + commentBoxes: [], + settings: { dataTransferBatchSize: 400 }, + }, + }); + + /** Routes each backend call the agent makes to a canned body, and records the URL. */ + function dispatch(execBody: any, seed: any = srcOp("op-1")) { + fetchSpy.mockImplementation((async (u: any) => { + const url = String(u); + urls.push(url); + if (url.includes("/api/compile")) + return { ok: true, json: async () => ({ operatorOutputSchemas: {}, operatorErrors: {} }) } as any; + if (url.includes("/api/workflow/persist")) return { ok: true, json: async () => wfBody([]) } as any; + if (url.includes("/api/workflow/")) return { ok: true, json: async () => wfBody([seed]) } as any; + return { ok: true, json: async () => execBody } as any; + }) as any); + } + + /** A completed run. The field is `operators`, not `operatorResults`; with the wrong key the + * result callback silently never fires and the assertions below all still look plausible. */ + const okExec = { + success: true, + state: "Completed", + operators: { + "op-1": { + state: "Completed", + inputTuples: 0, + outputTuples: 1, + resultMode: "table", + totalRowCount: 1, + result: [{ z: 9 }], + }, + }, + }; + + test("refreshes the workflow once, on the first turn only", async () => { + dispatch(okExec); + const model = textModel("ok", 1, 1); + const agent = makeAgentWith(model); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7 }); + await agent.sendMessage("one"); + const retrieves = () => urls.filter(u => u.includes("/api/workflow/7")).length; + expect(retrieves()).toBe(1); + expect( + agent + .getWorkflowState() + .getAllOperators() + .map((o: any) => o.operatorID) + ).toEqual(["op-1"]); + expect(agent.getAllSteps()[0].beforeWorkflowContent?.operators.length).toBe(1); + await agent.sendMessage("two"); + expect(retrieves()).toBe(1); + agent.destroy(); + }); + + test("a failed refresh is swallowed", async () => { + fetchSpy.mockImplementation((async (u: any) => { + urls.push(String(u)); + return { ok: false, status: 500, statusText: "err", text: async () => "boom" } as any; + }) as any); + const model = textModel("still-ok", 1, 1); + const agent = makeAgentWith(model); + agent.getWorkflowState().addOperator(srcOp("local-op") as any); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7 }); + const res = await agent.sendMessage("hi"); + expect(res.response).toBe("still-ok"); + expect( + agent + .getWorkflowState() + .getAllOperators() + .map((o: any) => o.operatorID) + ).toEqual(["local-op"]); + agent.destroy(); + }); + + test("auto-executes after modifyOperator and keys the result at the agent step", async () => { + dispatch(okExec); + const vSpy = spyOn(WorkflowSystemMetadata.getInstance(), "validateOperatorProperties").mockReturnValue({ + isValid: true, + } as any); + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + if (n === 1) + return { + content: [ + { + type: "tool-call", + toolCallId: "c1", + toolName: "modifyOperator", + input: JSON.stringify({ operatorId: "op-1", summary: "renamed" }), + }, + ], + finishReason: finish("tool-calls"), + usage: usage(1, 1), + warnings: [], + } as any; + return { + content: [{ type: "text", text: "d" }], + finishReason: finish("stop"), + usage: usage(1, 1), + warnings: [], + } as any; + }, + }); + const agent = makeAgentWith(model); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7, workflowName: "w" }); + await agent.sendMessage("modify it"); + const steps = agent.getAllSteps(); + const txt2 = (model as any).doGenerateCalls[1].prompt[1].content[0].text; + expect(urls.some(u => u.includes("/api/execution/7/0/run"))).toBe(true); + expect((agent.getWorkflowResultState() as any).get("op-1").stepId).toBe(steps[1].id); + vSpy.mockRestore(); + agent.destroy(); + }); + + test("executeOperator tool keys its result at the current head (the user step)", async () => { + // The contrast with the previous test is the point: an explicit executeOperator call has no agent + // step of its own yet, so its result belongs at the head rather than at a step that follows it. + dispatch(okExec); + const vSpy = spyOn(WorkflowSystemMetadata.getInstance(), "validateOperatorProperties").mockReturnValue({ + isValid: true, + } as any); + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + if (n === 1) + return { + content: [ + { + type: "tool-call", + toolCallId: "c1", + toolName: "executeOperator", + input: JSON.stringify({ operatorId: "op-1" }), + }, + ], + finishReason: finish("tool-calls"), + usage: usage(1, 1), + warnings: [], + } as any; + return { + content: [{ type: "text", text: "d" }], + finishReason: finish("stop"), + usage: usage(1, 1), + warnings: [], + } as any; + }, + }); + const agent = makeAgentWith(model); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7, workflowName: "w" }); + await agent.sendMessage("run it"); + const steps = agent.getAllSteps(); + expect(urls.filter(u => u.includes("/api/execution/")).length).toBe(1); + expect((agent.getWorkflowResultState() as any).get("op-1").stepId).toBe(steps[0].id); + vSpy.mockRestore(); + agent.destroy(); + }); + + test("a tool call missing operatorId does not trigger a whole-workflow run", async () => { + // Without the guard an incomplete tool call falls through to a run of the entire workflow, which is + // both expensive and not what was asked for. + dispatch(okExec); + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + if (n === 1) + return { + content: [{ type: "tool-call", toolCallId: "c1", toolName: "modifyOperator", input: JSON.stringify({}) }], + finishReason: finish("tool-calls"), + usage: usage(1, 1), + warnings: [], + } as any; + return { + content: [{ type: "text", text: "d" }], + finishReason: finish("stop"), + usage: usage(1, 1), + warnings: [], + } as any; + }, + }); + const agent = makeAgentWith(model); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7 }); + await agent.sendMessage("bad call"); + const step = agent.getAllSteps()[1]; + expect(step.toolResults).toEqual([]); + expect(urls.some(u => u.includes("/api/execution/"))).toBe(false); + agent.destroy(); + }); + + test("a rejected modification suppresses the follow-up execution", async () => { + // A modification the validator rejected did not change anything, so executing afterwards would run + // the old workflow and report it as the result of the change. + dispatch(okExec, srcOp("op-1", { fileName: "a.csv" })); + const saved = (WorkflowSystemMetadata as any).instance; + (WorkflowSystemMetadata as any).instance = undefined; + WorkflowSystemMetadata.getInstance().loadFromMetadata({ + operators: [ + { + operatorType: "CSVFileScan", + jsonSchema: { type: "object", properties: { fileName: { type: "string" } }, required: ["fileName"] }, + additionalMetadata: { userFriendlyName: "CSV", operatorDescription: "csv" }, + }, + ], + } as any); + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + if (n === 1) + return { + content: [ + { + type: "tool-call", + toolCallId: "c1", + toolName: "modifyOperator", + input: JSON.stringify({ operatorId: "op-1", properties: { fileName: 123 }, summary: "s" }), + }, + ], + finishReason: finish("tool-calls"), + usage: usage(1, 1), + warnings: [], + } as any; + return { + content: [{ type: "text", text: "d" }], + finishReason: finish("stop"), + usage: usage(1, 1), + warnings: [], + } as any; + }, + }); + const agent = makeAgentWith(model); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7 }); + await agent.sendMessage("bad props"); + const step = agent.getAllSteps()[1]; + expect(String(step.toolResults?.[0]?.output)).toStartWith("[ERROR]"); + expect(urls.some(u => u.includes("/api/execution/"))).toBe(false); + agent.destroy(); + (WorkflowSystemMetadata as any).instance = saved; + }); + + test("buildExecutionConfig projects the delegate config and live settings", async () => { + const agent = makeAgentWith(textModel("x")); + expect((agent as any).buildExecutionConfig()).toBeUndefined(); + (agent as any).delegateConfig = { userToken: "tok", workflowId: 5, computingUnitId: 2 }; + agent.updateSettings({ + executionTimeoutMs: 7000, + maxOperatorResultCharLimit: 11, + maxOperatorResultCellCharLimit: 13, + }); + expect((agent as any).buildExecutionConfig()).toEqual({ + userToken: "tok", + workflowId: 5, + computingUnitId: 2, + maxOperatorResultCharLimit: 11, + maxOperatorResultCellCharLimit: 13, + executionTimeoutMs: 7000, + }); + }); + + test("auto-persist coalesces a burst into one request under the delegate's name", async () => { + // The debounce is what keeps a burst of edits from becoming a burst of writes. + dispatch(okExec); + const agent = makeAgentWith(textModel("x")); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7, workflowName: "My Flow" }); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7, workflowName: "My Flow" }); + agent.getWorkflowState().addOperator(srcOp("o1") as any); + agent.getWorkflowState().addOperator(srcOp("o2") as any); + await new Promise(r => setTimeout(r, 700)); + const persists = fetchSpy.mock.calls.filter((c: any) => String(c[0]).includes("/api/workflow/persist")); + expect(persists.length).toBe(1); + expect(JSON.parse(persists[0][1].body).name).toBe("My Flow"); + expect(JSON.parse(JSON.parse(persists[0][1].body).content).operators.map((o: any) => o.operatorID)).toEqual([ + "o1", + "o2", + ]); + agent.destroy(); + }); + + test("a failed auto-persist is logged, not thrown", async () => { + const errs: any[] = []; + const agent = makeAgentWith(textModel("x")); + (agent as any).log = { error: (...a: any[]) => errs.push(a), debug: () => {}, warn: () => {}, info: () => {} }; + agent.setDelegateConfig({ userToken: "tok", workflowId: 7 }); + agent.getWorkflowState().addOperator(srcOp("o1") as any); + await new Promise(r => setTimeout(r, 700)); + expect(errs.length).toBe(1); + expect(errs[0][1]).toBe("failed to auto-persist workflow"); + agent.destroy(); + }); +}); From 80a8e5ce5dd2b1e59a05393e648a6addc08d3a1b Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Sun, 9 Aug 2026 18:47:04 -0700 Subject: [PATCH 2/2] test(agent-service): make spec cleanup survive failing assertions Per-test cleanup ran after the assertions, so a failing test skipped it and leaked state into the tests that follow: - agents built by makeAgentWith are now tracked and destroyed from the root afterEach, before the fetch spy is restored, so a pending auto-persist debounce can no longer fire a real request; the per-test destroy() calls are gone - the WorkflowSystemMetadata singleton swap and the two validator spies are restored in finally blocks --- agent-service/src/agent/texera-agent.spec.ts | 238 ++++++++++--------- 1 file changed, 126 insertions(+), 112 deletions(-) diff --git a/agent-service/src/agent/texera-agent.spec.ts b/agent-service/src/agent/texera-agent.spec.ts index 93c9500c401..416fee5444f 100644 --- a/agent-service/src/agent/texera-agent.spec.ts +++ b/agent-service/src/agent/texera-agent.spec.ts @@ -328,9 +328,15 @@ const textModel = (text: string, i = 11, o = 7) => }), }); -/** Sibling of makeAgent() that takes a stand-in model, so sendMessage runs with no network. */ +/** Sibling of makeAgent() that takes a stand-in model, so sendMessage runs with no network. + * Every agent built here is tracked and destroyed from `afterEach` rather than per test: a + * failing assertion would skip an in-test `destroy()`, and a pending auto-persist debounce + * could then fire after the fetch spy is restored and issue a real request. */ +const liveAgents: TexeraAgent[] = []; function makeAgentWith(model: any): TexeraAgent { - return new TexeraAgent({ model, modelType: "test-model", agentId: "agent-1", systemPrompt: "SYS-XYZ" }); + const agent = new TexeraAgent({ model, modelType: "test-model", agentId: "agent-1", systemPrompt: "SYS-XYZ" }); + liveAgents.push(agent); + return agent; } /** A source operator. `inputPorts: []` matters — a non-empty one fails validateOperatorConnection @@ -360,7 +366,11 @@ beforeEach(() => { throw new Error("unexpected fetch"); }) as any); }); -afterEach(() => fetchSpy.mockRestore()); +afterEach(() => { + // Destroy before restoring the spy — destruction is what cancels a pending auto-persist. + for (const agent of liveAgents.splice(0)) agent.destroy(); + fetchSpy.mockRestore(); +}); /** * sendMessage was the largest untested region in the file. Everything it needs is in-process: @@ -782,11 +792,10 @@ describe("sendMessage", () => { /** * With a delegate config the agent talks to the backend, so these drive it through `fetch`. * - * Two traps. Setting a delegate config makes the first turn refresh from the backend, and that + * One trap. Setting a delegate config makes the first turn refresh from the backend, and that * refresh replaces the whole workflow — so operators have to arrive through `dispatch`'s stub - * rather than being seeded on the agent. And every test here ends with `agent.destroy()`, - * because the auto-persist debounce would otherwise fire after the fetch spy is restored and - * issue a real request. + * rather than being seeded on the agent. Destruction, which is what cancels the auto-persist + * debounce, is centralized in the root `afterEach` so it runs even when a test fails mid-way. */ describe("delegate mode", () => { const wfBody = (ops: any[]) => ({ @@ -848,7 +857,6 @@ describe("delegate mode", () => { expect(agent.getAllSteps()[0].beforeWorkflowContent?.operators.length).toBe(1); await agent.sendMessage("two"); expect(retrieves()).toBe(1); - agent.destroy(); }); test("a failed refresh is swallowed", async () => { @@ -868,7 +876,6 @@ describe("delegate mode", () => { .getAllOperators() .map((o: any) => o.operatorID) ).toEqual(["local-op"]); - agent.destroy(); }); test("auto-executes after modifyOperator and keys the result at the agent step", async () => { @@ -876,41 +883,45 @@ describe("delegate mode", () => { const vSpy = spyOn(WorkflowSystemMetadata.getInstance(), "validateOperatorProperties").mockReturnValue({ isValid: true, } as any); - let n = 0; - const model = new MockLanguageModelV4({ - doGenerate: async () => { - n++; - if (n === 1) + try { + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + if (n === 1) + return { + content: [ + { + type: "tool-call", + toolCallId: "c1", + toolName: "modifyOperator", + input: JSON.stringify({ operatorId: "op-1", summary: "renamed" }), + }, + ], + finishReason: finish("tool-calls"), + usage: usage(1, 1), + warnings: [], + } as any; return { - content: [ - { - type: "tool-call", - toolCallId: "c1", - toolName: "modifyOperator", - input: JSON.stringify({ operatorId: "op-1", summary: "renamed" }), - }, - ], - finishReason: finish("tool-calls"), + content: [{ type: "text", text: "d" }], + finishReason: finish("stop"), usage: usage(1, 1), warnings: [], } as any; - return { - content: [{ type: "text", text: "d" }], - finishReason: finish("stop"), - usage: usage(1, 1), - warnings: [], - } as any; - }, - }); - const agent = makeAgentWith(model); - agent.setDelegateConfig({ userToken: "tok", workflowId: 7, workflowName: "w" }); - await agent.sendMessage("modify it"); - const steps = agent.getAllSteps(); - const txt2 = (model as any).doGenerateCalls[1].prompt[1].content[0].text; - expect(urls.some(u => u.includes("/api/execution/7/0/run"))).toBe(true); - expect((agent.getWorkflowResultState() as any).get("op-1").stepId).toBe(steps[1].id); - vSpy.mockRestore(); - agent.destroy(); + }, + }); + const agent = makeAgentWith(model); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7, workflowName: "w" }); + await agent.sendMessage("modify it"); + const steps = agent.getAllSteps(); + const txt2 = (model as any).doGenerateCalls[1].prompt[1].content[0].text; + expect(urls.some(u => u.includes("/api/execution/7/0/run"))).toBe(true); + expect((agent.getWorkflowResultState() as any).get("op-1").stepId).toBe(steps[1].id); + } finally { + // A leaked always-valid stub would let the rejected-modification test below pass validation + // and execute, so the restore has to survive a failed assertion. + vSpy.mockRestore(); + } }); test("executeOperator tool keys its result at the current head (the user step)", async () => { @@ -920,40 +931,42 @@ describe("delegate mode", () => { const vSpy = spyOn(WorkflowSystemMetadata.getInstance(), "validateOperatorProperties").mockReturnValue({ isValid: true, } as any); - let n = 0; - const model = new MockLanguageModelV4({ - doGenerate: async () => { - n++; - if (n === 1) + try { + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + if (n === 1) + return { + content: [ + { + type: "tool-call", + toolCallId: "c1", + toolName: "executeOperator", + input: JSON.stringify({ operatorId: "op-1" }), + }, + ], + finishReason: finish("tool-calls"), + usage: usage(1, 1), + warnings: [], + } as any; return { - content: [ - { - type: "tool-call", - toolCallId: "c1", - toolName: "executeOperator", - input: JSON.stringify({ operatorId: "op-1" }), - }, - ], - finishReason: finish("tool-calls"), + content: [{ type: "text", text: "d" }], + finishReason: finish("stop"), usage: usage(1, 1), warnings: [], } as any; - return { - content: [{ type: "text", text: "d" }], - finishReason: finish("stop"), - usage: usage(1, 1), - warnings: [], - } as any; - }, - }); - const agent = makeAgentWith(model); - agent.setDelegateConfig({ userToken: "tok", workflowId: 7, workflowName: "w" }); - await agent.sendMessage("run it"); - const steps = agent.getAllSteps(); - expect(urls.filter(u => u.includes("/api/execution/")).length).toBe(1); - expect((agent.getWorkflowResultState() as any).get("op-1").stepId).toBe(steps[0].id); - vSpy.mockRestore(); - agent.destroy(); + }, + }); + const agent = makeAgentWith(model); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7, workflowName: "w" }); + await agent.sendMessage("run it"); + const steps = agent.getAllSteps(); + expect(urls.filter(u => u.includes("/api/execution/")).length).toBe(1); + expect((agent.getWorkflowResultState() as any).get("op-1").stepId).toBe(steps[0].id); + } finally { + vSpy.mockRestore(); + } }); test("a tool call missing operatorId does not trigger a whole-workflow run", async () => { @@ -985,7 +998,6 @@ describe("delegate mode", () => { const step = agent.getAllSteps()[1]; expect(step.toolResults).toEqual([]); expect(urls.some(u => u.includes("/api/execution/"))).toBe(false); - agent.destroy(); }); test("a rejected modification suppresses the follow-up execution", async () => { @@ -994,49 +1006,53 @@ describe("delegate mode", () => { dispatch(okExec, srcOp("op-1", { fileName: "a.csv" })); const saved = (WorkflowSystemMetadata as any).instance; (WorkflowSystemMetadata as any).instance = undefined; - WorkflowSystemMetadata.getInstance().loadFromMetadata({ - operators: [ - { - operatorType: "CSVFileScan", - jsonSchema: { type: "object", properties: { fileName: { type: "string" } }, required: ["fileName"] }, - additionalMetadata: { userFriendlyName: "CSV", operatorDescription: "csv" }, - }, - ], - } as any); - let n = 0; - const model = new MockLanguageModelV4({ - doGenerate: async () => { - n++; - if (n === 1) + try { + WorkflowSystemMetadata.getInstance().loadFromMetadata({ + operators: [ + { + operatorType: "CSVFileScan", + jsonSchema: { type: "object", properties: { fileName: { type: "string" } }, required: ["fileName"] }, + additionalMetadata: { userFriendlyName: "CSV", operatorDescription: "csv" }, + }, + ], + } as any); + let n = 0; + const model = new MockLanguageModelV4({ + doGenerate: async () => { + n++; + if (n === 1) + return { + content: [ + { + type: "tool-call", + toolCallId: "c1", + toolName: "modifyOperator", + input: JSON.stringify({ operatorId: "op-1", properties: { fileName: 123 }, summary: "s" }), + }, + ], + finishReason: finish("tool-calls"), + usage: usage(1, 1), + warnings: [], + } as any; return { - content: [ - { - type: "tool-call", - toolCallId: "c1", - toolName: "modifyOperator", - input: JSON.stringify({ operatorId: "op-1", properties: { fileName: 123 }, summary: "s" }), - }, - ], - finishReason: finish("tool-calls"), + content: [{ type: "text", text: "d" }], + finishReason: finish("stop"), usage: usage(1, 1), warnings: [], } as any; - return { - content: [{ type: "text", text: "d" }], - finishReason: finish("stop"), - usage: usage(1, 1), - warnings: [], - } as any; - }, - }); - const agent = makeAgentWith(model); - agent.setDelegateConfig({ userToken: "tok", workflowId: 7 }); - await agent.sendMessage("bad props"); - const step = agent.getAllSteps()[1]; - expect(String(step.toolResults?.[0]?.output)).toStartWith("[ERROR]"); - expect(urls.some(u => u.includes("/api/execution/"))).toBe(false); - agent.destroy(); - (WorkflowSystemMetadata as any).instance = saved; + }, + }); + const agent = makeAgentWith(model); + agent.setDelegateConfig({ userToken: "tok", workflowId: 7 }); + await agent.sendMessage("bad props"); + const step = agent.getAllSteps()[1]; + expect(String(step.toolResults?.[0]?.output)).toStartWith("[ERROR]"); + expect(urls.some(u => u.includes("/api/execution/"))).toBe(false); + } finally { + // The swap must be undone even on a failed assertion, or the stub metadata leaks into + // every later test that touches the singleton. + (WorkflowSystemMetadata as any).instance = saved; + } }); test("buildExecutionConfig projects the delegate config and live settings", async () => { @@ -1074,7 +1090,6 @@ describe("delegate mode", () => { "o1", "o2", ]); - agent.destroy(); }); test("a failed auto-persist is logged, not thrown", async () => { @@ -1086,6 +1101,5 @@ describe("delegate mode", () => { await new Promise(r => setTimeout(r, 700)); expect(errs.length).toBe(1); expect(errs[0][1]).toBe("failed to auto-persist workflow"); - agent.destroy(); }); });