fix(agent-service): unify operator-failure semantics across consumers

getOperatorErrorText pins the one derivation of "operator failed" the new
errorMessages type forces (a fatal error carrying message text), replacing
the divergent .length and joined-truthiness checks. In the failure branch,
surface per-operator errors for any terminal state (success=false can come
with state "Completed"), label CompilationFailed results distinctly, and
preserve the Killed state on timeout instead of collapsing to Failed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: bobbai00

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

@@ -18,10 +18,15 @@
  */
 
 import { OperatorResultMode, type OperatorExecutionSummary, type SampleRow } from "../../types/execution";
-import { formatExecuteOperatorResult, getOperatorWarnings, getVisibleResultHeaders } from "./tools-utility";
+import {
+  formatExecuteOperatorResult,
+  getOperatorErrorText,
+  getOperatorWarnings,
+  getVisibleResultHeaders,
+} from "./tools-utility";
 
 export function formatOperatorResult(operatorId: string, opInfo: OperatorExecutionSummary): string {
-  const errorText = opInfo.errorMessages.map(e => e.message).join("; ");
+  const errorText = getOperatorErrorText(opInfo);
   if (errorText) {
     return `[ERROR] ${errorText}`;
   }

agent-service/src/agent/tools/tools-utility.spec.ts

@@ -25,8 +25,43 @@ import {
   formatModifyOperatorResult,
   formatExecuteOperatorResult,
   formatOperatorError,
+  getOperatorErrorText,
   getVisibleResultHeaders,
 } from "./tools-utility";
+import { OperatorState, WorkflowFatalErrorType, type WorkflowFatalError } from "../../types/execution";
+
+function makeFatal(message: string): WorkflowFatalError {
+  return {
+    type: { name: WorkflowFatalErrorType.EXECUTION_FAILURE },
+    timestamp: { seconds: 0, nanos: 0 },
+    message,
+    details: "",
+    operatorId: "",
+    workerId: "",
+  };
+}
+
+describe("getOperatorErrorText", () => {
+  test("joins the messages of every fatal error", () => {
+    const opInfo = { state: OperatorState.FAILED, errorMessages: [makeFatal("e1"), makeFatal("e2")] };
+    expect(getOperatorErrorText(opInfo)).toBe("e1; e2");
+  });
+
+  test("ignores errors whose message is empty", () => {
+    const opInfo = { state: OperatorState.FAILED, errorMessages: [makeFatal(""), makeFatal("real")] };
+    expect(getOperatorErrorText(opInfo)).toBe("real");
+  });
+
+  test("returns an empty string when every error message is empty", () => {
+    const opInfo = { state: OperatorState.COMPLETED, errorMessages: [makeFatal("")] };
+    expect(getOperatorErrorText(opInfo)).toBe("");
+  });
+
+  test("returns an empty string when there are no errors", () => {
+    const opInfo = { state: OperatorState.COMPLETED, errorMessages: [] };
+    expect(getOperatorErrorText(opInfo)).toBe("");
+  });
+});
 
 describe("getVisibleResultHeaders", () => {
   test("returns every key", () => {

agent-service/src/agent/tools/tools-utility.ts

@@ -19,6 +19,16 @@
 
 import type { OperatorExecutionSummary } from "../../types/execution";
 
+// The single definition of "this operator failed": some fatal error carries
+// message text. The engine can emit console ERRORs with empty text, which do
+// not count, matching the previous `error` field's truthiness semantics.
+export function getOperatorErrorText(opInfo: OperatorExecutionSummary): string {
+  return opInfo.errorMessages
+    .map(e => e.message)
+    .filter(Boolean)
+    .join("; ");
+}
+
 export function getVisibleResultHeaders(row: Record<string, any>): string[] {
   return Object.keys(row);
 }

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

@@ -320,6 +320,39 @@ describe("executeOperatorAndFormat - execution failures", () => {
     expect(info.errorMessages[0].message).toContain("kaboom");
   });
 
+  test("extracts per-operator errors even when the aggregate state stays Completed", async () => {
+    const { state, target } = makeLinearState();
+    // A console ERROR marks the run unsuccessful without flipping the state.
+    const summary: WorkflowExecutionSummary = {
+      success: false,
+      state: WorkflowExecutionState.COMPLETED,
+      operators: {
+        [target]: { state: OperatorState.COMPLETED, errorMessages: [makeFatal("console kaboom")] },
+      },
+      errors: [],
+    };
+    setFetchResolving(jsonResponse(summary));
+
+    const out = await executeOperatorAndFormat(state, makeConfig(), target);
+    expect(out).toContain("Execution error:");
+    expect(out).toContain(`${target}: console kaboom`);
+  });
+
+  test("labels compilation failures distinctly from runtime errors", async () => {
+    const { state, target } = makeLinearState();
+    const summary: WorkflowExecutionSummary = {
+      success: false,
+      state: WorkflowExecutionState.COMPILATION_FAILED,
+      operators: {},
+      errors: ["schema mismatch on port 0"],
+    };
+    setFetchResolving(jsonResponse(summary));
+
+    const out = await executeOperatorAndFormat(state, makeConfig(), target);
+    expect(out).toContain("Compilation error:");
+    expect(out).toContain("schema mismatch on port 0");
+  });
+
   test("reports a timeout when the workflow was KILLED (no onResult callback)", async () => {
     const { state, target } = makeLinearState();
     const summary: WorkflowExecutionSummary = {
@@ -335,6 +368,25 @@ describe("executeOperatorAndFormat - execution failures", () => {
     expect(out).toContain("Workflow execution was killed (timeout).");
   });
 
+  test("stores the Killed state in the synthetic summary when the workflow was KILLED", async () => {
+    const { state, target } = makeLinearState();
+    const summary: WorkflowExecutionSummary = {
+      success: false,
+      state: WorkflowExecutionState.KILLED,
+      operators: {},
+      errors: [],
+    };
+    setFetchResolving(jsonResponse(summary));
+
+    const onResult = mock(() => {});
+    await executeOperatorAndFormat(state, makeConfig(), target, { onResult });
+
+    expect(onResult).toHaveBeenCalledTimes(1);
+    const [, info] = onResult.mock.calls[0] as unknown as [string, OperatorExecutionSummary];
+    expect(info.state).toBe(OperatorState.KILLED);
+    expect(info.errorMessages[0].message).toContain("killed (timeout)");
+  });
+
   test("surfaces general errors when the HTTP request itself fails (ERROR state)", async () => {
     const { state, target } = makeLinearState();
     setFetchResolving(new Response("backend detail", { status: 500, statusText: "Server Error" }));

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

@@ -22,6 +22,7 @@ import { tool } from "ai";
 import {
   createErrorResult,
   formatExecuteOperatorResult,
+  getOperatorErrorText,
   getOperatorWarnings,
   getVisibleResultHeaders,
 } from "./tools-utility";
@@ -342,7 +343,8 @@ async function executeWorkflowHttp(
 
 function formatExecutionError(
   operatorErrors?: Array<{ operatorId: string; error: string }>,
-  generalErrors?: string[]
+  generalErrors?: string[],
+  generalErrorsLabel: string = "Error:"
 ): string {
   const lines: string[] = ["Execution failed due to the following error:"];
 
@@ -354,7 +356,7 @@ function formatExecutionError(
   }
 
   if (generalErrors && generalErrors.length > 0) {
-    lines.push("Error:");
+    lines.push(generalErrorsLabel);
     for (const error of generalErrors) {
       lines.push(`  ${error}`);
     }
@@ -449,21 +451,25 @@ export async function executeOperatorAndFormat(
     });
 
     if (!result.success) {
-      const operatorErrors =
-        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;
+      // Per-operator errors can accompany any terminal state (e.g. a console
+      // ERROR leaves state "Completed" with success=false), so never gate on it.
+      const operatorErrors = Object.entries(result.operators)
+        .map(([opId, op]) => ({ operatorId: opId, error: getOperatorErrorText(op) }))
+        .filter(({ error }) => error);
 
       const generalErrors =
         result.state === WorkflowExecutionState.KILLED ? ["Workflow execution was killed (timeout)."] : result.errors;
 
-      const errorText = formatExecutionError(operatorErrors, generalErrors);
+      const generalErrorsLabel =
+        result.state === WorkflowExecutionState.COMPILATION_FAILED ? "Compilation error:" : "Error:";
+
+      const errorText = formatExecutionError(operatorErrors, generalErrors, generalErrorsLabel);
 
       if (options.onResult) {
         const errorInfo: OperatorExecutionSummary = {
-          state: OperatorState.FAILED,
+          // Preserve the killed state so a timeout is distinguishable from a
+          // genuine operator failure downstream.
+          state: result.state === WorkflowExecutionState.KILLED ? OperatorState.KILLED : OperatorState.FAILED,
           errorMessages: [makeExecutionFailure(errorText, operatorId)],
         };
         options.onResult(operatorId, errorInfo);
@@ -477,11 +483,11 @@ export async function executeOperatorAndFormat(
       return createErrorResult(formatExecutionError(undefined, [`No result found for operator: ${operatorId}`]));
     }
 
-    if (opInfo.errorMessages.length) {
+    const opError = getOperatorErrorText(opInfo);
+    if (opError) {
       if (options.onResult) {
         options.onResult(operatorId, opInfo);
       }
-      const opError = opInfo.errorMessages.map(e => e.message).join("; ");
       return createErrorResult(formatExecutionError([{ operatorId, error: opError }]));
     }
 
@@ -496,7 +502,7 @@ export async function executeOperatorAndFormat(
     // Notify for every operator in the execution so upstream stats are also stored.
     if (options.onResult) {
       for (const [opId, info] of Object.entries(result.operators)) {
-        if (info && !info.errorMessages.length) {
+        if (info && !getOperatorErrorText(info)) {
           options.onResult(opId, info);
         }
       }