kunwp1 commented on code in PR #8637:
URL: https://github.com/apache/texera/pull/8637#discussion_r4213865804
##########
frontend/src/app/workspace/component/result-panel/result-panel.component.ts:
##########
@@ -241,7 +244,12 @@ export class ResultPanelComponent implements OnInit,
OnDestroy {
this.workflowCompilingService.getCompilationStateInfoChangedStream(),
this.workflowActionService.getJointGraphWrapper().getJointOperatorHighlightStream(),
this.workflowActionService.getJointGraphWrapper().getJointOperatorUnhighlightStream(),
- this.workflowResultService.getResultInitiateStream()
+ this.workflowResultService.getResultInitiateStream(),
+ // A Python UDF that raises while its operator is already selected
changes nothing in
+ // the streams above: the execution stays Running and the highlight
never moves, so
+ // without this the AI Fix tab would not appear until the user clicked
away and back.
+ // Debounced because a chatty UDF's prints share this stream.
+
this.workflowConsoleService.getConsoleMessageUpdateStream().pipe(debounceTime(200))
)
.pipe(untilDestroyed(this))
.subscribe(_ => {
Review Comment:
Related to the comment above. This subscriber calls
`registerOperatorDisplayNameChangeHandler()` every time and each call adds a
new `untilDestroyed` subscription without releasing the previous one. With the
console stream added it grows a lot. I think it's worth registering that
handler once.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-prompt.ts:
##########
@@ -0,0 +1,69 @@
+/**
+ * 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 { PortSchema } from "../../../types/workflow-compiling.interface";
+
+/**
+ * Builds the single user message sent to the LLM for one failed operator.
+ *
+ * The model is asked for exactly one minimal fix and for raw JSON. The reply
is
+ * still parsed defensively (fenced blocks are unwrapped) because "no markdown"
+ * is a request, not a guarantee.
+ */
+export function buildFixPrompt(
+ errorMessage: string,
+ operatorCode: string | undefined,
+ schema: PortSchema | undefined,
+ operatorProperties: Readonly<Record<string, unknown>>
+): string {
+ return `You are a debugging assistant for a Python data processing operator
in a workflow engine.
+
+Error
+${errorMessage}
+
+Operator code
+\`\`\`python
+${operatorCode ?? "(this operator has no user code)"}
+\`\`\`
+
+Input schema (column names and types)
+${JSON.stringify(schema ?? [], null, 2)}
+
+Operator configuration
+${JSON.stringify(operatorProperties, null, 2)}
Review Comment:
I have a concern that the credentials can leave the browser in the prompt.
`operatorProperties` is serialized in full, and some operators keep secrets
there, e.g. `hfApiToken` on HuggingFaceInference and the SQL/Reddit/Twitter
sources. I suggest sending only the fields the fix needs or dropping
password-widget fields.
##########
frontend/src/app/workspace/component/result-panel/result-panel.component.ts:
##########
@@ -241,7 +244,12 @@ export class ResultPanelComponent implements OnInit,
OnDestroy {
this.workflowCompilingService.getCompilationStateInfoChangedStream(),
this.workflowActionService.getJointGraphWrapper().getJointOperatorHighlightStream(),
this.workflowActionService.getJointGraphWrapper().getJointOperatorUnhighlightStream(),
- this.workflowResultService.getResultInitiateStream()
+ this.workflowResultService.getResultInitiateStream(),
+ // A Python UDF that raises while its operator is already selected
changes nothing in
+ // the streams above: the execution stays Running and the highlight
never moves, so
+ // without this the AI Fix tab would not appear until the user clicked
away and back.
+ // Debounced because a chatty UDF's prints share this stream.
+
this.workflowConsoleService.getConsoleMessageUpdateStream().pipe(debounceTime(200))
Review Comment:
This re-renders the whole panel on every console update, from any operator.
A rerender resets the `Console`/`Result` entries and `keyvalue` + `*ngFor`
rebuilds every tab, so the selected operator's Console frame (type filter,
debug input, scroll) and Result table get reset whenever any operator prints.
`debounceTime` can also starve the AI Fix tab while something keeps printing. I
suggest narrowing this to the selected operator's "has ERROR" boolean (with
`distinctUntilChanged`) and only adding/removing the `AI Fix` entry instead of
running the full `rerenderResultPanel()`.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.html:
##########
@@ -0,0 +1,121 @@
+<!--
+ 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.
+-->
+
+<ng-container *ngIf="state$ | async as state">
+ <div
+ class="ai-fix-status"
+ *ngIf="state.status === 'analyzing' || state.status === 'applying'">
+ <nz-spin nzSize="small"></nz-spin>
+ <span>{{ state.status === "analyzing" ? "Analyzing error with AI..." :
"Applying fix..." }}</span>
+ </div>
+
+ <!-- Nothing has been analysed yet (or the user discarded the last
suggestion). -->
+ <div
+ class="ai-fix-start"
+ *ngIf="
+ state.status === 'idle' ||
+ state.status === 'applied' ||
+ state.status === 'applied_without_run' ||
+ state.status === 'error'
+ ">
+ <button
+ nz-button
+ nzType="primary"
+ class="ai-fix-analyze"
+ [disabled]="!operatorId"
+ (click)="onAnalyze()">
+ <span
+ nz-icon
+ nzType="robot"></span>
+ Analyze with AI
+ </button>
+ </div>
+
+ <nz-alert
+ *ngIf="state.status === 'ready' && state.errorType === 'unsupported'"
Review Comment:
The "unsupported" result is a dead end. The Analyze button only renders for
idle/applied/applied_without_run/error so after an unsupported error there is
no button and nothing resets the state. If the user fixes the code and re-runs
into a supported error, this tab keeps showing the stale message until some
other operator is analyzed. Suggest showing Analyze for `ready`-unsupported too
or resetting the state when a new run starts.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) => pattern.test(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 warehouseService: WarehouseService,
+ private config: GuiConfigService
+ ) {}
+
+ 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,
+ schema: PortSchema | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const base: FixState = {
+ operatorId,
+ errorMessage,
+ errorType,
+ originalCode: code,
+ originalProperties: properties,
+ status: "analyzing",
+ };
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
+ this.stateSubject.next({ ...base, status: "ready" });
+ return;
+ }
+
+ const seq = ++this.analysisSeq;
+ this.stateSubject.next(base);
+ try {
+ const { text } = await
this.callModelWithTimeout(buildFixPrompt(errorMessage, code, schema,
properties));
+ if (seq !== this.analysisSeq) {
+ return;
+ }
+ this.stateSubject.next({ ...base, status: "ready", suggestedFix:
this.toFix(text, properties) });
+ } catch (err) {
+ console.error("AI workflow fixer: analysis failed", err);
+ if (seq !== this.analysisSeq) {
+ 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 applyFix(): void {
+ const current = this.stateSubject.getValue();
+ const fix = current.suggestedFix;
+ if (!fix || current.status !== "ready") {
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applying" });
+ try {
+ const properties: Record<string, unknown> = {
+
...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties,
+ };
+ if (fix.type === "code_change") {
+ const code = String(properties.code ?? "");
+ if (!code.includes(fix.original)) {
+ throw new Error("the code changed since the suggestion was
generated");
+ }
+ // split/join replaces every occurrence and keeps the match literal;
replace() with a
+ // string argument would patch only the first one and leave the rest
failing.
+ properties.code = code.split(fix.original).join(fix.suggested);
+ } else {
+ const field = String(fix.fieldName);
+ // Same staleness guard as the code branch: re-reading the properties
keeps the
+ // unrelated fields, but the field being replaced can itself have
moved on since
+ // the suggestion was generated, and overwriting it would discard that
newer value.
+ if (String(properties[field] ?? "") !== fix.original) {
+ throw new Error(`${field} changed since the suggestion was
generated`);
+ }
+ properties[field] = fix.suggested;
+ }
+ this.workflowActionService.setOperatorProperty(current.operatorId,
properties);
+ if (!this.canStartRun()) {
+ // ExecuteWorkflowService refuses the run and shows its own toast;
reporting
+ // "re-running" here would contradict it.
+ this.stateSubject.next({ ...current, status: "applied_without_run" });
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applied" });
+ // Same empty execution name the operator menu uses for an ad-hoc run.
+ this.executeWorkflowService.executeWorkflow("");
+ } catch (err) {
+ console.error("AI workflow fixer: apply failed", err);
+ this.stateSubject.next({ ...current, status: "error" });
+ }
+ }
+
+ /**
+ * Mirrors ExecuteWorkflowService's own entry guard: while the deployment
requires a
+ * warehouse and none is selected, a run request is refused before it starts.
+ */
+ private canStartRun(): boolean {
Review Comment:
This copies `executeWorkflow`'s refusal logic instead of asking it.
`canStartRun()` mirrors only the warehouse guard, but `ExecuteWorkflowService`
on `main` also has `refuseToRunOnUnavailableUnit()` (terminating/unavailable
computing unit). In that case the panel would say "workflow re-running..."
while a toast says the run was refused, which is the contradiction this PR is
trying to avoid. I suggest having `executeWorkflow` return whether it actually
started or expose a public `canRun()` so the guards live in one place.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) => pattern.test(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 warehouseService: WarehouseService,
+ private config: GuiConfigService
+ ) {}
+
+ 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,
+ schema: PortSchema | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const base: FixState = {
+ operatorId,
+ errorMessage,
+ errorType,
+ originalCode: code,
+ originalProperties: properties,
+ status: "analyzing",
+ };
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
Review Comment:
This early return doesn't bump `analysisSeq` so an in-flight analysis for
another operator can still land afterwards and overwrite this "unsupported"
state. Maybe consider incrementing `analysisSeq` before returning.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) => pattern.test(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 warehouseService: WarehouseService,
+ private config: GuiConfigService
+ ) {}
+
+ 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,
+ schema: PortSchema | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const base: FixState = {
+ operatorId,
+ errorMessage,
+ errorType,
+ originalCode: code,
+ originalProperties: properties,
+ status: "analyzing",
+ };
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
+ this.stateSubject.next({ ...base, status: "ready" });
+ return;
+ }
+
+ const seq = ++this.analysisSeq;
+ this.stateSubject.next(base);
+ try {
+ const { text } = await
this.callModelWithTimeout(buildFixPrompt(errorMessage, code, schema,
properties));
+ if (seq !== this.analysisSeq) {
+ return;
+ }
+ this.stateSubject.next({ ...base, status: "ready", suggestedFix:
this.toFix(text, properties) });
+ } catch (err) {
+ console.error("AI workflow fixer: analysis failed", err);
+ if (seq !== this.analysisSeq) {
+ 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 applyFix(): void {
+ const current = this.stateSubject.getValue();
+ const fix = current.suggestedFix;
+ if (!fix || current.status !== "ready") {
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applying" });
+ try {
+ const properties: Record<string, unknown> = {
+
...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties,
+ };
+ if (fix.type === "code_change") {
+ const code = String(properties.code ?? "");
+ if (!code.includes(fix.original)) {
+ throw new Error("the code changed since the suggestion was
generated");
+ }
+ // split/join replaces every occurrence and keeps the match literal;
replace() with a
+ // string argument would patch only the first one and leave the rest
failing.
+ properties.code = code.split(fix.original).join(fix.suggested);
+ } else {
+ const field = String(fix.fieldName);
+ // Same staleness guard as the code branch: re-reading the properties
keeps the
+ // unrelated fields, but the field being replaced can itself have
moved on since
+ // the suggestion was generated, and overwriting it would discard that
newer value.
+ if (String(properties[field] ?? "") !== fix.original) {
Review Comment:
The field name comes straight from the model and is never checked. For a
property change the model replies with a field name and a new value and Apply
writes that value to whatever field was named. If the model names a field the
operator doesn't have (say modelName instead of modelId), Apply adds a new,
unused key and the real field stays wrong, so the re-run fails the same way. If
it names `code`, the whole UDF source is replaced by the new value. And since
the new value is always a string, a numeric or boolean setting (e.g.
temperature: 0) would be overwritten with a string.
I suggest only allowing field names that already exist on the operator (and
aren't `code`), and keeping the original value's type when writing.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
Review Comment:
I think this line has an issue. For example a `File
".../model_selection/_split.py", line 404` line followed by a
`ZeroDivisionError` matches `model_not_found` (`\bmodel...404` matches any path
containing "model" on a line-number line), and a chained `KeyError ... During
handling of the above exception ... ZeroDivisionError` classifies as
`missing_column`. I suggest classifying on the final exception line of the
traceback.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) => pattern.test(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 warehouseService: WarehouseService,
+ private config: GuiConfigService
+ ) {}
+
+ 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,
+ schema: PortSchema | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const base: FixState = {
+ operatorId,
+ errorMessage,
+ errorType,
+ originalCode: code,
+ originalProperties: properties,
+ status: "analyzing",
+ };
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
+ this.stateSubject.next({ ...base, status: "ready" });
+ return;
+ }
+
+ const seq = ++this.analysisSeq;
+ this.stateSubject.next(base);
+ try {
+ const { text } = await
this.callModelWithTimeout(buildFixPrompt(errorMessage, code, schema,
properties));
+ if (seq !== this.analysisSeq) {
+ return;
+ }
+ this.stateSubject.next({ ...base, status: "ready", suggestedFix:
this.toFix(text, properties) });
+ } catch (err) {
+ console.error("AI workflow fixer: analysis failed", err);
+ if (seq !== this.analysisSeq) {
+ 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 applyFix(): void {
+ const current = this.stateSubject.getValue();
+ const fix = current.suggestedFix;
+ if (!fix || current.status !== "ready") {
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applying" });
+ try {
+ const properties: Record<string, unknown> = {
+
...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties,
+ };
+ if (fix.type === "code_change") {
+ const code = String(properties.code ?? "");
+ if (!code.includes(fix.original)) {
+ throw new Error("the code changed since the suggestion was
generated");
+ }
+ // split/join replaces every occurrence and keeps the match literal;
replace() with a
+ // string argument would patch only the first one and leave the rest
failing.
+ properties.code = code.split(fix.original).join(fix.suggested);
+ } else {
+ const field = String(fix.fieldName);
+ // Same staleness guard as the code branch: re-reading the properties
keeps the
+ // unrelated fields, but the field being replaced can itself have
moved on since
+ // the suggestion was generated, and overwriting it would discard that
newer value.
+ if (String(properties[field] ?? "") !== fix.original) {
+ throw new Error(`${field} changed since the suggestion was
generated`);
+ }
+ properties[field] = fix.suggested;
+ }
+ this.workflowActionService.setOperatorProperty(current.operatorId,
properties);
+ if (!this.canStartRun()) {
+ // ExecuteWorkflowService refuses the run and shows its own toast;
reporting
+ // "re-running" here would contradict it.
+ this.stateSubject.next({ ...current, status: "applied_without_run" });
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applied" });
Review Comment:
After Apply the stored status stays applied until the user discards or
analyzes again. If the re-run fails again, the AI Fix tab reappears showing the
green "Fix applied — workflow re-running..." message next to Analyze even
though nothing is re-running. I suggest resetting to idle when a new execution
starts (e.g. on `Initializing`) or when the operator's console is cleared.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) => pattern.test(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 warehouseService: WarehouseService,
+ private config: GuiConfigService
+ ) {}
+
+ 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,
+ schema: PortSchema | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const base: FixState = {
+ operatorId,
+ errorMessage,
+ errorType,
+ originalCode: code,
+ originalProperties: properties,
+ status: "analyzing",
+ };
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
+ this.stateSubject.next({ ...base, status: "ready" });
+ return;
+ }
+
+ const seq = ++this.analysisSeq;
+ this.stateSubject.next(base);
+ try {
+ const { text } = await
this.callModelWithTimeout(buildFixPrompt(errorMessage, code, schema,
properties));
+ if (seq !== this.analysisSeq) {
+ return;
+ }
+ this.stateSubject.next({ ...base, status: "ready", suggestedFix:
this.toFix(text, properties) });
+ } catch (err) {
+ console.error("AI workflow fixer: analysis failed", err);
+ if (seq !== this.analysisSeq) {
+ 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 applyFix(): void {
+ const current = this.stateSubject.getValue();
+ const fix = current.suggestedFix;
+ if (!fix || current.status !== "ready") {
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applying" });
+ try {
+ const properties: Record<string, unknown> = {
+
...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties,
+ };
+ if (fix.type === "code_change") {
+ const code = String(properties.code ?? "");
+ if (!code.includes(fix.original)) {
+ throw new Error("the code changed since the suggestion was
generated");
+ }
+ // split/join replaces every occurrence and keeps the match literal;
replace() with a
+ // string argument would patch only the first one and leave the rest
failing.
+ properties.code = code.split(fix.original).join(fix.suggested);
+ } else {
+ const field = String(fix.fieldName);
+ // Same staleness guard as the code branch: re-reading the properties
keeps the
+ // unrelated fields, but the field being replaced can itself have
moved on since
+ // the suggestion was generated, and overwriting it would discard that
newer value.
+ if (String(properties[field] ?? "") !== fix.original) {
+ throw new Error(`${field} changed since the suggestion was
generated`);
+ }
+ properties[field] = fix.suggested;
+ }
+ this.workflowActionService.setOperatorProperty(current.operatorId,
properties);
Review Comment:
While the execution is Running/Paused/Initializing,
`ExecuteWorkflowService.updateWorkflowActionLock` disables workflow
modification, and this PR's trigger is a UDF traceback while the execution is
still `Running`. Every other edit related features check
`WorkflowActionService.checkWorkflowModificationEnabled()` first. Here the
Apply button is never disabled and `setOperatorProperty` is called regardless.
I suggest disabling Apply while modification is locked, or stopping the run
first (see the earlier comment on re-running over a live execution).
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.ts:
##########
@@ -0,0 +1,154 @@
+/**
+ * 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 { Component, Input } from "@angular/core";
+import { AsyncPipe, NgFor, NgIf } from "@angular/common";
+import { BehaviorSubject, Observable, combineLatest } from "rxjs";
+import { map } from "rxjs/operators";
+import { NzAlertComponent } from "ng-zorro-antd/alert";
+import { NzSpinComponent } from "ng-zorro-antd/spin";
+import { NzTagComponent } from "ng-zorro-antd/tag";
+import { NzButtonComponent } from "ng-zorro-antd/button";
+import { NzWaveDirective } from "ng-zorro-antd/core/wave";
+import { NzIconDirective } from "ng-zorro-antd/icon";
+import { ExecuteWorkflowService } from
"../../../service/execute-workflow/execute-workflow.service";
+import { WorkflowActionService } from
"../../../service/workflow-graph/model/workflow-action.service";
+import { WorkflowCompilingService } from
"../../../service/compile-workflow/workflow-compiling.service";
+import { WorkflowConsoleService } from
"../../../service/workflow-console/workflow-console.service";
+import {
+ AiWorkflowFixerService,
+ FixConfidence,
+ FixState,
+ IDLE_STATE,
+ UNSUPPORTED_MESSAGE,
+} from "./ai-workflow-fixer.service";
+
+/**
+ * Result-panel frame that asks the LLM to fix the failed operator.
+ *
+ * The analysis is started by an explicit button: ResultPanelComponent rebuilds
+ * every frame on each re-render (clearResultPanel), so analysing on init would
+ * fire a model request per re-render. For the same reason all state lives in
+ * AiWorkflowFixerService and this component keeps none.
+ */
+@Component({
+ selector: "texera-ai-fix-frame",
+ templateUrl: "./ai-fix-frame.component.html",
+ styleUrls: ["./ai-fix-frame.component.scss"],
+ imports: [
+ NgIf,
+ NgFor,
+ AsyncPipe,
+ NzAlertComponent,
+ NzSpinComponent,
+ NzTagComponent,
+ NzButtonComponent,
+ NzWaveDirective,
+ NzIconDirective,
+ ],
+})
+export class AiFixFrameComponent {
+ private readonly operatorIdSubject = new BehaviorSubject<string |
undefined>(undefined);
+
+ @Input()
+ set operatorId(value: string | undefined) {
+ this.operatorIdSubject.next(value);
+ }
+
+ get operatorId(): string | undefined {
+ return this.operatorIdSubject.value;
+ }
+
+ public readonly state$: Observable<FixState>;
+ public readonly unsupportedMessage = UNSUPPORTED_MESSAGE;
+
+ constructor(
+ private aiWorkflowFixerService: AiWorkflowFixerService,
+ private executeWorkflowService: ExecuteWorkflowService,
+ private workflowActionService: WorkflowActionService,
+ private workflowCompilingService: WorkflowCompilingService,
+ private workflowConsoleService: WorkflowConsoleService
+ ) {
+ // The service is a root singleton holding one suggestion, but a failed
pipeline can
+ // put an AI Fix tab on several operators at once. Showing another
operator's
+ // suggestion here would let Apply patch an operator the user is not
looking at, so a
+ // frame only ever renders state that names its own operator.
+ this.state$ = combineLatest([aiWorkflowFixerService.getState$(),
this.operatorIdSubject]).pipe(
+ map(([state, operatorId]) => (state.operatorId === operatorId ? state :
IDLE_STATE))
+ );
+ }
+
+ /** Collects the operator's error, code, schema and config, then asks for a
fix. */
+ public onAnalyze(): void {
+ const operatorId = this.operatorId;
+ if (!operatorId) {
+ return;
+ }
+ const errorMessage = this.rawErrorMessage(operatorId);
+ if (!errorMessage) {
+ return;
+ }
+ const properties =
this.workflowActionService.getTexeraGraph().getOperator(operatorId).operatorProperties
?? {};
+ void this.aiWorkflowFixerService.analyzeError(
+ operatorId,
+ errorMessage,
+ this.workflowCompilingService.getPortInputSchema(operatorId, 0),
Review Comment:
Only input port 0's schema reaches the model.
`getPortInputSchema(operatorId, 0)` ignores the other ports, but Python UDFs
can have several input ports. Can you consider this case too?
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) => pattern.test(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 warehouseService: WarehouseService,
+ private config: GuiConfigService
+ ) {}
+
+ 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,
+ schema: PortSchema | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const base: FixState = {
+ operatorId,
+ errorMessage,
+ errorType,
+ originalCode: code,
+ originalProperties: properties,
+ status: "analyzing",
+ };
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
+ this.stateSubject.next({ ...base, status: "ready" });
+ return;
+ }
+
+ const seq = ++this.analysisSeq;
+ this.stateSubject.next(base);
+ try {
+ const { text } = await
this.callModelWithTimeout(buildFixPrompt(errorMessage, code, schema,
properties));
+ if (seq !== this.analysisSeq) {
+ return;
+ }
+ this.stateSubject.next({ ...base, status: "ready", suggestedFix:
this.toFix(text, properties) });
+ } catch (err) {
+ console.error("AI workflow fixer: analysis failed", err);
+ if (seq !== this.analysisSeq) {
+ 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 applyFix(): void {
+ const current = this.stateSubject.getValue();
+ const fix = current.suggestedFix;
+ if (!fix || current.status !== "ready") {
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applying" });
+ try {
+ const properties: Record<string, unknown> = {
+
...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties,
+ };
+ if (fix.type === "code_change") {
+ const code = String(properties.code ?? "");
+ if (!code.includes(fix.original)) {
+ throw new Error("the code changed since the suggestion was
generated");
+ }
+ // split/join replaces every occurrence and keeps the match literal;
replace() with a
+ // string argument would patch only the first one and leave the rest
failing.
+ properties.code = code.split(fix.original).join(fix.suggested);
+ } else {
+ const field = String(fix.fieldName);
+ // Same staleness guard as the code branch: re-reading the properties
keeps the
+ // unrelated fields, but the field being replaced can itself have
moved on since
+ // the suggestion was generated, and overwriting it would discard that
newer value.
+ if (String(properties[field] ?? "") !== fix.original) {
+ throw new Error(`${field} changed since the suggestion was
generated`);
+ }
+ properties[field] = fix.suggested;
+ }
+ this.workflowActionService.setOperatorProperty(current.operatorId,
properties);
+ if (!this.canStartRun()) {
+ // ExecuteWorkflowService refuses the run and shows its own toast;
reporting
+ // "re-running" here would contradict it.
+ this.stateSubject.next({ ...current, status: "applied_without_run" });
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applied" });
+ // Same empty execution name the operator menu uses for an ad-hoc run.
+ this.executeWorkflowService.executeWorkflow("");
+ } catch (err) {
+ console.error("AI workflow fixer: apply failed", err);
+ this.stateSubject.next({ ...current, status: "error" });
+ }
+ }
+
+ /**
+ * Mirrors ExecuteWorkflowService's own entry guard: while the deployment
requires a
+ * warehouse and none is selected, a run request is refused before it starts.
+ */
+ private canStartRun(): boolean {
+ return !this.config.env.warehouseEnabled ||
this.warehouseService.getSelectedWarehouseIdValue() !== undefined;
+ }
+
+ public discardFix(): void {
+ this.stateSubject.next(IDLE_STATE);
+ }
+
+ // Seam over the `ai` transport: specs spy this instead of mocking the "ai"
module,
+ // which leaks across specs sharing the import and hangs on a real network
call.
+ protected callModel(messages: ModelMessage[], abortSignal?: AbortSignal):
Promise<{ text: string }> {
+ // Built per call so a refreshed access token is used: caching the
provider on this root
+ // singleton would pin the JWT captured on the first Analyze, and a later
call after a
+ // token rotation would 401. The /api/chat/* LiteLLM proxy authenticates
with the Texera
+ // JWT and swaps in the master key upstream, so that token is the only
credential sent.
+ const model = createOpenAI({
+ baseURL: new URL(`${AppSettings.getApiEndpoint()}`,
document.baseURI).toString(),
+ apiKey: AuthService.getAccessToken() ?? "",
+ }).chat(AI_FIXER_MODEL);
+ return generateText({ model, messages, abortSignal });
+ }
+
+ private callModelWithTimeout(prompt: string): Promise<{ text: string }> {
+ const controller = new AbortController();
+ let timer: ReturnType<typeof setTimeout>;
+ const timeout = new Promise<never>((_resolve, reject) => {
+ timer = setTimeout(() => {
+ controller.abort();
+ reject(new Error("AI fix request timed out"));
+ }, AI_FIXER_TIMEOUT_MS);
+ });
+ return Promise.race([this.callModel([{ role: "user", content: prompt }],
controller.signal), timeout]).finally(() =>
+ clearTimeout(timer)
+ );
+ }
+
+ /** Maps the model's JSON contract onto SuggestedFix, rejecting incomplete
replies. */
+ private toFix(raw: string, properties: Readonly<Record<string, unknown>>):
SuggestedFix {
+ let text = raw.trim();
+ const fenced = text.match(/```(?:[a-zA-Z]+)?\s*([\s\S]*?)```/);
+ if (fenced) {
+ text = fenced[1].trim();
+ }
+ let parsed: any;
+ try {
+ parsed = JSON.parse(text);
+ } catch (err) {
+ throw new Error(`model did not return JSON: ${(err as Error).message}`);
+ }
+ const { explanation, fix_type, original_snippet, suggested_snippet,
confidence } = parsed ?? {};
Review Comment:
I suggest validating `typeof === "string"` for the text fields here and
treating anything else as an error. If `original_snippet`/`suggested_snippet`
come back as an array (or any non-string), `splitLines()` in the template
throws during change detection and the panel is stuck in `ready`.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) => pattern.test(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 warehouseService: WarehouseService,
+ private config: GuiConfigService
+ ) {}
+
+ 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,
+ schema: PortSchema | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const base: FixState = {
+ operatorId,
+ errorMessage,
+ errorType,
+ originalCode: code,
+ originalProperties: properties,
+ status: "analyzing",
+ };
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
+ this.stateSubject.next({ ...base, status: "ready" });
+ return;
+ }
+
+ const seq = ++this.analysisSeq;
+ this.stateSubject.next(base);
+ try {
+ const { text } = await
this.callModelWithTimeout(buildFixPrompt(errorMessage, code, schema,
properties));
+ if (seq !== this.analysisSeq) {
+ return;
+ }
+ this.stateSubject.next({ ...base, status: "ready", suggestedFix:
this.toFix(text, properties) });
+ } catch (err) {
+ console.error("AI workflow fixer: analysis failed", err);
+ if (seq !== this.analysisSeq) {
+ 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 applyFix(): void {
+ const current = this.stateSubject.getValue();
+ const fix = current.suggestedFix;
+ if (!fix || current.status !== "ready") {
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applying" });
+ try {
+ const properties: Record<string, unknown> = {
+
...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties,
+ };
+ if (fix.type === "code_change") {
+ const code = String(properties.code ?? "");
+ if (!code.includes(fix.original)) {
+ throw new Error("the code changed since the suggestion was
generated");
+ }
+ // split/join replaces every occurrence and keeps the match literal;
replace() with a
+ // string argument would patch only the first one and leave the rest
failing.
+ properties.code = code.split(fix.original).join(fix.suggested);
+ } else {
+ const field = String(fix.fieldName);
+ // Same staleness guard as the code branch: re-reading the properties
keeps the
+ // unrelated fields, but the field being replaced can itself have
moved on since
+ // the suggestion was generated, and overwriting it would discard that
newer value.
+ if (String(properties[field] ?? "") !== fix.original) {
+ throw new Error(`${field} changed since the suggestion was
generated`);
+ }
+ properties[field] = fix.suggested;
+ }
+ this.workflowActionService.setOperatorProperty(current.operatorId,
properties);
+ if (!this.canStartRun()) {
+ // ExecuteWorkflowService refuses the run and shows its own toast;
reporting
+ // "re-running" here would contradict it.
+ this.stateSubject.next({ ...current, status: "applied_without_run" });
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applied" });
+ // Same empty execution name the operator menu uses for an ad-hoc run.
+ this.executeWorkflowService.executeWorkflow("");
+ } catch (err) {
+ console.error("AI workflow fixer: apply failed", err);
+ this.stateSubject.next({ ...current, status: "error" });
Review Comment:
A failed Apply lands in the same `error` status as a failed analysis, so the
panel says "The AI could not suggest a fix..." and hides the suggestion forcing
another model call.
I suggest a different failure message (e.g., "the code changed since the
suggestion") that keeps the suggestion visible and ideally checking that the
snippet exists in the code when the reply is parsed rather than at Apply time.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
Review Comment:
`originalCode`, `originalProperties` and `errorMessage` are written in
`analyzeError` but never read. Can you consider dropping them?
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) => pattern.test(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 warehouseService: WarehouseService,
+ private config: GuiConfigService
+ ) {}
+
+ 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,
+ schema: PortSchema | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const base: FixState = {
+ operatorId,
+ errorMessage,
+ errorType,
+ originalCode: code,
+ originalProperties: properties,
+ status: "analyzing",
+ };
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
+ this.stateSubject.next({ ...base, status: "ready" });
+ return;
+ }
+
+ const seq = ++this.analysisSeq;
+ this.stateSubject.next(base);
+ try {
+ const { text } = await
this.callModelWithTimeout(buildFixPrompt(errorMessage, code, schema,
properties));
+ if (seq !== this.analysisSeq) {
+ return;
+ }
+ this.stateSubject.next({ ...base, status: "ready", suggestedFix:
this.toFix(text, properties) });
+ } catch (err) {
+ console.error("AI workflow fixer: analysis failed", err);
+ if (seq !== this.analysisSeq) {
+ 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 applyFix(): void {
+ const current = this.stateSubject.getValue();
+ const fix = current.suggestedFix;
+ if (!fix || current.status !== "ready") {
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applying" });
+ try {
+ const properties: Record<string, unknown> = {
+
...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties,
+ };
+ if (fix.type === "code_change") {
+ const code = String(properties.code ?? "");
+ if (!code.includes(fix.original)) {
+ throw new Error("the code changed since the suggestion was
generated");
+ }
+ // split/join replaces every occurrence and keeps the match literal;
replace() with a
+ // string argument would patch only the first one and leave the rest
failing.
+ properties.code = code.split(fix.original).join(fix.suggested);
+ } else {
+ const field = String(fix.fieldName);
+ // Same staleness guard as the code branch: re-reading the properties
keeps the
+ // unrelated fields, but the field being replaced can itself have
moved on since
+ // the suggestion was generated, and overwriting it would discard that
newer value.
+ if (String(properties[field] ?? "") !== fix.original) {
+ throw new Error(`${field} changed since the suggestion was
generated`);
+ }
+ properties[field] = fix.suggested;
+ }
+ this.workflowActionService.setOperatorProperty(current.operatorId,
properties);
+ if (!this.canStartRun()) {
+ // ExecuteWorkflowService refuses the run and shows its own toast;
reporting
+ // "re-running" here would contradict it.
+ this.stateSubject.next({ ...current, status: "applied_without_run" });
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applied" });
+ // Same empty execution name the operator menu uses for an ad-hoc run.
+ this.executeWorkflowService.executeWorkflow("");
Review Comment:
The PR description saysa UDF exception leaves the execution `Running`
(paused worker), and the menu only offers Pause/Resume in that state. We call
`executeWorkflow("")` here without `killWorkflow()` first. Could that leave the
old paused execution holding its computing-unit resources? If so, I think it's
better to kill it (and wait for Killed/Terminated) before re-running.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) => pattern.test(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 warehouseService: WarehouseService,
+ private config: GuiConfigService
+ ) {}
+
+ 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,
+ schema: PortSchema | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const base: FixState = {
+ operatorId,
+ errorMessage,
+ errorType,
+ originalCode: code,
+ originalProperties: properties,
+ status: "analyzing",
+ };
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
+ this.stateSubject.next({ ...base, status: "ready" });
+ return;
+ }
+
+ const seq = ++this.analysisSeq;
+ this.stateSubject.next(base);
+ try {
+ const { text } = await
this.callModelWithTimeout(buildFixPrompt(errorMessage, code, schema,
properties));
+ if (seq !== this.analysisSeq) {
+ return;
+ }
+ this.stateSubject.next({ ...base, status: "ready", suggestedFix:
this.toFix(text, properties) });
+ } catch (err) {
+ console.error("AI workflow fixer: analysis failed", err);
+ if (seq !== this.analysisSeq) {
+ 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 applyFix(): void {
+ const current = this.stateSubject.getValue();
+ const fix = current.suggestedFix;
+ if (!fix || current.status !== "ready") {
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applying" });
+ try {
+ const properties: Record<string, unknown> = {
+
...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties,
+ };
+ if (fix.type === "code_change") {
+ const code = String(properties.code ?? "");
+ if (!code.includes(fix.original)) {
+ throw new Error("the code changed since the suggestion was
generated");
+ }
+ // split/join replaces every occurrence and keeps the match literal;
replace() with a
+ // string argument would patch only the first one and leave the rest
failing.
+ properties.code = code.split(fix.original).join(fix.suggested);
Review Comment:
Apply can change more than the diff shows, and can break indentation.
Replacing every occurrence is reasonable but the panel shows only one -/+ pair.
A short snippet (a plain-text match, so count also hits account) can rewrite
lines the user never saw. Also, a multi-line suggestion is inserted after the
existing indentation so only the first line is indented and the rest start at
column 0 which might give an IndentationError right after "applied". I suggest
showing how many places will change, and indenting continuation lines to match
the line where the snippet was found.
##########
frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts:
##########
@@ -0,0 +1,267 @@
+/**
+ * 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 } from "rxjs";
+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 { WarehouseService } from
"../../../../common/service/warehouse/warehouse.service";
+import { GuiConfigService } from
"../../../../common/service/gui-config.service";
+import { PortSchema } from "../../../types/workflow-compiling.interface";
+import { buildFixPrompt } from "./ai-fix-prompt";
+
+export type FixErrorType = "missing_column" | "type_error" | "null_error" |
"model_not_found" | "unsupported";
+export type FixConfidence = "high" | "medium" | "low";
+
+export interface SuggestedFix {
+ 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;
+ errorMessage: string;
+ errorType: FixErrorType;
+ originalCode?: string;
+ originalProperties?: object;
+ suggestedFix?: SuggestedFix;
+ // `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.
+ status: "idle" | "analyzing" | "ready" | "applying" | "applied" |
"applied_without_run" | "error";
+}
+
+// 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 UNSUPPORTED_MESSAGE = "This error type is not yet supported for
automatic fixing.";
+
+export const IDLE_STATE: FixState = { operatorId: "", errorMessage: "",
errorType: "unsupported", status: "idle" };
+
+// Each pattern matches a line anywhere inside a traceback, not the whole
message.
+const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [
+ [/KeyError:\s*['"][^'"]+['"]/, "missing_column"],
+ [/TypeError:\s*unsupported operand/i, "type_error"],
+ [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"],
+ // `404` only counts beside a model reference. On its own it also matches a
line number
+ // in a traceback, an HTTP status from unrelated user code, or a literal in
the data --
+ // all of which would be sent off asking for a model-config fix that makes
no sense.
+ [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i,
"model_not_found"],
+];
+
+export function classifyError(errorMessage: string): FixErrorType {
+ return PATTERNS.find(([pattern]) => pattern.test(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 warehouseService: WarehouseService,
+ private config: GuiConfigService
+ ) {}
+
+ 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,
+ schema: PortSchema | undefined,
+ code: string | undefined,
+ properties: Readonly<Record<string, unknown>>
+ ): Promise<void> {
+ const errorType = classifyError(errorMessage);
+ const base: FixState = {
+ operatorId,
+ errorMessage,
+ errorType,
+ originalCode: code,
+ originalProperties: properties,
+ status: "analyzing",
+ };
+
+ // Out of scope: report it without spending a model call.
+ if (errorType === "unsupported") {
+ this.stateSubject.next({ ...base, status: "ready" });
+ return;
+ }
+
+ const seq = ++this.analysisSeq;
+ this.stateSubject.next(base);
+ try {
+ const { text } = await
this.callModelWithTimeout(buildFixPrompt(errorMessage, code, schema,
properties));
+ if (seq !== this.analysisSeq) {
+ return;
+ }
+ this.stateSubject.next({ ...base, status: "ready", suggestedFix:
this.toFix(text, properties) });
+ } catch (err) {
+ console.error("AI workflow fixer: analysis failed", err);
+ if (seq !== this.analysisSeq) {
+ 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 applyFix(): void {
+ const current = this.stateSubject.getValue();
+ const fix = current.suggestedFix;
+ if (!fix || current.status !== "ready") {
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applying" });
+ try {
+ const properties: Record<string, unknown> = {
+
...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties,
+ };
+ if (fix.type === "code_change") {
+ const code = String(properties.code ?? "");
+ if (!code.includes(fix.original)) {
+ throw new Error("the code changed since the suggestion was
generated");
+ }
+ // split/join replaces every occurrence and keeps the match literal;
replace() with a
+ // string argument would patch only the first one and leave the rest
failing.
+ properties.code = code.split(fix.original).join(fix.suggested);
+ } else {
+ const field = String(fix.fieldName);
+ // Same staleness guard as the code branch: re-reading the properties
keeps the
+ // unrelated fields, but the field being replaced can itself have
moved on since
+ // the suggestion was generated, and overwriting it would discard that
newer value.
+ if (String(properties[field] ?? "") !== fix.original) {
+ throw new Error(`${field} changed since the suggestion was
generated`);
+ }
+ properties[field] = fix.suggested;
+ }
+ this.workflowActionService.setOperatorProperty(current.operatorId,
properties);
+ if (!this.canStartRun()) {
+ // ExecuteWorkflowService refuses the run and shows its own toast;
reporting
+ // "re-running" here would contradict it.
+ this.stateSubject.next({ ...current, status: "applied_without_run" });
+ return;
+ }
+ this.stateSubject.next({ ...current, status: "applied" });
+ // Same empty execution name the operator menu uses for an ad-hoc run.
+ this.executeWorkflowService.executeWorkflow("");
+ } catch (err) {
+ console.error("AI workflow fixer: apply failed", err);
+ this.stateSubject.next({ ...current, status: "error" });
+ }
+ }
+
+ /**
+ * Mirrors ExecuteWorkflowService's own entry guard: while the deployment
requires a
+ * warehouse and none is selected, a run request is refused before it starts.
+ */
+ private canStartRun(): boolean {
+ return !this.config.env.warehouseEnabled ||
this.warehouseService.getSelectedWarehouseIdValue() !== undefined;
+ }
+
+ public discardFix(): void {
+ this.stateSubject.next(IDLE_STATE);
+ }
+
+ // Seam over the `ai` transport: specs spy this instead of mocking the "ai"
module,
+ // which leaks across specs sharing the import and hangs on a real network
call.
+ protected callModel(messages: ModelMessage[], abortSignal?: AbortSignal):
Promise<{ text: string }> {
+ // Built per call so a refreshed access token is used: caching the
provider on this root
+ // singleton would pin the JWT captured on the first Analyze, and a later
call after a
+ // token rotation would 401. The /api/chat/* LiteLLM proxy authenticates
with the Texera
+ // JWT and swaps in the master key upstream, so that token is the only
credential sent.
+ const model = createOpenAI({
+ baseURL: new URL(`${AppSettings.getApiEndpoint()}`,
document.baseURI).toString(),
+ apiKey: AuthService.getAccessToken() ?? "",
+ }).chat(AI_FIXER_MODEL);
+ return generateText({ model, messages, abortSignal });
+ }
+
+ private callModelWithTimeout(prompt: string): Promise<{ text: string }> {
Review Comment:
This duplicates `notebook-migration/migration-llm.ts` (the `createOpenAI`
provider/JWT setup, the abort + race timeout, the fenced-JSON parsing).
Consider a small shared helper (e.g. `createTexeraChatModel`, `withTimeout`,
`parseJsonReply`) so fixes land in one place.
--
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]
