kunwp1 commented on code in PR #8637: URL: https://github.com/apache/texera/pull/8637#discussion_r4328317580
########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts: ########## @@ -0,0 +1,481 @@ +/** + * 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"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "model_not_found" | "unsupported"; +export type FixConfidence = "high" | "medium" | "low"; + +export interface SuggestedFix { + /** How many places Apply will touch; the diff only ever shows one -/+ pair. */ + occurrences?: number; + type: "code_change" | "property_change"; + original: string; + suggested: string; + explanation: string; + // Rendered as the confidence badge; `fieldName` names the patched property + // (property_change only) so applyFix knows which field to write. + confidence: FixConfidence; + fieldName?: string; +} + +export interface FixState { + operatorId: string; + /** + * 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; + errorType: FixErrorType; + 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. + // `apply_failed` keeps the suggestion on screen: the analysis was fine, the write was + // refused, and re-analysing would spend another model call to produce the same fix. + status: "idle" | "analyzing" | "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 that reached one of these is over, so anything the panel still shows predates it. +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: "", errorType: "unsupported", status: "idle" }; + +// Matched against the exception line only, never the frame lines above it. +const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [ + [/KeyError:\s*['"][^'"]+['"]/, "missing_column"], + [/TypeError:\s*unsupported operand/i, "type_error"], + [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"], + [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i, "model_not_found"], +]; + +/** + * Replaces every occurrence, keeping each one at the indentation it was found at. + * + * The snippet the model matches on usually starts mid-line, after the existing indentation, + * so a multi-line replacement put its first line in the right place and every line after it + * at column 0 -- an IndentationError the moment the fix was "applied". Continuation lines now + * take the indentation of the line the match was found on, and keep whatever relative + * indentation the suggestion itself had. + */ +export function replaceAllIndented(code: string, original: string, suggested: string): { code: string; count: number } { + const parts = code.split(original); + const count = parts.length - 1; + if (count === 0 || !suggested.includes("\n")) { + return { code: parts.join(suggested), count }; + } + + const [head, ...rest] = suggested.split("\n"); + let out = parts[0]; + for (const tail of parts.slice(1)) { + // Whitespace since the last newline is what this occurrence sits at. + const lineStart = out.lastIndexOf("\n") + 1; + const indent = (out.slice(lineStart).match(/^[ \t]*/) ?? [""])[0]; + out += [head, ...rest.map(line => (line.length > 0 ? indent + line : line))].join("\n") + tail; + } Review Comment: When the original spans more than one line, its continuation lines must already carry the code's absolute indentation, and a suggestion written the same way gets indented twice. For example, `title = tuple_["lines"]\n yield tuple_` in an 8-space body puts `yield` at 16 spaces, which is an IndentationError right after "Fix applied". Anchoring the edit on whole lines (e.g. asking the model for a line range plus a dedented replacement) would avoid guessing at indentation at all. ########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts: ########## @@ -0,0 +1,481 @@ +/** + * 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"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "model_not_found" | "unsupported"; +export type FixConfidence = "high" | "medium" | "low"; + +export interface SuggestedFix { + /** How many places Apply will touch; the diff only ever shows one -/+ pair. */ + occurrences?: number; + type: "code_change" | "property_change"; + original: string; + suggested: string; + explanation: string; + // Rendered as the confidence badge; `fieldName` names the patched property + // (property_change only) so applyFix knows which field to write. + confidence: FixConfidence; + fieldName?: string; +} + +export interface FixState { + operatorId: string; + /** + * 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; + errorType: FixErrorType; + 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. + // `apply_failed` keeps the suggestion on screen: the analysis was fine, the write was + // refused, and re-analysing would spend another model call to produce the same fix. + status: "idle" | "analyzing" | "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 that reached one of these is over, so anything the panel still shows predates it. +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: "", errorType: "unsupported", status: "idle" }; + +// Matched against the exception line only, never the frame lines above it. +const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [ + [/KeyError:\s*['"][^'"]+['"]/, "missing_column"], + [/TypeError:\s*unsupported operand/i, "type_error"], + [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"], + [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i, "model_not_found"], +]; + +/** + * Replaces every occurrence, keeping each one at the indentation it was found at. + * + * The snippet the model matches on usually starts mid-line, after the existing indentation, + * so a multi-line replacement put its first line in the right place and every line after it + * at column 0 -- an IndentationError the moment the fix was "applied". Continuation lines now + * take the indentation of the line the match was found on, and keep whatever relative + * indentation the suggestion itself had. + */ +export function replaceAllIndented(code: string, original: string, suggested: string): { code: string; count: number } { + const parts = code.split(original); + const count = parts.length - 1; + if (count === 0 || !suggested.includes("\n")) { + return { code: parts.join(suggested), count }; + } + + const [head, ...rest] = suggested.split("\n"); + let out = parts[0]; + for (const tail of parts.slice(1)) { + // Whitespace since the last newline is what this occurrence sits at. + const lineStart = out.lastIndexOf("\n") + 1; + const indent = (out.slice(lineStart).match(/^[ \t]*/) ?? [""])[0]; + out += [head, ...rest.map(line => (line.length > 0 ? indent + line : line))].join("\n") + tail; + } + return { code: out, count }; +} + +/** Writes the model's string back in the type the field already held. */ +export function coerceToFieldType(currentValue: unknown, suggested: string): unknown { + // Arrays and objects are out of scope: the model replies with a string, and writing it + // into a structured field replaced e.g. an outputColumns array with the text "[]". + // Validating a replacement against the operator's schema is a bigger change than the four + // error shapes this panel handles need, so it refuses instead of guessing. + if (currentValue !== null && typeof currentValue === "object") { + throw new Error("this setting holds structured data, which an automatic fix does not edit"); + } + if (typeof currentValue === "number") { + const parsed = Number(suggested); + if (Number.isNaN(parsed)) { + throw new Error(`${suggested} is not a number`); + } + return parsed; + } + if (typeof currentValue === "boolean") { + if (suggested !== "true" && suggested !== "false") { + throw new Error(`${suggested} is not a boolean`); + } + return suggested === "true"; + } + return suggested; +} + +/** + * The exception a traceback ends on. + * + * Matching the whole message classified on whatever appeared first, which is wrong twice + * over: the frame lines carry paths and line numbers that read like patterns -- a path + * containing "model" above a `line 404` entry looks like a model 404 -- and a chained + * traceback reports the handled exception before the one that actually stopped the worker. + * Python puts the real exception last. + */ +export function finalExceptionLine(errorMessage: string): string { + const lines = errorMessage + .split("\n") + .map(line => line.trim()) + .filter(line => line.length > 0); + // Frame lines are the `File "...", line N` entries and the source line under them; the + // exception is the last line that is neither that nor the traceback header. + for (let i = lines.length - 1; i >= 0; i--) { + if (!/^File\s+"/.test(lines[i]) && !/^Traceback \(/.test(lines[i])) { + return lines[i]; + } + } + return errorMessage; +} + +export function classifyError(errorMessage: string): FixErrorType { + return PATTERNS.find(([pattern]) => pattern.test(finalExceptionLine(errorMessage)))?.[1] ?? "unsupported"; +} + +/** + * Holds the AI-fix loop for one failed operator: classify -> ask the LLM -> apply -> re-run. + * + * The state lives here rather than in the frame component because + * ResultPanelComponent.clearResultPanel() destroys and rebuilds the frames on + * every re-render, which would otherwise discard an in-flight suggestion. + */ +@Injectable({ providedIn: "root" }) +export class AiWorkflowFixerService { + private readonly stateSubject = new BehaviorSubject<FixState>(IDLE_STATE); + + constructor( + private workflowActionService: WorkflowActionService, + private executeWorkflowService: ExecuteWorkflowService, + private operatorMetadataService: OperatorMetadataService + ) { + // Without this the panel kept whatever it last showed: after Apply the green + // "re-running..." message stayed up, so a run that failed again brought the tab back + // claiming a re-run that had already finished. Resetting when the run ends rather than + // when it starts keeps that message truthful for as long as it is running. + 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()); + } + + public getState$(): Observable<FixState> { + return this.stateSubject.asObservable(); + } + + // Numbers the analyses so a slow reply cannot overwrite a newer one. A frame whose + // operator does not own the state renders idle, so its Analyze button stays live while + // another operator is mid-analysis -- two requests can be in flight at once. + private analysisSeq = 0; + + public async analyzeError( + operatorId: string, + errorMessage: string, + inputSchemas: OperatorPortSchemaMap | undefined, + code: string | undefined, + properties: Readonly<Record<string, unknown>> + ): Promise<void> { + const errorType = classifyError(errorMessage); + const workflowId = this.workflowActionService.getOpenWorkflowId(); + const base: FixState = { operatorId, workflowId, errorType, 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. + // 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: "ready" }); + return; + } + + this.stateSubject.next(base); + try { + const { text } = await this.callModelWithTimeout( + buildFixPrompt(errorMessage, code, inputSchemas, withoutSecrets(properties, this.schemaProperties(operatorId))) + ); + if (seq !== this.analysisSeq || workflowId !== this.workflowActionService.getOpenWorkflowId()) { + return; + } + const suggestedFix = this.toFix(text, properties); + // 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") { + suggestedFix.occurrences = replaceAllIndented(code ?? "", suggestedFix.original, suggestedFix.suggested).count; Review Comment: When the count here is 0 (the model's snippet isn't in the code, or the operator has no `code` at all), the panel still shows a ready suggestion, and Apply then says "the code changed since the suggestion was generated", which isn't what happened. Could a zero count be treated as a failed analysis here, so the user never gets offered a diff that can't apply? ########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts: ########## @@ -0,0 +1,481 @@ +/** + * 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"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "model_not_found" | "unsupported"; +export type FixConfidence = "high" | "medium" | "low"; + +export interface SuggestedFix { + /** How many places Apply will touch; the diff only ever shows one -/+ pair. */ + occurrences?: number; + type: "code_change" | "property_change"; + original: string; + suggested: string; + explanation: string; + // Rendered as the confidence badge; `fieldName` names the patched property + // (property_change only) so applyFix knows which field to write. + confidence: FixConfidence; + fieldName?: string; +} + +export interface FixState { + operatorId: string; + /** + * 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; + errorType: FixErrorType; + 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. + // `apply_failed` keeps the suggestion on screen: the analysis was fine, the write was + // refused, and re-analysing would spend another model call to produce the same fix. + status: "idle" | "analyzing" | "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 that reached one of these is over, so anything the panel still shows predates it. +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: "", errorType: "unsupported", status: "idle" }; + +// Matched against the exception line only, never the frame lines above it. +const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [ + [/KeyError:\s*['"][^'"]+['"]/, "missing_column"], + [/TypeError:\s*unsupported operand/i, "type_error"], + [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"], + [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i, "model_not_found"], +]; + +/** + * Replaces every occurrence, keeping each one at the indentation it was found at. + * + * The snippet the model matches on usually starts mid-line, after the existing indentation, + * so a multi-line replacement put its first line in the right place and every line after it + * at column 0 -- an IndentationError the moment the fix was "applied". Continuation lines now + * take the indentation of the line the match was found on, and keep whatever relative + * indentation the suggestion itself had. + */ +export function replaceAllIndented(code: string, original: string, suggested: string): { code: string; count: number } { + const parts = code.split(original); + const count = parts.length - 1; + if (count === 0 || !suggested.includes("\n")) { + return { code: parts.join(suggested), count }; + } + + const [head, ...rest] = suggested.split("\n"); + let out = parts[0]; + for (const tail of parts.slice(1)) { + // Whitespace since the last newline is what this occurrence sits at. + const lineStart = out.lastIndexOf("\n") + 1; + const indent = (out.slice(lineStart).match(/^[ \t]*/) ?? [""])[0]; + out += [head, ...rest.map(line => (line.length > 0 ? indent + line : line))].join("\n") + tail; + } + return { code: out, count }; +} + +/** Writes the model's string back in the type the field already held. */ +export function coerceToFieldType(currentValue: unknown, suggested: string): unknown { + // Arrays and objects are out of scope: the model replies with a string, and writing it + // into a structured field replaced e.g. an outputColumns array with the text "[]". + // Validating a replacement against the operator's schema is a bigger change than the four + // error shapes this panel handles need, so it refuses instead of guessing. + if (currentValue !== null && typeof currentValue === "object") { + throw new Error("this setting holds structured data, which an automatic fix does not edit"); + } + if (typeof currentValue === "number") { + const parsed = Number(suggested); + if (Number.isNaN(parsed)) { + throw new Error(`${suggested} is not a number`); + } + return parsed; + } + if (typeof currentValue === "boolean") { + if (suggested !== "true" && suggested !== "false") { + throw new Error(`${suggested} is not a boolean`); + } + return suggested === "true"; + } + return suggested; +} + +/** + * The exception a traceback ends on. + * + * Matching the whole message classified on whatever appeared first, which is wrong twice + * over: the frame lines carry paths and line numbers that read like patterns -- a path + * containing "model" above a `line 404` entry looks like a model 404 -- and a chained + * traceback reports the handled exception before the one that actually stopped the worker. + * Python puts the real exception last. + */ +export function finalExceptionLine(errorMessage: string): string { + const lines = errorMessage + .split("\n") + .map(line => line.trim()) + .filter(line => line.length > 0); + // Frame lines are the `File "...", line N` entries and the source line under them; the + // exception is the last line that is neither that nor the traceback header. + for (let i = lines.length - 1; i >= 0; i--) { + if (!/^File\s+"/.test(lines[i]) && !/^Traceback \(/.test(lines[i])) { + return lines[i]; + } + } + return errorMessage; +} + +export function classifyError(errorMessage: string): FixErrorType { + return PATTERNS.find(([pattern]) => pattern.test(finalExceptionLine(errorMessage)))?.[1] ?? "unsupported"; +} + +/** + * Holds the AI-fix loop for one failed operator: classify -> ask the LLM -> apply -> re-run. + * + * The state lives here rather than in the frame component because + * ResultPanelComponent.clearResultPanel() destroys and rebuilds the frames on + * every re-render, which would otherwise discard an in-flight suggestion. + */ +@Injectable({ providedIn: "root" }) +export class AiWorkflowFixerService { + private readonly stateSubject = new BehaviorSubject<FixState>(IDLE_STATE); + + constructor( + private workflowActionService: WorkflowActionService, + private executeWorkflowService: ExecuteWorkflowService, + private operatorMetadataService: OperatorMetadataService + ) { + // Without this the panel kept whatever it last showed: after Apply the green + // "re-running..." message stayed up, so a run that failed again brought the tab back + // claiming a re-run that had already finished. Resetting when the run ends rather than + // when it starts keeps that message truthful for as long as it is running. + 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()); Review Comment: Resetting on terminal states covers runs that end, but not the main case this panel is for. If the fix doesn't work and the re-run hits a UDF error again, the execution stays `Running` (paused), so nothing terminal arrives and the status stays `applied`. The new traceback then brings the tab back with "Fix applied — workflow re-running..." while the run is actually stuck on an error. Could the state remember which error it was made for (e.g. the ERROR message's timestamp), and treat a newer error on that operator as a reason to go back to idle? ########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts: ########## @@ -0,0 +1,481 @@ +/** + * 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"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "model_not_found" | "unsupported"; +export type FixConfidence = "high" | "medium" | "low"; + +export interface SuggestedFix { + /** How many places Apply will touch; the diff only ever shows one -/+ pair. */ + occurrences?: number; + type: "code_change" | "property_change"; + original: string; + suggested: string; + explanation: string; + // Rendered as the confidence badge; `fieldName` names the patched property + // (property_change only) so applyFix knows which field to write. + confidence: FixConfidence; + fieldName?: string; +} + +export interface FixState { + operatorId: string; + /** + * 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; + errorType: FixErrorType; + 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. + // `apply_failed` keeps the suggestion on screen: the analysis was fine, the write was + // refused, and re-analysing would spend another model call to produce the same fix. + status: "idle" | "analyzing" | "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 that reached one of these is over, so anything the panel still shows predates it. +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: "", errorType: "unsupported", status: "idle" }; + +// Matched against the exception line only, never the frame lines above it. +const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [ + [/KeyError:\s*['"][^'"]+['"]/, "missing_column"], + [/TypeError:\s*unsupported operand/i, "type_error"], + [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"], + [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i, "model_not_found"], +]; + +/** + * Replaces every occurrence, keeping each one at the indentation it was found at. + * + * The snippet the model matches on usually starts mid-line, after the existing indentation, + * so a multi-line replacement put its first line in the right place and every line after it + * at column 0 -- an IndentationError the moment the fix was "applied". Continuation lines now + * take the indentation of the line the match was found on, and keep whatever relative + * indentation the suggestion itself had. + */ +export function replaceAllIndented(code: string, original: string, suggested: string): { code: string; count: number } { + const parts = code.split(original); + const count = parts.length - 1; + if (count === 0 || !suggested.includes("\n")) { + return { code: parts.join(suggested), count }; + } + + const [head, ...rest] = suggested.split("\n"); + let out = parts[0]; + for (const tail of parts.slice(1)) { + // Whitespace since the last newline is what this occurrence sits at. + const lineStart = out.lastIndexOf("\n") + 1; + const indent = (out.slice(lineStart).match(/^[ \t]*/) ?? [""])[0]; + out += [head, ...rest.map(line => (line.length > 0 ? indent + line : line))].join("\n") + tail; + } + return { code: out, count }; +} + +/** Writes the model's string back in the type the field already held. */ +export function coerceToFieldType(currentValue: unknown, suggested: string): unknown { + // Arrays and objects are out of scope: the model replies with a string, and writing it + // into a structured field replaced e.g. an outputColumns array with the text "[]". + // Validating a replacement against the operator's schema is a bigger change than the four + // error shapes this panel handles need, so it refuses instead of guessing. + if (currentValue !== null && typeof currentValue === "object") { + throw new Error("this setting holds structured data, which an automatic fix does not edit"); + } + if (typeof currentValue === "number") { + const parsed = Number(suggested); + if (Number.isNaN(parsed)) { + throw new Error(`${suggested} is not a number`); + } + return parsed; + } + if (typeof currentValue === "boolean") { + if (suggested !== "true" && suggested !== "false") { + throw new Error(`${suggested} is not a boolean`); + } + return suggested === "true"; + } + return suggested; +} + +/** + * The exception a traceback ends on. + * + * Matching the whole message classified on whatever appeared first, which is wrong twice + * over: the frame lines carry paths and line numbers that read like patterns -- a path + * containing "model" above a `line 404` entry looks like a model 404 -- and a chained + * traceback reports the handled exception before the one that actually stopped the worker. + * Python puts the real exception last. + */ +export function finalExceptionLine(errorMessage: string): string { + const lines = errorMessage + .split("\n") + .map(line => line.trim()) + .filter(line => line.length > 0); + // Frame lines are the `File "...", line N` entries and the source line under them; the + // exception is the last line that is neither that nor the traceback header. + for (let i = lines.length - 1; i >= 0; i--) { + if (!/^File\s+"/.test(lines[i]) && !/^Traceback \(/.test(lines[i])) { + return lines[i]; + } + } + return errorMessage; +} + +export function classifyError(errorMessage: string): FixErrorType { + return PATTERNS.find(([pattern]) => pattern.test(finalExceptionLine(errorMessage)))?.[1] ?? "unsupported"; +} + +/** + * Holds the AI-fix loop for one failed operator: classify -> ask the LLM -> apply -> re-run. + * + * The state lives here rather than in the frame component because + * ResultPanelComponent.clearResultPanel() destroys and rebuilds the frames on + * every re-render, which would otherwise discard an in-flight suggestion. + */ +@Injectable({ providedIn: "root" }) +export class AiWorkflowFixerService { + private readonly stateSubject = new BehaviorSubject<FixState>(IDLE_STATE); + + constructor( + private workflowActionService: WorkflowActionService, + private executeWorkflowService: ExecuteWorkflowService, + private operatorMetadataService: OperatorMetadataService + ) { + // Without this the panel kept whatever it last showed: after Apply the green + // "re-running..." message stayed up, so a run that failed again brought the tab back + // claiming a re-run that had already finished. Resetting when the run ends rather than + // when it starts keeps that message truthful for as long as it is running. + 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()); + } + + public getState$(): Observable<FixState> { + return this.stateSubject.asObservable(); + } + + // Numbers the analyses so a slow reply cannot overwrite a newer one. A frame whose + // operator does not own the state renders idle, so its Analyze button stays live while + // another operator is mid-analysis -- two requests can be in flight at once. + private analysisSeq = 0; + + public async analyzeError( + operatorId: string, + errorMessage: string, + inputSchemas: OperatorPortSchemaMap | undefined, + code: string | undefined, + properties: Readonly<Record<string, unknown>> + ): Promise<void> { + const errorType = classifyError(errorMessage); + const workflowId = this.workflowActionService.getOpenWorkflowId(); + const base: FixState = { operatorId, workflowId, errorType, 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. + // 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: "ready" }); + return; + } + + this.stateSubject.next(base); + try { + const { text } = await this.callModelWithTimeout( + buildFixPrompt(errorMessage, code, inputSchemas, withoutSecrets(properties, this.schemaProperties(operatorId))) + ); + if (seq !== this.analysisSeq || workflowId !== this.workflowActionService.getOpenWorkflowId()) { + return; + } + const suggestedFix = this.toFix(text, properties); + // 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") { + suggestedFix.occurrences = replaceAllIndented(code ?? "", suggestedFix.original, suggestedFix.suggested).count; + } + this.stateSubject.next({ ...base, status: "ready", suggestedFix }); + } catch (err) { + console.error("AI workflow fixer: analysis failed", err); + if (seq !== this.analysisSeq || workflowId !== this.workflowActionService.getOpenWorkflowId()) { + return; + } + this.stateSubject.next({ ...base, status: "error" }); + } + } + + /** + * Applies the suggestion to the live operator and re-runs the workflow. + * Properties are re-read from the graph so a concurrent edit is not clobbered. + */ + public async applyFix(): Promise<void> { + const current = this.stateSubject.getValue(); + const fix = current.suggestedFix; + // `apply_failed` is retryable: the panel keeps the suggestion and its Apply button, so + // refusing here left the user clicking something that silently did nothing. Everything + // below re-reads the operator and re-runs every check, so a retry is not a shortcut. + if (!fix || (current.status !== "ready" && current.status !== "apply_failed")) { + return; + } + // Read, not bumped. The unlock wait below can take seconds and an analysis started in + // the meantime owns the panel, so this apply must not report over it. (Bumping would be + // equivalent today -- an analysis in flight has already left the status at `analyzing`, + // which the guard above turns away -- but reading says what is meant.) + const seq = this.analysisSeq; + this.stateSubject.next({ ...current, status: "applying" }); + try { + this.assertSameWorkflow(current); + await this.ensureModifiable(); + // Checked again: ensureModifiable can wait seconds for a kill to clear the lock, and + // the user is free to open another workflow while it does. + this.assertSameWorkflow(current); + const properties: Record<string, unknown> = { + ...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties, + }; + if (fix.type === "code_change") { + const code = String(properties.code ?? ""); + if (!code.includes(fix.original)) { + throw new Error("the code changed since the suggestion was generated"); + } + properties.code = replaceAllIndented(code, fix.original, fix.suggested).code; + } else { + const field = String(fix.fieldName); + // The field name comes from the model, so it is checked against the operator before + // anything is written. An invented name would add a dead key and leave the real + // setting wrong, and "code" would hand the whole UDF source to a property change. + if (field === "code" || !Object.prototype.hasOwnProperty.call(properties, field)) { + throw new Error(`${field} is not a configuration field of this operator`); + } + // 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`); + } + // The model always replies with a string; writing it raw would turn a numeric or + // boolean setting into one, which the operator then rejects at validation. + properties[field] = coerceToFieldType(properties[field], fix.suggested); + } + this.workflowActionService.setOperatorProperty(current.operatorId, properties); + // Asked rather than predicted: this used to mirror one of ExecuteWorkflowService's + // refusal conditions, which left the other one -- an unavailable computing unit -- + // reported as "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, + }); + } + } + + public discardFix(): void { + this.reset(); + } + + /** + * Clears the panel and supersedes whatever analysis is in flight. + * + * Publishing IDLE alone was not enough: the reply still matched the sequence it started + * with, so it repainted a suggestion the panel had already cleared -- reachable by + * analysing a failed operator and then stopping the run before the model answers. + */ + private reset(): void { + this.analysisSeq++; + 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) + ); + } + + /** + * Refuses to touch a graph the suggestion was not made against. + * + * The service outlives the workspace, so a suggestion survives a switch to another + * workflow -- and two imports of the same JSON share operator ids, so the id alone would + * let the write land on the wrong graph. + */ + private assertSameWorkflow(state: FixState): void { + if (state.workflowId !== this.workflowActionService.getOpenWorkflowId()) { + throw new Error("the workflow changed since the suggestion was generated"); + } + } + + /** + * Clears the workflow-modification lock before the fix is written. + * + * The traceback this panel exists for arrives while the execution is still `Running` -- + * the worker pauses, it does not fail -- and Texera locks modification for as long as a + * run is live, which every other editing feature checks before writing. Killing the run + * both lifts the lock and releases the computing unit the paused execution still holds. + */ + private async ensureModifiable(): Promise<void> { + if (this.workflowActionService.checkWorkflowModificationEnabled()) { + return; + } + this.executeWorkflowService.killWorkflow(); Review Comment: This assumes the lock always comes from a live run, but modification is also off for read-only workflows, and `enableWorkflowModification()` is a no-op there. Running only needs write access to the computing unit, so a read-only user can reach this with a paused run. Apply then kills their run, waits 15s, and reports "the running execution did not stop", which is wrong on both counts. When the lock has another cause (loading, version preview), `killWorkflow()` can also throw its internal "cannot kill workflow, the current execution state is ..." message, which then shows up as the reason. I'd suggest hiding or refusing Apply for read-only workflows, and only killing when there is actually a live run. ########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.ts: ########## @@ -0,0 +1,158 @@ +/** + * 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.workflowId === this.workflowActionService.getOpenWorkflowId() + ? 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.getOperatorInputSchemaMap(operatorId), + properties.code as string | undefined, + properties + ); + } + + public onApply(): void { + this.aiWorkflowFixerService.applyFix(); + } + + public onDiscard(): void { + this.aiWorkflowFixerService.discardFix(); + } + + public confidenceColor(confidence: FixConfidence): string { + if (confidence === "high") { + return "green"; + } + return confidence === "medium" ? "gold" : "red"; + } + + /** Splits a snippet so the template can prefix each line with - or +. */ + public splitLines(snippet: string): string[] { + return snippet.split("\n"); + } + + /** + * The untrimmed error text for this operator. + * + * Engine failures arrive as fatal errors, but a Python UDF exception never + * becomes one: it is delivered as an ERROR console message whose `message` + * holds the whole traceback, which is what the classifier needs. + */ + private rawErrorMessage(operatorId: string): string | undefined { + const fatal = this.executeWorkflowService.getErrorMessages().find(err => err.operatorId === operatorId); + if (fatal) { + return [fatal.message, fatal.details].filter(part => part).join("\n"); + } + const consoleError = (this.workflowConsoleService.getConsoleMessages(operatorId) ?? []) + .filter(msg => msg.msgType.name === "ERROR") + .pop(); + return consoleError ? consoleError.message || consoleError.title : undefined; Review Comment: For JVM operators the console ERROR keeps the exception in `title` and only the stack frames in `message` (`ErrorUtils.mkConsoleMessage`). Reading `message` first means the model never sees the exception text for those errors, and the classifier's last line is a stack frame. So the `NullPointerException` row can't actually match anything that reaches this tab. The spec only tests it as a bare string. For Python, `title` is already the final exception line so classifying on `title` and sending `title` plus `message` to the model may be simpler than re-deriving it. ########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts: ########## @@ -0,0 +1,481 @@ +/** + * 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"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "model_not_found" | "unsupported"; +export type FixConfidence = "high" | "medium" | "low"; + +export interface SuggestedFix { + /** How many places Apply will touch; the diff only ever shows one -/+ pair. */ + occurrences?: number; + type: "code_change" | "property_change"; + original: string; + suggested: string; + explanation: string; + // Rendered as the confidence badge; `fieldName` names the patched property + // (property_change only) so applyFix knows which field to write. + confidence: FixConfidence; + fieldName?: string; +} + +export interface FixState { + operatorId: string; + /** + * 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; + errorType: FixErrorType; + 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. + // `apply_failed` keeps the suggestion on screen: the analysis was fine, the write was + // refused, and re-analysing would spend another model call to produce the same fix. + status: "idle" | "analyzing" | "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 that reached one of these is over, so anything the panel still shows predates it. +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: "", errorType: "unsupported", status: "idle" }; + +// Matched against the exception line only, never the frame lines above it. +const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [ + [/KeyError:\s*['"][^'"]+['"]/, "missing_column"], + [/TypeError:\s*unsupported operand/i, "type_error"], + [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"], + [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i, "model_not_found"], +]; + +/** + * Replaces every occurrence, keeping each one at the indentation it was found at. + * + * The snippet the model matches on usually starts mid-line, after the existing indentation, + * so a multi-line replacement put its first line in the right place and every line after it + * at column 0 -- an IndentationError the moment the fix was "applied". Continuation lines now + * take the indentation of the line the match was found on, and keep whatever relative + * indentation the suggestion itself had. + */ +export function replaceAllIndented(code: string, original: string, suggested: string): { code: string; count: number } { + const parts = code.split(original); + const count = parts.length - 1; + if (count === 0 || !suggested.includes("\n")) { + return { code: parts.join(suggested), count }; + } + + const [head, ...rest] = suggested.split("\n"); + let out = parts[0]; + for (const tail of parts.slice(1)) { + // Whitespace since the last newline is what this occurrence sits at. + const lineStart = out.lastIndexOf("\n") + 1; + const indent = (out.slice(lineStart).match(/^[ \t]*/) ?? [""])[0]; + out += [head, ...rest.map(line => (line.length > 0 ? indent + line : line))].join("\n") + tail; + } + return { code: out, count }; +} + +/** Writes the model's string back in the type the field already held. */ +export function coerceToFieldType(currentValue: unknown, suggested: string): unknown { + // Arrays and objects are out of scope: the model replies with a string, and writing it + // into a structured field replaced e.g. an outputColumns array with the text "[]". + // Validating a replacement against the operator's schema is a bigger change than the four + // error shapes this panel handles need, so it refuses instead of guessing. + if (currentValue !== null && typeof currentValue === "object") { + throw new Error("this setting holds structured data, which an automatic fix does not edit"); + } + if (typeof currentValue === "number") { + const parsed = Number(suggested); + if (Number.isNaN(parsed)) { + throw new Error(`${suggested} is not a number`); + } + return parsed; + } + if (typeof currentValue === "boolean") { + if (suggested !== "true" && suggested !== "false") { + throw new Error(`${suggested} is not a boolean`); + } + return suggested === "true"; + } + return suggested; +} + +/** + * The exception a traceback ends on. + * + * Matching the whole message classified on whatever appeared first, which is wrong twice + * over: the frame lines carry paths and line numbers that read like patterns -- a path + * containing "model" above a `line 404` entry looks like a model 404 -- and a chained + * traceback reports the handled exception before the one that actually stopped the worker. + * Python puts the real exception last. + */ +export function finalExceptionLine(errorMessage: string): string { + const lines = errorMessage + .split("\n") + .map(line => line.trim()) + .filter(line => line.length > 0); + // Frame lines are the `File "...", line N` entries and the source line under them; the + // exception is the last line that is neither that nor the traceback header. + for (let i = lines.length - 1; i >= 0; i--) { + if (!/^File\s+"/.test(lines[i]) && !/^Traceback \(/.test(lines[i])) { + return lines[i]; + } + } + return errorMessage; +} + +export function classifyError(errorMessage: string): FixErrorType { + return PATTERNS.find(([pattern]) => pattern.test(finalExceptionLine(errorMessage)))?.[1] ?? "unsupported"; +} + +/** + * Holds the AI-fix loop for one failed operator: classify -> ask the LLM -> apply -> re-run. + * + * The state lives here rather than in the frame component because + * ResultPanelComponent.clearResultPanel() destroys and rebuilds the frames on + * every re-render, which would otherwise discard an in-flight suggestion. + */ +@Injectable({ providedIn: "root" }) +export class AiWorkflowFixerService { + private readonly stateSubject = new BehaviorSubject<FixState>(IDLE_STATE); + + constructor( + private workflowActionService: WorkflowActionService, + private executeWorkflowService: ExecuteWorkflowService, + private operatorMetadataService: OperatorMetadataService + ) { + // Without this the panel kept whatever it last showed: after Apply the green + // "re-running..." message stayed up, so a run that failed again brought the tab back + // claiming a re-run that had already finished. Resetting when the run ends rather than + // when it starts keeps that message truthful for as long as it is running. + 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()); + } + + public getState$(): Observable<FixState> { + return this.stateSubject.asObservable(); + } + + // Numbers the analyses so a slow reply cannot overwrite a newer one. A frame whose + // operator does not own the state renders idle, so its Analyze button stays live while + // another operator is mid-analysis -- two requests can be in flight at once. + private analysisSeq = 0; + + public async analyzeError( + operatorId: string, + errorMessage: string, + inputSchemas: OperatorPortSchemaMap | undefined, + code: string | undefined, + properties: Readonly<Record<string, unknown>> + ): Promise<void> { + const errorType = classifyError(errorMessage); + const workflowId = this.workflowActionService.getOpenWorkflowId(); + const base: FixState = { operatorId, workflowId, errorType, 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. + // 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: "ready" }); + return; + } + + this.stateSubject.next(base); + try { + const { text } = await this.callModelWithTimeout( + buildFixPrompt(errorMessage, code, inputSchemas, withoutSecrets(properties, this.schemaProperties(operatorId))) + ); + if (seq !== this.analysisSeq || workflowId !== this.workflowActionService.getOpenWorkflowId()) { + return; + } + const suggestedFix = this.toFix(text, properties); + // 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") { + suggestedFix.occurrences = replaceAllIndented(code ?? "", suggestedFix.original, suggestedFix.suggested).count; + } + this.stateSubject.next({ ...base, status: "ready", suggestedFix }); + } catch (err) { + console.error("AI workflow fixer: analysis failed", err); + if (seq !== this.analysisSeq || workflowId !== this.workflowActionService.getOpenWorkflowId()) { + return; + } + this.stateSubject.next({ ...base, status: "error" }); + } + } + + /** + * Applies the suggestion to the live operator and re-runs the workflow. + * Properties are re-read from the graph so a concurrent edit is not clobbered. + */ + public async applyFix(): Promise<void> { + const current = this.stateSubject.getValue(); + const fix = current.suggestedFix; + // `apply_failed` is retryable: the panel keeps the suggestion and its Apply button, so + // refusing here left the user clicking something that silently did nothing. Everything + // below re-reads the operator and re-runs every check, so a retry is not a shortcut. + if (!fix || (current.status !== "ready" && current.status !== "apply_failed")) { + return; + } + // Read, not bumped. The unlock wait below can take seconds and an analysis started in + // the meantime owns the panel, so this apply must not report over it. (Bumping would be + // equivalent today -- an analysis in flight has already left the status at `analyzing`, + // which the guard above turns away -- but reading says what is meant.) + const seq = this.analysisSeq; + this.stateSubject.next({ ...current, status: "applying" }); + try { + this.assertSameWorkflow(current); + await this.ensureModifiable(); + // Checked again: ensureModifiable can wait seconds for a kill to clear the lock, and + // the user is free to open another workflow while it does. + this.assertSameWorkflow(current); + const properties: Record<string, unknown> = { + ...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties, + }; + if (fix.type === "code_change") { + const code = String(properties.code ?? ""); + if (!code.includes(fix.original)) { + throw new Error("the code changed since the suggestion was generated"); + } + properties.code = replaceAllIndented(code, fix.original, fix.suggested).code; + } else { + const field = String(fix.fieldName); + // The field name comes from the model, so it is checked against the operator before + // anything is written. An invented name would add a dead key and leave the real + // setting wrong, and "code" would hand the whole UDF source to a property change. + if (field === "code" || !Object.prototype.hasOwnProperty.call(properties, field)) { + throw new Error(`${field} is not a configuration field of this operator`); + } Review Comment: The field check runs against the raw operator properties, which still include the secrets that `withoutSecrets` kept out of the prompt. If the model guesses a name like `password` or `hfApiToken`, the diff shows that field's current value on screen, and Apply overwrites it. Could the allowed targets be just the fields the model was actually shown, and maybe only for the `model_not_found` case, since that's the only one that calls for a property change? ########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts: ########## @@ -0,0 +1,481 @@ +/** + * 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"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "model_not_found" | "unsupported"; +export type FixConfidence = "high" | "medium" | "low"; + +export interface SuggestedFix { + /** How many places Apply will touch; the diff only ever shows one -/+ pair. */ + occurrences?: number; + type: "code_change" | "property_change"; + original: string; + suggested: string; + explanation: string; + // Rendered as the confidence badge; `fieldName` names the patched property + // (property_change only) so applyFix knows which field to write. + confidence: FixConfidence; + fieldName?: string; +} + +export interface FixState { + operatorId: string; + /** + * 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; + errorType: FixErrorType; + 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. + // `apply_failed` keeps the suggestion on screen: the analysis was fine, the write was + // refused, and re-analysing would spend another model call to produce the same fix. + status: "idle" | "analyzing" | "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 that reached one of these is over, so anything the panel still shows predates it. +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: "", errorType: "unsupported", status: "idle" }; + +// Matched against the exception line only, never the frame lines above it. +const PATTERNS: ReadonlyArray<[RegExp, FixErrorType]> = [ + [/KeyError:\s*['"][^'"]+['"]/, "missing_column"], + [/TypeError:\s*unsupported operand/i, "type_error"], + [/ValueError:[^\n]*\bNaN\b|NullPointerException/i, "null_error"], + [/ModelNotFound|\bmodel[^\n]*\b404\b|\b404\b[^\n]*\bmodel/i, "model_not_found"], +]; + +/** + * Replaces every occurrence, keeping each one at the indentation it was found at. + * + * The snippet the model matches on usually starts mid-line, after the existing indentation, + * so a multi-line replacement put its first line in the right place and every line after it + * at column 0 -- an IndentationError the moment the fix was "applied". Continuation lines now + * take the indentation of the line the match was found on, and keep whatever relative + * indentation the suggestion itself had. + */ +export function replaceAllIndented(code: string, original: string, suggested: string): { code: string; count: number } { + const parts = code.split(original); + const count = parts.length - 1; + if (count === 0 || !suggested.includes("\n")) { + return { code: parts.join(suggested), count }; + } + + const [head, ...rest] = suggested.split("\n"); + let out = parts[0]; + for (const tail of parts.slice(1)) { + // Whitespace since the last newline is what this occurrence sits at. + const lineStart = out.lastIndexOf("\n") + 1; + const indent = (out.slice(lineStart).match(/^[ \t]*/) ?? [""])[0]; + out += [head, ...rest.map(line => (line.length > 0 ? indent + line : line))].join("\n") + tail; + } + return { code: out, count }; +} + +/** Writes the model's string back in the type the field already held. */ +export function coerceToFieldType(currentValue: unknown, suggested: string): unknown { + // Arrays and objects are out of scope: the model replies with a string, and writing it + // into a structured field replaced e.g. an outputColumns array with the text "[]". + // Validating a replacement against the operator's schema is a bigger change than the four + // error shapes this panel handles need, so it refuses instead of guessing. + if (currentValue !== null && typeof currentValue === "object") { + throw new Error("this setting holds structured data, which an automatic fix does not edit"); + } + if (typeof currentValue === "number") { + const parsed = Number(suggested); + if (Number.isNaN(parsed)) { + throw new Error(`${suggested} is not a number`); + } + return parsed; + } + if (typeof currentValue === "boolean") { + if (suggested !== "true" && suggested !== "false") { + throw new Error(`${suggested} is not a boolean`); + } + return suggested === "true"; + } + return suggested; +} + +/** + * The exception a traceback ends on. + * + * Matching the whole message classified on whatever appeared first, which is wrong twice + * over: the frame lines carry paths and line numbers that read like patterns -- a path + * containing "model" above a `line 404` entry looks like a model 404 -- and a chained + * traceback reports the handled exception before the one that actually stopped the worker. + * Python puts the real exception last. + */ +export function finalExceptionLine(errorMessage: string): string { + const lines = errorMessage + .split("\n") + .map(line => line.trim()) + .filter(line => line.length > 0); + // Frame lines are the `File "...", line N` entries and the source line under them; the + // exception is the last line that is neither that nor the traceback header. + for (let i = lines.length - 1; i >= 0; i--) { + if (!/^File\s+"/.test(lines[i]) && !/^Traceback \(/.test(lines[i])) { + return lines[i]; + } + } + return errorMessage; +} + +export function classifyError(errorMessage: string): FixErrorType { + return PATTERNS.find(([pattern]) => pattern.test(finalExceptionLine(errorMessage)))?.[1] ?? "unsupported"; +} + +/** + * Holds the AI-fix loop for one failed operator: classify -> ask the LLM -> apply -> re-run. + * + * The state lives here rather than in the frame component because + * ResultPanelComponent.clearResultPanel() destroys and rebuilds the frames on + * every re-render, which would otherwise discard an in-flight suggestion. + */ +@Injectable({ providedIn: "root" }) +export class AiWorkflowFixerService { + private readonly stateSubject = new BehaviorSubject<FixState>(IDLE_STATE); + + constructor( + private workflowActionService: WorkflowActionService, + private executeWorkflowService: ExecuteWorkflowService, + private operatorMetadataService: OperatorMetadataService + ) { + // Without this the panel kept whatever it last showed: after Apply the green + // "re-running..." message stayed up, so a run that failed again brought the tab back + // claiming a re-run that had already finished. Resetting when the run ends rather than + // when it starts keeps that message truthful for as long as it is running. + 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()); + } + + public getState$(): Observable<FixState> { + return this.stateSubject.asObservable(); + } + + // Numbers the analyses so a slow reply cannot overwrite a newer one. A frame whose + // operator does not own the state renders idle, so its Analyze button stays live while + // another operator is mid-analysis -- two requests can be in flight at once. + private analysisSeq = 0; + + public async analyzeError( + operatorId: string, + errorMessage: string, + inputSchemas: OperatorPortSchemaMap | undefined, + code: string | undefined, + properties: Readonly<Record<string, unknown>> + ): Promise<void> { + const errorType = classifyError(errorMessage); + const workflowId = this.workflowActionService.getOpenWorkflowId(); + const base: FixState = { operatorId, workflowId, errorType, 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. + // 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. Review Comment: A lot of the comments in this file describe how the code used to behave or what an earlier review found ("used to", "Without this the panel kept..."). A short statement of what the code does now would be easier to maintain. Similarly, in `ai-fix-prompt.ts` the doc block for `buildFixPrompt` ended up above `formatInputSchema`. ########## frontend/src/app/workspace/component/result-panel/result-panel.component.ts: ########## @@ -252,8 +256,52 @@ export class ResultPanelComponent implements OnInit, OnDestroy { .subscribe(_ => { this.rerenderResultPanel(); this.changeDetectorRef.detectChanges(); - this.registerOperatorDisplayNameChangeHandler(); + this.updateOperatorTitle(); }); + + // A Python UDF that raises while its operator is already selected moves none of the + // streams above: the execution stays Running and the highlight never moves, so the AI + // Fix tab would not appear until the user clicked away and back. Only that one tab + // depends on the console, so this toggles it instead of calling rerenderResultPanel(): + // a full rerender rebuilds every frame, losing the Console frame's type filter, scroll + // position and debug input, and the Result table, whenever any operator prints. + this.workflowConsoleService + .getConsoleMessageUpdateStream() + .pipe( + // Keyed by operator as well as by the flag: comparing the flag alone would block a + // second operator's error after the first one's, since both map to `true`. + map(() => `${this.currentOperatorId ?? ""}:${this.shouldOfferAiFix()}`), + distinctUntilChanged(), + untilDestroyed(this) + ) + .subscribe(() => { + this.syncAiFixFrame(); + this.changeDetectorRef.detectChanges(); + }); + } + + /** True when the selected operator reports an error the AI panel can be opened on. */ + private shouldOfferAiFix(): boolean { + return ( + this.config.env.copilotEnabled && + this.currentOperatorId !== undefined && + this.hasConsoleError(this.currentOperatorId) + ); + } + + /** Adds or removes the AI Fix entry, leaving every other frame untouched. */ + private syncAiFixFrame(): void { + if (this.shouldOfferAiFix()) { + this.frameComponentConfigs.set("AI Fix", { + component: AiFixFrameComponent, + componentInputs: { operatorId: this.currentOperatorId }, + }); Review Comment: The separate subscription is a good change, but this doesn't fully leave the other frames untouched yet. The panel renders `frameComponentConfigs | keyvalue` with `*ngFor` and no `trackBy`, and `KeyValuePipe` rebuilds all of its pair objects whenever the map changes. So adding this entry or re-setting it with a fresh object still recreates every tab including the Console framewhen the traceback arrives. A `trackBy` on the key in the template plus only calling `set` when the tab is missing or for a different operator should make this match the comment. ########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-prompt.ts: ########## @@ -0,0 +1,116 @@ +/** + * 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 { OperatorPortSchemaMap } from "../../../types/workflow-compiling.interface"; + +// Names that carry a credential even when the operator forgot the password widget. +const SECRET_NAME = /token|password|secret|credential|api[-_]?key/i; + +/** + * Drops operator properties whose values are credentials, so they never reach the model. + * + * Operators keep secrets alongside ordinary settings -- `hfApiToken` on HuggingFaceInference, + * `password` on the SQL sources -- and the whole configuration used to be serialized into the + * prompt. The operator's own schema marks those fields with the password widget, which is the + * reliable signal; the name pattern is a second barrier so a schema that cannot be read, or an + * operator missing the annotation, does not leak anyway. + * + * Dropped rather than masked: none of the four supported fixes touches a credential, so the + * model has no reason to know the field is there. + */ +export function withoutSecrets( + operatorProperties: Readonly<Record<string, unknown>>, + schemaProperties?: Readonly<Record<string, unknown>> +): Record<string, unknown> { + const isPasswordWidget = (fieldSchema: unknown): boolean => + (fieldSchema as any)?.widget?.formlyConfig?.templateOptions?.type === "password"; + + return Object.fromEntries( + Object.entries(operatorProperties).filter( + ([name]) => !SECRET_NAME.test(name) && !isPasswordWidget(schemaProperties?.[name]) + ) + ); +} + +/** + * 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. + */ +/** + * Renders every input port, not just the first. + * + * A Python UDF can have several input ports and `process_tuple` is told which one a tuple + * came from, so a column that only exists on port 1 is not a typo for one on port 0. Only + * port 0 used to be sent, which made the model guess for anything downstream of a join. + * + * A single port renders exactly as it did before, since that is nearly every operator and + * there is nothing to disambiguate. + */ +export function formatInputSchema(inputSchemas: OperatorPortSchemaMap | undefined): string { + const ports = Object.entries(inputSchemas ?? {}).filter(([, schema]) => schema !== undefined); + if (ports.length <= 1) { + return JSON.stringify(ports[0]?.[1] ?? [], null, 2); + } + return ports.map(([key, schema]) => `Port ${key.split("_")[0]}:\n${JSON.stringify(schema, null, 2)}`).join("\n"); +} + +export function buildFixPrompt( + errorMessage: string, + operatorCode: string | undefined, + inputSchemas: OperatorPortSchemaMap | 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) +${formatInputSchema(inputSchemas)} + +Operator configuration +${JSON.stringify(operatorProperties, null, 2)} Review Comment: For a Python UDF, `code` is still part of these properties, so the whole source goes into the prompt twice: once above in the code section and again JSON-escaped here. Dropping `code` before this section would roughly halve the prompt for larger UDFs. ########## frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts: ########## @@ -0,0 +1,481 @@ +/** + * 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"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "model_not_found" | "unsupported"; +export type FixConfidence = "high" | "medium" | "low"; + +export interface SuggestedFix { + /** How many places Apply will touch; the diff only ever shows one -/+ pair. */ + occurrences?: number; + type: "code_change" | "property_change"; + original: string; + suggested: string; + explanation: string; + // Rendered as the confidence badge; `fieldName` names the patched property + // (property_change only) so applyFix knows which field to write. + confidence: FixConfidence; + fieldName?: string; +} + +export interface FixState { + operatorId: string; + /** + * 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; + errorType: FixErrorType; + 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. + // `apply_failed` keeps the suggestion on screen: the analysis was fine, the write was + // refused, and re-analysing would spend another model call to produce the same fix. + status: "idle" | "analyzing" | "ready" | "applying" | "applied" | "applied_without_run" | "error" | "apply_failed"; Review Comment: `errorType` is only ever compared against `"unsupported"`, and with eight statuses the template needs long `||` chains. Making "unsupported" its own status and dropping `errorType` and the placeholder values in `IDLE_STATE` would shrink both the state model and the template conditions. -- 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]
