diff --git a/common/config/src/main/resources/gui.conf b/common/config/src/main/resources/gui.conf index 7cb7fff6555..d770b4ffd20 100644 --- a/common/config/src/main/resources/gui.conf +++ b/common/config/src/main/resources/gui.conf @@ -120,6 +120,11 @@ gui { copilot-enabled = false copilot-enabled = ${?GUI_WORKFLOW_WORKSPACE_COPILOT_ENABLED} + # the model the AI fix panel asks for, named as the LiteLLM proxy exposes it + # (see llm.conf and the proxy's own model list), not as the provider names it + ai-fix-model = "claude-haiku-4.5" + ai-fix-model = ${?GUI_WORKFLOW_WORKSPACE_AI_FIX_MODEL} + # the limit of columns to be displayed in the result table limit-columns = 15 limit-columns = ${?GUI_WORKFLOW_WORKSPACE_LIMIT_COLUMNS} diff --git a/common/config/src/main/scala/org/apache/texera/common/config/GuiConfig.scala b/common/config/src/main/scala/org/apache/texera/common/config/GuiConfig.scala index d5295024dc8..6283a8c225d 100644 --- a/common/config/src/main/scala/org/apache/texera/common/config/GuiConfig.scala +++ b/common/config/src/main/scala/org/apache/texera/common/config/GuiConfig.scala @@ -73,6 +73,8 @@ object GuiConfig { conf.getInt("gui.workflow-workspace.active-time-in-minutes") val guiWorkflowWorkspaceCopilotEnabled: Boolean = conf.getBoolean("gui.workflow-workspace.copilot-enabled") + val guiWorkflowWorkspaceAiFixModel: String = + conf.getString("gui.workflow-workspace.ai-fix-model") val guiWorkflowWorkspaceLimitColumns: Int = conf.getInt("gui.workflow-workspace.limit-columns") val guiAttributionEnabled: Boolean = diff --git a/config-service/src/main/scala/org/apache/texera/service/resource/ConfigResource.scala b/config-service/src/main/scala/org/apache/texera/service/resource/ConfigResource.scala index d682fe8d026..e3dff08575b 100644 --- a/config-service/src/main/scala/org/apache/texera/service/resource/ConfigResource.scala +++ b/config-service/src/main/scala/org/apache/texera/service/resource/ConfigResource.scala @@ -101,6 +101,7 @@ class ConfigResource { "pythonLanguageServerPort" -> GuiConfig.guiWorkflowWorkspacePythonLanguageServerPort, "activeTimeInMinutes" -> GuiConfig.guiWorkflowWorkspaceActiveTimeInMinutes, "copilotEnabled" -> GuiConfig.guiWorkflowWorkspaceCopilotEnabled, + "aiFixModel" -> GuiConfig.guiWorkflowWorkspaceAiFixModel, "limitColumns" -> GuiConfig.guiWorkflowWorkspaceLimitColumns, "pythonNotebookMigrationEnabled" -> GuiConfig.guiWorkflowWorkspacePythonNotebookMigrationEnabled, "pythonNotebookMigrationTimeoutMinutes" -> GuiConfig.guiWorkflowWorkspacePythonNotebookMigrationTimeoutMinutes diff --git a/frontend/src/app/common/service/gui-config.service.mock.ts b/frontend/src/app/common/service/gui-config.service.mock.ts index 8591fc727a9..e108856beda 100644 --- a/frontend/src/app/common/service/gui-config.service.mock.ts +++ b/frontend/src/app/common/service/gui-config.service.mock.ts @@ -54,6 +54,8 @@ export class MockGuiConfigService { defaultLocalUser: { username: "", password: "" }, activeTimeInMinutes: 15, copilotEnabled: false, + + aiFixModel: "claude-haiku-4.5", limitColumns: 15, attributionEnabled: false, pythonNotebookMigrationEnabled: false, diff --git a/frontend/src/app/common/type/gui-config.ts b/frontend/src/app/common/type/gui-config.ts index 5b8b23db03d..c84c104416a 100644 --- a/frontend/src/app/common/type/gui-config.ts +++ b/frontend/src/app/common/type/gui-config.ts @@ -45,6 +45,7 @@ export interface GuiConfig { defaultLocalUser?: { username?: string; password?: string }; activeTimeInMinutes: number; copilotEnabled: boolean; + aiFixModel: string; limitColumns: number; attributionEnabled: boolean; pythonNotebookMigrationEnabled: boolean; diff --git a/frontend/src/app/workspace/component/menu/menu.component.spec.ts b/frontend/src/app/workspace/component/menu/menu.component.spec.ts index 3a5f8440b80..abcb1c619f7 100644 --- a/frontend/src/app/workspace/component/menu/menu.component.spec.ts +++ b/frontend/src/app/workspace/component/menu/menu.component.spec.ts @@ -900,7 +900,7 @@ describe("MenuComponent", () => { component.currentExecutionName = "exec-1"; const executeSpy = vi .spyOn(executeWorkflowService, "executeWorkflowWithEmailNotification") - .mockImplementation(() => {}); + .mockImplementation(() => true); component.runWorkflow(); @@ -927,7 +927,7 @@ describe("MenuComponent", () => { component.currentExecutionName = ""; const executeSpy = vi .spyOn(executeWorkflowService, "executeWorkflowWithEmailNotification") - .mockImplementation(() => {}); + .mockImplementation(() => true); component.runWorkflow(); @@ -971,7 +971,7 @@ describe("MenuComponent", () => { } as unknown as Mocked; const executeSpy = vi .spyOn(executeWorkflowService, "executeWorkflowWithEmailNotification") - .mockImplementation(() => {}); + .mockImplementation(() => true); component.runWorkflow(); diff --git a/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.html b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.html new file mode 100644 index 00000000000..ffaec57a458 --- /dev/null +++ b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.html @@ -0,0 +1,126 @@ + + + +
+ + {{ state.status === "analyzing" ? "Analyzing error with AI..." : "Applying fix..." }} +
+ + +
+ +
+ + + + + + + + + + + + +
🔍 Root cause
+

{{ fix.explanation }}

+ +
+ 📝 Proposed fix + {{ fix.confidence }} confidence + changes {{ fix.occurrences }} places +
+ +
+
+ - {{ line }} +
+
+ + {{ line }} +
+
+ +
+ + +
+ + diff --git a/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.scss b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.scss new file mode 100644 index 00000000000..caa0cd376e5 --- /dev/null +++ b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.scss @@ -0,0 +1,81 @@ +/** + * 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. + */ + +:host { + display: block; + width: 100%; + padding: 8px; +} + +.ai-fix-status { + align-items: center; + display: flex; + gap: 8px; + padding: 8px 0; +} + +.ai-fix-section-title { + font-weight: bold; + margin-top: 8px; +} + +.ai-fix-explanation { + margin: 4px 0 8px; + user-select: text; +} + +.ai-fix-field { + margin-bottom: 4px; + user-select: text; +} + +/* + * Same monospace treatment the other result-panel frames use for code, except for the + * fallback: they name `serif`, which a diff cannot use -- the +/- columns stop lining up + * the moment monaco is missing. + */ +.ai-fix-diff { + border: 1px solid #e8e8e8; + border-radius: 2px; + font-family: monaco, monospace; + font-size: 11px; + line-height: 16px; + margin: 4px 0 8px; + overflow-x: auto; + white-space: pre; + user-select: text; +} + +.ai-fix-removed { + background-color: #ffeef0; + color: #82071e; + padding: 0 6px; +} + +.ai-fix-added { + background-color: #e6ffed; + color: #116329; + padding: 0 6px; +} + +.ai-fix-actions { + display: flex; + gap: 8px; + margin-top: 8px; +} diff --git a/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.spec.ts b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.spec.ts new file mode 100644 index 00000000000..d983909b779 --- /dev/null +++ b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.spec.ts @@ -0,0 +1,365 @@ +/** + * 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 { ComponentFixture, TestBed } from "@angular/core/testing"; +import { By } from "@angular/platform-browser"; +import { BehaviorSubject } from "rxjs"; +import { HttpClientTestingModule } from "@angular/common/http/testing"; +import { AiFixFrameComponent } from "./ai-fix-frame.component"; +import { AiWorkflowFixerService, FixState, UNSUPPORTED_MESSAGE } from "./ai-workflow-fixer.service"; +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 { commonTestProviders } from "../../../../common/testing/test-utils"; +import { WorkflowFatalError } from "../../../types/workflow-websocket.interface"; +import { ConsoleMessage } from "../../../types/workflow-common.interface"; + +const IDLE: FixState = { operatorId: "", status: "idle" }; + +const OPEN_WORKFLOW = 11; + +const READY_CODE_FIX: FixState = { + operatorId: "op1", + workflowId: OPEN_WORKFLOW, + status: "ready", + suggestedFix: { + original: "yield t['email']", + suggested: "yield t['user_email']\nyield t", + explanation: "the column is user_email", + confidence: "high", + }, +}; + +const fatalError = (over: Partial = {}) => + ({ + message: "KeyError: 'email'", + details: "Traceback ...", + operatorId: "op1", + workerId: "w1", + type: { name: "EXECUTION_FAILURE" }, + timestamp: { nanos: 0, seconds: 0 }, + ...over, + }) as WorkflowFatalError; + +const consoleMessage = (name: string, message: string, title = "KeyError: 'email'") => + ({ + workerId: "w1", + timestamp: { nanos: 0, seconds: 0 }, + msgType: { name }, + source: "udf.py:process_tuple:7", + title, + message, + }) as ConsoleMessage; + +describe("AiFixFrameComponent", () => { + let component: AiFixFrameComponent; + let fixture: ComponentFixture; + let state$: BehaviorSubject; + let analyzeError: ReturnType; + let applyFix: ReturnType; + let discardFix: ReturnType; + let getErrorMessages: ReturnType; + let getConsoleMessages: ReturnType; + let getOperatorInputSchemaMap: ReturnType; + let operatorProperties: Record; + + const render = (state: FixState) => { + state$.next(state); + fixture.detectChanges(); + }; + const query = (css: string) => fixture.debugElement.query(By.css(css)); + // Angular surrounds an interpolation with layout whitespace, so compare the content itself. + const texts = (css: string) => + fixture.debugElement.queryAll(By.css(css)).map(el => el.nativeElement.textContent.trim()); + + beforeEach(async () => { + state$ = new BehaviorSubject(IDLE); + analyzeError = vi.fn().mockResolvedValue(undefined); + applyFix = vi.fn(); + discardFix = vi.fn(); + getErrorMessages = vi.fn().mockReturnValue([]); + getConsoleMessages = vi.fn().mockReturnValue(undefined); + getOperatorInputSchemaMap = vi + .fn() + .mockReturnValue({ "0_false": [{ attributeName: "user_email", attributeType: "string" }] }); + operatorProperties = { code: "yield t['email']", workers: 1 }; + + await TestBed.configureTestingModule({ + imports: [AiFixFrameComponent, HttpClientTestingModule], + providers: [ + { + provide: AiWorkflowFixerService, + useValue: { getState$: () => state$.asObservable(), analyzeError, applyFix, discardFix }, + }, + { provide: ExecuteWorkflowService, useValue: { getErrorMessages } }, + { + provide: WorkflowActionService, + useValue: { + getTexeraGraph: () => ({ getOperator: () => ({ operatorProperties }) }), + getOpenWorkflowId: () => OPEN_WORKFLOW, + }, + }, + { provide: WorkflowCompilingService, useValue: { getOperatorInputSchemaMap } }, + { provide: WorkflowConsoleService, useValue: { getConsoleMessages } }, + ...commonTestProviders, + ], + }).compileComponents(); + + fixture = TestBed.createComponent(AiFixFrameComponent); + component = fixture.componentInstance; + component.operatorId = "op1"; + fixture.detectChanges(); + }); + + afterEach(() => vi.restoreAllMocks()); + + it("should create", () => { + expect(component).toBeTruthy(); + }); + + describe("the analyze button", () => { + it("passes the fatal error, schema and code to the service", () => { + getErrorMessages.mockReturnValue([fatalError({ operatorId: "other" }), fatalError()]); + + query(".ai-fix-analyze").nativeElement.click(); + + expect(analyzeError).toHaveBeenCalledWith( + "op1", + // A fatal error already separates them: `message` is the exception, `details` the stack. + { header: "KeyError: 'email'", full: "KeyError: 'email'\nTraceback ..." }, + // Every input port, not just port 0: a UDF downstream of a join has more than one. + { "0_false": [{ attributeName: "user_email", attributeType: "string" }] }, + "yield t['email']" + ); + }); + + it("falls back to the Python traceback, which never becomes a fatal error", () => { + getConsoleMessages.mockReturnValue([ + consoleMessage("PRINT", "hello"), + consoleMessage("ERROR", "Traceback (most recent call last):\nKeyError: 'email'"), + ]); + + query(".ai-fix-analyze").nativeElement.click(); + + // The exception is handed over on its own, for classification, and again inside the + // full text, which is what the model is shown. + expect(analyzeError.mock.calls[0][1]).toEqual({ + header: "KeyError: 'email'", + full: "KeyError: 'email'\nTraceback (most recent call last):\nKeyError: 'email'", + }); + }); + + it("classifies a JVM error on its title, never on the frames below it", () => { + // ErrorUtils.mkConsoleMessage puts `err.toString()` in title and only + // `getStackTrace.mkString` in message -- no `at ` prefix, and on Java 9+ the JDK's own + // frames carry a module name. A real worker trace ends on one of those, so anything + // that tried to find the exception inside the joined text picked the frame and left + // NullPointerException unclassifiable. The title is handed over directly instead. + const frames = [ + "org.apache.texera.udf.Op.process(Op.scala:31)", + "java.base/java.lang.Thread.run(Thread.java:840)", + ].join("\n"); + getConsoleMessages.mockReturnValue([ + consoleMessage("ERROR", frames, "java.lang.NullPointerException: value is null"), + ]); + + query(".ai-fix-analyze").nativeElement.click(); + + const passed = analyzeError.mock.calls[0][1]; + expect(passed.header).toEqual("java.lang.NullPointerException: value is null"); + expect(passed.full).toContain("java.base/java.lang.Thread.run"); + }); + + it("does nothing when the operator has no error at all", () => { + query(".ai-fix-analyze").nativeElement.click(); + expect(analyzeError).not.toHaveBeenCalled(); + }); + + it("is disabled without an operator, and hidden once a suggestion is on screen", () => { + component.operatorId = undefined; + fixture.detectChanges(); + expect(query(".ai-fix-analyze").nativeElement.disabled).toBe(true); + + component.operatorId = "op1"; + render(READY_CODE_FIX); + expect(query(".ai-fix-analyze")).toBeNull(); + }); + }); + + describe("operator scoping", () => { + // The service is a root singleton with one suggestion, but a failed pipeline can put a + // tab on several operators at once. A frame must ignore state naming another operator, + // or Apply would patch an operator the user is not looking at. + it("ignores a suggestion analyzed for a different operator", () => { + component.operatorId = "op2"; + + render(READY_CODE_FIX); // READY_CODE_FIX names op1 + + expect(query(".ai-fix-diff")).toBeNull(); + expect(query(".ai-fix-actions")).toBeNull(); + expect(query(".ai-fix-analyze")).toBeTruthy(); + }); + + it("cannot apply another operator's suggestion", () => { + component.operatorId = "op2"; + + render(READY_CODE_FIX); + + expect(query(".ai-fix-apply")).toBeNull(); + expect(applyFix).not.toHaveBeenCalled(); + }); + + it("still shows the suggestion on the operator it was analyzed for", () => { + component.operatorId = "op1"; + + render(READY_CODE_FIX); + + expect(query(".ai-fix-diff")).toBeTruthy(); + expect(query(".ai-fix-apply")).toBeTruthy(); + }); + + it("drops a suggestion that stops matching when the bound operator changes", () => { + component.operatorId = "op1"; + render(READY_CODE_FIX); + expect(query(".ai-fix-diff")).toBeTruthy(); + + component.operatorId = "op2"; + fixture.detectChanges(); + + expect(query(".ai-fix-diff")).toBeNull(); + }); + }); + + it("leaves a way out of the not-supported message", () => { + // Without the button this state is a dead end: nothing resets it, so after the user + // fixes the code and re-runs into a supported error the tab keeps showing the stale + // message until some other operator is analyzed. + render({ ...READY_CODE_FIX, status: "unsupported", suggestedFix: undefined }); + + expect(fixture.nativeElement.textContent).toContain(UNSUPPORTED_MESSAGE); + expect(query(".ai-fix-analyze")).toBeTruthy(); + }); + + it("warns when the fix touches more places than the diff shows", () => { + // The diff is one -/+ pair however many places change, and a short snippet matched as + // plain text can reach lines the user never saw. + render({ ...READY_CODE_FIX, suggestedFix: { ...READY_CODE_FIX.suggestedFix!, occurrences: 3 } }); + + expect(fixture.nativeElement.textContent).toContain("changes 3 places"); + }); + + it("stays quiet when the fix touches a single place", () => { + render({ ...READY_CODE_FIX, suggestedFix: { ...READY_CODE_FIX.suggestedFix!, occurrences: 1 } }); + + expect(fixture.nativeElement.textContent).not.toContain("places"); + }); + + it("keeps the suggestion on screen when the write is refused", () => { + // A refused write is not a failed analysis: the fix is still good, the operator moved + // under it. Hiding the diff sent the user back for another model call to get it again. + render({ + ...READY_CODE_FIX, + status: "apply_failed", + applyError: "the code changed since the suggestion was generated", + }); + + expect(fixture.nativeElement.textContent).toContain("the code changed since the suggestion"); + expect(query(".ai-fix-diff")).toBeTruthy(); + expect(query(".ai-fix-apply")).toBeTruthy(); + }); + + it("does not name a reason the panel no longer knows", () => { + // The panel asks whether the run started, not why it did not: naming the warehouse + // would be wrong whenever the refusal came from a computing unit instead, which is the + // contradiction this whole panel exists to avoid. + render({ ...READY_CODE_FIX, status: "applied_without_run", suggestedFix: undefined }); + + const text = fixture.nativeElement.textContent; + expect(text).toContain("the re-run was refused"); + expect(text).not.toContain("warehouse"); + }); + + it("ignores a suggestion made for another workflow", () => { + // Two imports of the same JSON share operator ids, so the id alone would let the stale + // suggestion render here. Refusing it only at Apply would still show the user a diff + // that belongs to a graph they are no longer looking at. + render({ ...READY_CODE_FIX, workflowId: 99 }); + + expect(query(".ai-fix-diff")).toBeNull(); + expect(query(".ai-fix-apply")).toBeNull(); + expect(query(".ai-fix-analyze")).toBeTruthy(); + }); + + it("shows a spinner while analyzing", () => { + render({ ...READY_CODE_FIX, status: "analyzing", suggestedFix: undefined }); + + expect(query("nz-spin")).toBeTruthy(); + expect(query(".ai-fix-status").nativeElement.textContent).toContain("Analyzing error with AI"); + expect(query(".ai-fix-actions")).toBeNull(); + }); + + it("shows the not-supported message for an out-of-scope error", () => { + render({ ...READY_CODE_FIX, status: "unsupported", suggestedFix: undefined }); + + expect(query("nz-alert")).toBeTruthy(); + expect(fixture.nativeElement.textContent).toContain(UNSUPPORTED_MESSAGE); + expect(query(".ai-fix-actions")).toBeNull(); + }); + + it("renders the root cause, a confidence badge, and one diff line per line of the fix", () => { + render(READY_CODE_FIX); + + expect(query(".ai-fix-explanation").nativeElement.textContent).toContain("the column is user_email"); + expect(query("nz-tag").nativeElement.textContent).toContain("high"); + expect(texts(".ai-fix-removed")).toEqual(["- yield t['email']"]); + // The suggestion spans two lines, so it renders two added rows. + expect(texts(".ai-fix-added")).toEqual(["+ yield t['user_email']", "+ yield t"]); + }); + + it("delegates the apply and discard buttons to the service", () => { + render(READY_CODE_FIX); + + query(".ai-fix-apply").nativeElement.click(); + query(".ai-fix-discard").nativeElement.click(); + + expect(applyFix).toHaveBeenCalledTimes(1); + expect(discardFix).toHaveBeenCalledTimes(1); + }); + + it("reports the applied state and offers a fresh analysis", () => { + render({ ...READY_CODE_FIX, status: "applied" }); + + expect(fixture.nativeElement.textContent).toContain("Fix applied"); + expect(query(".ai-fix-actions")).toBeNull(); + expect(query(".ai-fix-analyze")).toBeTruthy(); + }); + + it("reports a failed analysis", () => { + render({ ...READY_CODE_FIX, status: "error", suggestedFix: undefined }); + + expect(fixture.nativeElement.textContent).toContain("could not suggest a fix"); + }); + + it("maps confidence to a badge colour", () => { + expect(component.confidenceColor("high")).toEqual("green"); + expect(component.confidenceColor("medium")).toEqual("gold"); + expect(component.confidenceColor("low")).toEqual("red"); + }); +}); diff --git a/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.ts b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.ts new file mode 100644 index 00000000000..acfb357d475 --- /dev/null +++ b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-frame.component.ts @@ -0,0 +1,169 @@ +/** + * 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, + OperatorError, + 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(undefined); + + @Input() + set operatorId(value: string | undefined) { + this.operatorIdSubject.next(value); + } + + get operatorId(): string | undefined { + return this.operatorIdSubject.value; + } + + public readonly state$: Observable; + 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 error = this.operatorError(operatorId); + if (!error) { + return; + } + const properties = this.workflowActionService.getTexeraGraph().getOperator(operatorId).operatorProperties ?? {}; + void this.aiWorkflowFixerService.analyzeError( + operatorId, + error, + this.workflowCompilingService.getOperatorInputSchemaMap(operatorId), + properties.code as string | undefined + ); + } + + public onApply(): void { + // Awaited by the service's own state stream, not here: the template renders `applying` + // and then whatever the apply settles on. `void` says the promise is deliberately dropped. + 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"); + } + + /** + * This operator's error, split the way its source already splits it. + * + * Engine failures arrive as fatal errors, but a Python UDF exception never becomes one: it + * is delivered as an ERROR console message while the execution stays `Running`. Either way + * the exception is in a field of its own -- `message` on a fatal error, `title` on a console + * message -- so the classifier is handed it directly instead of searching the joined text. + */ + private operatorError(operatorId: string): OperatorError | undefined { + const fatal = this.executeWorkflowService.getErrorMessages().find(err => err.operatorId === operatorId); + if (fatal) { + return { header: fatal.message, full: [fatal.message, fatal.details].filter(part => part).join("\n") }; + } + const consoleError = (this.workflowConsoleService.getConsoleMessages(operatorId) ?? []) + .filter(msg => msg.msgType.name === "ERROR") + .pop(); + if (!consoleError) { + return undefined; + } + // `message` holds only the frames for a JVM operator and the whole traceback for a Python + // one, so the model is sent both parts with the exception first in either case. + return { + header: consoleError.title, + full: [consoleError.title, consoleError.message].filter(part => part).join("\n"), + }; + } +} diff --git a/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-prompt.spec.ts b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-prompt.spec.ts new file mode 100644 index 00000000000..d5776751411 --- /dev/null +++ b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-prompt.spec.ts @@ -0,0 +1,60 @@ +/** + * 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 { describe, expect, it } from "vitest"; +import { buildFixPrompt, formatInputSchema } from "./ai-fix-prompt"; +import { PortSchema } from "../../../types/workflow-compiling.interface"; + +describe("buildFixPrompt", () => { + it("sends no operator configuration at all, so a UI parameter cannot leak", () => { + // The configuration block is not sent at all. Filtering it by field name could never be + // enough: a Python UDF's uiParameters is a list of {attribute, value} pairs, so the name + // says nothing about whether a value is an API key the user typed in. + const prompt = buildFixPrompt("KeyError: 'email'", "print(1)", {}); + + expect(prompt).not.toContain("hf_live_abc"); + expect(prompt).not.toContain("Operator configuration"); + }); +}); + +describe("formatInputSchema", () => { + const port0: PortSchema = [{ attributeName: "user_email", attributeType: "string" }]; + const port1: PortSchema = [{ attributeName: "order_total", attributeType: "double" }]; + + it("renders a single port exactly as before, with no port label", () => { + // Nearly every operator has one input port and there is nothing to disambiguate, so + // this keeps the prompt that was verified end to end byte for byte. + expect(formatInputSchema({ "0_false": port0 })).toEqual(JSON.stringify(port0, null, 2)); + }); + + it("labels every port once there is more than one", () => { + // process_tuple is told which port a tuple came from, so a column that only exists on + // port 1 is not a typo for one on port 0. + const rendered = formatInputSchema({ "0_false": port0, "1_false": port1 }); + + expect(rendered).toContain("Port 0:"); + expect(rendered).toContain("Port 1:"); + expect(rendered).toContain("order_total"); + }); + + it("survives an operator whose schema has not been compiled yet", () => { + expect(formatInputSchema(undefined)).toEqual("[]"); + expect(formatInputSchema({ "0_false": undefined })).toEqual("[]"); + }); +}); diff --git a/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-prompt.ts b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-prompt.ts new file mode 100644 index 00000000000..cbc9bca6199 --- /dev/null +++ b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-fix-prompt.ts @@ -0,0 +1,92 @@ +/** + * 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"; + +/** + * 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. Sending + * port 0 alone leaves the model guessing for anything downstream of a join. + * + * A single port renders with no label at all, 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"); +} + +/** + * Builds the single user message sent to the LLM for one failed operator. + * + * The operator's configuration is not among the things sent. Every fix this panel writes is + * a change to `code`, so the block served nothing -- and it could not be made safe by + * filtering field names: a Python UDF's `uiParameters` is a list of attribute/value pairs, + * so the name says nothing about whether a value is an API key the user typed in. + * + * The model is asked for exactly one minimal fix and for raw JSON. The reply is still parsed + * defensively (fenced blocks are unwrapped) because "no markdown" is a request, not a + * guarantee. + */ +export function buildFixPrompt( + errorMessage: string, + operatorCode: string | undefined, + inputSchemas: OperatorPortSchemaMap | undefined +): 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)} + + +Task +Diagnose the root cause of the error. Then propose exactly ONE fix. + +Rules: +- If it is a missing column error: suggest the closest column name from the schema. +- If it is a type error: add an explicit cast (e.g., int(), str(), float()) at the point of failure. +- If it is a null/NaN error: add a .dropna() call or an explicit null guard. +- Do NOT rewrite the whole function. Change the minimum number of lines. +- original_snippet and suggested_snippet must be WHOLE LINES, copied starting at the + beginning of a line, with the leading indentation included exactly as it appears. Do not + start a snippet in the middle of a line, and do not re-indent: every line you return must + already carry the indentation it should have in the file. Ending a snippet with a newline + is fine; just end both snippets the same way. + +Respond ONLY with valid JSON, no markdown, no explanation outside the JSON: +{ + "explanation": "one sentence describing the root cause", + "original_snippet": "whole lines to replace, indentation included", + "suggested_snippet": "replacement lines, each already indented as it should appear", + "confidence": "high" | "medium" | "low" +}`; +} diff --git a/frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.spec.ts b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.spec.ts new file mode 100644 index 00000000000..b5c82437eb1 --- /dev/null +++ b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.spec.ts @@ -0,0 +1,920 @@ +/** + * 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 { TestBed } from "@angular/core/testing"; +import { firstValueFrom, Subject } from "rxjs"; +import { + AiWorkflowFixerService, + classifyError, + FixState, + replaceAll, + matchesWholeLines, + OperatorError, +} from "./ai-workflow-fixer.service"; +import { WorkflowActionService } from "../../../service/workflow-graph/model/workflow-action.service"; +import { ExecuteWorkflowService } from "../../../service/execute-workflow/execute-workflow.service"; +import { GuiConfigService } from "../../../../common/service/gui-config.service"; +import { commonTestProviders } from "../../../../common/testing/test-utils"; +import { OperatorMetadataService } from "../../../service/operator-metadata/operator-metadata.service"; +import { WorkflowConsoleService } from "../../../service/workflow-console/workflow-console.service"; +import { OperatorPortSchemaMap } from "../../../types/workflow-compiling.interface"; +import { ExecutionState } from "../../../types/execute-workflow.interface"; + +const OP = "PythonUDFV2-op-1"; +const SCHEMA: OperatorPortSchemaMap = { + "0_false": [ + { attributeName: "user_email", attributeType: "string" }, + { attributeName: "follower_count", attributeType: "integer" }, + ], +}; +const CODE = + "class ProcessTupleOperator(UDFOperatorV2):\n def process_tuple(self, t, port):\n yield t['email']\n"; + +// A UDF traceback as the Python worker sends it: the whole stack in one string, +// so the regexes must match a line inside it rather than the start of the message. +const KEY_ERROR = + "Traceback (most recent call last):\n File \"udf.py\", line 3\n yield t['email']\nKeyError: 'email'"; + +const QUOTE = String.fromCharCode(39); +const NL = String.fromCharCode(10); + +// Both sources hand the exception over on its own, so a test supplies it the same way: +// `header` is what classification reads, `full` is what the model is shown. +const err = (header: string, full: string = header): OperatorError => ({ header, full }); +const KEY_ERROR_EVENT = err("KeyError: " + QUOTE + "email" + QUOTE, KEY_ERROR); + +describe("AiWorkflowFixerService", () => { + let service: AiWorkflowFixerService; + let setOperatorProperty: ReturnType; + let executeWorkflow: ReturnType; + let killWorkflow: ReturnType; + let executionState$: Subject<{ previous: any; current: any }>; + let operatorProperties: Record; + let modificationEnabled: boolean; + let modificationEnabled$: Subject; + let executionState: ExecutionState; + let openWorkflowId: number | undefined; + let workflowReadonly: boolean; + let operatorSchemaProperties: Record; + let consoleMessages: Array<{ + msgType: { name: string }; + timestamp: { seconds: number; nanos: number }; + title: string; + }>; + let consoleUpdates$: Subject; + + // Stub the transport at the class seam (callModel), as migration-llm.spec.ts does: + // mocking the "ai" module leaks across specs sharing the import and hits the network. + const stubModel = (raw: string) => vi.spyOn(service as any, "callModel").mockResolvedValue({ text: raw }); + const state = () => (service as any).stateSubject.getValue() as FixState; + + const suggestion = (over: object = {}) => + JSON.stringify({ + explanation: "the column is user_email", + fix_type: "code_change", + // Whole lines with their indentation, which is what the prompt requires. + original_snippet: " yield t['email']", + suggested_snippet: " yield t['user_email']", + confidence: "high", + ...over, + }); + + beforeEach(() => { + setOperatorProperty = vi.fn(); + executeWorkflow = vi.fn().mockReturnValue(true); + killWorkflow = vi.fn(); + executionState$ = new Subject<{ previous: any; current: any }>(); + operatorProperties = { code: CODE, workers: 1 }; + modificationEnabled = true; + modificationEnabled$ = new Subject(); + // A live run is what holds the modification lock in production, so every test that + // sets modificationEnabled = false needs one for the lock to be the kind ensureModifiable + // is willing to break. + executionState = ExecutionState.Running; + openWorkflowId = 11; + workflowReadonly = false; + operatorSchemaProperties = {}; + consoleMessages = []; + consoleUpdates$ = new Subject(); + + TestBed.configureTestingModule({ + providers: [ + AiWorkflowFixerService, + { + provide: WorkflowActionService, + useValue: { + setOperatorProperty, + getTexeraGraph: () => ({ getOperator: () => ({ operatorProperties, operatorType: "PythonUDFV2" }) }), + getOpenWorkflowId: () => openWorkflowId, + getWorkflowMetadata: () => ({ readonly: workflowReadonly }), + // Unlocked by default; the locked path has its own test. + checkWorkflowModificationEnabled: () => modificationEnabled, + getWorkflowModificationEnabledStream: () => modificationEnabled$.asObservable(), + }, + }, + { + provide: ExecuteWorkflowService, + useValue: { + executeWorkflow, + killWorkflow, + getExecutionState: () => ({ state: executionState }), + getExecutionStateStream: () => executionState$.asObservable(), + }, + }, + { + provide: WorkflowConsoleService, + useValue: { + getConsoleMessages: () => consoleMessages, + getConsoleMessageUpdateStream: () => consoleUpdates$.asObservable(), + }, + }, + { + // Stubbed rather than left to the real service, which fetches operator metadata + // from the constructor: 46 TestBeds meant 46 live requests, and the suite died + // waiting on them instead of reporting. + provide: OperatorMetadataService, + useValue: { getOperatorSchema: () => ({ jsonSchema: { properties: operatorSchemaProperties } }) }, + }, + ...commonTestProviders, + ], + }); + service = TestBed.inject(AiWorkflowFixerService); + }); + + describe("classifyError", () => { + it("detects each supported pattern", () => { + expect(classifyError(KEY_ERROR_EVENT.header)).toEqual("missing_column"); + expect(classifyError("TypeError: unsupported operand type(s) for *: 'str' and 'int'")).toEqual("type_error"); + expect(classifyError("ValueError: input contains NaN")).toEqual("null_error"); + // A bare 404 is a line number, an unrelated HTTP status, or data -- not a model lookup. + expect(classifyError(' File "udf.py", line 404, in process_tuple')).toEqual("unsupported"); + expect(classifyError("requests.HTTPError: 404 for https://example.com/report")).toEqual("unsupported"); + }); + + it("returns unsupported for out-of-scope, empty, and near-miss messages", () => { + expect(classifyError("ZeroDivisionError: division by zero")).toEqual("unsupported"); + // Only a JVM operator raises this, and a JVM operator has no code for Apply to patch. + expect(classifyError("java.lang.NullPointerException: boom")).toEqual("unsupported"); + expect(classifyError("")).toEqual("unsupported"); + expect(classifyError("TypeError: 'int' object is not iterable")).toEqual("unsupported"); + }); + }); + + describe("analyzeError", () => { + it("moves idle -> analyzing -> ready and maps the LLM response into the fix", async () => { + stubModel(suggestion()); + const seen: string[] = []; + service.getState$().subscribe(s => seen.push(s.status)); + + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + expect(seen).toEqual(["idle", "analyzing", "ready"]); + expect(state().suggestedFix).toEqual({ + original: " yield t['email']", + suggested: " yield t['user_email']", + explanation: "the column is user_email", + confidence: "high", + // CODE holds the snippet once, and the panel only shows this when it is more. + occurrences: 1, + }); + }); + + it("accepts snippets quoted with their trailing newline", async () => { + // Whether the model includes the final newline is not something the prompt settles, + // and with it `code[end]` is the first character of the next line -- so the same valid + // reply was refused mid-file and accepted on the last line. One trailing newline is + // dropped from each snippet, independently, so neither the check nor the replacement + // sees it and no lines are merged. + // A line in the MIDDLE of the file: quoted with its newline, the next character is the + // start of the following line rather than a newline. The last line of a file passes + // either way, which is what hid this. + const code = "def f():" + NL + " a = read()" + NL + " yield a" + NL; + stubModel(suggestion({ original_snippet: " a = read()" + NL, suggested_snippet: " a = f()" + NL })); + + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, code); + + expect(state().status).toEqual("ready"); + expect(state().suggestedFix?.original).toEqual(" a = read()"); + expect(state().suggestedFix?.suggested).toEqual(" a = f()"); + }); + + it("still refuses a mid-line match quoted with a newline", async () => { + // Tolerating the newline must not tolerate what the end-of-line check was added for: + // once it is dropped, `n = 1` is again only the start of `n = 10`. + const code = "def f():\n n = 10\n"; + stubModel(suggestion({ original_snippet: " n = 1\n", suggested_snippet: " n = 2\n" })); + + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, code); + + expect(state().status).toEqual("error"); + }); + it("refuses a snippet quoted from the middle of a line", async () => { + // The prompt asks for whole lines carrying their indentation. A mid-line snippet says + // nothing about the column its continuation lines belong at, and inferring one is the + // guess that put `yield` at sixteen spaces in an eight-space body. + stubModel(suggestion({ original_snippet: "yield t['email']", suggested_snippet: "yield t['user_email']" })); + + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + expect(state().status).toEqual("error"); + expect(state().suggestedFix).toBeUndefined(); + }); + + it("reports a snippet that is not in the code instead of offering an Apply that cannot work", async () => { + // The panel only shows the occurrence count when it is above one, so a snippet the + // model invented looked like any other suggestion. Apply then failed with "the code + // changed since the suggestion was generated", which blames an edit that never + // happened. + stubModel(suggestion({ original_snippet: " yield t['address']" })); + + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + expect(state().status).toEqual("error"); + expect(state().suggestedFix).toBeUndefined(); + }); + + it("refuses a reply that names a configuration field instead of quoting code", async () => { + // Not by reading fix_type -- that field is gone, since one legal value could only ever + // refuse a good reply. A field name is not in the code, so the occurrence count turns + // it away on its own. + stubModel(suggestion({ fix_type: "property_change", original_snippet: "modelId", suggested_snippet: "gpt-4" })); + + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + expect(state().status).toEqual("error"); + expect(state().suggestedFix).toBeUndefined(); + }); + + it("accepts a reply that omits fix_type", async () => { + // There is one kind of fix, so demanding the field only added a way for a good answer + // to be thrown away. + const { fix_type, ...withoutKind } = JSON.parse(suggestion()); + stubModel(JSON.stringify(withoutKind)); + + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + expect(state().status).toEqual("ready"); + }); + + it("tolerates a fenced JSON response even though the prompt forbids markdown", async () => { + stubModel("```json\n" + suggestion() + "\n```"); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + expect(state().status).toEqual("ready"); + }); + + it("never calls the model for an out-of-scope error", async () => { + const callModel = vi.spyOn(service as any, "callModel"); + await service.analyzeError(OP, err("ZeroDivisionError: division by zero"), SCHEMA, CODE); + + expect(callModel).not.toHaveBeenCalled(); + expect(state().status).toEqual("unsupported"); + expect(state().suggestedFix).toBeUndefined(); + }); + + it("ends in error on transport failure, malformed JSON, or a missing field", async () => { + vi.spyOn(service as any, "callModel").mockRejectedValue(new Error("502")); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + expect(state().status).toEqual("error"); + + stubModel("I cannot help with that."); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + expect(state().status).toEqual("error"); + + stubModel(JSON.stringify({ explanation: "e", fix_type: "code_change", confidence: "low" })); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + expect(state().status).toEqual("error"); + }); + }); + + it("ignores a slow analysis that a newer one has superseded", async () => { + // A frame whose operator does not own the state renders idle, so its Analyze button + // stays live while another operator is mid-analysis and two requests can overlap. + // The first one landing last must not replace the newer result. + let resolveFirst: (value: { text: string }) => void = () => {}; + const slowFirst = new Promise<{ text: string }>(resolve => (resolveFirst = resolve)); + vi.spyOn(service as any, "callModel") + .mockReturnValueOnce(slowFirst) + .mockResolvedValueOnce({ text: suggestion({ explanation: "for the second operator" }) }); + + const first = service.analyzeError("op-first", KEY_ERROR_EVENT, SCHEMA, CODE); + await service.analyzeError("op-second", KEY_ERROR_EVENT, SCHEMA, CODE); + expect(state().operatorId).toEqual("op-second"); + + resolveFirst({ text: suggestion({ explanation: "for the first operator" }) }); + await first; + + expect(state().operatorId).toEqual("op-second"); + expect(state().suggestedFix?.explanation).toEqual("for the second operator"); + }); + + it("sends the operator's configuration nowhere, so a credential cannot ride along", async () => { + // Operators keep secrets beside ordinary settings, and filtering by field name was never + // enough: a Python UDF's uiParameters is a list of attribute/value pairs, so the name + // says nothing about the value. None of the three fixes reads configuration, so none is + // sent -- this asserts the wiring, and ai-fix-prompt.spec covers the prompt itself. + operatorProperties = { + code: CODE, + hfApiToken: "hf_live_abc", + uiParameters: [{ attribute: { attributeName: "hf_token" }, value: "hf_ui_xyz" }], + }; + const callModel = vi.spyOn(service as any, "callModel").mockResolvedValue({ text: suggestion() }); + + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + const prompt = JSON.stringify(callModel.mock.calls[0][0]); + expect(prompt).not.toContain("hf_live_abc"); + expect(prompt).not.toContain("hf_ui_xyz"); + expect(prompt).not.toContain("Operator configuration"); + }); + + describe("classifying a real traceback", () => { + it("reads a JVM exception from the header rather than the frames below it", () => { + // mkConsoleMessage puts `err.toString` in `title` and only `getStackTrace.mkString` in + // `message`, so the frames arrive with no `at ` prefix -- and on Java 9+ the JDK's own + // carry a module name. Nothing has to find the exception any more, because the header + // is handed over separately. It classifies as unsupported, which is the honest answer: + // the fix this panel writes is a change to `code`, and a JVM operator has none. + expect(classifyError("java.lang.NullPointerException: value is null")).toEqual("unsupported"); + expect(classifyError("java.base/java.lang.Thread.run(Thread.java:840)")).toEqual("unsupported"); + }); + + it("classifies a chained traceback on the exception that stopped the worker", () => { + // Nothing here has to unpick the chain: the Python worker sets `title` from + // `format_exception[-1]`, which is already the final exception. + expect(classifyError("ZeroDivisionError: integer division or modulo by zero")).toEqual("unsupported"); + expect(classifyError("KeyError: " + QUOTE + "email" + QUOTE)).toEqual("missing_column"); + }); + + it("never sees the frame lines that read like patterns on their own", () => { + // A path with "model" above a `line 404` entry used to look like a model 404. Those + // lines live in `full`, which only the prompt is given. + const event = err( + "ZeroDivisionError: integer division or modulo by zero", + [ + "Traceback (most recent call last):", + ' File "/usr/lib/python3/sklearn/model_selection/_split.py", line 404, in split', + " n = len(indices) // 0", + "ZeroDivisionError: integer division or modulo by zero", + ].join("\n") + ); + + expect(classifyError(event.header)).toEqual("unsupported"); + expect(event.full).toContain("line 404"); + }); + + it("still classifies the plain case the panel is built for", () => { + expect(classifyError(KEY_ERROR_EVENT.header)).toEqual("missing_column"); + }); + }); + + it("lets an unsupported error supersede an analysis already in flight", async () => { + // The unsupported branch answers without calling the model, but it still replaces the + // panel state, so a slower analysis started earlier must not land on top of it. + let resolveFirst: (value: { text: string }) => void = () => {}; + const slowFirst = new Promise<{ text: string }>(resolve => (resolveFirst = resolve)); + vi.spyOn(service as any, "callModel").mockReturnValueOnce(slowFirst); + + const first = service.analyzeError("op-first", KEY_ERROR_EVENT, SCHEMA, CODE); + await service.analyzeError("op-second", err("ZeroDivisionError: division by zero"), SCHEMA, CODE); + expect(state().operatorId).toEqual("op-second"); + expect(state().status).toEqual("unsupported"); + + resolveFirst({ text: suggestion() }); + await first; + + expect(state().operatorId).toEqual("op-second"); + expect(state().status).toEqual("unsupported"); + }); + + it("rejects a reply whose snippets are not strings", async () => { + // The template splits these into diff lines; an array would throw inside change + // detection, leaving the panel on `ready` with nothing rendered and no way out. + stubModel( + JSON.stringify({ + explanation: "the column is user_email", + fix_type: "code_change", + original_snippet: [" yield t['email']"], + suggested_snippet: " yield t['user_email']", + confidence: "high", + }) + ); + + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + expect(state().status).toEqual("error"); + expect(state().suggestedFix).toBeUndefined(); + }); + + describe("a finished run clears the panel", () => { + const ended = (state: string) => executionState$.next({ previous: {}, current: { state } }); + + it("drops the applied message once the re-run is over", async () => { + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + await service.applyFix(); + expect(state().status).toEqual("applied"); + + ended("Completed"); + + // Otherwise a run that failed again brought the tab back still claiming a re-run. + expect(state().status).toEqual("idle"); + }); + + it("drops a suggestion the user never applied", async () => { + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + expect(state().status).toEqual("ready"); + + ended("Failed"); + + expect(state().status).toEqual("idle"); + }); + + it("does not revive a suggestion the reset already cleared", async () => { + // Reachable from the toolbar: analyse a failed operator, then stop the run while the + // model is still answering. The panel resets, and the late reply used to repaint a + // suggestion for an execution that no longer exists. + let resolveAnalysis: (value: { text: string }) => void = () => {}; + const slow = new Promise<{ text: string }>(resolve => (resolveAnalysis = resolve)); + vi.spyOn(service as any, "callModel").mockReturnValueOnce(slow); + + const analysing = service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + ended("Killed"); + expect(state().status).toEqual("idle"); + + resolveAnalysis({ text: suggestion() }); + await analysing; + + expect(state().status).toEqual("idle"); + expect(state().suggestedFix).toBeUndefined(); + }); + + it("does not revive a suggestion the user discarded", async () => { + let resolveAnalysis: (value: { text: string }) => void = () => {}; + const slow = new Promise<{ text: string }>(resolve => (resolveAnalysis = resolve)); + vi.spyOn(service as any, "callModel").mockReturnValueOnce(slow); + + const analysing = service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + service.discardFix(); + + resolveAnalysis({ text: suggestion() }); + await analysing; + + expect(state().status).toEqual("idle"); + }); + + it("never flashes the panel back to idle during an apply", async () => { + // applyFix kills a live execution before it writes, and that Killed state reaches + // this very subscription. The end state is "applied" either way because the apply + // publishes last, so what the guard protects is the moment in between: without it + // the panel drops to idle and offers Analyze while the apply is still running. + modificationEnabled = false; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + const seen: string[] = []; + const watching = service.getState$().subscribe(current => seen.push(current.status)); + + const applying = service.applyFix(); + ended("Killed"); + modificationEnabled = true; + modificationEnabled$.next(true); + await applying; + watching.unsubscribe(); + + expect(seen).toEqual(["ready", "applying", "applied"]); + expect(setOperatorProperty).toHaveBeenCalled(); + }); + }); + + describe("replaceAll", () => { + it("writes the suggestion verbatim, adding no indentation of its own", () => { + // The snippet is a whole line carrying its own eight spaces, and so is the + // replacement. Adding the context indentation on top is what put `yield` at sixteen. + const code = "def f():\n a = read()\n yield a\n"; + + const out = replaceAll(code, " a = read()", " a = read().strip()").code; + + expect(out).toEqual("def f():\n a = read().strip()\n yield a\n"); + }); + + it("places a multi-line suggestion at the column each of its lines asks for", () => { + const code = "def f():\n x = read()\n"; + + const out = replaceAll(code, " x = read()", " x = read()\n x = x.strip()").code; + + expect(out).toEqual("def f():\n x = read()\n x = x.strip()\n"); + }); + + it("counts the places it changed", () => { + const out = replaceAll("p()\np()\n", "p()", "q()"); + + expect(out.code).toEqual("q()\nq()\n"); + expect(out.count).toEqual(2); + }); + }); + + describe("matchesWholeLines", () => { + it("accepts a snippet that begins a line, including the first one", () => { + expect(matchesWholeLines(" a()\n b()\n", " a()")).toBe(true); + expect(matchesWholeLines("a()\n", "a()")).toBe(true); + }); + + it("rejects a snippet quoted from the middle of a line", () => { + // Without its leading indentation the snippet says nothing about what column the + // lines after it belong at, which is exactly what used to be guessed. + expect(matchesWholeLines(" a()\n", "a()")).toBe(false); + }); + + it("rejects when any one occurrence is mid-line, not just the first", () => { + // Every match is replaced, so one mid-line occurrence is enough to make the result + // depend on a guess -- and the first match looking fine is what would hide it. + expect(matchesWholeLines("a()\nif x: a()\n", "a()")).toBe(false); + }); + + it("rejects a whole line that is only the start of a longer one", () => { + // The dangerous half: the match begins a line, so the indentation is unambiguous and + // the count is 1, which hides the badge -- but it stops mid-line. Replacing it leaves + // the rest of the line stuck on the end, so `n = 1` becoming `n = 2` silently rewrites + // `n = 10` as `n = 20` while the diff only ever showed `n = 1`. + expect(matchesWholeLines(" n = 10\n", " n = 1")).toBe(false); + }); + + it("accepts a match that ends at the end of the code, with no trailing newline", () => { + expect(matchesWholeLines(" a()", " a()")).toBe(true); + }); + + it("rejects when only one of several occurrences stops mid-line", () => { + expect(matchesWholeLines(" a()\n a() + 1\n", " a()")).toBe(false); + }); + it("says nothing about a snippet the code does not contain", () => { + // Absence is a separate refusal, with its own message. + expect(matchesWholeLines("a()\n", "zzz()")).toBe(true); + }); + }); + + describe("a newer error clears a stale result", () => { + const errorAt = (seconds: number) => ({ + msgType: { name: "ERROR" }, + timestamp: { seconds, nanos: 0 }, + title: "KeyError: 'lines'", + }); + + it("goes back to idle when the operator reports a newer error", async () => { + // The main case this panel exists for never reaches a terminal state: a UDF that + // raises leaves the execution Running, so resetting on Completed/Failed/Killed misses + // it entirely and the tab came back saying "re-running" over a run stuck on an error. + consoleMessages = [errorAt(100)]; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + await service.applyFix(); + expect(state().status).toEqual("applied"); + + consoleMessages = [errorAt(100), errorAt(250)]; + consoleUpdates$.next(); + + expect(state().status).toEqual("idle"); + }); + + it("leaves the result alone while the error is still the one it was made for", async () => { + consoleMessages = [errorAt(100)]; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + await service.applyFix(); + + consoleUpdates$.next(); + + expect(state().status).toEqual("applied"); + }); + }); + + describe("applyFix", () => { + it("replaces only the snippet, keeps other properties, and re-runs", async () => { + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + await service.applyFix(); + + expect(setOperatorProperty).toHaveBeenCalledWith(OP, { + code: CODE.replace("yield t['email']", "yield t['user_email']"), + workers: 1, + }); + expect(executeWorkflow).toHaveBeenCalledWith(""); + expect(state().status).toEqual("applied"); + }); + + it("replaces every occurrence of the snippet, not just the first", async () => { + // A UDF that reads the same wrong column twice must come back fixed in both places, + // or the re-run fails again on the second read. + const twice = CODE + " yield t['email']\n"; + operatorProperties = { code: twice, workers: 1 }; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, twice); + + await service.applyFix(); + + const patched = setOperatorProperty.mock.calls[0][1].code; + expect(patched).not.toContain("t['email']"); + expect(patched.match(/t\['user_email'\]/g)).toHaveLength(2); + }); + + it("does nothing without a ready suggestion", async () => { + await service.applyFix(); + expect(setOperatorProperty).not.toHaveBeenCalled(); + expect(executeWorkflow).not.toHaveBeenCalled(); + expect(state().status).toEqual("idle"); + }); + + it("does not stop the user's run for a fix it was going to refuse anyway", async () => { + // Every refusal used to run after ensureModifiable(), which kills the paused execution + // to clear the modification lock. So a fix that could never be written cost the user + // their run: the panel reported the refusal over a workflow it had just stopped. + modificationEnabled = false; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + operatorProperties.code = "class ProcessTupleOperator(UDFOperatorV2):\n pass\n"; + + await service.applyFix(); + + expect(killWorkflow).not.toHaveBeenCalled(); + expect(setOperatorProperty).not.toHaveBeenCalled(); + expect(state().status).toEqual("apply_failed"); + }); + + it("refuses when the snippet now appears fewer times than the panel promised", async () => { + // Deliberate, not an oversight in the comparison: the badge said two places, so the + // user approved an edit to two. Patching the one that is left is a different change + // from the one on screen, and the diff for it is one model call away. + const twice = CODE + " yield t['email']\n"; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, twice); + expect(state().suggestedFix?.occurrences).toEqual(2); + + operatorProperties.code = CODE; + + await service.applyFix(); + + expect(setOperatorProperty).not.toHaveBeenCalled(); + expect(state().applyError).toContain("1 place instead of 2 places"); + }); + + it("refuses when the snippet now appears more times than the panel promised", async () => { + // The occurrence count is captured at the analysis and drives the "changes N places" + // badge, which stays hidden at one. Apply only checked that the snippet was still + // THERE, so a copy added in between was patched too -- an edit the user was never + // warned about. Reachable because a kill lifts the modification lock without + // resetting the panel: `Uninitialized` is deliberately not a terminal state. + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + expect(state().suggestedFix?.occurrences).toEqual(1); + + operatorProperties.code = CODE + " yield t['email']\n"; + + await service.applyFix(); + + expect(setOperatorProperty).not.toHaveBeenCalled(); + expect(executeWorkflow).not.toHaveBeenCalled(); + expect(state().status).toEqual("apply_failed"); + expect(state().applyError).toContain("2 places"); + }); + + it("reports an apply failure, without re-running, when the snippet is no longer in the code", async () => { + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + // The user edits the UDF while reading the suggestion -- the snippet it quoted was in + // the code when it was generated and is gone by the time Apply runs. A snippet that + // was never there is a different fault, and the analysis refuses that one outright. + operatorProperties.code = "class ProcessTupleOperator(UDFOperatorV2):\n pass\n"; + + await service.applyFix(); + + expect(setOperatorProperty).not.toHaveBeenCalled(); + expect(executeWorkflow).not.toHaveBeenCalled(); + expect(state().status).toEqual("apply_failed"); + }); + + it("writes the fix but claims no re-run when the deployment refuses one", async () => { + // Whatever the reason -- no warehouse, a computing unit shutting down -- the panel + // asks instead of predicting, so it cannot report "re-running" over the toast that + // says the run was refused. + executeWorkflow.mockReturnValue(false); + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + await service.applyFix(); + + expect(setOperatorProperty).toHaveBeenCalled(); + expect(state().status).toEqual("applied_without_run"); + }); + + it("reports the re-run once the deployment accepts it", async () => { + executeWorkflow.mockReturnValue(true); + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + await service.applyFix(); + + expect(executeWorkflow).toHaveBeenCalledWith(""); + expect(state().status).toEqual("applied"); + }); + + it("kills the live execution before writing, since modification is locked while it runs", async () => { + // The traceback this panel exists for arrives while the execution is still Running -- + // the worker pauses, it does not fail -- and Texera locks workflow modification for as + // long as a run is live. Writing through the lock is what every other editing feature + // avoids by checking first. + modificationEnabled = false; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + const applying = service.applyFix(); + expect(killWorkflow).toHaveBeenCalled(); + expect(setOperatorProperty).not.toHaveBeenCalled(); + + modificationEnabled = true; + modificationEnabled$.next(true); + await applying; + + expect(setOperatorProperty).toHaveBeenCalled(); + expect(executeWorkflow).toHaveBeenCalledWith(""); + }); + + it("explains a lock that never clears, instead of reporting a bare timeout", async () => { + // rxjs reports this as "Timeout has occurred", which says nothing about what the + // panel was waiting for. + vi.useFakeTimers(); + modificationEnabled = false; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + const applying = service.applyFix(); + await vi.advanceTimersByTimeAsync(20_000); + await applying; + + expect(setOperatorProperty).not.toHaveBeenCalled(); + expect(state().status).toEqual("apply_failed"); + expect(state().applyError).toContain("stayed locked"); + vi.useRealTimers(); + }); + + it("retries from a failed apply once the operator is back in shape", async () => { + // The panel keeps the suggestion and its Apply button after a refused write, so the + // button has to work: refusing silently because the status is no longer `ready` + // leaves the user clicking something that does nothing. + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + operatorProperties.code = "something else entirely"; + await service.applyFix(); + expect(state().status).toEqual("apply_failed"); + expect(setOperatorProperty).not.toHaveBeenCalled(); + + operatorProperties.code = CODE; + await service.applyFix(); + + expect(setOperatorProperty).toHaveBeenCalled(); + expect(executeWorkflow).toHaveBeenCalledWith(""); + expect(state().status).toEqual("applied"); + }); + + it("refuses to write a suggestion into a different workflow", async () => { + // Two imports of the same JSON carry the same operator ids, and this service outlives + // the workspace, so the operator id alone would let the write land on the wrong graph. + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + openWorkflowId = 12; + await service.applyFix(); + + expect(setOperatorProperty).not.toHaveBeenCalled(); + expect(executeWorkflow).not.toHaveBeenCalled(); + expect(state().applyError).toContain("workflow changed"); + }); + + it("refuses when the workflow changes during the unlock wait", async () => { + // ensureModifiable can wait seconds for a kill to clear the lock, and the user is + // free to open another workflow while it does. + modificationEnabled = false; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + const applying = service.applyFix(); + openWorkflowId = 12; + modificationEnabled = true; + modificationEnabled$.next(true); + await applying; + + expect(setOperatorProperty).not.toHaveBeenCalled(); + expect(state().applyError).toContain("workflow changed"); + }); + + it("drops a reply that lands after the user switched workflows", async () => { + let resolveAnalysis: (value: { text: string }) => void = () => {}; + const slow = new Promise<{ text: string }>(resolve => (resolveAnalysis = resolve)); + vi.spyOn(service as any, "callModel").mockReturnValueOnce(slow); + + const analysing = service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + openWorkflowId = 12; + resolveAnalysis({ text: suggestion() }); + await analysing; + + expect(state().suggestedFix).toBeUndefined(); + }); + + it("does not report an apply over an analysis started while it waited", async () => { + // Reachable: Apply on one operator, then switch to another whose frame shows idle and + // click Analyze there. The order is forced here so the newer analysis settles first, + // leaving the apply to publish last -- which is when it overwrites what the user is + // now looking at. + modificationEnabled = false; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + const applying = service.applyFix(); + await service.analyzeError("other-op", KEY_ERROR_EVENT, SCHEMA, CODE); + expect(state().operatorId).toEqual("other-op"); + + modificationEnabled = true; + modificationEnabled$.next(true); + await applying; + + expect(state().operatorId).toEqual("other-op"); + expect(state().status).toEqual("ready"); + }); + + it("refuses on a read-only workflow instead of killing someone else's run", async () => { + // enableWorkflowModification() is a no-op while the workflow is read-only, so the + // unlock this waits for never arrives. Running only needs write access to the + // computing unit, so a read-only user can reach a paused run and used to have it + // killed, then wait fifteen seconds to be told the execution would not stop. + workflowReadonly = true; + modificationEnabled = false; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + await service.applyFix(); + + expect(killWorkflow).not.toHaveBeenCalled(); + expect(setOperatorProperty).not.toHaveBeenCalled(); + expect(state().applyError).toContain("read-only"); + }); + + it("refuses a lock that no running execution explains", async () => { + // killWorkflow() only clears the lock a live run holds. Any other source -- the + // workflow being shared without write access, a version being viewed -- survives the + // kill, so the wait would run out while the user watched a spinner. + modificationEnabled = false; + executionState = ExecutionState.Uninitialized; + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + await service.applyFix(); + + expect(killWorkflow).not.toHaveBeenCalled(); + expect(setOperatorProperty).not.toHaveBeenCalled(); + expect(state().applyError).toContain("other than a running execution"); + }); + + it("reports an apply failure when the graph rejects the change", async () => { + setOperatorProperty.mockImplementation(() => { + throw new Error("read-only workflow"); + }); + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + await service.applyFix(); + + expect(state().status).toEqual("apply_failed"); + expect(executeWorkflow).not.toHaveBeenCalled(); + }); + }); + + it("discards to idle and keeps state across re-subscription", async () => { + stubModel(suggestion()); + await service.analyzeError(OP, KEY_ERROR_EVENT, SCHEMA, CODE); + + // A frame rebuilt by clearResultPanel() re-subscribes and must still see the fix. + expect((await firstValueFrom(service.getState$())).status).toEqual("ready"); + + service.discardFix(); + const after = await firstValueFrom(service.getState$()); + expect(after).toEqual({ operatorId: "", status: "idle" }); + }); +}); diff --git a/frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts new file mode 100644 index 00000000000..306ba4df35a --- /dev/null +++ b/frontend/src/app/workspace/component/result-panel/ai-fix/ai-workflow-fixer.service.ts @@ -0,0 +1,548 @@ +/** + * 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 } from "./ai-fix-prompt"; +import { WorkflowConsoleService } from "../../../service/workflow-console/workflow-console.service"; +import { GuiConfigService } from "../../../../common/service/gui-config.service"; + +export type FixErrorType = "missing_column" | "type_error" | "null_error" | "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; + /** Whole lines quoted from the operator's code, and what to put in their place. */ + original: string; + suggested: string; + explanation: string; + /** Rendered as the confidence badge. */ + confidence: FixConfidence; +} + +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; + suggestedFix?: SuggestedFix; + // `unsupported`: the error is outside the three 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; +} + +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 = 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 = 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/i, "null_error"], +]; + +/** + * 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 `matchesWholeLines` 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 covers whole lines. + * + * Both ends matter, for different reasons. Starting a line is what makes the indentation + * unambiguous: the snippet carries its own, so the suggestion carries the indentation it + * wants and nothing is inferred. Ending a line is what keeps the replacement honest -- a + * snippet that begins a line but stops inside it still replaces cleanly, so ` n = 1` + * matches ` n = 10` once, shows no occurrence badge, and leaves the trailing `0` + * behind: the diff offers `n = 1` -> `n = 2` and the operator is left saying `n = 20`. + */ +export function matchesWholeLines(code: string, original: string): boolean { + let from = 0; + for (;;) { + const at = code.indexOf(original, from); + if (at === -1) { + return true; + } + const end = at + original.length; + const startsLine = at === 0 || code[at - 1] === "\n"; + const endsLine = end === code.length || code[end] === "\n"; + if (!startsLine || !endsLine) { + return false; + } + from = end; + } +} + +/** + * 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(IDLE_STATE); + + constructor( + private config: GuiConfigService, + private workflowActionService: WorkflowActionService, + private executeWorkflowService: ExecuteWorkflowService, + 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 { + 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 + ): Promise { + const errorType = classifyError(error.header); + const workflowId = this.workflowActionService.getOpenWorkflowId(); + const base: FixState = { + operatorId, + workflowId, + 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 { + const { text } = await this.callModelWithTimeout(buildFixPrompt(error.full, code, inputSchemas)); + if (seq !== this.analysisSeq || workflowId !== this.workflowActionService.getOpenWorkflowId()) { + return; + } + const suggestedFix = this.toFix(text); + // 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 (!matchesWholeLines(code ?? "", suggestedFix.original)) { + throw new Error("the model did not quote whole lines, so the fix cannot be placed"); + } + // 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. + 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 { + 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 { + const properties: Record = { + ...this.workflowActionService.getTexeraGraph().getOperator(current.operatorId).operatorProperties, + }; + 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; + return properties; + } + + public discardFix(): void { + this.reset(); + } + + /** + * Clears the panel and supersedes whatever analysis is in flight. + * + * Publishing IDLE is not enough on its own: a reply still in flight matches the sequence + * it started with and repaints a suggestion over the cleared panel. 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() ?? "", + // Named as the LiteLLM proxy exposes it, not as the provider does, and read from the + // deployment's own config: a proxy publishing a different alias would otherwise need a + // rebuild to be reachable. + }).chat(this.config.env.aiFixModel); + return generateText({ model, messages, abortSignal }); + } + + private callModelWithTimeout(prompt: string): Promise<{ text: string }> { + const controller = new AbortController(); + let timer: ReturnType; + const timeout = new Promise((_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) + ); + } + + /** The operator's most recent ERROR, as a value that changes when a new one arrives. */ + private latestErrorKey(operatorId: string): string | undefined { + const last = (this.workflowConsoleService.getConsoleMessages(operatorId) ?? []) + .filter(message => message.msgType?.name === "ERROR") + .pop(); + return last ? `${last.timestamp.seconds}.${last.timestamp.nanos}:${last.title}` : undefined; + } + + /** + * 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 { + if (this.workflowActionService.checkWorkflowModificationEnabled()) { + return; + } + // enableWorkflowModification() is a no-op while the workflow is read-only, so the unlock + // waited for below never arrives. Running only needs write access to the computing unit, + // so a read-only user can reach a paused run: refused here rather than below, or their + // run is killed on their behalf and they wait fifteen seconds to hear it would not stop. + if (this.workflowActionService.getWorkflowMetadata().readonly) { + throw new Error("this workflow is read-only, so the fix cannot be written"); + } + // The lock has other causes -- loading, a version preview -- and killing then both fails + // and reports the wrong reason. + if (!LIVE_STATES.has(this.executeWorkflowService.getExecutionState().state)) { + throw new Error("the workflow is locked by something other than a running execution"); + } + try { + this.executeWorkflowService.killWorkflow(); + } catch (err) { + throw new Error(`the running execution could not be stopped: ${(err as Error).message}`); + } + try { + await firstValueFrom( + this.workflowActionService.getWorkflowModificationEnabledStream().pipe( + filter(enabled => enabled), + timeout(UNLOCK_TIMEOUT_MS) + ) + ); + } catch { + // rxjs reports this as "Timeout has occurred", which tells the user nothing about + // what the panel was waiting for. + throw new Error("the running execution did not stop, so the workflow stayed locked"); + } + } + + /** Maps the model's JSON contract onto SuggestedFix, rejecting incomplete replies. */ + private toFix(raw: string): SuggestedFix { + let text = raw.trim(); + const fenced = text.match(/```(?:[a-zA-Z]+)?\s*([\s\S]*?)```/); + if (fenced) { + text = fenced[1].trim(); + } + let parsed: any; + try { + parsed = JSON.parse(text); + } catch (err) { + throw new Error(`model did not return JSON: ${(err as Error).message}`); + } + const { explanation, original_snippet, suggested_snippet, confidence } = parsed ?? {}; + if (!explanation || !original_snippet || !suggested_snippet) { + throw new Error("model response is missing a required field"); + } + // Checked as well as present: the template splits these into diff lines, and a reply + // that sends an array or an object would throw inside change detection, which leaves the + // panel stuck on `ready` with nothing rendered and no way back. + for (const [name, value] of [ + ["explanation", explanation], + ["original_snippet", original_snippet], + ["suggested_snippet", suggested_snippet], + ] as const) { + if (typeof value !== "string") { + throw new Error(`model returned ${typeof value} for ${name}, expected a string`); + } + } + // One trailing newline off each, independently. The prompt asks for whole lines and does + // not settle whether the final newline is part of the quote, so both forms arrive. Kept, + // it makes the verdict depend on where the line sits: `code[end]` is the next line's + // first character everywhere except the last line of the file, which passes. Dropped + // from the replacement too, so the line's own newline stays put and nothing is merged. + const withoutTrailingNewline = (snippet: string) => (snippet.endsWith("\n") ? snippet.slice(0, -1) : snippet); + return { + original: withoutTrailingNewline(original_snippet), + suggested: withoutTrailingNewline(suggested_snippet), + explanation, + confidence: confidence === "high" || confidence === "low" ? confidence : "medium", + }; + } +} diff --git a/frontend/src/app/workspace/component/result-panel/result-panel.component.html b/frontend/src/app/workspace/component/result-panel/result-panel.component.html index d2c6a535493..0ad9f54eb9a 100644 --- a/frontend/src/app/workspace/component/result-panel/result-panel.component.html +++ b/frontend/src/app/workspace/component/result-panel/result-panel.component.html @@ -103,7 +103,7 @@

No results available to display.

-
+
{ let component: ResultPanelComponent; @@ -111,6 +113,15 @@ describe("ResultPanelComponent", () => { timestamp: { nanos: 0, seconds: 0 }, }); + const consoleMessage = (msgTypeName: string): ConsoleMessage => ({ + workerId: "w1", + timestamp: { nanos: 0, seconds: 0 }, + msgType: { name: msgTypeName }, + source: "udf.py:process_tuple:10", + title: "KeyError: 'lines'", + message: "Traceback (most recent call last):\nKeyError: 'lines'", + }); + it("should create", () => expect(component).toBeTruthy()); // Every other re-render waits for an event, and none follows a hand-over between a workflow's @@ -417,6 +428,205 @@ describe("ResultPanelComponent", () => { expect(errorConfig?.componentInputs).toEqual({ operatorId: "3" }); }); + it("registers the AI fix frame for a console ERROR while the execution is still running", () => { + // A Python UDF exception never becomes a fatal error: the worker pauses and the + // traceback arrives as an ERROR console message with the execution still Running, + // so the Failed branch never fires for the case the AI panel exists to fix. + workflowActionService.addOperator(mockResultPredicate, mockPoint); + // The fix this panel writes is a change to `code`, so it only offers itself on a + // Python UDF. The test metadata does not know the V2 types, so the graph is told. + vi.spyOn(workflowActionService.getTexeraGraph(), "getOperator").mockReturnValue({ + ...mockResultPredicate, + operatorID: "3", + operatorType: PYTHON_UDF_V2_OP_TYPE, + }); + const consoleService = TestBed.inject(WorkflowConsoleService); + const config = TestBed.inject(GuiConfigService); + (config.env as any).copilotEnabled = true; + vi.spyOn(workflowActionService.getJointGraphWrapper(), "getCurrentHighlightedOperatorIDs").mockReturnValue(["3"]); + vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([consoleMessage("ERROR")]); + + component.rerenderResultPanel(); + + const aiConfig = component.frameComponentConfigs.get("AI Fix"); + expect(aiConfig?.component).toBe(AiFixFrameComponent); + expect(aiConfig?.componentInputs).toEqual({ operatorId: "3" }); + }); + + it("registers the AI fix frame when the traceback arrives while the operator is already selected", () => { + // The common case: the user is watching the operator that fails. The highlight never + // moves and the execution stays Running, so the console update is the only signal the + // panel gets. Driving the stream instead of calling rerenderResultPanel() is the point + // of this test -- the wiring, not the branch, was what was missing. + vi.useFakeTimers(); + workflowActionService.addOperator(mockResultPredicate, mockPoint); + // The fix this panel writes is a change to `code`, so it only offers itself on a + // Python UDF. The test metadata does not know the V2 types, so the graph is told. + vi.spyOn(workflowActionService.getTexeraGraph(), "getOperator").mockReturnValue({ + ...mockResultPredicate, + operatorID: "3", + operatorType: PYTHON_UDF_V2_OP_TYPE, + }); + const consoleService = TestBed.inject(WorkflowConsoleService); + (TestBed.inject(GuiConfigService).env as any).copilotEnabled = true; + vi.spyOn(workflowActionService.getJointGraphWrapper(), "getCurrentHighlightedOperatorIDs").mockReturnValue(["3"]); + vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([consoleMessage("ERROR")]); + expect(component.frameComponentConfigs.has("AI Fix")).toBe(false); + + consoleService.clearConsoleMessages(); + vi.advanceTimersByTime(300); + + expect(component.frameComponentConfigs.get("AI Fix")?.component).toBe(AiFixFrameComponent); + vi.useRealTimers(); + }); + + it("tears down a stale AI fix frame once the operator stops reporting an error", () => { + // A re-run clears the console. Without the teardown the tab lingered, offering to fix + // an error that is no longer there. + workflowActionService.addOperator(mockResultPredicate, mockPoint); + // The fix this panel writes is a change to `code`, so it only offers itself on a + // Python UDF. The test metadata does not know the V2 types, so the graph is told. + vi.spyOn(workflowActionService.getTexeraGraph(), "getOperator").mockReturnValue({ + ...mockResultPredicate, + operatorID: "3", + operatorType: PYTHON_UDF_V2_OP_TYPE, + }); + const consoleService = TestBed.inject(WorkflowConsoleService); + (TestBed.inject(GuiConfigService).env as any).copilotEnabled = true; + vi.spyOn(workflowActionService.getJointGraphWrapper(), "getCurrentHighlightedOperatorIDs").mockReturnValue(["3"]); + const messages = vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([consoleMessage("ERROR")]); + component.rerenderResultPanel(); + expect(component.frameComponentConfigs.has("AI Fix")).toBe(true); + + messages.mockReturnValue([]); + component.rerenderResultPanel(); + + expect(component.frameComponentConfigs.has("AI Fix")).toBe(false); + }); + + it("toggles only the AI Fix entry on a console update, leaving the other frames alone", () => { + // A full rerender rebuilds every frame, so the Console frame's type filter, scroll + // position and debug input and the Result table were reset whenever any operator + // printed. The console stream must touch the AI Fix entry and nothing else. + workflowActionService.addOperator(mockResultPredicate, mockPoint); + // The fix this panel writes is a change to `code`, so it only offers itself on a + // Python UDF. The test metadata does not know the V2 types, so the graph is told. + vi.spyOn(workflowActionService.getTexeraGraph(), "getOperator").mockReturnValue({ + ...mockResultPredicate, + operatorID: "3", + operatorType: PYTHON_UDF_V2_OP_TYPE, + }); + const consoleService = TestBed.inject(WorkflowConsoleService); + (TestBed.inject(GuiConfigService).env as any).copilotEnabled = true; + vi.spyOn(workflowActionService.getJointGraphWrapper(), "getCurrentHighlightedOperatorIDs").mockReturnValue(["3"]); + vi.spyOn(consoleService, "hasConsoleMessages").mockReturnValue(true); + vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([consoleMessage("PRINT")]); + component.rerenderResultPanel(); + const consoleFrame = component.frameComponentConfigs.get("Console"); + expect(consoleFrame).toBeDefined(); + expect(component.frameComponentConfigs.has("AI Fix")).toBe(false); + + vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([consoleMessage("ERROR")]); + consoleService.clearConsoleMessages(); + + expect(component.frameComponentConfigs.has("AI Fix")).toBe(true); + // Same object, so the frame was never torn down and rebuilt. + expect(component.frameComponentConfigs.get("Console")).toBe(consoleFrame); + }); + + it("brings the tab back when the same operator fails again on a later run", () => { + // The map is written by two paths -- this stream and rerenderResultPanel -- but only + // this one fed the distinctUntilChanged key. Starting a run clears the console and the + // rerender drops the tab, while the pipe still remembers `true`, so the next error maps + // to `true` again, is read as "no change", and the tab never comes back. That is the + // fix-and-re-run loop this panel exists for. + workflowActionService.addOperator(mockResultPredicate, mockPoint); + // The fix this panel writes is a change to `code`, so it only offers itself on a + // Python UDF. The test metadata does not know the V2 types, so the graph is told. + vi.spyOn(workflowActionService.getTexeraGraph(), "getOperator").mockReturnValue({ + ...mockResultPredicate, + operatorID: "3", + operatorType: PYTHON_UDF_V2_OP_TYPE, + }); + const consoleService = TestBed.inject(WorkflowConsoleService); + (TestBed.inject(GuiConfigService).env as any).copilotEnabled = true; + vi.spyOn(workflowActionService.getJointGraphWrapper(), "getCurrentHighlightedOperatorIDs").mockReturnValue(["3"]); + vi.spyOn(consoleService, "hasConsoleMessages").mockReturnValue(true); + + // First run fails and the error reaches this stream, so the pipe now remembers `true`. + vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([consoleMessage("ERROR")]); + component.rerenderResultPanel(); + consoleService.clearConsoleMessages(); + expect(component.frameComponentConfigs.has("AI Fix")).toBe(true); + + // A new run starts. The console is emptied without notifying this stream -- the engine + // clears it on its own -- and the rerender takes the tab down, so the pipe's memory is + // now stale. + vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([]); + component.rerenderResultPanel(); + expect(component.frameComponentConfigs.has("AI Fix")).toBe(false); + + // It fails again, with the operator still selected. + vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([consoleMessage("ERROR")]); + consoleService.clearConsoleMessages(); + + expect(component.frameComponentConfigs.has("AI Fix")).toBe(true); + }); + + it("registers the display-name handler once rather than on every rerender", () => { + // It used to be called from the rerender subscriber, adding a subscription each time + // without releasing the previous one, so handlers piled up for the panel's lifetime. + const displayNameStream = vi.spyOn(workflowActionService.getTexeraGraph(), "getOperatorDisplayNameChangedStream"); + + component.rerenderResultPanel(); + component.rerenderResultPanel(); + + expect(displayNameStream).not.toHaveBeenCalled(); + }); + + it("registers no AI fix frame on an operator whose fix would be unwritable", () => { + // The only fix this panel writes is a change to `code`. A JVM operator can report an + // error the classifier recognises and still have nothing to patch, so offering the tab + // there spends a model call to reach a refusal. mockResultPredicate is not a UDF. + workflowActionService.addOperator(mockResultPredicate, mockPoint); + const consoleService = TestBed.inject(WorkflowConsoleService); + (TestBed.inject(GuiConfigService).env as any).copilotEnabled = true; + vi.spyOn(workflowActionService.getJointGraphWrapper(), "getCurrentHighlightedOperatorIDs").mockReturnValue(["3"]); + vi.spyOn(consoleService, "hasConsoleMessages").mockReturnValue(true); + vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([consoleMessage("ERROR")]); + + component.rerenderResultPanel(); + + expect(component.frameComponentConfigs.has("Console")).toBe(true); + expect(component.frameComponentConfigs.has("AI Fix")).toBe(false); + }); + + it("registers no AI fix frame when the operator's console holds no ERROR", () => { + workflowActionService.addOperator(mockResultPredicate, mockPoint); + const consoleService = TestBed.inject(WorkflowConsoleService); + const config = TestBed.inject(GuiConfigService); + (config.env as any).copilotEnabled = true; + vi.spyOn(workflowActionService.getJointGraphWrapper(), "getCurrentHighlightedOperatorIDs").mockReturnValue(["3"]); + vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([consoleMessage("PRINT")]); + + component.rerenderResultPanel(); + + expect(component.frameComponentConfigs.has("AI Fix")).toBe(false); + }); + + it("registers no AI fix frame while the copilot feature is disabled", () => { + workflowActionService.addOperator(mockResultPredicate, mockPoint); + const consoleService = TestBed.inject(WorkflowConsoleService); + const config = TestBed.inject(GuiConfigService); + (config.env as any).copilotEnabled = false; + vi.spyOn(workflowActionService.getJointGraphWrapper(), "getCurrentHighlightedOperatorIDs").mockReturnValue(["3"]); + vi.spyOn(consoleService, "getConsoleMessages").mockReturnValue([consoleMessage("ERROR")]); + + component.rerenderResultPanel(); + + expect(component.frameComponentConfigs.has("AI Fix")).toBe(false); + }); + it("displays a console frame when the selected operator has console messages", () => { workflowActionService.addOperator(mockResultPredicate, mockPoint); const consoleService = TestBed.inject(WorkflowConsoleService); diff --git a/frontend/src/app/workspace/component/result-panel/result-panel.component.ts b/frontend/src/app/workspace/component/result-panel/result-panel.component.ts index 88c29eed760..692dae2e78d 100644 --- a/frontend/src/app/workspace/component/result-panel/result-panel.component.ts +++ b/frontend/src/app/workspace/component/result-panel/result-panel.component.ts @@ -40,6 +40,8 @@ import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy"; import { isPythonUdf, isSink } from "../../service/workflow-graph/model/workflow-graph"; import { WorkflowVersionService } from "../../../dashboard/service/user/workflow-version/workflow-version.service"; import { ErrorFrameComponent } from "./error-frame/error-frame.component"; +import { AiFixFrameComponent } from "./ai-fix/ai-fix-frame.component"; +import { GuiConfigService } from "../../../common/service/gui-config.service"; import { WorkflowConsoleService } from "../../service/workflow-console/workflow-console.service"; import { NzResizeEvent, NzResizableDirective, NzResizeHandlesComponent } from "ng-zorro-antd/resizable"; import { VisualizationFrameContentComponent } from "../visualization-panel-content/visualization-frame-content.component"; @@ -118,7 +120,8 @@ export class ResultPanelComponent implements OnInit, OnDestroy { private changeDetectorRef: ChangeDetectorRef, private workflowConsoleService: WorkflowConsoleService, private resizeService: PanelResizeService, - private panelService: PanelService + private panelService: PanelService, + private config: GuiConfigService ) { this.width = 0; this.height = Number(localStorage.getItem("result-panel-height")) || this.height; @@ -133,6 +136,7 @@ export class ResultPanelComponent implements OnInit, OnDestroy { this.returnPosition = { x: -xOffset, y: -yOffset }; this.updateReturnPosition(DEFAULT_HEIGHT, this.height); this.registerAutoRerenderResultPanel(); + this.registerOperatorDisplayNameChangeHandler(); this.registerAutoOpenResultPanel(); this.registerResultClearedHandler(); this.handleResultPanelForVersionPreview(); @@ -252,8 +256,90 @@ 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(untilDestroyed(this)) + .subscribe(() => { + // Nothing is remembered between emissions here. syncAiFixFrame compares against the + // map itself, which rerenderResultPanel writes to as well: a key remembered by the + // pipe went stale the moment the rerender dropped the tab, so the next error mapped + // to the same value, read as "no change", and the tab never came back -- in the + // fix-and-re-run loop this panel is for. + if (this.syncAiFixFrame()) { + this.changeDetectorRef.detectChanges(); + } + }); + } + + /** True when the selected operator reports an error the AI panel can be opened on. */ + private shouldOfferAiFix(): boolean { + if (!this.config.env.copilotEnabled || this.currentOperatorId === undefined) { + return false; + } + if (!this.hasConsoleError(this.currentOperatorId)) { + return false; + } + // Python UDFs only, because the fix this panel writes is a change to `code`. A JVM + // operator can report an error the classifier recognises and still have nothing to + // patch, so offering the tab there spends a model call to reach a refusal. + const graph = this.workflowActionService.getTexeraGraph(); + return graph.hasOperator(this.currentOperatorId) && isPythonUdf(graph.getOperator(this.currentOperatorId)); + } + + /** + * Keys the frame tabs by their config object. + * + * `keyvalue` rebuilds its pair objects on every change to the map, so without a trackBy + * every tab is torn down and recreated whenever one is added or removed -- a result table + * losing its scroll position and paging because a console error appeared elsewhere. + * + * By the config object rather than by the tab name: `NgComponentOutlet` only recreates a + * component when its type changes, so keying by name would also keep every other frame + * alive across re-renders and operator switches, which is more than this needs to change. + * A re-render re-`set`s each entry with a fresh object, so those frames are recreated as + * before, while syncAiFixFrame touching only its own entry leaves the others' identity -- + * and so their components -- untouched. + */ + public trackFrameByConfig(_index: number, config: { value: unknown }): unknown { + return config.value; + } + + /** + * Adds or removes the AI Fix entry, leaving every other frame untouched. + * + * Returns whether it changed anything, so a caller can skip change detection. The decision + * is made against the entry that is actually there, including which operator it points at: + * this runs from the console stream and from rerenderResultPanel, and anything remembered + * outside the map goes stale as soon as the other path writes to it. + */ + private syncAiFixFrame(): boolean { + const operatorId = this.shouldOfferAiFix() ? this.currentOperatorId : undefined; + const existing = this.frameComponentConfigs.get("AI Fix"); + const shown = existing ? (existing.componentInputs as { operatorId?: string }).operatorId : undefined; + if (operatorId === shown) { + return false; + } + if (operatorId) { + this.frameComponentConfigs.set("AI Fix", { + component: AiFixFrameComponent, + componentInputs: { operatorId }, + }); + } else { + // Mirrors the Static Error frame's teardown: once the operator no longer reports an + // error -- a re-run cleared the console, or the selection moved to a healthy operator + // -- a tab offering to fix it has nothing to act on. + this.frameComponentConfigs.delete("AI Fix"); + } + return true; } rerenderResultPanel(): void { @@ -300,9 +386,20 @@ export class ResultPanelComponent implements OnInit, OnDestroy { if (this.workflowConsoleService.hasConsoleMessages(this.currentOperatorId) || isPythonUdf(operator)) { this.displayConsole(this.currentOperatorId, isPythonUdf(operator)); } + // A Python UDF exception never becomes a fatal error: the worker pauses and the + // traceback arrives as an ERROR console message while the execution stays Running, + // so the Failed branch above never fires for the case this panel exists to fix. + this.syncAiFixFrame(); } } + /** True when the operator reported an ERROR console message (e.g. a Python UDF traceback). */ + private hasConsoleError(operatorId: string): boolean { + return (this.workflowConsoleService.getConsoleMessages(operatorId) ?? []).some( + message => message.msgType.name === "ERROR" + ); + } + clearResultPanel(): void { this.frameComponentConfigs.clear(); } @@ -351,22 +448,31 @@ export class ResultPanelComponent implements OnInit, OnDestroy { return errorMessages; } - private registerOperatorDisplayNameChangeHandler(): void { + /** Refreshes the title for whichever operator the panel now shows. */ + private updateOperatorTitle(): void { if (this.currentOperatorId) { - const operator = this.workflowActionService.getTexeraGraph().getOperator(this.currentOperatorId); - this.operatorTitle = operator.customDisplayName ?? ""; - this.workflowActionService - .getTexeraGraph() - .getOperatorDisplayNameChangedStream() - .pipe(untilDestroyed(this)) - .subscribe(({ operatorID, newDisplayName }) => { - if (operatorID === this.currentOperatorId) { - this.operatorTitle = newDisplayName; - } - }); + this.operatorTitle = + this.workflowActionService.getTexeraGraph().getOperator(this.currentOperatorId).customDisplayName ?? ""; } } + /** + * Registered once from ngOnInit, not from the rerender subscriber: this takes out a + * subscription without releasing a previous one, so calling it per rerender grows the + * handler count for as long as the panel lives. + */ + private registerOperatorDisplayNameChangeHandler(): void { + this.workflowActionService + .getTexeraGraph() + .getOperatorDisplayNameChangedStream() + .pipe(untilDestroyed(this)) + .subscribe(({ operatorID, newDisplayName }) => { + if (operatorID === this.currentOperatorId) { + this.operatorTitle = newDisplayName; + } + }); + } + private static needRerenderOnStateChange(event: { previous: ExecutionStateInfo; current: ExecutionStateInfo; diff --git a/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.spec.ts b/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.spec.ts index 650a2d32a61..6e20b0c880a 100644 --- a/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.spec.ts +++ b/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.spec.ts @@ -484,7 +484,7 @@ describe("ExecuteWorkflowService", () => { // ---- execution entry points ---------------------------------------------------------------- it("executeWorkflow delegates to executeWorkflowWithEmailNotification with email disabled", () => { - const delegateSpy = vi.spyOn(service, "executeWorkflowWithEmailNotification").mockImplementation(() => {}); + const delegateSpy = vi.spyOn(service, "executeWorkflowWithEmailNotification").mockImplementation(() => true); service.executeWorkflow("run-1", "op-9"); expect(delegateSpy).toHaveBeenCalledWith("run-1", false, "op-9"); }); diff --git a/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.ts b/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.ts index c513ebe7c60..21519374f9c 100644 --- a/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.ts +++ b/frontend/src/app/workspace/service/execute-workflow/execute-workflow.service.ts @@ -258,29 +258,38 @@ export class ExecuteWorkflowService { return []; } + /** + * Returns whether the request was sent, so a caller that reports on it does not have to + * repeat the refusal conditions. False means this method turned the run down locally -- + * no computing unit, no warehouse -- and showed its own toast. True only means the request + * left the browser: the deployment can still reject it, which arrives on the execution + * state stream. Callers that only fire and forget can keep ignoring the result. + */ public executeWorkflowWithEmailNotification( executionName: string, emailNotificationEnabled: boolean, targetOperatorId?: string - ): void { + ): boolean { const logicalPlan = ExecuteWorkflowService.getLogicalPlanRequest( this.workflowActionService.getTexeraGraph(), targetOperatorId ); const settings = this.workflowActionService.getWorkflowSettings(); if (this.refuseToRunOnUnavailableUnit()) { - return; + return false; } if (this.refuseToRunWithoutWarehouse()) { - return; + return false; } this.resetExecutionState(); this.workflowStatusService.resetStatus(); this.sendExecutionRequest(executionName, logicalPlan, settings, emailNotificationEnabled); + return true; } - public executeWorkflow(executionName: string, targetOperatorId?: string): void { - this.executeWorkflowWithEmailNotification(executionName, false, targetOperatorId); + /** Returns whether the request was sent; see executeWorkflowWithEmailNotification. */ + public executeWorkflow(executionName: string, targetOperatorId?: string): boolean { + return this.executeWorkflowWithEmailNotification(executionName, false, targetOperatorId); } public executeWorkflowWithReplay(replayExecutionInfo: ReplayExecutionInfo): void { diff --git a/frontend/src/app/workspace/service/operator-menu/operator-menu.service.spec.ts b/frontend/src/app/workspace/service/operator-menu/operator-menu.service.spec.ts index 0f1282a4a5f..987ddf47e69 100644 --- a/frontend/src/app/workspace/service/operator-menu/operator-menu.service.spec.ts +++ b/frontend/src/app/workspace/service/operator-menu/operator-menu.service.spec.ts @@ -252,7 +252,7 @@ describe("OperatorMenuService", () => { const notificationService = TestBed.inject(NotificationService); const executeWorkflowService = TestBed.inject(ExecuteWorkflowService); const errorSpy = vi.spyOn(notificationService, "error").mockImplementation(() => {}); - const executeSpy = vi.spyOn(executeWorkflowService, "executeWorkflow").mockImplementation(() => undefined); + const executeSpy = vi.spyOn(executeWorkflowService, "executeWorkflow").mockImplementation(() => true); service.executeUpToOperator(); @@ -264,7 +264,7 @@ describe("OperatorMenuService", () => { const notificationService = TestBed.inject(NotificationService); const executeWorkflowService = TestBed.inject(ExecuteWorkflowService); const errorSpy = vi.spyOn(notificationService, "error").mockImplementation(() => {}); - const executeSpy = vi.spyOn(executeWorkflowService, "executeWorkflow").mockImplementation(() => undefined); + const executeSpy = vi.spyOn(executeWorkflowService, "executeWorkflow").mockImplementation(() => true); workflowActionService.addOperatorsAndLinks( [ { op: mockScanPredicate, pos: mockPoint }, @@ -285,7 +285,7 @@ describe("OperatorMenuService", () => { it("executes the workflow up to the single highlighted operator", () => { const executeWorkflowService = TestBed.inject(ExecuteWorkflowService); - const executeSpy = vi.spyOn(executeWorkflowService, "executeWorkflow").mockImplementation(() => undefined); + const executeSpy = vi.spyOn(executeWorkflowService, "executeWorkflow").mockImplementation(() => true); workflowActionService.addOperator(mockScanPredicate, mockPoint); workflowActionService.getJointGraphWrapper().highlightOperators(mockScanPredicate.operatorID);