kunwp1 commented on code in PR #8637:
URL: https://github.com/apache/texera/pull/8637#discussion_r4355132496
##########
frontend/src/app/workspace/component/result-panel/result-panel.component.ts:
##########
@@ -252,8 +256,63 @@ export class ResultPanelComponent implements OnInit,
OnDestroy {
.subscribe(_ => {
this.rerenderResultPanel();
this.changeDetectorRef.detectChanges();
- this.registerOperatorDisplayNameChangeHandler();
+ this.updateOperatorTitle();
});
+
+ // A Python UDF that raises while its operator is already selected moves
none of the
+ // streams above: the execution stays Running and the highlight never
moves, so the AI
+ // Fix tab would not appear until the user clicked away and back. Only
that one tab
+ // depends on the console, so this toggles it instead of calling
rerenderResultPanel():
+ // a full rerender rebuilds every frame, losing the Console frame's type
filter, scroll
+ // position and debug input, and the Result table, whenever any operator
prints.
+ this.workflowConsoleService
+ .getConsoleMessageUpdateStream()
+ .pipe(
+ // Keyed by operator as well as by the flag: comparing the flag alone
would block a
+ // second operator's error after the first one's, since both map to
`true`.
+ map(() => `${this.currentOperatorId ??
""}:${this.shouldOfferAiFix()}`),
+ distinctUntilChanged(),
+ untilDestroyed(this)
+ )
+ .subscribe(() => {
+ this.syncAiFixFrame();
+ this.changeDetectorRef.detectChanges();
+ });
+ }
+
+ /** True when the selected operator reports an error the AI panel can be
opened on. */
+ private shouldOfferAiFix(): boolean {
+ return (
+ this.config.env.copilotEnabled &&
+ this.currentOperatorId !== undefined &&
+ this.hasConsoleError(this.currentOperatorId)
+ );
+ }
+
+ /**
+ * Keys the frame tabs by name.
+ *
+ * `keyvalue` rebuilds its pair objects on every change to the map, so
without this every
+ * tab was torn down and recreated whenever one was added or removed -- a
result table lost
+ * its scroll position and paging because a console error appeared elsewhere.
+ */
+ public trackFrameByName(_index: number, config: { key: string }): string {
+ return config.key;
+ }
Review Comment:
Keying by name fixes the rebuild when the AI Fix tab is added, but it also
keeps every other frame alive across re-renders and operator switches.
`NgComponentOutlet` only recreates a component when its type changes and
otherwise just updates the inputs. Tracking by the config object instead (e.g.
returning `config.value`) seems to give the intended behavior. A re-render
builds fresh config objects, so frames are recreated as before, while
`syncAiFixFrame` adding or removing its entry leaves the others untouched.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,613 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { Injectable } from "@angular/core";
+import { BehaviorSubject, Observable, firstValueFrom } from "rxjs";
+import { filter, timeout } from "rxjs/operators";
+import { createOpenAI } from "@ai-sdk/openai";
+import { generateText, type ModelMessage } from "ai";
+import { AppSettings } from "../../../../common/app-setting";
+import { AuthService } from "../../../../common/service/user/auth.service";
+import { WorkflowActionService } from
"../../../service/workflow-graph/model/workflow-action.service";
+import { ExecuteWorkflowService } from
"../../../service/execute-workflow/execute-workflow.service";
+import { OperatorPortSchemaMap } from
"../../../types/workflow-compiling.interface";
+import { ExecutionState } from "../../../types/execute-workflow.interface";
+import { buildFixPrompt, withoutSecrets } from "./ai-fix-prompt";
+import { OperatorMetadataService } from
"../../../service/operator-metadata/operator-metadata.service";
+import { WorkflowConsoleService } from
"../../../service/workflow-console/workflow-console.service";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ /** How many places Apply will touch; the diff only ever shows one -/+ pair.
*/
+ occurrences?: number;
+ type: "code_change" | "property_change";
+ original: string;
+ suggested: string;
+ explanation: string;
+ // Rendered as the confidence badge; `fieldName` names the patched property
+ // (property_change only) so applyFix knows which field to write.
+ confidence: FixConfidence;
+ fieldName?: string;
+}
+
+export interface FixState {
+ operatorId: string;
+ /**
+ * Identifies the ERROR the suggestion was made for, so a newer one on the
same operator
+ * can clear it. A terminal execution state is not enough on its own: the
case this panel
+ * exists for leaves the run `Running` -- the UDF raises and the worker
pauses -- so a
+ * re-run that fails again never reaches one, and the error key is the only
signal left.
+ */
+ errorKey?: string;
+ /**
+ * The workflow the analysis belongs to. The operator id alone does not
identify it: two
+ * imports of the same JSON carry the same ids, so a reply that arrived
after the user
+ * switched workflows could be applied to the wrong one.
+ */
+ workflowId?: number;
+ /**
+ * Which of the four supported shapes the error was classified as. Only the
apply path
+ * reads it, to refuse a configuration change for anything but
`model_not_found`; an error
+ * outside those shapes is the `unsupported` status, not a value here.
+ */
+ errorType?: FixErrorType;
+ suggestedFix?: SuggestedFix;
+ // `unsupported`: the error is outside the four shapes this panel handles,
reported without
+ // spending a model call. `applied_without_run`: the fix was written but the
deployment
+ // refused to start a run (a warehouse is required and none is selected), so
the panel must
+ // not claim one. `apply_failed` keeps the suggestion on screen: the
analysis was fine, the
+ // write was refused, and re-analysing would spend another model call for
the same fix.
+ status:
+ | "idle"
+ | "analyzing"
+ | "unsupported"
+ | "ready"
+ | "applying"
+ | "applied"
+ | "applied_without_run"
+ | "error"
+ | "apply_failed";
+ /** Why the write was refused, shown as-is when the status is
`apply_failed`. */
+ applyError?: string;
+}
+
+// Model id as LiteLLM exposes it (bin/single-node/litellm-config.yaml), not
the provider's
+// id. A deployment whose proxy publishes a different alias has to change
this: there is no
+// GuiConfig field for it today, and adding one is a backend change.
+export const AI_FIXER_MODEL = "claude-haiku-4.5";
+export const AI_FIXER_TIMEOUT_MS = 120_000;
+export const UNLOCK_TIMEOUT_MS = 15_000;
+
+// A run in one of these is live enough to be worth stopping; a lock coming
from anything
+// else is not something killing an execution can clear.
+const LIVE_STATES: ReadonlySet<ExecutionState> = new Set([
+ ExecutionState.Initializing,
+ ExecutionState.Running,
+ ExecutionState.Pausing,
+ ExecutionState.Paused,
+ ExecutionState.Resuming,
+ ExecutionState.Recovering,
+]);
+
+// A run that reached one of these is over, so anything the panel still shows
predates it.
+// `Uninitialized` is deliberately absent even though updateWorkflowActionLock
treats it as
+// over: a kill lands there, and a kill is how Apply itself clears the lock,
so resetting on
+// it would throw away a suggestion that is still the right one for the error
on screen.
+const TERMINAL_STATES: ReadonlySet<ExecutionState> = new Set([
+ ExecutionState.Completed,
+ ExecutionState.Failed,
+ ExecutionState.Killed,
+ ExecutionState.Terminated,
+]);
+
+export const UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", status: "idle" };
+
+// Matched against the exception line only, never the frame lines above it.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+/**
+ * Replaces every occurrence, keeping each one at the indentation it was found
at.
+ *
+ * A single-line match starts mid-line, after the existing indentation, so a
multi-line
+ * replacement would put its first line in the right place and every line
after it at column
+ * 0 -- an IndentationError the moment the fix was applied. Those continuation
lines take the
+ * indentation of the line the match was found on, and keep whatever relative
indentation the
+ * suggestion itself had.
+ *
+ * A match that spans lines already carries the absolute indentation of the
region it came
+ * from, and the model's reply mirrors it, so there is nothing to add:
prefixing the context
+ * indentation on top would double it.
+ */
+export function replaceAllIndented(code: string, original: string, suggested:
string): { code: string; count: number } {
+ const parts = code.split(original);
+ const count = parts.length - 1;
+ if (count === 0 || !suggested.includes("\n") || original.includes("\n")) {
Review Comment:
This avoids doubling indentation for multi-line originals but the result now
depends on which convention the model happens to use for new lines and the
prompt never states one. A single-line original with a suggestion whose new
lines carry the code's absolute indentation still gets indented twice (I tried
`return` landing at 20 spaces instead of 12). Could the prompt ask for whole
lines copied with their exact indentation with originals that don't start at a
line boundary rejected at analysis? Then nothing would need to be re-indented
here.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,613 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { Injectable } from "@angular/core";
+import { BehaviorSubject, Observable, firstValueFrom } from "rxjs";
+import { filter, timeout } from "rxjs/operators";
+import { createOpenAI } from "@ai-sdk/openai";
+import { generateText, type ModelMessage } from "ai";
+import { AppSettings } from "../../../../common/app-setting";
+import { AuthService } from "../../../../common/service/user/auth.service";
+import { WorkflowActionService } from
"../../../service/workflow-graph/model/workflow-action.service";
+import { ExecuteWorkflowService } from
"../../../service/execute-workflow/execute-workflow.service";
+import { OperatorPortSchemaMap } from
"../../../types/workflow-compiling.interface";
+import { ExecutionState } from "../../../types/execute-workflow.interface";
+import { buildFixPrompt, withoutSecrets } from "./ai-fix-prompt";
+import { OperatorMetadataService } from
"../../../service/operator-metadata/operator-metadata.service";
+import { WorkflowConsoleService } from
"../../../service/workflow-console/workflow-console.service";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ /** How many places Apply will touch; the diff only ever shows one -/+ pair.
*/
+ occurrences?: number;
+ type: "code_change" | "property_change";
+ original: string;
+ suggested: string;
+ explanation: string;
+ // Rendered as the confidence badge; `fieldName` names the patched property
+ // (property_change only) so applyFix knows which field to write.
+ confidence: FixConfidence;
+ fieldName?: string;
+}
+
+export interface FixState {
+ operatorId: string;
+ /**
+ * Identifies the ERROR the suggestion was made for, so a newer one on the
same operator
+ * can clear it. A terminal execution state is not enough on its own: the
case this panel
+ * exists for leaves the run `Running` -- the UDF raises and the worker
pauses -- so a
+ * re-run that fails again never reaches one, and the error key is the only
signal left.
+ */
+ errorKey?: string;
+ /**
+ * The workflow the analysis belongs to. The operator id alone does not
identify it: two
+ * imports of the same JSON carry the same ids, so a reply that arrived
after the user
+ * switched workflows could be applied to the wrong one.
+ */
+ workflowId?: number;
+ /**
+ * Which of the four supported shapes the error was classified as. Only the
apply path
+ * reads it, to refuse a configuration change for anything but
`model_not_found`; an error
+ * outside those shapes is the `unsupported` status, not a value here.
+ */
+ errorType?: FixErrorType;
+ suggestedFix?: SuggestedFix;
+ // `unsupported`: the error is outside the four shapes this panel handles,
reported without
+ // spending a model call. `applied_without_run`: the fix was written but the
deployment
+ // refused to start a run (a warehouse is required and none is selected), so
the panel must
+ // not claim one. `apply_failed` keeps the suggestion on screen: the
analysis was fine, the
+ // write was refused, and re-analysing would spend another model call for
the same fix.
+ status:
+ | "idle"
+ | "analyzing"
+ | "unsupported"
+ | "ready"
+ | "applying"
+ | "applied"
+ | "applied_without_run"
+ | "error"
+ | "apply_failed";
+ /** Why the write was refused, shown as-is when the status is
`apply_failed`. */
+ applyError?: string;
+}
+
+// Model id as LiteLLM exposes it (bin/single-node/litellm-config.yaml), not
the provider's
+// id. A deployment whose proxy publishes a different alias has to change
this: there is no
+// GuiConfig field for it today, and adding one is a backend change.
+export const AI_FIXER_MODEL = "claude-haiku-4.5";
+export const AI_FIXER_TIMEOUT_MS = 120_000;
+export const UNLOCK_TIMEOUT_MS = 15_000;
+
+// A run in one of these is live enough to be worth stopping; a lock coming
from anything
+// else is not something killing an execution can clear.
+const LIVE_STATES: ReadonlySet<ExecutionState> = new Set([
+ ExecutionState.Initializing,
+ ExecutionState.Running,
+ ExecutionState.Pausing,
+ ExecutionState.Paused,
+ ExecutionState.Resuming,
+ ExecutionState.Recovering,
+]);
+
+// A run that reached one of these is over, so anything the panel still shows
predates it.
+// `Uninitialized` is deliberately absent even though updateWorkflowActionLock
treats it as
+// over: a kill lands there, and a kill is how Apply itself clears the lock,
so resetting on
+// it would throw away a suggestion that is still the right one for the error
on screen.
+const TERMINAL_STATES: ReadonlySet<ExecutionState> = new Set([
+ ExecutionState.Completed,
+ ExecutionState.Failed,
+ ExecutionState.Killed,
+ ExecutionState.Terminated,
+]);
+
+export const UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", status: "idle" };
+
+// Matched against the exception line only, never the frame lines above it.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+/**
+ * Replaces every occurrence, keeping each one at the indentation it was found
at.
+ *
+ * A single-line match starts mid-line, after the existing indentation, so a
multi-line
+ * replacement would put its first line in the right place and every line
after it at column
+ * 0 -- an IndentationError the moment the fix was applied. Those continuation
lines take the
+ * indentation of the line the match was found on, and keep whatever relative
indentation the
+ * suggestion itself had.
+ *
+ * A match that spans lines already carries the absolute indentation of the
region it came
+ * from, and the model's reply mirrors it, so there is nothing to add:
prefixing the context
+ * indentation on top would double it.
+ */
+export function replaceAllIndented(code: string, original: string, suggested:
string): { code: string; count: number } {
+ const parts = code.split(original);
+ const count = parts.length - 1;
+ if (count === 0 || !suggested.includes("\n") || original.includes("\n")) {
+ return { code: parts.join(suggested), count };
+ }
+
+ const [head, ...rest] = suggested.split("\n");
+ let out = parts[0];
+ for (const tail of parts.slice(1)) {
+ // Whitespace since the last newline is what this occurrence sits at.
+ const lineStart = out.lastIndexOf("\n") + 1;
+ const indent = (out.slice(lineStart).match(/^[ \t]*/) ?? [""])[0];
+ out += [head, ...rest.map(line => (line.length > 0 ? indent + line :
line))].join("\n") + tail;
+ }
+ return { code: out, count };
+}
+
+/** Writes the model's string back in the type the field already held. */
+export function coerceToFieldType(currentValue: unknown, suggested: string):
unknown {
+ // Arrays and objects are out of scope: the model replies with a string, and
writing it
+ // into a structured field replaced e.g. an outputColumns array with the
text "[]".
+ // Validating a replacement against the operator's schema is a bigger change
than the four
+ // error shapes this panel handles need, so it refuses instead of guessing.
+ if (currentValue !== null && typeof currentValue === "object") {
+ throw new Error("this setting holds structured data, which an automatic
fix does not edit");
+ }
+ if (typeof currentValue === "number") {
+ const parsed = Number(suggested);
+ if (Number.isNaN(parsed)) {
+ throw new Error(`${suggested} is not a number`);
+ }
+ return parsed;
+ }
+ if (typeof currentValue === "boolean") {
+ if (suggested !== "true" && suggested !== "false") {
+ throw new Error(`${suggested} is not a boolean`);
+ }
+ return suggested === "true";
+ }
+ return suggested;
+}
+
+// What a stack frame looks like, so it is never mistaken for the exception.
Python writes
+// File "udf.py", line 7, in process_tuple
+// and the JVM writes
+// org.apache.texera.Foo.run(Foo.scala:12)
+// which carries no colon-separated exception name, so a Python-only pattern
lets a Java
+// frame through and classifies on it.
+const FRAME_LINE = /^File\s+"|^Traceback \(|^at\s|^[\w$.]+\([\w.]+:\d+\)$/;
Review Comment:
JVM errors still won't classify with this pattern. On Java 9+, JDK frames
print with a module prefix (`java.base/java.lang.Thread.run(Thread.java:840)`),
and native frames read `(Native Method)`. Neither matches and a real worker
trace ends on exactly such a frame so the final-line search returns it and
`NullPointerException` stays unsupported. Both sources already separate the
exception header: `title` for console errors and `message` for fatal errors. So
classifying on that header directly and sending the full text only to the
prompt, may be simpler and would let `FRAME_LINE`/`finalExceptionLine` go.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,613 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { Injectable } from "@angular/core";
+import { BehaviorSubject, Observable, firstValueFrom } from "rxjs";
+import { filter, timeout } from "rxjs/operators";
+import { createOpenAI } from "@ai-sdk/openai";
+import { generateText, type ModelMessage } from "ai";
+import { AppSettings } from "../../../../common/app-setting";
+import { AuthService } from "../../../../common/service/user/auth.service";
+import { WorkflowActionService } from
"../../../service/workflow-graph/model/workflow-action.service";
+import { ExecuteWorkflowService } from
"../../../service/execute-workflow/execute-workflow.service";
+import { OperatorPortSchemaMap } from
"../../../types/workflow-compiling.interface";
+import { ExecutionState } from "../../../types/execute-workflow.interface";
+import { buildFixPrompt, withoutSecrets } from "./ai-fix-prompt";
+import { OperatorMetadataService } from
"../../../service/operator-metadata/operator-metadata.service";
+import { WorkflowConsoleService } from
"../../../service/workflow-console/workflow-console.service";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ /** How many places Apply will touch; the diff only ever shows one -/+ pair.
*/
+ occurrences?: number;
+ type: "code_change" | "property_change";
+ original: string;
+ suggested: string;
+ explanation: string;
+ // Rendered as the confidence badge; `fieldName` names the patched property
+ // (property_change only) so applyFix knows which field to write.
+ confidence: FixConfidence;
+ fieldName?: string;
+}
+
+export interface FixState {
+ operatorId: string;
+ /**
+ * Identifies the ERROR the suggestion was made for, so a newer one on the
same operator
+ * can clear it. A terminal execution state is not enough on its own: the
case this panel
+ * exists for leaves the run `Running` -- the UDF raises and the worker
pauses -- so a
+ * re-run that fails again never reaches one, and the error key is the only
signal left.
+ */
+ errorKey?: string;
+ /**
+ * The workflow the analysis belongs to. The operator id alone does not
identify it: two
+ * imports of the same JSON carry the same ids, so a reply that arrived
after the user
+ * switched workflows could be applied to the wrong one.
+ */
+ workflowId?: number;
+ /**
+ * Which of the four supported shapes the error was classified as. Only the
apply path
+ * reads it, to refuse a configuration change for anything but
`model_not_found`; an error
+ * outside those shapes is the `unsupported` status, not a value here.
+ */
+ errorType?: FixErrorType;
+ suggestedFix?: SuggestedFix;
+ // `unsupported`: the error is outside the four shapes this panel handles,
reported without
+ // spending a model call. `applied_without_run`: the fix was written but the
deployment
+ // refused to start a run (a warehouse is required and none is selected), so
the panel must
+ // not claim one. `apply_failed` keeps the suggestion on screen: the
analysis was fine, the
+ // write was refused, and re-analysing would spend another model call for
the same fix.
+ status:
+ | "idle"
+ | "analyzing"
+ | "unsupported"
+ | "ready"
+ | "applying"
+ | "applied"
+ | "applied_without_run"
+ | "error"
+ | "apply_failed";
+ /** Why the write was refused, shown as-is when the status is
`apply_failed`. */
+ applyError?: string;
+}
+
+// Model id as LiteLLM exposes it (bin/single-node/litellm-config.yaml), not
the provider's
+// id. A deployment whose proxy publishes a different alias has to change
this: there is no
+// GuiConfig field for it today, and adding one is a backend change.
+export const AI_FIXER_MODEL = "claude-haiku-4.5";
+export const AI_FIXER_TIMEOUT_MS = 120_000;
+export const UNLOCK_TIMEOUT_MS = 15_000;
+
+// A run in one of these is live enough to be worth stopping; a lock coming
from anything
+// else is not something killing an execution can clear.
+const LIVE_STATES: ReadonlySet<ExecutionState> = new Set([
+ ExecutionState.Initializing,
+ ExecutionState.Running,
+ ExecutionState.Pausing,
+ ExecutionState.Paused,
+ ExecutionState.Resuming,
+ ExecutionState.Recovering,
+]);
+
+// A run that reached one of these is over, so anything the panel still shows
predates it.
+// `Uninitialized` is deliberately absent even though updateWorkflowActionLock
treats it as
+// over: a kill lands there, and a kill is how Apply itself clears the lock,
so resetting on
+// it would throw away a suggestion that is still the right one for the error
on screen.
+const TERMINAL_STATES: ReadonlySet<ExecutionState> = new Set([
+ ExecutionState.Completed,
+ ExecutionState.Failed,
+ ExecutionState.Killed,
+ ExecutionState.Terminated,
+]);
+
+export const UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", status: "idle" };
+
+// Matched against the exception line only, never the frame lines above it.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+/**
+ * Replaces every occurrence, keeping each one at the indentation it was found
at.
+ *
+ * A single-line match starts mid-line, after the existing indentation, so a
multi-line
+ * replacement would put its first line in the right place and every line
after it at column
+ * 0 -- an IndentationError the moment the fix was applied. Those continuation
lines take the
+ * indentation of the line the match was found on, and keep whatever relative
indentation the
+ * suggestion itself had.
+ *
+ * A match that spans lines already carries the absolute indentation of the
region it came
+ * from, and the model's reply mirrors it, so there is nothing to add:
prefixing the context
+ * indentation on top would double it.
+ */
+export function replaceAllIndented(code: string, original: string, suggested:
string): { code: string; count: number } {
+ const parts = code.split(original);
+ const count = parts.length - 1;
+ if (count === 0 || !suggested.includes("\n") || original.includes("\n")) {
+ return { code: parts.join(suggested), count };
+ }
+
+ const [head, ...rest] = suggested.split("\n");
+ let out = parts[0];
+ for (const tail of parts.slice(1)) {
+ // Whitespace since the last newline is what this occurrence sits at.
+ const lineStart = out.lastIndexOf("\n") + 1;
+ const indent = (out.slice(lineStart).match(/^[ \t]*/) ?? [""])[0];
+ out += [head, ...rest.map(line => (line.length > 0 ? indent + line :
line))].join("\n") + tail;
+ }
+ return { code: out, count };
+}
+
+/** Writes the model's string back in the type the field already held. */
+export function coerceToFieldType(currentValue: unknown, suggested: string):
unknown {
+ // Arrays and objects are out of scope: the model replies with a string, and
writing it
+ // into a structured field replaced e.g. an outputColumns array with the
text "[]".
+ // Validating a replacement against the operator's schema is a bigger change
than the four
+ // error shapes this panel handles need, so it refuses instead of guessing.
+ if (currentValue !== null && typeof currentValue === "object") {
+ throw new Error("this setting holds structured data, which an automatic
fix does not edit");
+ }
+ if (typeof currentValue === "number") {
+ const parsed = Number(suggested);
+ if (Number.isNaN(parsed)) {
+ throw new Error(`${suggested} is not a number`);
+ }
+ return parsed;
+ }
+ if (typeof currentValue === "boolean") {
+ if (suggested !== "true" && suggested !== "false") {
+ throw new Error(`${suggested} is not a boolean`);
+ }
+ return suggested === "true";
+ }
+ return suggested;
+}
+
+// What a stack frame looks like, so it is never mistaken for the exception.
Python writes
+// File "udf.py", line 7, in process_tuple
+// and the JVM writes
+// org.apache.texera.Foo.run(Foo.scala:12)
+// which carries no colon-separated exception name, so a Python-only pattern
lets a Java
+// frame through and classifies on it.
+const FRAME_LINE = /^File\s+"|^Traceback \(|^at\s|^[\w$.]+\([\w.]+:\d+\)$/;
+
+/**
+ * The exception a traceback ends on.
+ *
+ * Matching the whole message would classify on whatever appears first, which
is wrong twice
+ * over: the frame lines carry paths and line numbers that read like patterns
-- a path
+ * containing "model" above a `line 404` entry looks like a model 404 -- and a
chained
+ * traceback reports the handled exception before the one that actually
stopped the worker.
+ * Python puts the real exception last.
+ */
+export function finalExceptionLine(errorMessage: string): string {
+ const lines = errorMessage
+ .split("\n")
+ .map(line => line.trim())
+ .filter(line => line.length > 0);
+ // Frame lines are the `File "...", line N` entries and the source line
under them; the
+ // exception is the last line that is neither that nor the traceback header.
+ for (let i = lines.length - 1; i >= 0; i--) {
+ if (!FRAME_LINE.test(lines[i])) {
+ return lines[i];
+ }
+ }
+ return errorMessage;
+}
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) =>
pattern.test(finalExceptionLine(errorMessage)))?.[1] ?? "unsupported";
+}
+
+/**
+ * Holds the AI-fix loop for one failed operator: classify -> ask the LLM ->
apply -> re-run.
+ *
+ * The state lives here rather than in the frame component because
+ * ResultPanelComponent.clearResultPanel() destroys and rebuilds the frames on
+ * every re-render, which would otherwise discard an in-flight suggestion.
+ */
+@Injectable({ providedIn: "root" })
+export class AiWorkflowFixerService {
+ private readonly stateSubject = new BehaviorSubject<FixState>(IDLE_STATE);
+
+ constructor(
+ private workflowActionService: WorkflowActionService,
+ private executeWorkflowService: ExecuteWorkflowService,
+ private operatorMetadataService: OperatorMetadataService,
+ private workflowConsoleService: WorkflowConsoleService
+ ) {
+ // The panel holds whatever it last showed until something clears it, and
after Apply
+ // that is the green "re-running..." message. Resetting when the run ends
-- rather than
+ // when the next one starts -- keeps that message truthful for exactly as
long as the run
+ // it describes, so a run that fails again does not reopen the tab on a
finished claim.
+ this.executeWorkflowService
+ .getExecutionStateStream()
+ .pipe(
+ filter(event => TERMINAL_STATES.has(event.current.state)),
+ // applyFix kills a live execution before it writes; that kill must
not wipe the
+ // state the apply is in the middle of publishing.
+ filter(() => this.stateSubject.getValue().status !== "applying")
+ )
+ .subscribe(() => this.reset());
+
+ // A re-run that fails again never reaches a terminal state, so the
subscription above
+ // never fires for it. The operator reporting a different error is the
signal that
+ // whatever the panel is showing belongs to a run that has already been
superseded.
+ this.workflowConsoleService.getConsoleMessageUpdateStream().subscribe(()
=> {
+ const current = this.stateSubject.getValue();
+ if (current.status === "idle" || current.status === "analyzing" ||
current.status === "applying") {
+ return;
+ }
+ const latest = this.latestErrorKey(current.operatorId);
+ if (latest && latest !== current.errorKey) {
+ this.reset();
+ }
+ });
+ }
+
+ public getState$(): Observable<FixState> {
+ return this.stateSubject.asObservable();
+ }
+
+ // Numbers the analyses so a slow reply cannot overwrite a newer one. A
frame whose
+ // operator does not own the state renders idle, so its Analyze button stays
live while
+ // another operator is mid-analysis -- two requests can be in flight at once.
+ private analysisSeq = 0;
+
+ public async analyzeError(
+ operatorId: string,
+ errorMessage: string,
+ inputSchemas: OperatorPortSchemaMap | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const workflowId = this.workflowActionService.getOpenWorkflowId();
+ const base: FixState = {
+ operatorId,
+ workflowId,
+ errorType,
+ errorKey: this.latestErrorKey(operatorId),
+ status: "analyzing",
+ };
+
+ // Numbered before the early return as well: this still supersedes
whatever is in flight,
+ // and without the bump a slower analysis for another operator would land
on top of it.
+ const seq = ++this.analysisSeq;
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
+ this.stateSubject.next({ ...base, status: "unsupported" });
+ return;
+ }
+
+ this.stateSubject.next(base);
+ try {
+ // One redacted view, used for both the prompt and the suggestion: toFix
reads the
+ // field's current value to show in the diff, and reading it from the
raw properties
+ // put a credential on screen whenever the model guessed one of those
names.
+ const visible = withoutSecrets(properties,
this.schemaProperties(operatorId));
+ const { text } = await
this.callModelWithTimeout(buildFixPrompt(errorMessage, code, inputSchemas,
visible));
+ if (seq !== this.analysisSeq || workflowId !==
this.workflowActionService.getOpenWorkflowId()) {
+ return;
+ }
+ const suggestedFix = this.toFix(text, visible);
+ // Counted here so the panel can say it before Apply: a short snippet
matched as plain
+ // text can appear in lines the user never saw in the diff.
+ if (suggestedFix.type === "code_change") {
+ const occurrences = replaceAllIndented(code ?? "",
suggestedFix.original, suggestedFix.suggested).count;
+ // Zero means the model quoted something the operator does not
contain. The panel
+ // only surfaces this count when it is above one, so such a suggestion
is
+ // indistinguishable from any other, and Apply turns it away as a
stale snippet --
+ // blaming an edit that never happened.
+ if (occurrences === 0) {
+ throw new Error(`the code does not contain
${JSON.stringify(suggestedFix.original)}`);
+ }
+ suggestedFix.occurrences = occurrences;
+ }
+ this.stateSubject.next({ ...base, status: "ready", suggestedFix });
+ } catch (err) {
+ console.error("AI workflow fixer: analysis failed", err);
+ if (seq !== this.analysisSeq || workflowId !==
this.workflowActionService.getOpenWorkflowId()) {
+ return;
+ }
+ this.stateSubject.next({ ...base, status: "error" });
+ }
+ }
+
+ /**
+ * Applies the suggestion to the live operator and re-runs the workflow.
+ * Properties are re-read from the graph so a concurrent edit is not
clobbered.
+ */
+ public async applyFix(): Promise<void> {
+ const current = this.stateSubject.getValue();
+ const fix = current.suggestedFix;
+ // `apply_failed` is retryable: the panel keeps the suggestion and its
Apply button, so
+ // refusing here left the user clicking something that silently did
nothing. Everything
+ // below re-reads the operator and re-runs every check, so a retry is not
a shortcut.
+ if (!fix || (current.status !== "ready" && current.status !==
"apply_failed")) {
+ return;
+ }
+ // Read, not bumped. The unlock wait below can take seconds and an
analysis started in
+ // the meantime owns the panel, so this apply must not report over it.
(Bumping would be
+ // equivalent today -- an analysis in flight has already left the status
at `analyzing`,
+ // which the guard above turns away -- but reading says what is meant.)
+ const seq = this.analysisSeq;
+ this.stateSubject.next({ ...current, status: "applying" });
+ try {
+ this.assertSameWorkflow(current);
+ await this.ensureModifiable();
+ // Checked again: ensureModifiable can wait seconds for a kill to clear
the lock, and
+ // the user is free to open another workflow while it does.
+ this.assertSameWorkflow(current);
+ const properties: Record<string, unknown> = {
+
...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties,
+ };
+ if (fix.type === "code_change") {
+ const code = String(properties.code ?? "");
+ const patched = replaceAllIndented(code, fix.original, fix.suggested);
+ if (patched.count === 0) {
+ throw new Error("the code changed since the suggestion was
generated");
+ }
+ // Counted, not just found. The count behind the "changes N places"
badge was taken at
+ // the analysis, and this replaces every match -- so a copy added or
removed in between
+ // means the write touches a different number of lines than the panel
promised. A kill
+ // lifts the modification lock without resetting the panel, so the
user can edit the
+ // UDF with the suggestion still on screen.
+ if (fix.occurrences !== undefined && patched.count !==
fix.occurrences) {
+ const places = (n: number) => `${n} place${n === 1 ? "" : "s"}`;
+ throw new Error(
+ `the snippet is now in ${places(patched.count)} instead of
${places(fix.occurrences)}, ` +
+ "so the fix was not applied"
+ );
+ }
+ properties.code = patched.code;
+ } else {
+ // Only the model-not-found case asks for a configuration change; a
property fix for
+ // any other error is the model leaving the four shapes this panel
supports.
+ if (current.errorType !== "model_not_found") {
+ throw new Error("a configuration change is not the fix for this
error");
+ }
Review Comment:
This check only runs at Apply so a configuration-change suggestion for a
non-model error is still shown and then refused which is what the new comment
in `toFix` says shouldn't happen. Also, every refusal in this method (this one,
the field checks, the snippet count, the type coercion) now runs after
`ensureModifiable()` has already killed the paused run so a refused fix stops
the user's run for nothing. One shared validation step, used at analysis and
again before the kill, would cover both.
--
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]
