Yicong-Huang commented on code in PR #6437:
URL: https://github.com/apache/texera/pull/6437#discussion_r3786771925
##########
frontend/src/app/workspace/service/drag-drop/drag-drop.service.ts:
##########
@@ -75,10 +75,16 @@ export class DragDropService {
this.workflowActionService.addOperatorsAndLinks([{ op: this.op, pos:
coordinates }], newLinks);
this.resetSuggestions();
- this.operatorDroppedSubject.next();
+ this.operatorDroppedSubject.next(this.op);
}
- get operatorDropStream() {
+ /**
+ * Emits the operator a user just dropped onto the canvas, after it has been
+ * added to the graph. Unlike the graph's own operator-add stream, this fires
+ * only for interactive drag-drop placement — not for workflow load,
undo/redo,
+ * paste, or a remote co-editor's edits.
+ */
+ get operatorDropStream(): Observable<OperatorPredicate> {
Review Comment:
The payload is the point of this change, and nothing asserts it.
`drag-drop.service.spec.ts:397` does `subscribe(() => (dropped = true))`. A
`dragDropped` that emitted the wrong operator — or a stale `this.op` — would
still pass, and the recommender specs stub this producer entirely.
One line in that existing test closes it: capture the emission and assert it
is the operator that was dropped. Optional, and my miss.
##########
frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts:
##########
@@ -1717,6 +1764,205 @@ export class WorkflowEditorComponent implements OnInit,
AfterViewInit, OnDestroy
return
this.operatorSummaries.get(operatorId)?.sampleRecords?.[0]?.["__is_visualization__"]
=== true;
}
+ /**
+ * Ambient operator recommender (apache/texera#5240). When the user drops an
+ * operator onto the canvas, ask the recommender for likely next operators
and
+ * float them as suggestion chips on the operator's output port; clicking one
+ * materializes it. The whole feature is opt-in and self-effacing: if it is
+ * disabled or the backend returns nothing, the canvas is untouched.
+ */
+ private handleNextOperatorSuggestions(): void {
+ if (!this.operatorRecommendationService.isEnabled()) {
+ return;
+ }
+
+ // Repositioning is throttled: change:position fires once per drag frame.
+ this.repositionNextOperatorSuggestion$
+ .pipe(auditTime(100), untilDestroyed(this))
+ .subscribe(() => this.repositionNextOperatorSuggestions());
+
+ // Every suggestion request — from a drop or from chaining after a click —
+ // goes through this one pipeline. switchMap unsubscribes the previous
+ // request, so a slow response can neither overwrite newer suggestions nor
+ // re-open the overlay after the user dismissed it; `null` means "cancel".
+ this.nextOperatorSuggestionRequest$
+ .pipe(
+ switchMap(operator =>
+ operator === null
+ ? of(null)
+ : this.operatorRecommendationService
+ .getRecommendations(operator)
+ .pipe(map(recommendations => ({ operator, recommendations })))
+ ),
+ untilDestroyed(this)
+ )
+ .subscribe(result => this.showNextOperatorSuggestions(result));
+
+ // Trigger: the user interactively dropped an operator onto the canvas.
+ // Deliberately not the graph's operator-add stream, which also fires on
+ // workflow load, undo/redo, paste, and remote co-editor edits — none of
+ // which are a user authoring a next step.
+ this.dragDropService.operatorDropStream
+ .pipe(untilDestroyed(this))
+ .subscribe(operator => this.requestNextOperatorSuggestionsFor(operator));
+
+ // Dismiss when the user clicks on blank canvas.
+ fromJointPaperEvent(this.paper, "blank:pointerdown")
+ .pipe(untilDestroyed(this))
+ .subscribe(() => this.closeNextOperatorSuggestions());
+
+ // Dismiss if the anchor operator is deleted out from under the
suggestions.
+ this.workflowActionService
+ .getTexeraGraph()
+ .getOperatorDeleteStream()
+ .pipe(untilDestroyed(this))
+ .subscribe(({ deletedOperatorID }) => {
+ if (this.nextOperatorSuggestion?.operatorId === deletedOperatorID) {
+ this.closeNextOperatorSuggestions();
+ }
+ });
+
+ // Keep the suggestions anchored to the operator's output port as it moves.
+ this.paper.model.on("change:position", (cell: joint.dia.Cell) => {
Review Comment:
This one listener isn't torn down. `attachMainJointPaper` points the paper
at `JointGraphWrapper`'s own graph (`joint-graph-wrapper.ts:201`), which is
root-provided. So the closure outlives the component, while the six
subscriptions around it all use `untilDestroyed(this)`.
It matches lines 496 and 1673, so this is a consistency note rather than
something this PR broke. One shared helper would fix all three. Also my miss.
##########
frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts:
##########
@@ -1717,6 +1764,205 @@ export class WorkflowEditorComponent implements OnInit,
AfterViewInit, OnDestroy
return
this.operatorSummaries.get(operatorId)?.sampleRecords?.[0]?.["__is_visualization__"]
=== true;
}
+ /**
+ * Ambient operator recommender (apache/texera#5240). When the user drops an
+ * operator onto the canvas, ask the recommender for likely next operators
and
+ * float them as suggestion chips on the operator's output port; clicking one
+ * materializes it. The whole feature is opt-in and self-effacing: if it is
+ * disabled or the backend returns nothing, the canvas is untouched.
+ */
+ private handleNextOperatorSuggestions(): void {
+ if (!this.operatorRecommendationService.isEnabled()) {
+ return;
+ }
+
+ // Repositioning is throttled: change:position fires once per drag frame.
+ this.repositionNextOperatorSuggestion$
+ .pipe(auditTime(100), untilDestroyed(this))
+ .subscribe(() => this.repositionNextOperatorSuggestions());
+
+ // Every suggestion request — from a drop or from chaining after a click —
+ // goes through this one pipeline. switchMap unsubscribes the previous
+ // request, so a slow response can neither overwrite newer suggestions nor
+ // re-open the overlay after the user dismissed it; `null` means "cancel".
+ this.nextOperatorSuggestionRequest$
+ .pipe(
+ switchMap(operator =>
+ operator === null
+ ? of(null)
+ : this.operatorRecommendationService
+ .getRecommendations(operator)
+ .pipe(map(recommendations => ({ operator, recommendations })))
+ ),
+ untilDestroyed(this)
+ )
+ .subscribe(result => this.showNextOperatorSuggestions(result));
+
+ // Trigger: the user interactively dropped an operator onto the canvas.
+ // Deliberately not the graph's operator-add stream, which also fires on
+ // workflow load, undo/redo, paste, and remote co-editor edits — none of
+ // which are a user authoring a next step.
+ this.dragDropService.operatorDropStream
+ .pipe(untilDestroyed(this))
+ .subscribe(operator => this.requestNextOperatorSuggestionsFor(operator));
+
+ // Dismiss when the user clicks on blank canvas.
+ fromJointPaperEvent(this.paper, "blank:pointerdown")
+ .pipe(untilDestroyed(this))
+ .subscribe(() => this.closeNextOperatorSuggestions());
+
+ // Dismiss if the anchor operator is deleted out from under the
suggestions.
+ this.workflowActionService
+ .getTexeraGraph()
+ .getOperatorDeleteStream()
+ .pipe(untilDestroyed(this))
+ .subscribe(({ deletedOperatorID }) => {
+ if (this.nextOperatorSuggestion?.operatorId === deletedOperatorID) {
+ this.closeNextOperatorSuggestions();
+ }
+ });
+
+ // Keep the suggestions anchored to the operator's output port as it moves.
+ this.paper.model.on("change:position", (cell: joint.dia.Cell) => {
+ if (this.nextOperatorSuggestion && cell.id.toString() ===
this.nextOperatorSuggestion.operatorId) {
+ this.repositionNextOperatorSuggestion$.next();
+ }
+ });
+
+ // Keep the suggestions anchored on zoom.
+ this.wrapper
+ .getWorkflowEditorZoomStream()
+ .pipe(untilDestroyed(this))
+ .subscribe(() => {
+ if (this.nextOperatorSuggestion) {
+ this.repositionNextOperatorSuggestions();
+ }
+ });
+ }
+
+ /** Ask for suggestions on `operator`, cancelling whatever was in flight. */
+ private requestNextOperatorSuggestionsFor(operator: OperatorPredicate): void
{
+ this.closeNextOperatorSuggestions();
+ // An operator with no output ports (e.g. a chart sink) has no port to
+ // anchor suggestions on, so there is nothing to ask for.
+ if (operator.outputPorts.length === 0) {
+ return;
+ }
+ // A drop can arrive already wired: onto an existing edge, or auto-linked
to
+ // a nearby operator. The next step is chosen in that case, so suggesting
+ // another one would both be noise and risk placing it on top of the
+ // successor the drop just created.
+ if (this.isOutputPortLinked(operator.operatorID,
operator.outputPorts[0].portID)) {
+ return;
+ }
+ this.nextOperatorSuggestionRequest$.next(operator);
+ }
+
+ /** Whether `portID` on `operatorID` already has a link leaving it. */
+ private isOutputPortLinked(operatorID: string, portID: string): boolean {
+ return this.workflowActionService
+ .getTexeraGraph()
+ .getAllLinks()
+ .some(link => link.source.operatorID === operatorID &&
link.source.portID === portID);
+ }
+
+ /** Render the result of the most recent, uncancelled suggestion request. */
+ private showNextOperatorSuggestions(
+ result: { operator: OperatorPredicate; recommendations:
OperatorRecommendation[] } | null
+ ): void {
+ if (result === null || result.recommendations.length === 0) {
+ return;
+ }
+ const { operator, recommendations } = result;
+ // The operator may have been deleted while the request was in flight.
+ if
(!this.workflowActionService.getTexeraGraph().hasOperator(operator.operatorID))
{
+ return;
+ }
+ const position =
this.getNextOperatorSuggestionPosition(operator.operatorID);
+ if (!position) {
+ return;
+ }
+ this.nextOperatorSuggestion = {
+ operatorId: operator.operatorID,
+ sourceOutputPortID: operator.outputPorts[0].portID,
+ position,
+ recommendations,
+ };
+ this.changeDetectorRef.detectChanges();
+ }
+
+ /**
+ * Materialize a clicked suggestion into a real operator wired onto the
+ * source operator's output port, then suggest what could follow that new
+ * operator in turn — accepting a suggestion leaves the canvas ready for the
+ * next one, the way accepting a code completion does.
+ */
+ materializeNextOperatorSuggestion(recommendation: OperatorRecommendation):
void {
+ if (!this.nextOperatorSuggestion) {
+ return;
+ }
+ const graph = this.workflowActionService.getTexeraGraph();
+ let newOperatorID: string | undefined;
+ // Re-check the port: nothing dismisses the chips when the user hand-draws
a
+ // link out of the anchor port, because that gesture starts on the port and
+ // fires element:pointerdown rather than the blank:pointerdown we listen
for.
+ // Without this the click would add a second successor on top of the first.
+ if (
+ graph.hasOperator(this.nextOperatorSuggestion.operatorId) &&
Review Comment:
This guard covers the anchor operator and its port, but not the
workflow-modification lock. The chips are the only authoring affordance in the
workspace that skips it.
Drop an operator, click Run, then click a chip.
`execute-workflow.service.ts:400` has already called
`disableWorkflowModification()`. Nothing here dismisses the chips — only
`blank:pointerdown`, an anchor delete, or a new request do. And
`addOperatorsAndLinks` checks nothing itself, so the operator and link land in
the shared model mid-execution. `undoRedoService.undoAction()` then refuses
while the lock is on, so the user can't take it back. Version preview
(`workflow-version.service.ts:109`) and read-only hub views reach it the same
way.
I'd subscribe to `getWorkflowModificationEnabledStream()` here and close the
suggestions when it emits false. That also stops the chips looking clickable
while the canvas is frozen, which is what the paper does at 307-321. Adding the
check to this condition is smaller but leaves dead chips on screen.
My miss from earlier rounds.
##########
frontend/src/app/workspace/service/operator-recommendation/operator-recommendation.service.ts:
##########
@@ -0,0 +1,166 @@
+/**
+ * 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 { HttpClient } from "@angular/common/http";
+import { Observable, catchError, map, of } from "rxjs";
+import { OperatorLink, OperatorPredicate, Point } from
"../../types/workflow-common.interface";
+import { WorkflowActionService } from
"../workflow-graph/model/workflow-action.service";
+import { WorkflowUtilService } from
"../workflow-graph/util/workflow-util.service";
+import { JointUIService } from "../joint-ui/joint-ui.service";
+import { GuiConfigService } from "../../../common/service/gui-config.service";
+
+/**
+ * A single operator suggestion returned by the agent-service `/api/recommend`
+ * endpoint (apache/texera#5240). Mirrors the backend `OperatorRecommendation`.
+ */
+export interface OperatorRecommendation {
+ /** Recommended operator type (validated against the live catalog when
available). */
+ operatorType: string;
+ /** Confidence in `[0, 1]`, monotonically non-increasing down the list. */
+ score: number;
+ /** Short, human-readable rationale shown alongside the suggested operator.
*/
+ reason: string;
+ /** Display name from operator metadata, when available. */
+ userFriendlyName?: string;
+}
+
+interface RecommendationResponse {
+ recommendations: OperatorRecommendation[];
+ strategy: "hardcoded" | "llm";
+}
+
+/**
+ * Client for the ambient operator recommender. Asks the stateless
agent-service
+ * endpoint what operators are likely to follow the one just added, and turns a
+ * chosen suggestion into a real operator wired onto the source's output port.
+ *
+ * The service never fails loudly: the recommender is a non-essential, ambient
+ * aid, so a backend error or a disabled feature simply yields no suggestions
+ * and the canvas behaves exactly as before.
+ */
+@Injectable({
+ providedIn: "root",
+})
+export class OperatorRecommendationService {
+ private static readonly RECOMMEND_API_URL = "/api/recommend";
+
+ // Horizontal gap between the source operator and a materialized suggestion.
+ private static readonly MATERIALIZE_GAP_X = 100;
+
+ constructor(
+ private http: HttpClient,
+ private config: GuiConfigService,
+ private workflowActionService: WorkflowActionService,
+ private workflowUtilService: WorkflowUtilService
+ ) {}
+
+ /** Whether the opt-in recommender feature is turned on for this deployment.
*/
+ public isEnabled(): boolean {
+ return this.config.env.operatorRecommendationEnabled === true;
+ }
+
+ /**
+ * Fetch ranked next-operator suggestions for the operator just added.
+ *
+ * Returns an empty list (never errors) when the feature is disabled, the
+ * operator has no output port to suggest from, or the backend call fails.
+ *
+ * How many suggestions come back is the backend's call: it defaults to three
+ * and clamps anything larger, so there is nothing useful to send from here.
Review Comment:
"anything larger" has no antecedent here — the method sends no count, so a
reader can't tell what is being clamped.
```suggestion
* How many suggestions come back is the backend's call: it defaults to
three and
* clamps any larger requested limit, so there is nothing useful to send
from here.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
