aglinxinyuan commented on code in PR #7487:
URL: https://github.com/apache/texera/pull/7487#discussion_r3755274047
##########
agent-service/src/agent/texera-agent.spec.ts:
##########
@@ -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<string, any> = {}) => ({
+ 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<void>(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.
Review Comment:
Both are fixed in 80a8e5ce5d, and you were right about the failure mode —
per-test cleanup is exactly the wrong place for it.
Agents built by the helper are now tracked in a module-level list and
destroyed from the root `afterEach`, before `fetchSpy.mockRestore()`, since
destruction is what cancels the pending debounce. The metadata-singleton swap
is wrapped in `try/finally` so the original instance is restored even on a
failed assertion.
Verified rather than assumed: I injected a deliberate
`expect("PROBE").toBe("FORCE-FAIL")` at the top of the auto-persist test — the
point where a debounce is pending and the old in-test `destroy()` would have
been skipped — and re-ran the suite. Exactly one test fails (the one I broke),
252 pass, and there are zero `unexpected fetch` errors, so nothing escaped to
the restored spy and nothing cascaded.
I also re-ran the mutation sample afterwards, since a 126/112 restructure of
a spec can quietly cost it its discriminating power. All six outcomes are
unchanged, including the two that are expected to survive because they are
unobservable in `[email protected]`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
