refactor: align execution result summary contract

Author: bobbai00

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

@@ -19,7 +19,15 @@
 
 import { describe, expect, test } from "bun:test";
 import { formatOperatorResult } from "./result-formatting";
-import type { OperatorExecutionSummary, OperatorState, SampleRow } from "../../types/execution";
+import {
+  ConsoleMessageType,
+  OperatorState,
+  OperatorResultMode,
+  WorkflowFatalErrorType,
+  type OperatorExecutionSummary,
+  type WorkflowFatalError,
+  type SampleRow,
+} from "../../types/execution";
 
 // Convert flat test rows (with an optional embedded __row_index__) into the
 // structured SampleRow[] the summary now carries.
@@ -36,29 +44,40 @@ interface OpInfoOverrides {
   state?: OperatorState;
   error?: string;
   outputTuples?: number;
-  totalRowCount?: number;
+  tuplesCount?: number;
   warnings?: string[];
   result?: Record<string, any>[];
 }
 
+function makeExecutionFailure(message: string): WorkflowFatalError {
+  return {
+    type: { name: WorkflowFatalErrorType.EXECUTION_FAILURE },
+    timestamp: { seconds: 0, nanos: 0 },
+    message,
+    details: "",
+    operatorId: "",
+    workerId: "",
+  };
+}
+
 function makeOpInfo(overrides: OpInfoOverrides = {}): OperatorExecutionSummary {
   const summary: OperatorExecutionSummary = {
-    state: overrides.state ?? "Completed",
-    errorMessages: overrides.error ? [{ type: "EXECUTION_FAILURE", message: overrides.error }] : [],
+    state: overrides.state ?? OperatorState.COMPLETED,
+    errorMessages: overrides.error ? [makeExecutionFailure(overrides.error)] : [],
   };
   // The result summary is present only when the operator produced a result.
   if (overrides.result !== undefined) {
     summary.resultSummary = {
-      resultMode: "table",
+      resultMode: OperatorResultMode.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,
+      tuplesCount: overrides.tuplesCount ?? 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: "" })),
+      messages: overrides.warnings.map(w => ({ msgType: ConsoleMessageType.PRINT, title: w, message: "" })),
     };
   }
   return summary;
@@ -93,15 +112,15 @@ describe("formatOperatorResult - early returns", () => {
 });
 
 describe("formatOperatorResult - table shape and metadata", () => {
-  test("uses outputTuples for row count when totalRowCount missing", () => {
+  test("uses outputTuples for row count when tuplesCount missing", () => {
     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", () => {
+  test("tuplesCount overrides outputTuples in output shape", () => {
     const out = formatOperatorResult(
       "op1",
-      makeOpInfo({ outputTuples: 7, totalRowCount: 999, result: [{ a: 1, b: 2 }] })
+      makeOpInfo({ outputTuples: 7, tuplesCount: 999, result: [{ a: 1, b: 2 }] })
     );
     expect(out).toContain("Output table shape: (999, 2)");
   });

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

@@ -54,7 +54,7 @@ export function formatOperatorResult(operatorId: string, opInfo: OperatorExecuti
 
   // Output shape only; input-port shapes are derivable by the agent from the DAG
   // links plus each upstream operator's own output shape shown in context.
-  const outputRows = opInfo.resultSummary?.totalRowCount ?? 0;
+  const outputRows = opInfo.resultSummary?.tuplesCount ?? 0;
   const metadataLines = [`Output table shape: (${outputRows}, ${columns})`, ...getOperatorWarnings(opInfo)].filter(
     Boolean
   );

agent-service/src/agent/tools/workflow-execution-tools.ts

@@ -29,7 +29,15 @@ import type { WorkflowState } from "../workflow-state";
 import { getBackendConfig } from "../../api/backend-api";
 import { env } from "../../config/env";
 import type { LogicalPlan, LogicalLink } from "../../api/execution-api";
-import type { OperatorExecutionSummary, SampleRow, WorkflowExecutionSummary } from "../../types/execution";
+import {
+  OperatorState,
+  WorkflowFatalErrorType,
+  WorkflowExecutionState,
+  type OperatorExecutionSummary,
+  type SampleRow,
+  type WorkflowFatalError,
+  type WorkflowExecutionSummary,
+} from "../../types/execution";
 import { WorkflowSystemMetadata } from "../util/workflow-system-metadata";
 import { DEFAULT_AGENT_SETTINGS } from "../../types/agent";
 import { createLogger } from "../../logger";
@@ -325,7 +333,7 @@ async function executeWorkflowHttp(
     log.error({ err: error }, "execution failed");
     return {
       success: false,
-      state: "Error",
+      state: WorkflowExecutionState.ERROR,
       operators: {},
       errors: [error instanceof Error ? error.message : "Unknown error"],
     };
@@ -355,6 +363,18 @@ function formatExecutionError(
   return lines.join("\n");
 }
 
+function makeExecutionFailure(message: string, operatorId: string): WorkflowFatalError {
+  const now = Date.now();
+  return {
+    type: { name: WorkflowFatalErrorType.EXECUTION_FAILURE },
+    timestamp: { seconds: Math.floor(now / 1000), nanos: (now % 1000) * 1_000_000 },
+    message,
+    details: "",
+    operatorId,
+    workerId: "",
+  };
+}
+
 function jsonToTableFormat(rows: SampleRow[]): string {
   if (!rows || rows.length === 0) return "";
 
@@ -430,20 +450,21 @@ export async function executeOperatorAndFormat(
 
     if (!result.success) {
       const operatorErrors =
-        result.state === "Failed"
+        result.state === WorkflowExecutionState.FAILED
           ? Object.entries(result.operators)
               .filter(([_, op]) => op.errorMessages.length)
               .map(([opId, op]) => ({ operatorId: opId, error: op.errorMessages.map(e => e.message).join("; ") }))
           : undefined;
 
-      const generalErrors = result.state === "Killed" ? ["Workflow execution was killed (timeout)."] : result.errors;
+      const generalErrors =
+        result.state === WorkflowExecutionState.KILLED ? ["Workflow execution was killed (timeout)."] : result.errors;
 
       const errorText = formatExecutionError(operatorErrors, generalErrors);
 
       if (options.onResult) {
         const errorInfo: OperatorExecutionSummary = {
-          state: "Failed",
-          errorMessages: [{ type: "EXECUTION_FAILURE", message: errorText, operatorId }],
+          state: OperatorState.FAILED,
+          errorMessages: [makeExecutionFailure(errorText, operatorId)],
         };
         options.onResult(operatorId, errorInfo);
       }
@@ -519,7 +540,7 @@ export async function executeOperatorAndFormat(
 
     // Output shape only; the agent derives input-port shapes from the DAG + the
     // upstream operators' own output shapes shown in context.
-    const shapeLine = `Output table shape: (${opInfo.resultSummary?.totalRowCount ?? 0}, ${columns})`;
+    const shapeLine = `Output table shape: (${opInfo.resultSummary?.tuplesCount ?? 0}, ${columns})`;
 
     const warningLines = getOperatorWarnings(opInfo);
 

agent-service/src/agent/workflow-result-state.spec.ts

@@ -19,13 +19,13 @@
 
 import { describe, expect, test } from "bun:test";
 import { WorkflowResultState } from "./workflow-result-state";
-import type { OperatorExecutionSummary } from "../types/execution";
+import { OperatorResultMode, OperatorState, type OperatorExecutionSummary } from "../types/execution";
 
-function makeInfo(totalRowCount: number): OperatorExecutionSummary {
+function makeInfo(tuplesCount: number): OperatorExecutionSummary {
   return {
-    state: "Completed",
+    state: OperatorState.COMPLETED,
     errorMessages: [],
-    resultSummary: { resultMode: "table", sampleTuples: [], totalRowCount },
+    resultSummary: { resultMode: OperatorResultMode.TABLE, sampleTuples: [], tuplesCount },
   };
 }
 
@@ -39,15 +39,15 @@ describe("WorkflowResultState - ancestor walk", () => {
     state.set("op1", "step-C", makeInfo(3));
 
     path = ["step-A", "step-B", "step-C"];
-    expect(state.get("op1")?.operatorInfo.resultSummary?.totalRowCount).toBe(3);
+    expect(state.get("op1")?.operatorInfo.resultSummary?.tuplesCount).toBe(3);
 
     // Rewind to step-B; step-C is no longer an ancestor.
     path = ["step-A", "step-B"];
-    expect(state.get("op1")?.operatorInfo.resultSummary?.totalRowCount).toBe(2);
+    expect(state.get("op1")?.operatorInfo.resultSummary?.tuplesCount).toBe(2);
 
     // Rewind further.
     path = ["step-A"];
-    expect(state.get("op1")?.operatorInfo.resultSummary?.totalRowCount).toBe(1);
+    expect(state.get("op1")?.operatorInfo.resultSummary?.tuplesCount).toBe(1);
   });
 
   test("returns undefined when no ancestor has a result", () => {
@@ -73,8 +73,8 @@ describe("WorkflowResultState - ancestor walk", () => {
     path = ["step-A", "step-B"];
     const visible = state.getAllVisible();
     expect(visible.size).toBe(2);
-    expect(visible.get("op1")?.operatorInfo.resultSummary?.totalRowCount).toBe(1);
-    expect(visible.get("op2")?.operatorInfo.resultSummary?.totalRowCount).toBe(7);
+    expect(visible.get("op1")?.operatorInfo.resultSummary?.tuplesCount).toBe(1);
+    expect(visible.get("op2")?.operatorInfo.resultSummary?.tuplesCount).toBe(7);
   });
 
   test("clear drops all stored results", () => {
@@ -89,6 +89,6 @@ describe("WorkflowResultState - ancestor walk", () => {
     const state = new WorkflowResultState(() => ["step-A"]);
     state.set("op1", "step-A", makeInfo(1));
     state.set("op1", "step-A", makeInfo(42));
-    expect(state.get("op1")?.operatorInfo.resultSummary?.totalRowCount).toBe(42);
+    expect(state.get("op1")?.operatorInfo.resultSummary?.tuplesCount).toBe(42);
   });
 });

agent-service/src/server.spec.ts

@@ -21,6 +21,7 @@ import { beforeEach, describe, expect, spyOn, test } from "bun:test";
 import { buildApp, start, _resetAgentStoreForTests, _getAgentForTests } from "./server";
 import { WorkflowSystemMetadata } from "./agent/util/workflow-system-metadata";
 import { env } from "./config/env";
+import { OperatorResultMode, OperatorState } from "./types/execution";
 
 const API = env.API_PREFIX;
 const app = buildApp();
@@ -331,12 +332,12 @@ describe("agent read routes", () => {
             "op-1",
             {
               operatorInfo: {
-                state: "Completed",
+                state: OperatorState.COMPLETED,
                 errorMessages: [],
                 resultSummary: {
-                  resultMode: "table",
+                  resultMode: OperatorResultMode.TABLE,
                   sampleTuples: [{ rowIndex: 0, tuple: { a: 1 } }],
-                  totalRowCount: 2,
+                  tuplesCount: 2,
                 },
               },
             },
@@ -345,10 +346,10 @@ describe("agent read routes", () => {
     });
 
     const body = await readJson<{
-      results: Record<string, { state: string; resultSummary: { totalRowCount: number; sampleTuples: unknown[] } }>;
+      results: Record<string, { state: string; resultSummary: { tuplesCount: number; sampleTuples: unknown[] } }>;
     }>(await getJson(`${API}/agents/${id}/operator-results`));
     expect(body.results["op-1"].state).toBe("Completed");
-    expect(body.results["op-1"].resultSummary.totalRowCount).toBe(2);
+    expect(body.results["op-1"].resultSummary.tuplesCount).toBe(2);
     expect(body.results["op-1"].resultSummary.sampleTuples).toEqual([{ rowIndex: 0, tuple: { a: 1 } }]);
   });
 });

agent-service/src/types/execution.ts

@@ -17,44 +17,70 @@
  * under the License.
  */
 
+export enum WorkflowFatalErrorType {
+  COMPILATION_ERROR = "COMPILATION_ERROR",
+  EXECUTION_FAILURE = "EXECUTION_FAILURE",
+}
+
 // A fatal error reported for one operator. Reuses the engine's wire shape
-// (workflowruntimestate.proto): `type` is the FatalErrorType enum name. The same
-// type the workflow-compiling service returns for compilation errors, so compile
-// and execution errors share one shape. Re-exported by api/compile-api.ts.
+// (workflowruntimestate.proto). The same type the workflow-compiling service
+// returns for compilation errors, so compile and execution errors share one
+// shape. Re-exported by api/compile-api.ts.
 export interface WorkflowFatalError {
-  // FatalErrorType enum name, e.g. "COMPILATION_ERROR" | "EXECUTION_FAILURE".
-  type: string;
+  type: { name: WorkflowFatalErrorType };
+  timestamp: { seconds: number; nanos: number };
   message: string;
-  details?: string;
-  operatorId?: string;
-  workerId?: string;
-  timestamp?: { seconds: number; nanos: number };
+  details: string;
+  operatorId: string;
+  workerId: string;
 }
 
 // Lifecycle state of a single operator, as reported by the engine
 // (mirrors the backend's WorkflowAggregatedState string mapping).
-export type OperatorState =
-  | "Uninitialized"
-  | "Ready"
-  | "Running"
-  | "Pausing"
-  | "Paused"
-  | "Resuming"
-  | "Completed"
-  | "Failed"
-  | "Killed"
-  | "Terminated"
-  | "Unknown";
+export enum OperatorState {
+  UNINITIALIZED = "Uninitialized",
+  READY = "Ready",
+  RUNNING = "Running",
+  PAUSING = "Pausing",
+  PAUSED = "Paused",
+  RESUMING = "Resuming",
+  COMPLETED = "Completed",
+  FAILED = "Failed",
+  KILLED = "Killed",
+  TERMINATED = "Terminated",
+  UNKNOWN = "Unknown",
+}
 
 // Aggregated state of a whole workflow execution: the OperatorState values the
 // engine reports, plus the synthetic outcomes the sync-execution endpoint adds.
-export type WorkflowExecutionState = OperatorState | "Error" | "CompilationFailed";
+export enum WorkflowExecutionState {
+  UNINITIALIZED = "Uninitialized",
+  READY = "Ready",
+  RUNNING = "Running",
+  PAUSING = "Pausing",
+  PAUSED = "Paused",
+  RESUMING = "Resuming",
+  COMPLETED = "Completed",
+  FAILED = "Failed",
+  KILLED = "Killed",
+  TERMINATED = "Terminated",
+  UNKNOWN = "Unknown",
+  ERROR = "Error",
+  COMPILATION_FAILED = "CompilationFailed",
+}
+
+export enum ConsoleMessageType {
+  PRINT = "PRINT",
+  ERROR = "ERROR",
+  COMMAND = "COMMAND",
+  DEBUGGER = "DEBUGGER",
+}
 
 // A single console message emitted by an operator during execution.
 // `title` is the short header (Scala errors put their text here); `message` is
 // the body (Python errors / stack traces).
 export interface ConsoleMessage {
-  msgType: string;
+  msgType: ConsoleMessageType;
   title: string;
   message: string;
 }
@@ -66,16 +92,20 @@ export interface SampleRow {
   tuple: Record<string, unknown>;
 }
 
+export enum OperatorResultMode {
+  TABLE = "table",
+  VISUALIZATION = "visualization",
+}
+
 // An operator's output, summarized for the agent. `sampleTuples` are the
 // symmetrically-truncated output rows (the middle is dropped, so `rowIndex`
 // values may have gaps). `outputSchema` / per-column statistics are intended
 // future additions — the engine does not produce them yet.
 export interface OperatorResultSummary {
-  // "table" or "visualization".
-  resultMode: string;
+  resultMode: OperatorResultMode;
   sampleTuples: SampleRow[];
   // Total output rows before truncation (sampleTuples may hold fewer).
-  totalRowCount: number;
+  tuplesCount: number;
 }
 
 // An operator's console output. Warnings are not a separate field: they are the