inout change vars regime

models/workflow/workflow.go

@@ -959,7 +959,8 @@ const (
 	ViolationVariableNotFound   ViolationType = "variable_not_found"
 	ViolationMissingComputeUnit ViolationType = "missing_compute_unit"
 	ViolationCycle              ViolationType = "cycle"
-	ViolationMissingDataStorage ViolationType = "missing_data_storage"
+	ViolationMissingDataStorage   ViolationType = "missing_data_storage"
+	ViolationRequiredOutputMissing ViolationType = "required_output_missing"
 
 	// Warnings — non-blocking, reported for UX
 	ViolationInvertedArrow                ViolationType = "inverted_arrow"
@@ -997,6 +998,7 @@ func (v IntegrityViolation) IsWarning() bool { return v.Severity == SeverityWarn
 func (w *Workflow) ValidateIntegrity() []IntegrityViolation {
 	var violations []IntegrityViolation
 	violations = append(violations, w.validateVariables()...)
+	violations = append(violations, w.validateRequiredInputs()...)
 	violations = append(violations, w.validateComputeLinks()...)
 	violations = append(violations, w.detectCycles()...)
 	violations = append(violations, w.validateDataStorageLinks()...)
@@ -1234,6 +1236,97 @@ func (w *Workflow) validateDataStorageLinks() []IntegrityViolation {
 	return violations
 }
 
+// validateRequiredInputs checks that for each processing node with a required
+// input, every immediate predecessor outputs a parameter with that name.
+// Mirrors the requiredOutputMissing check in oc-front's checkTopology().
+func (w *Workflow) validateRequiredInputs() []IntegrityViolation {
+	var violations []IntegrityViolation
+
+	procIDs := map[string]struct{}{}
+	for id, item := range w.Graph.Items {
+		if w.Graph.IsProcessing(item) || w.Graph.IsService(item) || w.Graph.IsNativeTool(item) {
+			procIDs[id] = struct{}{}
+		}
+	}
+
+	// Build direct predecessors map.
+	predecessors := map[string][]string{}
+	for id := range procIDs {
+		predecessors[id] = []string{}
+	}
+	for _, link := range w.Graph.Links {
+		src, dst := link.Source.ID, link.Destination.ID
+		_, srcIsProc := procIDs[src]
+		_, dstIsProc := procIDs[dst]
+		if !srcIsProc || !dstIsProc {
+			continue
+		}
+		dir := int64(0)
+		if link.Style != nil {
+			dir = link.Style.ArrowDirection
+		}
+		if dir == arrowDirectionBackward {
+			predecessors[src] = append(predecessors[src], dst)
+		} else {
+			predecessors[dst] = append(predecessors[dst], src)
+		}
+	}
+
+	for id, reqInputs := range w.Inputs {
+		if _, isProc := procIDs[id]; !isProc {
+			continue
+		}
+		for _, inp := range reqInputs {
+			if !inp.Required || inp.Name == "" {
+				continue
+			}
+			for _, predID := range predecessors[id] {
+				if !w.nodeHasOutput(predID, inp.Name) {
+					violations = append(violations, IntegrityViolation{
+						Severity: SeverityError,
+						Type:     ViolationRequiredOutputMissing,
+						ItemIDs:  []string{id, predID},
+						Message: fmt.Sprintf(
+							`"%s" requires input "%s" but "%s" does not output it`,
+							w.itemName(id), inp.Name, w.itemName(predID),
+						),
+					})
+				}
+			}
+		}
+	}
+	return violations
+}
+
+// nodeHasOutput returns true if the given node outputs a parameter named name,
+// either via workflow-level outputs or its resource's own outputs.
+func (w *Workflow) nodeHasOutput(nodeID, name string) bool {
+	for _, p := range w.Outputs[nodeID] {
+		if p.Name == name {
+			return true
+		}
+	}
+	item, ok := w.Graph.Items[nodeID]
+	if !ok {
+		return false
+	}
+	var res resources.ResourceInterface
+	switch {
+	case item.Processing != nil:
+		res = item.Processing
+	case item.Service != nil:
+		res = item.Service
+	}
+	if res != nil {
+		for _, p := range res.GetOutputs() {
+			if p.Name == name {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // detectInvertedArrows warns when a link between two processing nodes uses a
 // backward arrow direction — mirroring the invertedArrow warning in oc-front.
 func (w *Workflow) detectInvertedArrows() []IntegrityViolation {