Yicong-Huang commented on code in PR #6009:
URL: https://github.com/apache/texera/pull/6009#discussion_r3739325596
##########
agent-service/src/agent/tools/tools-utility.ts:
##########
@@ -17,10 +17,90 @@
* under the License.
*/
-export const INTERNAL_RESULT_KEYS: ReadonlySet<string> = new
Set(["__row_index__", "__is_visualization__"]);
+import type { IndexedTuple, OperatorExecutionSummary, Tuple, WebOutputMode }
from "../../types/execution";
-export function getVisibleResultHeaders(row: Record<string, any>): string[] {
- return Object.keys(row).filter(k => !INTERNAL_RESULT_KEYS.has(k));
+// 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.
Review Comment:
The justification points at the wrong field: this function reads only
`opInfo.errorMessages`, never `consoleMessages`, so what `filter(Boolean)`
drops is a fatal error with an empty `message` — console ERRORs never reach it.
```suggestion
// not count, matching the previous `error` field's truthiness semantics.
```
(The fix is on line 23: "The engine can emit console ERRORs with empty text"
→ "Fatal errors can arrive with empty message text".)
##########
agent-service/src/agent/tools/result-formatting.ts:
##########
@@ -17,119 +17,42 @@
* under the License.
*/
-import type { OperatorInfo } from "../../types/execution";
-import type { WorkflowState } from "../workflow-state";
-import { formatExecuteOperatorResult, getVisibleResultHeaders } from
"./tools-utility";
-
-export function formatOperatorResult(operatorId: string, opInfo: OperatorInfo,
workflowState: WorkflowState): string {
- if (opInfo.error) {
- return `[ERROR] ${opInfo.error}`;
+import type { OperatorExecutionSummary } from "../../types/execution";
+import {
+ formatExecuteOperatorResult,
+ formatSampleTuplesAsTsv,
+ getOperatorErrorText,
+ getOperatorWarnings,
+ redactVisualizationPayloads,
+ tupleColumns,
+} from "./tools-utility";
+
+export function formatOperatorResult(operatorId: string, opInfo:
OperatorExecutionSummary): string {
+ const errorText = getOperatorErrorText(opInfo);
+ if (errorText) {
+ return `[ERROR] ${errorText}`;
}
- if (!opInfo.result || !Array.isArray(opInfo.result)) {
+ const resultSummary = opInfo.resultSummary;
+ const sampleTuples = resultSummary?.sampleTuples;
+ if (!sampleTuples || !Array.isArray(sampleTuples)) {
Review Comment:
`sampleTuples` is now a typed `ReadonlyArray<IndexedTuple>` coming out of
the zod-validated adapter, so `Array.isArray` can no longer be false when the
value is present. Leftover from the `opInfo.result as Record<string, any>[]`
days; same guard at `workflow-execution-tools.ts:460`.
```suggestion
if (!sampleTuples) {
```
##########
agent-service/src/types/execution.ts:
##########
@@ -17,56 +17,118 @@
* under the License.
*/
-interface ConsoleMessage {
- msgType: string;
- message: string;
+export enum WorkflowFatalErrorType {
+ COMPILATION_ERROR = "COMPILATION_ERROR",
+ EXECUTION_FAILURE = "EXECUTION_FAILURE",
}
-interface PortShape {
- portIndex: number;
- rows: number;
- columns: number;
+// Canonical agent-service error projection. It follows the engine's
+// workflowruntimestate.proto shape so compile and execution errors share one
model.
+// Re-exported by api/compile-api.ts.
+export interface WorkflowFatalError {
+ readonly type: Readonly<{ name: WorkflowFatalErrorType }>;
+ readonly timestamp: Readonly<{ seconds: number; nanos: number }>;
+ readonly message: string;
+ readonly details: string;
+ readonly operatorId: string;
+ readonly workerId: string;
}
-export interface OperatorInfo {
- state: string;
- inputTuples: number;
- outputTuples: number;
- inputPortShapes?: PortShape[];
- resultMode: string;
- result?: Record<string, any>[];
- totalRowCount?: number;
- displayedRows?: number;
- truncated?: boolean;
- consoleLogs?: ConsoleMessage[];
- error?: string;
- warnings?: string[];
- resultStatistics?: Record<string, string>;
+// Lifecycle state of a single operator, as reported by the engine
+// (mirrors the backend's WorkflowAggregatedState string mapping).
+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",
}
-export interface SyncExecutionResult {
- success: boolean;
- state: string;
- operators: Record<string, OperatorInfo>;
- compilationErrors?: Record<string, string>;
- errors?: string[];
+// Aggregated state of a whole workflow execution: the OperatorState values the
+// engine reports, plus agent-service execution outcomes.
+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",
}
-/**
- * Wire projection of one operator's execution result, summarized for the
- * client: counts and a small record sample instead of full payloads. Returned
- * by the REST route `GET /agents/:id/operator-results`.
- */
+export enum ConsoleMessageType {
+ PRINT = "PRINT",
+ ERROR = "ERROR",
+ COMMAND = "COMMAND",
+ DEBUGGER = "DEBUGGER",
+}
+
+// A reduced console-message projection. The engine proto also has
+// workerId/timestamp/source; this summary keeps only the fields consumed by
agent-service.
+export interface ConsoleMessageSummary {
+ readonly msgType: ConsoleMessageType;
+ readonly title: string;
+ readonly message: string;
+}
+
+// A normalized result tuple using the engine Tuple shape: a schema plus
positional fields.
Review Comment:
This line describes `Tuple`, but sits on `Attribute` — every other comment
in the block documents the declaration directly below it. Same misattachment on
the frontend copy at `agent.service.ts:100`.
Moving it above `export interface Tuple` (or making it an explicit group
header for the three) would read correctly.
##########
agent-service/src/agent/tools/workflow-execution-tools.ts:
##########
@@ -438,7 +382,7 @@ export async function executeOperatorAndFormat(
operatorId: string,
options: {
abortSignal?: AbortSignal;
- onResult?: (operatorId: string, operatorInfo: OperatorInfo) => void;
+ onResult?: (operatorId: string, operatorInfo: OperatorExecutionSummary) =>
void;
onResultLegacy?: (operatorId: string, backendStats?: Record<string,
string>) => void;
Review Comment:
`onResultLegacy` is never invoked in this function and no caller supplies it
— a repo-wide grep finds only this declaration. It was already dead before this
PR, but the sibling `onResult` type was rewritten on the line above, so this is
the moment to drop it.
##########
frontend/src/app/workspace/component/agent/agent-interaction/agent-interaction.component.ts:
##########
@@ -176,112 +185,43 @@ export class AgentInteractionComponent implements
OnInit, OnChanges {
return !!this.selectedAgentId && !!this.feedbackMessage.trim();
}
- /**
- * Check if sample records represent a visualization (has
__is_visualization__ flag).
- */
public isVisualization(): boolean {
- if (!this.sampleRecords || this.sampleRecords.length === 0) return false;
- return this.sampleRecords[0]["__is_visualization__"] === true;
+ return this.resultMode?.type === "SetSnapshotMode";
Review Comment:
`SetDeltaMode` is the mode this PR adds to the union, and every consumer
treats it as tabular: here, `workflow-editor.component.ts:1709`, and the
redaction gate at `tools-utility.ts:52`. The producer documents it as "used by
visualization in delta mode" (`ExecutionResultService.scala:285-286`), so once
the backend returns the canonical model a delta-mode chart will render its raw
HTML as table cells and reach the LLM unredacted — the leak you fixed in
31c817fe8, via the other mode.
The frontend already has the right grouping: `isWebDataUpdate` in
`execute-workflow.interface.ts:137-139` covers snapshot and delta together. I'd
reuse that shape rather than repeat the discriminant a third time, and update
the `SetDeltaMode` case in `agent-interaction.component.spec.ts:167-168`, which
currently asserts the classification you'd be fixing.
##########
agent-service/src/server.ts:
##########
@@ -388,25 +387,12 @@ const agentsRouter = new Elysia({ prefix: "/agents" })
}
);
-function getOperatorResultSummaries(agent: TexeraAgent): Record<string,
OperatorResultSummary> {
+function getOperatorResultSummaries(agent: TexeraAgent): Record<string,
OperatorExecutionSummary> {
const resultState = agent.getWorkflowResultState();
const visible = resultState.getAllVisible();
- const results: Record<string, OperatorResultSummary> = {};
+ const results: Record<string, OperatorExecutionSummary> = {};
for (const [opId, entry] of visible) {
- const info = entry.operatorInfo;
- results[opId] = {
- state: info.state,
- inputTuples: info.inputTuples,
- outputTuples: info.outputTuples,
- inputPortShapes: info.inputPortShapes,
- outputColumns: info.result && info.result.length > 0 ?
getVisibleResultHeaders(info.result[0]).length : undefined,
- error: info.error,
- warnings: info.warnings,
- consoleLogCount: info.consoleLogs?.length,
- totalRowCount: info.totalRowCount,
- sampleRecords: info.result,
- resultStatistics: info.resultStatistics,
- };
+ results[opId] = entry.operatorInfo;
Review Comment:
The projection this replaced sent `consoleLogCount`; the pass-through sends
every console message body (stack traces included) plus `state` and
`errorMessages`, on a route the popover polls. The frontend mirror's own
comment says those fields have no consumer (`agent.service.ts:112-114`).
Returning `{ resultSummary: entry.operatorInfo.resultSummary }` would keep
the route's payload matched to what the mirror declares.
##########
agent-service/src/agent/tools/result-formatting.ts:
##########
@@ -17,119 +17,42 @@
* under the License.
*/
-import type { OperatorInfo } from "../../types/execution";
-import type { WorkflowState } from "../workflow-state";
-import { formatExecuteOperatorResult, getVisibleResultHeaders } from
"./tools-utility";
-
-export function formatOperatorResult(operatorId: string, opInfo: OperatorInfo,
workflowState: WorkflowState): string {
- if (opInfo.error) {
- return `[ERROR] ${opInfo.error}`;
+import type { OperatorExecutionSummary } from "../../types/execution";
+import {
+ formatExecuteOperatorResult,
+ formatSampleTuplesAsTsv,
+ getOperatorErrorText,
+ getOperatorWarnings,
+ redactVisualizationPayloads,
+ tupleColumns,
+} from "./tools-utility";
+
+export function formatOperatorResult(operatorId: string, opInfo:
OperatorExecutionSummary): string {
+ const errorText = getOperatorErrorText(opInfo);
+ if (errorText) {
+ return `[ERROR] ${errorText}`;
}
- if (!opInfo.result || !Array.isArray(opInfo.result)) {
+ const resultSummary = opInfo.resultSummary;
+ const sampleTuples = resultSummary?.sampleTuples;
+ if (!sampleTuples || !Array.isArray(sampleTuples)) {
return "(no result data)";
}
- const jsonArray = opInfo.result as Record<string, any>[];
- const headers = jsonArray.length > 0 ? getVisibleResultHeaders(jsonArray[0])
: [];
- const columns = headers.length;
+ const displayTuples = redactVisualizationPayloads(sampleTuples,
resultSummary.resultMode);
Review Comment:
This sequence — redact, derive headers, TSV, `Output table shape: (...)`,
warnings — is the same one `executeOperatorAndFormat` runs at
`workflow-execution-tools.ts:463-521`. Centralising `formatSampleTuplesAsTsv`
was the right move; the block around it stayed in two copies, so anything that
changes what the agent sees has to be written twice. This PR already paid that
cost once, mirroring the visualization redaction into both.
Extracting the whole block into `tools-utility.ts` would leave each caller
with just the summary line it prepends.
##########
agent-service/src/types/execution.ts:
##########
@@ -17,56 +17,118 @@
* under the License.
*/
-interface ConsoleMessage {
- msgType: string;
- message: string;
+export enum WorkflowFatalErrorType {
+ COMPILATION_ERROR = "COMPILATION_ERROR",
+ EXECUTION_FAILURE = "EXECUTION_FAILURE",
}
-interface PortShape {
- portIndex: number;
- rows: number;
- columns: number;
+// Canonical agent-service error projection. It follows the engine's
+// workflowruntimestate.proto shape so compile and execution errors share one
model.
+// Re-exported by api/compile-api.ts.
+export interface WorkflowFatalError {
+ readonly type: Readonly<{ name: WorkflowFatalErrorType }>;
+ readonly timestamp: Readonly<{ seconds: number; nanos: number }>;
+ readonly message: string;
+ readonly details: string;
+ readonly operatorId: string;
+ readonly workerId: string;
}
-export interface OperatorInfo {
- state: string;
- inputTuples: number;
- outputTuples: number;
- inputPortShapes?: PortShape[];
- resultMode: string;
- result?: Record<string, any>[];
- totalRowCount?: number;
- displayedRows?: number;
- truncated?: boolean;
- consoleLogs?: ConsoleMessage[];
- error?: string;
- warnings?: string[];
- resultStatistics?: Record<string, string>;
+// Lifecycle state of a single operator, as reported by the engine
+// (mirrors the backend's WorkflowAggregatedState string mapping).
+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",
}
-export interface SyncExecutionResult {
- success: boolean;
- state: string;
- operators: Record<string, OperatorInfo>;
- compilationErrors?: Record<string, string>;
- errors?: string[];
+// Aggregated state of a whole workflow execution: the OperatorState values the
+// engine reports, plus agent-service execution outcomes.
+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",
}
-/**
- * Wire projection of one operator's execution result, summarized for the
- * client: counts and a small record sample instead of full payloads. Returned
- * by the REST route `GET /agents/:id/operator-results`.
- */
+export enum ConsoleMessageType {
+ PRINT = "PRINT",
+ ERROR = "ERROR",
+ COMMAND = "COMMAND",
+ DEBUGGER = "DEBUGGER",
+}
+
+// A reduced console-message projection. The engine proto also has
+// workerId/timestamp/source; this summary keeps only the fields consumed by
agent-service.
+export interface ConsoleMessageSummary {
+ readonly msgType: ConsoleMessageType;
+ readonly title: string;
+ readonly message: string;
+}
+
+// A normalized result tuple using the engine Tuple shape: a schema plus
positional fields.
+export interface Attribute {
Review Comment:
This concept already exists one directory over: `SchemaAttribute` in
`api/compile-api.ts:30-33`, and again on the frontend as `SchemaAttribute` in
`workspace/types/workflow-compiling.interface.ts:75-79`. Both carry the
`"string" | "integer" | ... | "binary"` union, so the new copies are the
weakest of the three — `attributeType: string` gives a future consumer no
exhaustiveness and no signal when the engine's attribute-type set moves.
Can `Schema` be defined over the existing `SchemaAttribute` in both apps
instead?
##########
agent-service/src/api/execution-api.ts:
##########
@@ -36,3 +49,172 @@ export interface LogicalPlan {
opsToViewResult?: string[];
opsToReuseResult?: string[];
}
+
+/**
+ * TEMPORARY BACKEND COMPATIBILITY LAYER.
+ *
+ * The schemas and inferred wire-format types below exist only to validate and
+ * convert the backend's current execution response. They MUST NOT be reused
+ * outside this file; all other code must use the canonical types from
+ * `types/execution`. Remove this compatibility layer once backend execution is
+ * refactored to return the canonical model.
+ *
+ * TypeScript types are erased at runtime, so Zod validates the untrusted JSON
+ * response before it enters the canonical agent-service execution model.
+ */
+const legacyConsoleMessageSchema = z.object({
+ msgType: z.nativeEnum(ConsoleMessageType),
Review Comment:
`z.nativeEnum` closes this over exactly today's four proto values, and the
schema also requires `inputTuples`/`outputTuples` (lines 82-83) that the
adapter never reads. A rejection here isn't local: the throw is caught at
`workflow-execution-tools.ts:329-338` and returned as `state: ERROR`, so a
backend that adds a console-message type would turn successful executions into
reported failures.
Both enums do match their producers today. Validating only the fields the
adapter consumes — and letting an unknown `msgType` fall through rather than
reject — would keep the boundary from failing closed on an additive change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
