kunwp1 commented on code in PR #8637: URL: https://github.com/apache/texera/pull/8637#discussion_r4362847494
########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts: ########## @@ -0,0 +1,642 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { Injectable } from "@angular/core"; +import { BehaviorSubject, Observable, firstValueFrom } from "rxjs"; +import { filter, timeout } from "rxjs/operators"; +import { createOpenAI } from "@ai-sdk/openai"; +import { generateText, type ModelMessage } from "ai"; +import { AppSettings } from "../../../../common/app-setting"; +import { AuthService } from "../../../../common/service/user/auth.service"; +import { WorkflowActionService } from "../../../service/workflow-graph/model/workflow-action.service"; +import { ExecuteWorkflowService } from "../../../service/execute-workflow/execute-workflow.service"; +import { OperatorPortSchemaMap } from "../../../types/workflow-compiling.interface"; +import { ExecutionState } from "../../../types/execute-workflow.interface"; +import { buildFixPrompt, withoutSecrets } from "./ai-fix-prompt"; +import { OperatorMetadataService } from "../../../service/operator-metadata/operator-metadata.service"; +import { WorkflowConsoleService } from "../../../service/workflow-console/workflow-console.service"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "model_not_found" | "unsupported"; +/** + * One operator error, kept as the two parts its source already distinguishes. + * + * Classification reads `header` and the prompt gets `full`: collapsing them would put the + * burden of finding the exception inside the joined text back on this service. + */ +export interface OperatorError { + /** The exception line on its own -- a console message's `title`, a fatal error's `message`. */ + header: string; + /** Exception and stack together, which is what the model is shown. */ + full: string; +} + +export type FixConfidence = "high" | "medium" | "low"; + +export interface SuggestedFix { + /** How many places Apply will touch; the diff only ever shows one -/+ pair. */ + occurrences?: number; + type: "code_change" | "property_change"; + original: string; + suggested: string; + explanation: string; + // Rendered as the confidence badge; `fieldName` names the patched property + // (property_change only) so applyFix knows which field to write. + confidence: FixConfidence; + fieldName?: string; +} + +export interface FixState { + operatorId: string; + /** + * Identifies the ERROR the suggestion was made for, so a newer one on the same operator + * can clear it. A terminal execution state is not enough on its own: the case this panel + * exists for leaves the run `Running` -- the UDF raises and the worker pauses -- so a + * re-run that fails again never reaches one, and the error key is the only signal left. + */ + errorKey?: string; + /** + * The workflow the analysis belongs to. The operator id alone does not identify it: two + * imports of the same JSON carry the same ids, so a reply that arrived after the user + * switched workflows could be applied to the wrong one. + */ + workflowId?: number; + /** + * Which of the four supported shapes the error was classified as. Only the apply path + * reads it, to refuse a configuration change for anything but `model_not_found`; an error + * outside those shapes is the `unsupported` status, not a value here. + */ + errorType?: FixErrorType; + suggestedFix?: SuggestedFix; + // `unsupported`: the error is outside the four shapes this panel handles, reported without + // spending a model call. `applied_without_run`: the fix was written but the deployment + // refused to start a run (a warehouse is required and none is selected), so the panel must + // not claim one. `apply_failed` keeps the suggestion on screen: the analysis was fine, the + // write was refused, and re-analysing would spend another model call for the same fix. + status: + | "idle" + | "analyzing" + | "unsupported" + | "ready" + | "applying" + | "applied" + | "applied_without_run" + | "error" + | "apply_failed"; + /** Why the write was refused, shown as-is when the status is `apply_failed`. */ + applyError?: string; +} + +// Model id as LiteLLM exposes it (bin/single-node/litellm-config.yaml), not the provider's +// id. A deployment whose proxy publishes a different alias has to change this: there is no +// GuiConfig field for it today, and adding one is a backend change. +export const AI_FIXER_MODEL = "claude-haiku-4.5"; +export const AI_FIXER_TIMEOUT_MS = 120_000; +export const UNLOCK_TIMEOUT_MS = 15_000; + +// A run in one of these is live enough to be worth stopping; a lock coming from anything +// else is not something killing an execution can clear. +const LIVE_STATES: ReadonlySet<ExecutionState> = new Set([ + ExecutionState.Initializing, + ExecutionState.Running, + ExecutionState.Pausing, + ExecutionState.Paused, + ExecutionState.Resuming, + ExecutionState.Recovering, +]); + +// A run that reached one of these is over, so anything the panel still shows predates it. +// `Uninitialized` is deliberately absent even though updateWorkflowActionLock treats it as +// over: a kill lands there, and a kill is how Apply itself clears the lock, so resetting on +// it would throw away a suggestion that is still the right one for the error on screen. +const TERMINAL_STATES: ReadonlySet<ExecutionState> = new Set([ + ExecutionState.Completed, + ExecutionState.Failed, + ExecutionState.Killed, + ExecutionState.Terminated, +]); + +export const UNSUPPORTED_MESSAGE = "This error type is not yet supported for automatic fixing."; + +export const IDLE_STATE: FixState = { operatorId: "", status: "idle" }; + +// Matched against the exception line only, never the frame lines above it. +const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [ + [/KeyError:\s*['"][^'"]+['"]/, "missing_column"], + [/TypeError:\s*unsupported operand/i, "type_error"], + [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"], + [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i, "model_not_found"], +]; + +/** + * Replaces every occurrence of a snippet, and reports how many there were. + * + * No indentation work: the prompt requires whole lines carrying the indentation they should + * have, and `startsAtLineBoundary` turns away a reply that does not, so each line of the + * suggestion already sits at the right column. Re-indenting continuation lines here instead + * meant guessing which convention the model had used -- absolute or relative -- and the + * prompt never stated one, so either guess doubled the indentation for the other. + */ +export function replaceAll(code: string, original: string, suggested: string): { code: string; count: number } { + const parts = code.split(original); + return { code: parts.join(suggested), count: parts.length - 1 }; +} + +/** + * Whether every occurrence of the snippet begins a line. + * + * This is what makes the replacement unambiguous: a snippet that starts at column 0 of its + * line carries its own leading indentation, so the suggestion replacing it must carry the + * indentation it wants and nothing has to be inferred. A snippet starting mid-line leaves + * the indentation of its continuation lines undetermined. + */ +export function startsAtLineBoundary(code: string, original: string): boolean { + let from = 0; + for (;;) { + const at = code.indexOf(original, from); + if (at === -1) { + return true; + } + if (at !== 0 && code[at - 1] !== "\n") { + return false; Review Comment: This checks that each match starts a line but not that it ends one. So a whole line can still match the beginning of a longer line. For example, an original of ` n = 1` in code containing ` n = 10` passes the check with a count of 1 so no badge appears. "Apply" then turns that line into `n = 20` while the diff shows only `n = 1`. Requiring each match to also end at a newline (or the end of the code) could fix this issue. ########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts: ########## @@ -0,0 +1,642 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { Injectable } from "@angular/core"; +import { BehaviorSubject, Observable, firstValueFrom } from "rxjs"; +import { filter, timeout } from "rxjs/operators"; +import { createOpenAI } from "@ai-sdk/openai"; +import { generateText, type ModelMessage } from "ai"; +import { AppSettings } from "../../../../common/app-setting"; +import { AuthService } from "../../../../common/service/user/auth.service"; +import { WorkflowActionService } from "../../../service/workflow-graph/model/workflow-action.service"; +import { ExecuteWorkflowService } from "../../../service/execute-workflow/execute-workflow.service"; +import { OperatorPortSchemaMap } from "../../../types/workflow-compiling.interface"; +import { ExecutionState } from "../../../types/execute-workflow.interface"; +import { buildFixPrompt, withoutSecrets } from "./ai-fix-prompt"; +import { OperatorMetadataService } from "../../../service/operator-metadata/operator-metadata.service"; +import { WorkflowConsoleService } from "../../../service/workflow-console/workflow-console.service"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "model_not_found" | "unsupported"; +/** + * One operator error, kept as the two parts its source already distinguishes. + * + * Classification reads `header` and the prompt gets `full`: collapsing them would put the + * burden of finding the exception inside the joined text back on this service. + */ +export interface OperatorError { + /** The exception line on its own -- a console message's `title`, a fatal error's `message`. */ + header: string; + /** Exception and stack together, which is what the model is shown. */ + full: string; +} + +export type FixConfidence = "high" | "medium" | "low"; + +export interface SuggestedFix { + /** How many places Apply will touch; the diff only ever shows one -/+ pair. */ + occurrences?: number; + type: "code_change" | "property_change"; + original: string; + suggested: string; + explanation: string; + // Rendered as the confidence badge; `fieldName` names the patched property + // (property_change only) so applyFix knows which field to write. + confidence: FixConfidence; + fieldName?: string; +} + +export interface FixState { + operatorId: string; + /** + * Identifies the ERROR the suggestion was made for, so a newer one on the same operator + * can clear it. A terminal execution state is not enough on its own: the case this panel + * exists for leaves the run `Running` -- the UDF raises and the worker pauses -- so a + * re-run that fails again never reaches one, and the error key is the only signal left. + */ + errorKey?: string; + /** + * The workflow the analysis belongs to. The operator id alone does not identify it: two + * imports of the same JSON carry the same ids, so a reply that arrived after the user + * switched workflows could be applied to the wrong one. + */ + workflowId?: number; + /** + * Which of the four supported shapes the error was classified as. Only the apply path + * reads it, to refuse a configuration change for anything but `model_not_found`; an error + * outside those shapes is the `unsupported` status, not a value here. + */ + errorType?: FixErrorType; + suggestedFix?: SuggestedFix; + // `unsupported`: the error is outside the four shapes this panel handles, reported without + // spending a model call. `applied_without_run`: the fix was written but the deployment + // refused to start a run (a warehouse is required and none is selected), so the panel must + // not claim one. `apply_failed` keeps the suggestion on screen: the analysis was fine, the + // write was refused, and re-analysing would spend another model call for the same fix. + status: + | "idle" + | "analyzing" + | "unsupported" + | "ready" + | "applying" + | "applied" + | "applied_without_run" + | "error" + | "apply_failed"; + /** Why the write was refused, shown as-is when the status is `apply_failed`. */ + applyError?: string; +} + +// Model id as LiteLLM exposes it (bin/single-node/litellm-config.yaml), not the provider's +// id. A deployment whose proxy publishes a different alias has to change this: there is no +// GuiConfig field for it today, and adding one is a backend change. +export const AI_FIXER_MODEL = "claude-haiku-4.5"; +export const AI_FIXER_TIMEOUT_MS = 120_000; +export const UNLOCK_TIMEOUT_MS = 15_000; + +// A run in one of these is live enough to be worth stopping; a lock coming from anything +// else is not something killing an execution can clear. +const LIVE_STATES: ReadonlySet<ExecutionState> = new Set([ + ExecutionState.Initializing, + ExecutionState.Running, + ExecutionState.Pausing, + ExecutionState.Paused, + ExecutionState.Resuming, + ExecutionState.Recovering, +]); + +// A run that reached one of these is over, so anything the panel still shows predates it. +// `Uninitialized` is deliberately absent even though updateWorkflowActionLock treats it as +// over: a kill lands there, and a kill is how Apply itself clears the lock, so resetting on +// it would throw away a suggestion that is still the right one for the error on screen. +const TERMINAL_STATES: ReadonlySet<ExecutionState> = new Set([ + ExecutionState.Completed, + ExecutionState.Failed, + ExecutionState.Killed, + ExecutionState.Terminated, +]); + +export const UNSUPPORTED_MESSAGE = "This error type is not yet supported for automatic fixing."; + +export const IDLE_STATE: FixState = { operatorId: "", status: "idle" }; + +// Matched against the exception line only, never the frame lines above it. +const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [ + [/KeyError:\s*['"][^'"]+['"]/, "missing_column"], + [/TypeError:\s*unsupported operand/i, "type_error"], + [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"], + [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i, "model_not_found"], +]; + +/** + * Replaces every occurrence of a snippet, and reports how many there were. + * + * No indentation work: the prompt requires whole lines carrying the indentation they should + * have, and `startsAtLineBoundary` turns away a reply that does not, so each line of the + * suggestion already sits at the right column. Re-indenting continuation lines here instead + * meant guessing which convention the model had used -- absolute or relative -- and the + * prompt never stated one, so either guess doubled the indentation for the other. + */ +export function replaceAll(code: string, original: string, suggested: string): { code: string; count: number } { + const parts = code.split(original); + return { code: parts.join(suggested), count: parts.length - 1 }; +} + +/** + * Whether every occurrence of the snippet begins a line. + * + * This is what makes the replacement unambiguous: a snippet that starts at column 0 of its + * line carries its own leading indentation, so the suggestion replacing it must carry the + * indentation it wants and nothing has to be inferred. A snippet starting mid-line leaves + * the indentation of its continuation lines undetermined. + */ +export function startsAtLineBoundary(code: string, original: string): boolean { + let from = 0; + for (;;) { + const at = code.indexOf(original, from); + if (at === -1) { + return true; + } + if (at !== 0 && code[at - 1] !== "\n") { + return false; + } + from = at + original.length; + } +} + +/** Writes the model's string back in the type the field already held. */ +export function coerceToFieldType(currentValue: unknown, suggested: string): unknown { + // Arrays and objects are out of scope: the model replies with a string, and writing it + // into a structured field replaced e.g. an outputColumns array with the text "[]". + // Validating a replacement against the operator's schema is a bigger change than the four + // error shapes this panel handles need, so it refuses instead of guessing. + if (currentValue !== null && typeof currentValue === "object") { + throw new Error("this setting holds structured data, which an automatic fix does not edit"); + } + if (typeof currentValue === "number") { + const parsed = Number(suggested); + if (Number.isNaN(parsed)) { + throw new Error(`${suggested} is not a number`); + } + return parsed; + } + if (typeof currentValue === "boolean") { + if (suggested !== "true" && suggested !== "false") { + throw new Error(`${suggested} is not a boolean`); + } + return suggested === "true"; + } + return suggested; +} + +/** + * Classifies on the exception alone, never on the stack around it. + * + * Both sources already separate the exception from its frames, so nothing here has to find + * it: a console message carries `err.toString` / the traceback's last line in `title`, and a + * fatal error carries it in `message`. Deducing it from the joined text instead meant + * guessing at frame shapes, and the JDK's own are unguessable -- on Java 9+ they carry a + * module prefix (`java.base/java.lang.Thread.run(Thread.java:840)`) and native ones read + * `(Native Method)`, so a real worker trace ended on a frame no pattern recognised and + * `NullPointerException` never classified. Python's `title` is the LAST line of the formatted + * traceback, so a chained one arrives already reduced to the exception that stopped the + * worker. + */ +export function classifyError(exceptionHeader: string): FixErrorType { + return PATTERNS.find(([pattern]) => pattern.test(exceptionHeader))?.[1] ?? "unsupported"; +} + +/** + * Holds the AI-fix loop for one failed operator: classify -> ask the LLM -> apply -> re-run. + * + * The state lives here rather than in the frame component because + * ResultPanelComponent.clearResultPanel() destroys and rebuilds the frames on + * every re-render, which would otherwise discard an in-flight suggestion. + */ +@Injectable({ providedIn: "root" }) +export class AiWorkflowFixerService { + private readonly stateSubject = new BehaviorSubject<FixState>(IDLE_STATE); + + constructor( + private workflowActionService: WorkflowActionService, + private executeWorkflowService: ExecuteWorkflowService, + private operatorMetadataService: OperatorMetadataService, + private workflowConsoleService: WorkflowConsoleService + ) { + // The panel holds whatever it last showed until something clears it, and after Apply + // that is the green "re-running..." message. Resetting when the run ends -- rather than + // when the next one starts -- keeps that message truthful for exactly as long as the run + // it describes, so a run that fails again does not reopen the tab on a finished claim. + this.executeWorkflowService + .getExecutionStateStream() + .pipe( + filter(event => TERMINAL_STATES.has(event.current.state)), + // applyFix kills a live execution before it writes; that kill must not wipe the + // state the apply is in the middle of publishing. + filter(() => this.stateSubject.getValue().status !== "applying") + ) + .subscribe(() => this.reset()); + + // A re-run that fails again never reaches a terminal state, so the subscription above + // never fires for it. The operator reporting a different error is the signal that + // whatever the panel is showing belongs to a run that has already been superseded. + this.workflowConsoleService.getConsoleMessageUpdateStream().subscribe(() => { + const current = this.stateSubject.getValue(); + if (current.status === "idle" || current.status === "analyzing" || current.status === "applying") { + return; + } + const latest = this.latestErrorKey(current.operatorId); + if (latest && latest !== current.errorKey) { + this.reset(); + } + }); + } + + public getState$(): Observable<FixState> { + return this.stateSubject.asObservable(); + } + + // Numbers the analyses so a slow reply cannot overwrite a newer one. A frame whose + // operator does not own the state renders idle, so its Analyze button stays live while + // another operator is mid-analysis -- two requests can be in flight at once. + private analysisSeq = 0; + + public async analyzeError( + operatorId: string, + error: OperatorError, + inputSchemas: OperatorPortSchemaMap | undefined, + code: string | undefined, + properties: Readonly<Record<string, unknown>> + ): Promise<void> { + const errorType = classifyError(error.header); + const workflowId = this.workflowActionService.getOpenWorkflowId(); + const base: FixState = { + operatorId, + workflowId, + errorType, + errorKey: this.latestErrorKey(operatorId), + status: "analyzing", + }; + + // Numbered before the early return as well: this still supersedes whatever is in flight, + // and without the bump a slower analysis for another operator would land on top of it. + const seq = ++this.analysisSeq; + + // Out of scope: report it without spending a model call. + if (errorType === "unsupported") { + this.stateSubject.next({ ...base, status: "unsupported" }); + return; + } + + this.stateSubject.next(base); + try { + // One redacted view, used for both the prompt and the suggestion: toFix reads the + // field's current value to show in the diff, and reading it from the raw properties + // put a credential on screen whenever the model guessed one of those names. + const visible = withoutSecrets(properties, this.schemaProperties(operatorId)); + const { text } = await this.callModelWithTimeout(buildFixPrompt(error.full, code, inputSchemas, visible)); + if (seq !== this.analysisSeq || workflowId !== this.workflowActionService.getOpenWorkflowId()) { + return; + } + const suggestedFix = this.toFix(text, visible); + // The prompt asks for a configuration change in exactly one case, so a property fix + // for any other error is the model going off script. Refused here rather than only at + // Apply: the panel must not show a diff the write will never honour. + if (suggestedFix.type === "property_change" && errorType !== "model_not_found") { + throw new Error("a configuration change is not the fix for this error"); + } + // Counted here so the panel can say it before Apply: a short snippet matched as plain + // text can appear in lines the user never saw in the diff. + if (suggestedFix.type === "code_change") { + // The prompt asks for whole lines with their indentation; a snippet starting mid-line + // means the model ignored that, and applying it would put the lines after the first + // at a column nothing here can work out. + if (!startsAtLineBoundary(code ?? "", suggestedFix.original)) { + throw new Error("the model did not quote whole lines, so the fix cannot be placed"); + } + const occurrences = replaceAll(code ?? "", suggestedFix.original, suggestedFix.suggested).count; + // Zero means the model quoted something the operator does not contain. The panel + // only surfaces this count when it is above one, so such a suggestion is + // indistinguishable from any other, and Apply turns it away as a stale snippet -- + // blaming an edit that never happened. + if (occurrences === 0) { + throw new Error(`the code does not contain ${JSON.stringify(suggestedFix.original)}`); + } + suggestedFix.occurrences = occurrences; + } + this.stateSubject.next({ ...base, status: "ready", suggestedFix }); + } catch (err) { + console.error("AI workflow fixer: analysis failed", err); + if (seq !== this.analysisSeq || workflowId !== this.workflowActionService.getOpenWorkflowId()) { + return; + } + this.stateSubject.next({ ...base, status: "error" }); + } + } + + /** + * Applies the suggestion to the live operator and re-runs the workflow. + * Properties are re-read from the graph so a concurrent edit is not clobbered. + */ + public async applyFix(): Promise<void> { + const current = this.stateSubject.getValue(); + const fix = current.suggestedFix; + // `apply_failed` is retryable: the panel keeps the suggestion and its Apply button, so + // refusing here left the user clicking something that silently did nothing. Everything + // below re-reads the operator and re-runs every check, so a retry is not a shortcut. + if (!fix || (current.status !== "ready" && current.status !== "apply_failed")) { + return; + } + // Read, not bumped. The unlock wait below can take seconds and an analysis started in + // the meantime owns the panel, so this apply must not report over it. (Bumping would be + // equivalent today -- an analysis in flight has already left the status at `analyzing`, + // which the guard above turns away -- but reading says what is meant.) + const seq = this.analysisSeq; + this.stateSubject.next({ ...current, status: "applying" }); + try { + this.assertSameWorkflow(current); + // Run before the kill as well as after it. ensureModifiable stops the paused execution + // to clear the modification lock, so a fix that was going to be refused anyway cost the + // user their run -- and then the unlock wait before the refusal even reached the panel. + this.patchedProperties(current, fix); + await this.ensureModifiable(); + // Checked again: ensureModifiable can wait seconds for a kill to clear the lock, and + // the user is free to open another workflow while it does. + this.assertSameWorkflow(current); + // Re-read and re-validated rather than reusing the pre-flight result, which describes + // the operator as it was before that wait. + const properties = this.patchedProperties(current, fix); + this.workflowActionService.setOperatorProperty(current.operatorId, properties); + // Asked rather than predicted: mirroring ExecuteWorkflowService's refusal conditions + // here means every one it gains has to be repeated, and missing one reports + // "re-running" over a toast saying the opposite. The empty execution name is the one + // the operator menu uses for an ad-hoc run. + const started = this.executeWorkflowService.executeWorkflow(""); + if (seq !== this.analysisSeq) { + return; + } + this.stateSubject.next({ ...current, status: started ? "applied" : "applied_without_run" }); + } catch (err) { + console.error("AI workflow fixer: apply failed", err); + // Not the same as a failed analysis: the suggestion is still good, the operator moved + // under it. Reporting "could not suggest a fix" and hiding the diff sent the user back + // for another model call to get the fix they already had. + if (seq !== this.analysisSeq) { + return; + } + this.stateSubject.next({ + ...current, + status: "apply_failed", + applyError: (err as Error).message, + }); + } + } + + /** + * The operator's properties with the fix written into them, or a refusal. + * + * Reads only the graph and returns a fresh object, so it is safe to run twice: once as a + * pre-flight before ensureModifiable, so a refusal does not cost the user the execution + * that call kills, and once after it, because the operator can move during the wait. + */ + private patchedProperties(current: FixState, fix: SuggestedFix): Record<string, unknown> { + const properties: Record<string, unknown> = { + ...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties, + }; + if (fix.type === "code_change") { + const code = String(properties.code ?? ""); + const patched = replaceAll(code, fix.original, fix.suggested); + if (patched.count === 0) { + throw new Error("the code changed since the suggestion was generated"); + } + // Counted, not just found. The count behind the "changes N places" badge was taken at + // the analysis, and this replaces every match -- so a copy added or removed in between + // means the write touches a different number of lines than the panel promised. A kill + // lifts the modification lock without resetting the panel, so the user can edit the + // UDF with the suggestion still on screen. + if (fix.occurrences !== undefined && patched.count !== fix.occurrences) { + const places = (n: number) => `${n} place${n === 1 ? "" : "s"}`; + throw new Error( + `the snippet is now in ${places(patched.count)} instead of ${places(fix.occurrences)}, ` + + "so the fix was not applied" + ); + } + properties.code = patched.code; + } else { + // Only the model-not-found case asks for a configuration change; a property fix for + // any other error is the model leaving the four shapes this panel supports. + if (current.errorType !== "model_not_found") { Review Comment: Now that analysis refuses a configuration change for anything but `model_not_found`, this check can't pass. ########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts: ########## @@ -0,0 +1,642 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { Injectable } from "@angular/core"; +import { BehaviorSubject, Observable, firstValueFrom } from "rxjs"; +import { filter, timeout } from "rxjs/operators"; +import { createOpenAI } from "@ai-sdk/openai"; +import { generateText, type ModelMessage } from "ai"; +import { AppSettings } from "../../../../common/app-setting"; +import { AuthService } from "../../../../common/service/user/auth.service"; +import { WorkflowActionService } from "../../../service/workflow-graph/model/workflow-action.service"; +import { ExecuteWorkflowService } from "../../../service/execute-workflow/execute-workflow.service"; +import { OperatorPortSchemaMap } from "../../../types/workflow-compiling.interface"; +import { ExecutionState } from "../../../types/execute-workflow.interface"; +import { buildFixPrompt, withoutSecrets } from "./ai-fix-prompt"; +import { OperatorMetadataService } from "../../../service/operator-metadata/operator-metadata.service"; +import { WorkflowConsoleService } from "../../../service/workflow-console/workflow-console.service"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "model_not_found" | "unsupported"; +/** + * One operator error, kept as the two parts its source already distinguishes. + * + * Classification reads `header` and the prompt gets `full`: collapsing them would put the + * burden of finding the exception inside the joined text back on this service. + */ +export interface OperatorError { + /** The exception line on its own -- a console message's `title`, a fatal error's `message`. */ + header: string; + /** Exception and stack together, which is what the model is shown. */ + full: string; +} + +export type FixConfidence = "high" | "medium" | "low"; + +export interface SuggestedFix { + /** How many places Apply will touch; the diff only ever shows one -/+ pair. */ + occurrences?: number; + type: "code_change" | "property_change"; + original: string; + suggested: string; + explanation: string; + // Rendered as the confidence badge; `fieldName` names the patched property + // (property_change only) so applyFix knows which field to write. + confidence: FixConfidence; + fieldName?: string; +} + +export interface FixState { + operatorId: string; + /** + * Identifies the ERROR the suggestion was made for, so a newer one on the same operator + * can clear it. A terminal execution state is not enough on its own: the case this panel + * exists for leaves the run `Running` -- the UDF raises and the worker pauses -- so a + * re-run that fails again never reaches one, and the error key is the only signal left. + */ + errorKey?: string; + /** + * The workflow the analysis belongs to. The operator id alone does not identify it: two + * imports of the same JSON carry the same ids, so a reply that arrived after the user + * switched workflows could be applied to the wrong one. + */ + workflowId?: number; + /** + * Which of the four supported shapes the error was classified as. Only the apply path + * reads it, to refuse a configuration change for anything but `model_not_found`; an error + * outside those shapes is the `unsupported` status, not a value here. + */ + errorType?: FixErrorType; + suggestedFix?: SuggestedFix; + // `unsupported`: the error is outside the four shapes this panel handles, reported without + // spending a model call. `applied_without_run`: the fix was written but the deployment + // refused to start a run (a warehouse is required and none is selected), so the panel must + // not claim one. `apply_failed` keeps the suggestion on screen: the analysis was fine, the + // write was refused, and re-analysing would spend another model call for the same fix. + status: + | "idle" + | "analyzing" + | "unsupported" + | "ready" + | "applying" + | "applied" + | "applied_without_run" + | "error" + | "apply_failed"; + /** Why the write was refused, shown as-is when the status is `apply_failed`. */ + applyError?: string; +} + +// Model id as LiteLLM exposes it (bin/single-node/litellm-config.yaml), not the provider's +// id. A deployment whose proxy publishes a different alias has to change this: there is no +// GuiConfig field for it today, and adding one is a backend change. +export const AI_FIXER_MODEL = "claude-haiku-4.5"; +export const AI_FIXER_TIMEOUT_MS = 120_000; +export const UNLOCK_TIMEOUT_MS = 15_000; + +// A run in one of these is live enough to be worth stopping; a lock coming from anything +// else is not something killing an execution can clear. +const LIVE_STATES: ReadonlySet<ExecutionState> = new Set([ + ExecutionState.Initializing, + ExecutionState.Running, + ExecutionState.Pausing, + ExecutionState.Paused, + ExecutionState.Resuming, + ExecutionState.Recovering, +]); + +// A run that reached one of these is over, so anything the panel still shows predates it. +// `Uninitialized` is deliberately absent even though updateWorkflowActionLock treats it as +// over: a kill lands there, and a kill is how Apply itself clears the lock, so resetting on +// it would throw away a suggestion that is still the right one for the error on screen. +const TERMINAL_STATES: ReadonlySet<ExecutionState> = new Set([ + ExecutionState.Completed, + ExecutionState.Failed, + ExecutionState.Killed, + ExecutionState.Terminated, +]); + +export const UNSUPPORTED_MESSAGE = "This error type is not yet supported for automatic fixing."; + +export const IDLE_STATE: FixState = { operatorId: "", status: "idle" }; + +// Matched against the exception line only, never the frame lines above it. +const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [ + [/KeyError:\s*['"][^'"]+['"]/, "missing_column"], + [/TypeError:\s*unsupported operand/i, "type_error"], + [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"], + [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i, "model_not_found"], Review Comment: I couldn't find an operator where two of these rows can finish end to end: - **`model_not_found`.** The HuggingFace operator never raises for an unknown model. It writes "All inference providers failed" into its result rows, so there's no ERROR message and no tab. A malformed id raises `ValueError: Invalid Hugging Face model ID ...`, which none of these patterns match. In a Python UDF the model name lives in `code`, but the prompt tells the model to change a configuration field, which the new analysis check then refuses. - **`NullPointerException`.** It now classifies, but a JVM operator other than a UDF has no `code` and no allowed configuration change, so Analyze can only end in an error after a model call. Could the supported list (and when the tab is offered) be narrowed to what can actually be fixed today? Alternatively, the model rule could allow a code change when the operator has code. -- 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]
