This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7505-42d08a3701cd06542bcfa92728116302723b2536 in repository https://gitbox.apache.org/repos/asf/texera.git
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"); + }); });
