bobbai00 commented on code in PR #6009:
URL: https://github.com/apache/texera/pull/6009#discussion_r4342457310
##########
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:
Done in 94579a2.
##########
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:
Done in 5450e20.
##########
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:
Done in 17aa921.
##########
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:
Done in 8343cfc.
--
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]
