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 a61e1ee4a8 test(agent-service): pin the context serializer and the
execution tools (#7505)
a61e1ee4a8 is described below
commit a61e1ee4a8b08f196cf509171d6372f7ab5895b8
Author: Xinyuan Lin <[email protected]>
AuthorDate: Mon Aug 10 09:43:43 2026 -0700
test(agent-service): pin the context serializer and the execution tools
(#7505)
### What changes were proposed in this PR?
`context-utils.ts` builds the prompt the agent reasons over and
`workflow-execution-tools.ts` turns
an execution result into the table it reads back. Both looked reasonably
covered and were not:
`bun` credits a whole function body once entered, so blocks inside
`jsonToTableFormat` and
`executeOperatorAndFormat` counted as covered while nothing asserted
them.
Adds 15 tests across the two existing specs (no new spec files — one
spec per source class, both
already existed).
| File | Before | After |
|---|---|---|
| `context-utils.ts` | 60.00% funcs / 77.83% lines | **100% / 100%** |
| `workflow-execution-tools.ts` | 58.97% funcs / 91.24% lines | 85.71% /
95.16% |
**Ten of the killed mutations target lines `bun` already called
covered** — the DAG topological
ordering and its target-rank tie-break, port-ordinal mapping, the
`NULL`/`null`/`undefined`/object
cell rendering, the row-gap ellipsis, the leading-tab header, and the
schema-violation messages.
The file percentage barely moves for those; the pinning is the point.
### Verification
45 mutations were applied to the production files and reverted, each
revert confirmed with
`git diff --quiet` before the next. All 45 turned the suite red.
An adversarial re-check then found **two assertions that were vacuous
anyway**, which is the part
worth reporting:
| Survivor | Why it passed | Fix |
|---|---|---|
| `Math.ceil` → `Math.round` on the execution timeout | the fixture was
`4500 ms`, and `ceil(4.5) == round(4.5) == 5`, so the assertion
commented "rounds up" proved nothing | fixture changed to `4200 ms`,
which separates ceil (5) from round and floor (4) |
| `getConfig()` hoisted out of the `execute` closure | the test invoked
the tool once, so `toHaveBeenCalledTimes(1)` holds whether the config is
resolved per invocation or captured once at construction | invoke twice
with the workflow id changing in between, and assert the second request
URL reflects the second config |
Both mutations are now red, and so is `Math.floor`. Production diff
empty.
### Deliberately not included
Three regions of `workflow-execution-tools.ts` are left uncovered
because they are dead, confirmed
by inserting `throw new Error(...)` at the top of each and running the
whole 273-test module — all
still passed, so nothing reaches them:
- `formatWorkflowValidationErrors` (169–178) — no call site.
- lines 221–226 and 229–233.
Testing them would cement code that should be deleted instead.
No production file is touched.
### Any related issues, documentation, discussions?
Closes #7504
### How was this PR tested?
```
bun test
```
```
273 pass
0 fail
```
15 new on top of the existing 258. `bun run typecheck` and `bun run
format:check` both pass.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Co-authored-by: Meng Wang <[email protected]>
---
.../agent/tools/workflow-execution-tools.spec.ts | 240 ++++++++++++++++++++-
agent-service/src/agent/util/context-utils.spec.ts | 145 ++++++++++++-
2 files changed, 380 insertions(+), 5 deletions(-)
diff --git a/agent-service/src/agent/tools/workflow-execution-tools.spec.ts
b/agent-service/src/agent/tools/workflow-execution-tools.spec.ts
index f3623e2a4d..36313f181a 100644
--- a/agent-service/src/agent/tools/workflow-execution-tools.spec.ts
+++ b/agent-service/src/agent/tools/workflow-execution-tools.spec.ts
@@ -18,13 +18,17 @@
*/
import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from
"bun:test";
-import { executeOperatorAndFormat, type ExecutionConfig } from
"./workflow-execution-tools";
+import { createExecuteOperatorTool, executeOperatorAndFormat, type
ExecutionConfig } from "./workflow-execution-tools";
import { WorkflowState } from "../workflow-state";
import { WorkflowSystemMetadata } from "../util/workflow-system-metadata";
-import type { OperatorPredicate, PortDescription } from "../../types/workflow";
+import type { OperatorLink, OperatorPredicate, PortDescription } from
"../../types/workflow";
import type { OperatorInfo, SyncExecutionResult } from "../../types/execution";
-function makeOperator(id: string, inputPorts: PortDescription[] = []):
OperatorPredicate {
+function makeOperator(
+ id: string,
+ inputPorts: PortDescription[] = [],
+ overrides: Partial<OperatorPredicate> = {}
+): OperatorPredicate {
return {
operatorID: id,
operatorType: "TestOp",
@@ -33,6 +37,15 @@ function makeOperator(id: string, inputPorts:
PortDescription[] = []): OperatorP
inputPorts,
outputPorts: [],
showAdvanced: false,
+ ...overrides,
+ };
+}
+
+function makeLink(source: string, sourcePort: string, target: string,
targetPort: string): OperatorLink {
+ return {
+ linkID: `${source}.${sourcePort}->${target}.${targetPort}`,
+ source: { operatorID: source, portID: sourcePort },
+ target: { operatorID: target, portID: targetPort },
};
}
@@ -42,6 +55,11 @@ function stateWith(...operators: OperatorPredicate[]):
WorkflowState {
return state;
}
+// The JSON body of the nth fetch the code under test issued.
+function requestBody(spy: ReturnType<typeof spyOn>, callIndex = 0): any {
+ return JSON.parse((spy.mock.calls[callIndex][1] as RequestInit).body as
string);
+}
+
function cfg(overrides: Partial<ExecutionConfig> = {}): ExecutionConfig {
return { userToken: "tok", workflowId: 1, ...overrides };
}
@@ -90,6 +108,30 @@ describe("executeOperatorAndFormat — guards & validation",
() => {
expect(result).toBe("[ERROR] Operator op1:\n - inputs: input-0 requires
at least 1 input, has 0.");
expect(fetchSpy).not.toHaveBeenCalled();
});
+
+ test("merges schema and connection violations into one blocking message",
async () => {
+ validateSpy.mockReturnValue({ isValid: false, messages: { limit: "must be
a number" } });
+ const state = stateWith(makeOperator("op1", [{ portID: "input-0" }]));
+
+ const result = await executeOperatorAndFormat(state, cfg(), "op1");
+
+ expect(result).toBe(
+ "[ERROR] Operator op1:\n - limit: must be a number\n - inputs: input-0
requires at least 1 input, has 0."
+ );
+ expect(fetchSpy).not.toHaveBeenCalled();
+ });
+
+ test("rejects a single-input port that has more than one incoming link,
naming it by display name", async () => {
+ const dst = makeOperator("dst", [{ portID: "input-0", displayName: "Left",
disallowMultiInputs: true }]);
+ const state = stateWith(makeOperator("a"), makeOperator("b"), dst);
+ state.addLink(makeLink("a", "output-0", "dst", "input-0"));
+ state.addLink(makeLink("b", "output-0", "dst", "input-0"));
+
+ const result = await executeOperatorAndFormat(state, cfg(), "dst");
+
+ expect(result).toBe("[ERROR] Operator dst:\n - inputs: Left requires 1
input, has 2.");
+ expect(fetchSpy).not.toHaveBeenCalled();
+ });
});
describe("executeOperatorAndFormat — execution-level failures", () => {
@@ -277,6 +319,198 @@ describe("executeOperatorAndFormat — operator result
handling", () => {
});
});
+describe("executeOperatorAndFormat — request construction", () => {
+ test("sends the upstream sub-DAG with port ordinals resolved from each
operator's port list", async () => {
+ const src = makeOperator("src", [], {
+ operatorProperties: { limit: 3 },
+ outputPorts: [{ portID: "output-0" }, { portID: "output-1" }],
+ });
+ const dst = makeOperator("dst", [{ portID: "input-0" }, { portID:
"input-1" }]);
+ const state = stateWith(src, dst);
+ state.addLink(makeLink("src", "output-1", "dst", "input-0"));
+ // An unknown source port falls back to ordinal 0 rather than -1.
+ state.addLink(makeLink("src", "output-9", "dst", "input-1"));
+ resolveFetch(fetchSpy, {
+ success: true,
+ state: "Completed",
+ operators: {
+ dst: { state: "Completed", inputTuples: 0, outputTuples: 1,
resultMode: "table", result: [{ a: 1 }] },
+ },
+ });
+
+ await executeOperatorAndFormat(state, cfg({ workflowId: 7,
computingUnitId: 3, executionTimeoutMs: 4200 }), "dst");
+
+
expect(String(fetchSpy.mock.calls[0][0])).toBe("http://localhost:8085/api/execution/7/3/run";);
+ expect((fetchSpy.mock.calls[0][1] as RequestInit).headers).toMatchObject({
Authorization: "Bearer tok" });
+
+ const body = requestBody(fetchSpy);
+ expect(body.executionName).toBe("agent-execution");
+ // 4200 rather than a round 4500: with 4500 the assertion holds for ceil,
round *and*
+ // floor-plus-one, so it would not notice the rounding being changed. 4200
separates them.
+ expect(body.timeoutSeconds).toBe(5);
+ expect(body.targetOperatorIds).toEqual(["dst"]);
+ expect(body.logicalPlan.opsToViewResult).toEqual(["dst"]);
+ expect(body.logicalPlan.links).toEqual([
+ {
+ fromOpId: "src",
+ fromPortId: { id: 1, internal: false },
+ toOpId: "dst",
+ toPortId: { id: 0, internal: false },
+ },
+ {
+ fromOpId: "src",
+ fromPortId: { id: 0, internal: false },
+ toOpId: "dst",
+ toPortId: { id: 1, internal: false },
+ },
+ ]);
+ // Operator properties are flattened into the wire operator alongside its
ports.
+ expect(body.logicalPlan.operators).toContainEqual(
+ expect.objectContaining({ operatorID: "src", operatorType: "TestOp",
limit: 3 })
+ );
+ });
+});
+
+describe("executeOperatorAndFormat — result rendering", () => {
+ test("labels input shapes with the upstream operator id, ordered by port
index", async () => {
+ const dst = makeOperator("dst", [{ portID: "input-0" }, { portID:
"input-1" }]);
+ const state = stateWith(makeOperator("a"), makeOperator("b"), dst);
+ state.addLink(makeLink("a", "output-0", "dst", "input-0"));
+ state.addLink(makeLink("b", "output-0", "dst", "input-1"));
+ resolveFetch(fetchSpy, {
+ success: true,
+ state: "Completed",
+ operators: {
+ dst: {
+ state: "Completed",
+ inputTuples: 0,
+ outputTuples: 99,
+ resultMode: "table",
+ totalRowCount: 4,
+ // Deliberately out of order to pin the sort by port index.
+ inputPortShapes: [
+ { portIndex: 1, rows: 20, columns: 2 },
+ { portIndex: 0, rows: 5, columns: 3 },
+ ],
+ result: [{ x: 1 }],
+ },
+ },
+ });
+
+ const result = await executeOperatorAndFormat(state, cfg(), "dst");
+
+ expect(result).toContain("Input operator(table shape): a(5, 3), b(20, 2)");
+ expect(result).toContain("Output table shape: (4, 1)"); // totalRowCount
wins over outputTuples
+ });
+
+ test("falls back to outputTuples for the row count and appends operator
warnings", async () => {
+ const state = stateWith(makeOperator("op1"));
+ resolveFetch(fetchSpy, {
+ success: true,
+ state: "Completed",
+ operators: {
+ op1: {
+ state: "Completed",
+ inputTuples: 0,
+ outputTuples: 3,
+ resultMode: "table",
+ warnings: ["result truncated"],
+ result: [{ x: 1 }],
+ },
+ },
+ });
+
+ const result = await executeOperatorAndFormat(state, cfg(), "op1");
+
+ expect(result.split("\n").slice(0, 3)).toEqual([
+ "Executed operator op1",
+ "Output table shape: (3, 1)",
+ "result truncated",
+ ]);
+ });
+
+ test("renders backend row indices, inserting an ellipsis row where they
skip", async () => {
+ const state = stateWith(makeOperator("op1"));
+ resolveFetch(fetchSpy, {
+ success: true,
+ state: "Completed",
+ operators: {
+ op1: {
+ state: "Completed",
+ inputTuples: 0,
+ outputTuples: 4,
+ resultMode: "table",
+ totalRowCount: 4,
+ result: [
+ { __row_index__: 0, a: 1, b: "x" },
+ { __row_index__: 1, a: null, b: "NULL" },
+ { __row_index__: 10, a: true, b: "has\ttab\nand newline" },
+ { __row_index__: 11, a: { k: 1 }, b: undefined },
+ ],
+ },
+ },
+ });
+
+ const result = await executeOperatorAndFormat(state, cfg(), "op1");
+
+ expect(result.split("\n").slice(1)).toEqual([
+ "Output table shape: (4, 2)", // __row_index__ is internal and not
counted as a column
+ "\ta\tb",
+ "0\t1\tx",
+ "1\tNaN\tNaN",
+ "...\t...\t...",
+ "10\ttrue\thas\\ttab\\nand newline",
+ '11\t{"k":1}\t',
+ ]);
+ });
+
+ test("emits only the summary and shape line when the operator produced zero
rows", async () => {
+ const state = stateWith(makeOperator("op1"));
+ resolveFetch(fetchSpy, {
+ success: true,
+ state: "Completed",
+ operators: {
+ op1: { state: "Completed", inputTuples: 0, outputTuples: 0,
resultMode: "table", result: [] },
+ },
+ });
+
+ const result = await executeOperatorAndFormat(state, cfg(), "op1");
+
+ expect(result).toBe("Executed operator op1\nOutput table shape: (0, 0)");
+ });
+});
+
+describe("createExecuteOperatorTool", () => {
+ test("resolves the config per invocation and forwards the operator id and
onResult hook", async () => {
+ const state = stateWith(makeOperator("op1"));
+ resolveFetch(fetchSpy, {
+ success: true,
+ state: "Completed",
+ operators: {
+ op1: { state: "Completed", inputTuples: 0, outputTuples: 1,
resultMode: "table", result: [{ a: 1 }] },
+ },
+ });
+ const onResult = mock((_id: string, _info: OperatorInfo) => {});
+ let workflowId = 42;
+ const getConfig = mock(() => cfg({ workflowId }));
+
+ const executeTool = createExecuteOperatorTool(state, getConfig, onResult);
+ const output = await executeTool.execute!({ operatorId: "op1" }, {} as
any);
+ // Invoked twice, with the workflow id changing in between. One invocation
cannot tell
+ // per-invocation resolution from a config captured once when the tool was
built: the call
+ // count is 1 either way. The second URL is what proves the later config
is the one used.
+ workflowId = 43;
+ await executeTool.execute!({ operatorId: "op1" }, {} as any);
+
+ expect(getConfig).toHaveBeenCalledTimes(2);
+ expect(output).toContain("Executed operator op1");
+
expect(String(fetchSpy.mock.calls[0][0])).toContain("/api/execution/42/0/run");
+
expect(String(fetchSpy.mock.calls[1][0])).toContain("/api/execution/43/0/run");
+ expect(onResult).toHaveBeenCalledTimes(2);
+ expect(onResult.mock.calls[0][0]).toBe("op1");
+ });
+});
+
describe("executeOperatorAndFormat — cancellation", () => {
test("re-throws AbortError instead of formatting it as a result", async ()
=> {
const state = stateWith(makeOperator("op1"));
diff --git a/agent-service/src/agent/util/context-utils.spec.ts
b/agent-service/src/agent/util/context-utils.spec.ts
index 84af9df3bb..9ad90492d1 100644
--- a/agent-service/src/agent/util/context-utils.spec.ts
+++ b/agent-service/src/agent/util/context-utils.spec.ts
@@ -22,7 +22,8 @@ import type { ModelMessage } from "ai";
import { assembleContext } from "./context-utils";
import { WorkflowState } from "../workflow-state";
import type { ReActStep } from "../../types/agent";
-import type { OperatorPredicate } from "../../types/workflow";
+import type { OperatorLink, OperatorPredicate } from "../../types/workflow";
+import type { WorkflowCompilationResponse } from "../../api/compile-api";
function step(messageId: string, role: "user" | "agent", stepId: number,
content: string, isEnd: boolean): ReActStep {
return {
@@ -37,7 +38,7 @@ function step(messageId: string, role: "user" | "agent",
stepId: number, content
};
}
-function makeOperator(id: string): OperatorPredicate {
+function makeOperator(id: string, overrides: Partial<OperatorPredicate> = {}):
OperatorPredicate {
return {
operatorID: id,
operatorType: "TestOp",
@@ -46,6 +47,15 @@ function makeOperator(id: string): OperatorPredicate {
inputPorts: [],
outputPorts: [],
showAdvanced: false,
+ ...overrides,
+ };
+}
+
+function makeLink(source: string, sourcePort: string, target: string,
targetPort: string): OperatorLink {
+ return {
+ linkID: `${source}.${sourcePort}->${target}.${targetPort}`,
+ source: { operatorID: source, portID: sourcePort },
+ target: { operatorID: target, portID: targetPort },
};
}
@@ -100,4 +110,135 @@ describe("assembleContext", () => {
expect(content).toContain("### Turn 1");
expect(content).toContain("### Turn 2");
});
+
+ test("labels each tool call with the status of the result at the same
index", () => {
+ const agentStep: ReActStep = {
+ ...step("m1", "agent", 1, "acting", true),
+ toolCalls: [
+ { toolName: "addOperator", toolCallId: "c1", input: {} },
+ { toolName: "executeOperator", toolCallId: "c2", input: {} },
+ { toolName: "deleteOperator", toolCallId: "c3", input: {} },
+ ],
+ // Deliberately shorter than toolCalls: a call still awaiting its result
+ // must not be reported as failed.
+ toolResults: [
+ { toolCallId: "c1", output: "ok" },
+ { toolCallId: "c2", output: "bad", isError: true },
+ ],
+ };
+ const content = contentOf(
+ assembleContext([step("m1", "user", 0, "req", true), agentStep], new
WorkflowState(), new Map())
+ );
+ expect(content).toContain("- addOperator (succeeded)");
+ expect(content).toContain("- executeOperator (failed)");
+ expect(content).toContain("- deleteOperator (succeeded)");
+ });
+});
+
+describe("assembleContext — dataflow serialization", () => {
+ test("orders operators and links topologically rather than by insertion
order", () => {
+ const workflowState = new WorkflowState();
+ // Inserted leaf-first, so insertion order is the reverse of the dataflow
order.
+ workflowState.addOperator(makeOperator("c"));
+ workflowState.addOperator(makeOperator("b"));
+ workflowState.addOperator(makeOperator("a"));
+ workflowState.addLink(makeLink("b", "output-0", "c", "input-0"));
+ workflowState.addLink(makeLink("a", "output-0", "c", "input-1"));
+ workflowState.addLink(makeLink("a", "output-0", "b", "input-0"));
+
+ const content = contentOf(assembleContext([], workflowState, new Map()));
+
+ expect(content.indexOf("### Operator
`a`")).toBeLessThan(content.indexOf("### Operator `b`"));
+ expect(content.indexOf("### Operator
`b`")).toBeLessThan(content.indexOf("### Operator `c`"));
+ // Links are sorted by source rank, then by target rank — a → b precedes a
→ c.
+ const linkLines = content.slice(content.indexOf("##
Links")).split("\n").slice(1);
+ expect(linkLines).toEqual(["- a → b", "- a → c", "- b → c"]);
+ });
+
+ test("renders input schemas by port ordinal and falls through to the first
defined output schema", () => {
+ const workflowState = new WorkflowState();
+ workflowState.addOperator(makeOperator("src", { outputPorts: [{ portID:
"output-0" }] }));
+ workflowState.addOperator(makeOperator("dst", { inputPorts: [{ portID:
"input-0" }] }));
+ workflowState.addLink(makeLink("src", "output-0", "dst", "input-0"));
+
+ const compilationResult: WorkflowCompilationResponse = {
+ operatorOutputSchemas: {
+ src: {
+ "0_false": [
+ { attributeName: "a", attributeType: "string" },
+ { attributeName: "n", attributeType: "integer" },
+ ],
+ },
+ // Port 0 carries no schema, so the output line must come from port 1.
+ dst: { "0_false": undefined, "1_false": [{ attributeName: "flag",
attributeType: "boolean" }] },
+ },
+ operatorErrors: {},
+ };
+
+ const content = contentOf(assembleContext([], workflowState, new Map(),
false, compilationResult));
+
+ expect(content).toContain("Input Schema (port 0): [a: string, n:
integer]");
+ expect(content).toContain("Output Schema: [a: string, n: integer]");
+ expect(content).toContain("Output Schema: [flag: boolean]");
+ // The wire port id "0_false" is trimmed down to the bare ordinal.
+ expect(content).not.toContain("port 0_false");
+ });
+
+ test("renders non-empty properties and JSON-encodes non-string values", ()
=> {
+ const workflowState = new WorkflowState();
+ workflowState.addOperator(
+ makeOperator("op1", {
+ operatorProperties: { name: "alice", limit: 5, tags: ["x", "y"],
blank: "", missing: null, absent: undefined },
+ })
+ );
+
+ const content = contentOf(assembleContext([], workflowState, new Map()));
+
+ expect(content).toContain("Summary: op1"); // no display name -> falls
back to the id
+ expect(content).toContain(" name: alice");
+ expect(content).toContain(" limit: 5");
+ expect(content).toContain(' tags: ["x","y"]');
+ expect(content).not.toContain("blank");
+ expect(content).not.toContain("missing");
+ expect(content).not.toContain("absent");
+ });
+
+ test("omits properties under redaction unless the operator's result reports
an error", () => {
+ const workflowState = new WorkflowState();
+ workflowState.addOperator(makeOperator("op1", { operatorProperties: {
secret: "s3cr3t" } }));
+
+ const redacted = contentOf(assembleContext([], workflowState, new Map(),
true));
+ expect(redacted).toContain("(TestOp, not-executed)");
+ expect(redacted).not.toContain("Properties:");
+
+ // A failing operator keeps its properties even when redacting, so the
model
+ // can see what caused the failure.
+ const failed = contentOf(assembleContext([], workflowState, new
Map([["op1", "[ERROR] boom"]]), true));
+ expect(failed).toContain("(TestOp, failed)");
+ expect(failed).toContain(" secret: s3cr3t");
+ });
+
+ test("marks an operator executed and indents every line of its result", ()
=> {
+ const workflowState = new WorkflowState();
+ workflowState.addOperator(makeOperator("op1", { customDisplayName: "My
Filter" }));
+
+ const content = contentOf(assembleContext([], workflowState, new
Map([["op1", "header\nrow1"]])));
+
+ expect(content).toContain("### Operator `op1` (TestOp, executed)");
+ expect(content).toContain("Summary: My Filter");
+ expect(content).toContain("Result:\n header\n row1");
+ });
+
+ test("surfaces a per-operator compilation error", () => {
+ const workflowState = new WorkflowState();
+ workflowState.addOperator(makeOperator("op1"));
+ const compilationResult: WorkflowCompilationResponse = {
+ operatorOutputSchemas: {},
+ operatorErrors: { op1: { type: "CompilationError", message: "attribute x
not found" } },
+ };
+
+ const content = contentOf(assembleContext([], workflowState, new Map(),
false, compilationResult));
+
+ expect(content).toContain("Compilation Error: attribute x not found");
+ });
});
