This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 561cd0ef1f test(agent-service): drive sendMessage with a stand-in
language model (#7487)
561cd0ef1f is described below
commit 561cd0ef1fb85da3f2f59fe846947d93f5c14262
Author: Xinyuan Lin <[email protected]>
AuthorDate: Mon Aug 10 22:00:57 2026 -0700
test(agent-service): drive sendMessage with a stand-in language model
(#7487)
### What changes were proposed in this PR?
`sendMessage` held 238 of the 292 uncovered lines in `texera-agent.ts`,
and the existing spec stops
at the model boundary. Nothing in there actually needs the network:
`ai/test` ships a
`MockLanguageModelV4` satisfying the same `LanguageModel` type the
constructor already takes, so the
loop runs in-process. `fetch` is spied on as a tripwire, and the
no-delegate tests assert that not a
single call escapes.
Adds 25 tests in two blocks:
| Block | Covers |
|---|---|
| `sendMessage` | branch bookkeeping and ancestor path, per-step and
summed usage, the assembled context replacing the raw message, tool
projection and rolling before/after snapshots, the `maxSteps` cap, turn
chaining, an abandoned branch staying invisible to the model, DAG
compilation feeding schemas and cached results into the prompt |
| `sendMessage` failures | a thrown model recorded as an error step, a
non-`Error` throw stringified, a failed turn staying on the branch,
cancellation reported as stopped, an `AbortError`-named provider error
read as a user stop, `stop()` mid-run preventing the next call,
`GENERATING` for the duration |
| `delegate mode` | the one-time backend refresh and a failed refresh
being swallowed, auto-execution after `modifyOperator` and where its
result is keyed, the two guards that suppress it,
`buildExecutionConfig`, and the debounced auto-persist plus its failure
path |
`texera-agent.ts` goes from **51.01% to 99.82% lines** (80.43% → 98.21%
funcs).
### On what these tests actually pin
Line coverage overstates this, and it is worth being precise. `bun`
credits a whole function body
once it is entered, so the first test alone takes the uncovered count
from 292 to 27. Most of the
remaining 24 buy no additional lines — they exist because each kills a
mutation nothing else kills.
They are mutation guards, not coverage.
The matrix ran 44 mutations against a pristine source with a
checkout-and-verify between each. Six
were re-run independently after the tests were merged into the existing
spec:
| Mutation | Expected | Result |
|---|---|---|
| a tool call without `operatorId` still auto-executes | red | red |
| an `[ERROR]` tool result no longer suppresses the follow-up run | red
| red |
| the `EXECUTE_AFTER_TOOLS` filter dropped | red | red |
| input tokens not mapped | red | red |
| `totalUsage` preferred over `usage` | **survives** | survives |
| the `content: text \|\| ""` fallback | **survives** | survives |
The last two are listed deliberately. They survive because they are
unobservable in `[email protected]` —
`totalUsage` and `usage` are the same object, and the SDK already hands
`""` to a text-less step.
No test claims to pin them, and no test was written to cement them.
Five further fragments are line-covered but not behaviourally pinned,
for the same reason:
`lastPreparedMessages = undefined` (re-assigned before every step),
`isError: !!(tr.output)?.error`
(no tool ever returns an object), the `?? finalUsage?.promptTokens` /
`?? completionTokens` arms
(v4-era key names that no longer exist), and the delegate guard at
408–410 (masked by the catch
below it).
### Deliberately not included
- **`getStepsById` (line 261)** — the one line left uncovered. No call
site anywhere in the repo;
`server.ts` uses `getReActSteps` / `getAllSteps` /
`getVisibleReActSteps`. It also hands out the
live private `Map` by reference. Deleting it beats testing it, and that
belongs in its own change.
- **`currentMessageId`** — five writes, zero reads. An assertion on it
would cement dead state.
- **`maxSteps: 0`** — writing that test hangs the suite rather than
failing it. Filed as #7484.
Two defects surfaced while writing these and are filed rather than fixed
here, since this PR touches no production code: #7484 (a `maxSteps` of 0
silently disables the step cap) and #7485 (a falsy throw from the model
makes `sendMessage` reject instead of reporting an error step).
No production file is touched.
### Any related issues, documentation, discussions?
Closes #7486
### How was this PR tested?
```
bun test
```
```
253 pass
0 fail
```
25 new on top of the existing 228. `bun run typecheck` passes.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---
agent-service/src/agent/texera-agent.spec.ts | 808 ++++++++++++++++++++++++++-
1 file changed, 807 insertions(+), 1 deletion(-)
diff --git a/agent-service/src/agent/texera-agent.spec.ts
b/agent-service/src/agent/texera-agent.spec.ts
index 27aa37b813..416fee5444 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,807 @@ 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.
+ * 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 {
+ 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
+ * 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(() => {
+ // 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:
+ * `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`.
+ *
+ * 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. 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[]) => ({
+ 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);
+ });
+
+ 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"]);
+ });
+
+ 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);
+ 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: "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);
+ } 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 () => {
+ // 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);
+ 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: "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);
+ } finally {
+ vSpy.mockRestore();
+ }
+ });
+
+ 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);
+ });
+
+ 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;
+ 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: "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);
+ } 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 () => {
+ 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",
+ ]);
+ });
+
+ 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");
+ });
+});
