refactor(agent-service): redesign sync-execution result and error model

Restructure the per-operator summary the sync-execution backend returns and the
agent-service/frontend consume, for a leaner and consistent wire contract:

- Replace flat OperatorInfo with OperatorExecutionSummary: state, errorMessages,
  resultSummary?, consoleLogsSummary? (orthogonal sub-summaries; no shape stats).
- Rename SyncExecutionResult -> WorkflowExecutionSummary; drop compilationErrors
  (folded into errors). errors and per-op errorMessages are non-optional (empty
  means none).
- OperatorResultSummary.sampleTuples is now List[SampleRow] ({rowIndex, tuple})
  instead of a JSON array with an embedded __row_index__. Drop the table-shape
  types (TableShape/InputPortTableShape): the agent derives input-port shapes
  from the DAG + each upstream's output shape; output shape comes from the result
  summary.
- Reuse the engine's WorkflowFatalError for per-operator errors (the same type
  the compiling service returns for compilation errors), replacing the bespoke
  OperatorError so compile and execution errors share one wire shape.
- Collapse console messages onto one type; derive warnings from WARNING-titled
  messages rather than a separate field.
- Replace OperatorResultSummaryWs: the WS operatorResults payload now carries the
  canonical OperatorExecutionSummary; the frontend maps it to its flat display
  type (re-flattening sampleTuples to keep the display components unchanged).

Touches the Scala producer (SyncExecutionResource), the agent-service consumers
(result-formatting, workflow-execution-tools, workflow-result-state, server) and
the frontend WS mapping. Representation/type-level change; behavior preserved,
except input-port shape lines are now derived rather than explicitly rendered.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012qFkyrpTd5PrkNBPcBeo4Q

Author: bobbai00

agent-service/src/agent/texera-agent.ts

@@ -722,7 +722,7 @@ export class TexeraAgent {
     const result = new Map<string, string>();
     const visible = this.workflowResultState.getAllVisible();
     for (const [operatorId, entry] of visible) {
-      result.set(operatorId, formatOperatorResult(operatorId, entry.operatorInfo, this.workflowState));
+      result.set(operatorId, formatOperatorResult(operatorId, entry.operatorInfo));
     }
     return result;
   }

agent-service/src/agent/tools/result-formatting.spec.ts

@@ -19,89 +19,89 @@
 
 import { describe, expect, test } from "bun:test";
 import { formatOperatorResult } from "./result-formatting";
-import { WorkflowState } from "../workflow-state";
-import type { OperatorInfo } from "../../types/execution";
-import type { OperatorPredicate, OperatorLink, PortDescription } from "../../types/workflow";
-
-function makeOpInfo(overrides: Partial<OperatorInfo> = {}): OperatorInfo {
-  return {
-    state: "completed",
-    inputTuples: 0,
-    outputTuples: 0,
-    resultMode: "table",
-    ...overrides,
-  };
+import type { OperatorExecutionSummary, OperatorState, SampleRow } from "../../types/execution";
+
+// Convert flat test rows (with an optional embedded __row_index__) into the
+// structured SampleRow[] the summary now carries.
+function toSampleRows(rows: Record<string, any>[]): SampleRow[] {
+  return rows.map((row, i) => {
+    const { __row_index__, ...tuple } = row;
+    return { rowIndex: typeof __row_index__ === "number" ? __row_index__ : i, tuple };
+  });
 }
 
-function makeOperator(id: string, inputPortIDs: string[] = []): OperatorPredicate {
-  const inputPorts: PortDescription[] = inputPortIDs.map((portID, i) => ({
-    portID,
-    displayName: `Input ${i}`,
-  }));
-  return {
-    operatorID: id,
-    operatorType: "TestOp",
-    operatorVersion: "1.0",
-    operatorProperties: {},
-    inputPorts,
-    outputPorts: [{ portID: "output-0", displayName: "Output 0" }],
-    showAdvanced: false,
-  };
+// Test convenience: accept the (old) flat fields and assemble the structured
+// OperatorExecutionSummary, so the cases below stay terse.
+interface OpInfoOverrides {
+  state?: OperatorState;
+  error?: string;
+  outputTuples?: number;
+  totalRowCount?: number;
+  warnings?: string[];
+  result?: Record<string, any>[];
 }
 
-function makeLink(linkID: string, source: [string, string], target: [string, string]): OperatorLink {
-  return {
-    linkID,
-    source: { operatorID: source[0], portID: source[1] },
-    target: { operatorID: target[0], portID: target[1] },
+function makeOpInfo(overrides: OpInfoOverrides = {}): OperatorExecutionSummary {
+  const summary: OperatorExecutionSummary = {
+    state: overrides.state ?? "Completed",
+    errorMessages: overrides.error ? [{ type: "EXECUTION_FAILURE", message: overrides.error }] : [],
   };
+  // The result summary is present only when the operator produced a result.
+  if (overrides.result !== undefined) {
+    summary.resultSummary = {
+      resultMode: "table",
+      // Non-arrays are passed through to exercise the "(no result data)" guard.
+      sampleTuples: Array.isArray(overrides.result) ? toSampleRows(overrides.result) : (overrides.result as any),
+      totalRowCount: overrides.totalRowCount ?? overrides.outputTuples ?? 0,
+    };
+  }
+  if (overrides.warnings) {
+    // Warnings are derived from console messages whose title is "WARNING: ...".
+    summary.consoleLogsSummary = {
+      messages: overrides.warnings.map(w => ({ msgType: "PRINT", title: w, message: "" })),
+    };
+  }
+  return summary;
 }
 
-const EMPTY_STATE = new WorkflowState();
-
 describe("formatOperatorResult - early returns", () => {
   test("returns [ERROR] prefix when error field is set", () => {
-    const out = formatOperatorResult("op1", makeOpInfo({ error: "boom" }), EMPTY_STATE);
+    const out = formatOperatorResult("op1", makeOpInfo({ error: "boom" }));
     expect(out).toBe("[ERROR] boom");
   });
 
   test("treats empty-string error as falsy and continues to result path", () => {
-    const out = formatOperatorResult("op1", makeOpInfo({ error: "" }), EMPTY_STATE);
+    const out = formatOperatorResult("op1", makeOpInfo({ error: "" }));
     expect(out).not.toContain("[ERROR]");
     expect(out).toContain("(no result data)");
   });
 
   test("returns (no result data) when result is undefined", () => {
-    const out = formatOperatorResult("op1", makeOpInfo(), EMPTY_STATE);
+    const out = formatOperatorResult("op1", makeOpInfo());
     expect(out).toBe("(no result data)");
   });
 
   test("returns (no result data) when result is not an array", () => {
-    const out = formatOperatorResult(
-      "op1",
-      makeOpInfo({ result: { rows: [] } as unknown as Record<string, any>[] }),
-      EMPTY_STATE
-    );
+    const out = formatOperatorResult("op1", makeOpInfo({ result: { rows: [] } as unknown as Record<string, any>[] }));
     expect(out).toBe("(no result data)");
   });
 
   test("empty array result emits brief summary plus zero-column shape only", () => {
-    const out = formatOperatorResult("op1", makeOpInfo({ result: [], outputTuples: 0 }), EMPTY_STATE);
+    const out = formatOperatorResult("op1", makeOpInfo({ result: [], outputTuples: 0 }));
     expect(out.split("\n")).toEqual(["Executed operator op1", "Output table shape: (0, 0)"]);
   });
 });
 
 describe("formatOperatorResult - table shape and metadata", () => {
   test("uses outputTuples for row count when totalRowCount missing", () => {
-    const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 7, result: [{ a: 1, b: 2 }] }), EMPTY_STATE);
+    const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 7, result: [{ a: 1, b: 2 }] }));
     expect(out).toContain("Output table shape: (7, 2)");
   });
 
   test("totalRowCount overrides outputTuples in output shape", () => {
     const out = formatOperatorResult(
       "op1",
-      makeOpInfo({ outputTuples: 7, totalRowCount: 999, result: [{ a: 1, b: 2 }] }),
-      EMPTY_STATE
+      makeOpInfo({ outputTuples: 7, totalRowCount: 999, result: [{ a: 1, b: 2 }] })
     );
     expect(out).toContain("Output table shape: (999, 2)");
   });
@@ -112,8 +112,7 @@ describe("formatOperatorResult - table shape and metadata", () => {
       makeOpInfo({
         outputTuples: 1,
         result: [{ __is_visualization__: true, "html-content": "<x/>" }],
-      }),
-      EMPTY_STATE
+      })
     );
     // 1 visible column ("html-content") since __is_visualization__ is filtered.
     expect(out).toContain("Output table shape: (1, 1)");
@@ -125,85 +124,14 @@ describe("formatOperatorResult - table shape and metadata", () => {
       makeOpInfo({
         outputTuples: 1,
         result: [{ a: 1 }],
-        warnings: ["truncated to 1 row", "something else"],
-      }),
-      EMPTY_STATE
+        warnings: ["WARNING: truncated to 1 row", "WARNING: something else"],
+      })
     );
     const lines = out.split("\n");
     expect(lines[0]).toBe("Executed operator op1");
     expect(lines[1]).toBe("Output table shape: (1, 1)");
-    expect(lines[2]).toBe("truncated to 1 row");
-    expect(lines[3]).toBe("something else");
-  });
-});
-
-describe("formatOperatorResult - input port metadata", () => {
-  test("omits input metadata when inputPortShapes is missing", () => {
-    const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 1, result: [{ a: 1 }] }), EMPTY_STATE);
-    expect(out).not.toContain("Input operator");
-  });
-
-  test("omits input metadata when inputPortShapes is empty", () => {
-    const out = formatOperatorResult(
-      "op1",
-      makeOpInfo({ outputTuples: 1, result: [{ a: 1 }], inputPortShapes: [] }),
-      EMPTY_STATE
-    );
-    expect(out).not.toContain("Input operator");
-  });
-
-  test("falls back to inputN placeholder when no upstream link matches the port", () => {
-    const out = formatOperatorResult(
-      "op1",
-      makeOpInfo({
-        outputTuples: 1,
-        result: [{ a: 1 }],
-        inputPortShapes: [{ portIndex: 0, rows: 5, columns: 3 }],
-      }),
-      EMPTY_STATE
-    );
-    expect(out).toContain("Input operator(table shape): input0(5, 3)");
-  });
-
-  test("uses upstream operator id when an input link matches the port", () => {
-    const state = new WorkflowState();
-    state.addOperator(makeOperator("upstream"));
-    state.addOperator(makeOperator("op1", ["input-0"]));
-    state.addLink(makeLink("l1", ["upstream", "output-0"], ["op1", "input-0"]));
-
-    const out = formatOperatorResult(
-      "op1",
-      makeOpInfo({
-        outputTuples: 4,
-        result: [{ a: 1, b: 2 }],
-        inputPortShapes: [{ portIndex: 0, rows: 10, columns: 2 }],
-      }),
-      state
-    );
-    expect(out).toContain("Input operator(table shape): upstream(10, 2)");
-  });
-
-  test("sorts multiple input ports by portIndex regardless of input order", () => {
-    const state = new WorkflowState();
-    state.addOperator(makeOperator("up0"));
-    state.addOperator(makeOperator("up1"));
-    state.addOperator(makeOperator("op1", ["input-0", "input-1"]));
-    state.addLink(makeLink("l0", ["up0", "output-0"], ["op1", "input-0"]));
-    state.addLink(makeLink("l1", ["up1", "output-0"], ["op1", "input-1"]));
-
-    const out = formatOperatorResult(
-      "op1",
-      makeOpInfo({
-        outputTuples: 1,
-        result: [{ a: 1 }],
-        inputPortShapes: [
-          { portIndex: 1, rows: 2, columns: 2 },
-          { portIndex: 0, rows: 1, columns: 1 },
-        ],
-      }),
-      state
-    );
-    expect(out).toContain("Input operator(table shape): up0(1, 1), up1(2, 2)");
+    expect(lines[2]).toBe("WARNING: truncated to 1 row");
+    expect(lines[3]).toBe("WARNING: something else");
   });
 });
 
@@ -221,8 +149,7 @@ describe("formatOperatorResult - visualization rows", () => {
             label: "chart",
           },
         ],
-      }),
-      EMPTY_STATE
+      })
     );
     expect(out).toContain("<skipped: visualization content>");
     expect(out).not.toContain("<div>hidden</div>");
@@ -236,8 +163,7 @@ describe("formatOperatorResult - visualization rows", () => {
       makeOpInfo({
         outputTuples: 1,
         result: [{ __is_visualization__: false, "html-content": "<keep/>" }],
-      }),
-      EMPTY_STATE
+      })
     );
     expect(out).toContain("<keep/>");
     expect(out).not.toContain("<skipped: visualization content>");
@@ -249,8 +175,7 @@ describe("formatOperatorResult - visualization rows", () => {
       makeOpInfo({
         outputTuples: 1,
         result: [{ __is_visualization__: false, value: 1 }],
-      }),
-      EMPTY_STATE
+      })
     );
     const lines = out.split("\n");
     expect(out).toContain("Output table shape: (1, 1)");
@@ -262,8 +187,8 @@ describe("formatOperatorResult - visualization rows", () => {
 });
 
 describe("jsonToTableFormat - cell coercion via formatOperatorResult", () => {
-  function tableLines(opInfo: Partial<OperatorInfo>): string[] {
-    const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 1, ...opInfo }), EMPTY_STATE);
+  function tableLines(opInfo: OpInfoOverrides): string[] {
+    const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 1, ...opInfo }));
     // Skip brief summary + shape line.
     return out.split("\n").slice(2);
   }
@@ -305,8 +230,7 @@ describe("jsonToTableFormat - row index gaps", () => {
           { __row_index__: 0, v: "a" },
           { __row_index__: 5, v: "b" },
         ],
-      }),
-      EMPTY_STATE
+      })
     );
     const lines = out.split("\n");
     // header, row0, gap marker, row5
@@ -325,18 +249,13 @@ describe("jsonToTableFormat - row index gaps", () => {
           { __row_index__: 0, v: "a" },
           { __row_index__: 1, v: "b" },
         ],
-      }),
-      EMPTY_STATE
+      })
     );
     expect(out).not.toContain("...\t...");
   });
 
   test("non-zero starting __row_index__ does not emit a leading gap marker", () => {
-    const out = formatOperatorResult(
-      "op1",
-      makeOpInfo({ outputTuples: 1, result: [{ __row_index__: 9, v: "z" }] }),
-      EMPTY_STATE
-    );
+    const out = formatOperatorResult("op1", makeOpInfo({ outputTuples: 1, result: [{ __row_index__: 9, v: "z" }] }));
     expect(out).not.toContain("...\t...");
     expect(out.endsWith("9\tz")).toBe(true);
   });