wfctl validate: add field-level cross-reference checking for step outputs, SQL aliases, and plain-string step refs

cmd/wfctl/infra_state.go

@@ -181,11 +181,11 @@ func runInfraStateImport(args []string) error {
 // --- tfstate export ---
 
 type tfState struct {
-	Version          int              `json:"version"`
-	TerraformVersion string           `json:"terraform_version"`
-	Serial           int              `json:"serial"`
-	Lineage          string           `json:"lineage"`
-	Outputs          map[string]any   `json:"outputs"`
+	Version          int               `json:"version"`
+	TerraformVersion string            `json:"terraform_version"`
+	Serial           int               `json:"serial"`
+	Lineage          string            `json:"lineage"`
+	Outputs          map[string]any    `json:"outputs"`
 	Resources        []tfStateResource `json:"resources"`
 }
 
@@ -304,11 +304,11 @@ func importFromPulumi(srcFile, stateDir string) error {
 	var checkpoint struct {
 		Latest struct {
 			Resources []struct {
-				URN      string         `json:"urn"`
-				Type     string         `json:"type"`
-				ID       string         `json:"id"`
-				Inputs   map[string]any `json:"inputs"`
-				Outputs  map[string]any `json:"outputs"`
+				URN     string         `json:"urn"`
+				Type    string         `json:"type"`
+				ID      string         `json:"id"`
+				Inputs  map[string]any `json:"inputs"`
+				Outputs map[string]any `json:"outputs"`
 			} `json:"resources"`
 		} `json:"latest"`
 	}

cmd/wfctl/main.go

@@ -53,30 +53,30 @@ func isHelpRequested(err error) bool {
 // the runtime functions that are registered in the CLICommandRegistry service
 // and invoked by step.cli_invoke from within each command's pipeline.
 var commands = map[string]func([]string) error{
-	"init":      runInit,
-	"validate":  runValidate,
-	"inspect":   runInspect,
-	"run":       runRun,
-	"plugin":    runPlugin,
-	"pipeline":  runPipeline,
-	"schema":    runSchema,
-	"snippets":  runSnippets,
-	"manifest":  runManifest,
-	"migrate":   runMigrate,
-	"build-ui":  runBuildUI,
-	"ui":        runUI,
-	"publish":   runPublish,
-	"deploy":    runDeploy,
-	"api":       runAPI,
-	"diff":      runDiff,
-	"template":  runTemplate,
-	"contract":  runContract,
-	"compat":    runCompat,
-	"generate":  runGenerate,
-	"git":       runGit,
-	"registry":  runRegistry,
-	"update":    runUpdate,
-	"mcp":       runMCP,
+	"init":           runInit,
+	"validate":       runValidate,
+	"inspect":        runInspect,
+	"run":            runRun,
+	"plugin":         runPlugin,
+	"pipeline":       runPipeline,
+	"schema":         runSchema,
+	"snippets":       runSnippets,
+	"manifest":       runManifest,
+	"migrate":        runMigrate,
+	"build-ui":       runBuildUI,
+	"ui":             runUI,
+	"publish":        runPublish,
+	"deploy":         runDeploy,
+	"api":            runAPI,
+	"diff":           runDiff,
+	"template":       runTemplate,
+	"contract":       runContract,
+	"compat":         runCompat,
+	"generate":       runGenerate,
+	"git":            runGit,
+	"registry":       runRegistry,
+	"update":         runUpdate,
+	"mcp":            runMCP,
 	"modernize":      runModernize,
 	"infra":          runInfra,
 	"docs":           runDocs,

cmd/wfctl/plugin_install_new_test.go

@@ -24,8 +24,8 @@ func buildPluginTarGz(t *testing.T, pluginName string, binaryContent []byte, pjC
 	t.Helper()
 	topDir := pluginName + "-" + runtime.GOOS + "-" + runtime.GOARCH
 	entries := map[string][]byte{
-		topDir + "/" + pluginName:    binaryContent,
-		topDir + "/plugin.json":      pjContent,
+		topDir + "/" + pluginName: binaryContent,
+		topDir + "/plugin.json":   pjContent,
 	}
 	return buildTarGz(t, entries, 0755)
 }
@@ -166,8 +166,8 @@ func TestInstallFromURL_NameNormalization(t *testing.T) {
 
 	pjContent := minimalPluginJSON(fullName, "0.1.0")
 	entries := map[string][]byte{
-		"top/" + fullName:    []byte("#!/bin/sh\n"),
-		"top/plugin.json":    pjContent,
+		"top/" + fullName: []byte("#!/bin/sh\n"),
+		"top/plugin.json": pjContent,
 	}
 	tarball := buildTarGz(t, entries, 0755)
 

cmd/wfctl/template_validate.go

@@ -565,8 +565,10 @@ var _ fs.FS = templateFS
 // templateExprRe matches template actions {{ ... }}.
 var templateExprRe = regexp.MustCompile(`\{\{(.*?)\}\}`)
 
-// stepRefDotRe matches .steps.STEP_NAME patterns (dot access).
-var stepRefDotRe = regexp.MustCompile(`\.steps\.([a-zA-Z_][a-zA-Z0-9_-]*)`)
+// stepRefDotRe matches .steps.STEP_NAME and captures an optional field path.
+// Group 1: step name (may contain hyphens).
+// Group 2: remaining dot-path (e.g. ".row.auth_token"), field names without hyphens.
+var stepRefDotRe = regexp.MustCompile(`\.steps\.([a-zA-Z_][a-zA-Z0-9_-]*)((?:\.[a-zA-Z_][a-zA-Z0-9_]*)*)`)
 
 // stepRefIndexRe matches index .steps "STEP_NAME" patterns.
 var stepRefIndexRe = regexp.MustCompile(`index\s+\.steps\s+"([^"]+)"`)
@@ -579,11 +581,66 @@ var stepRefFuncRe = regexp.MustCompile(`(?:^|\||\()\s*step\s+"([^"]+)"`)
 // including continuation segments after the hyphenated part.
 var hyphenDotRe = regexp.MustCompile(`\.[a-zA-Z_][a-zA-Z0-9_]*-[a-zA-Z0-9_-]*(?:\.[a-zA-Z_][a-zA-Z0-9_-]*)*`)
 
+// plainStepPathRe matches bare step context-key references such as
+// "steps.STEP_NAME.field.subfield" used in plain-string config values (no {{ }}).
+var plainStepPathRe = regexp.MustCompile(`^steps\.([a-zA-Z_][a-zA-Z0-9_-]*)((?:\.[a-zA-Z_][a-zA-Z0-9_]*)*)`)
+
+// stepBuildInfo holds the type and config of a pipeline step, used for output field validation.
+type stepBuildInfo struct {
+	stepType   string
+	stepConfig map[string]any
+}
+
+// dbQueryStepTypes is the set of step types that produce a "row" or "rows" output
+// from a SQL query and support SQL alias extraction.
+var dbQueryStepTypes = map[string]bool{
+	"step.db_query":        true,
+	"step.db_query_cached": true,
+}
+
+// isDBQueryStep reports whether a step type is a DB query step.
+func isDBQueryStep(t string) bool { return dbQueryStepTypes[t] }
+
+// joinOutputKeys returns a comma-joined list of output key names for error messages,
+// omitting placeholder/wildcard entries like "(key)", "(dynamic)", "(nested)".
+func joinOutputKeys(outputs []schema.InferredOutput) string {
+	keys := make([]string, 0, len(outputs))
+	for _, o := range outputs {
+		if !isPlaceholderOutputKey(o.Key) {
+			keys = append(keys, o.Key)
+		}
+	}
+	return strings.Join(keys, ", ")
+}
+
+// isPlaceholderOutputKey reports whether an output key is a dynamic/wildcard
+// placeholder (e.g. "(key)", "(dynamic)", "(nested)").  Steps that expose
+// such placeholders produce outputs whose field names cannot be statically
+// determined, so field-path validation should be skipped for them.
+func isPlaceholderOutputKey(key string) bool {
+	return len(key) >= 2 && key[0] == '(' && key[len(key)-1] == ')'
+}
+
+// hasDynamicOutputs reports whether any output in the list is a wildcard
+// placeholder, meaning the step emits fields that are not statically known.
+func hasDynamicOutputs(outputs []schema.InferredOutput) bool {
+	for _, o := range outputs {
+		if isPlaceholderOutputKey(o.Key) {
+			return true
+		}
+	}
+	return false
+}
+
 // validatePipelineTemplates checks template expressions in pipeline step configs for
 // references to nonexistent or forward-declared steps and common template pitfalls.
 func validatePipelineTemplates(pipelineName string, stepsRaw []any, result *templateValidationResult) {
-	// Build ordered step name list
-	stepNames := make(map[string]int) // step name -> index in pipeline
+	// Build ordered step name list and per-step type/config info.
+	stepNames := make(map[string]int)           // step name -> index in pipeline
+	stepInfos := make(map[string]stepBuildInfo) // step name -> type and config
+
+	reg := schema.NewStepSchemaRegistry()
+
 	for i, stepRaw := range stepsRaw {
 		stepMap, ok := stepRaw.(map[string]any)
 		if !ok {
@@ -592,6 +649,12 @@ func validatePipelineTemplates(pipelineName string, stepsRaw []any, result *temp
 		name, _ := stepMap["name"].(string)
 		if name != "" {
 			stepNames[name] = i
+			sType, _ := stepMap["type"].(string)
+			sCfg, _ := stepMap["config"].(map[string]any)
+			if sCfg == nil {
+				sCfg = map[string]any{}
+			}
+			stepInfos[name] = stepBuildInfo{stepType: sType, stepConfig: sCfg}
 		}
 	}
 
@@ -624,25 +687,29 @@ func validatePipelineTemplates(pipelineName string, stepsRaw []any, result *temp
 					continue
 				}
 
-				// Check for step name references via dot-access
+				// Check for step name references via dot-access (captures optional field path)
 				dotMatches := stepRefDotRe.FindAllStringSubmatch(actionContent, -1)
 				for _, m := range dotMatches {
 					refName := m[1]
-					validateStepRef(pipelineName, stepName, refName, i, stepNames, result)
+					fieldPath := ""
+					if len(m) > 2 {
+						fieldPath = m[2]
+					}
+					validateStepRef(pipelineName, stepName, refName, fieldPath, i, stepNames, stepInfos, reg, result)
 				}
 
-				// Check for step name references via index
+				// Check for step name references via index (no field path resolvable)
 				indexMatches := stepRefIndexRe.FindAllStringSubmatch(actionContent, -1)
 				for _, m := range indexMatches {
 					refName := m[1]
-					validateStepRef(pipelineName, stepName, refName, i, stepNames, result)
+					validateStepRef(pipelineName, stepName, refName, "", i, stepNames, stepInfos, reg, result)
 				}
 
-				// Check for step name references via step function
+				// Check for step name references via step function (no field path resolvable)
 				funcMatches := stepRefFuncRe.FindAllStringSubmatch(actionContent, -1)
 				for _, m := range funcMatches {
 					refName := m[1]
-					validateStepRef(pipelineName, stepName, refName, i, stepNames, result)
+					validateStepRef(pipelineName, stepName, refName, "", i, stepNames, stepInfos, reg, result)
 				}
 
 				// Warn on hyphenated dot-access (auto-fixed but suggest preferred syntax)
@@ -652,23 +719,122 @@ func validatePipelineTemplates(pipelineName string, stepsRaw []any, result *temp
 				}
 			}
 		}
+
+		// Validate plain-string step references in specific config fields
+		// (e.g. secret_from, backend_url_key, field in conditional/branch).
+		if stepCfg, ok := stepMap["config"].(map[string]any); ok {
+			validatePlainStepRefs(pipelineName, stepName, i, stepCfg, stepNames, stepInfos, reg, result)
+		}
 	}
 }
 
 // validateStepRef checks that a referenced step name exists and appears before the
-// current step in the pipeline execution order.
-func validateStepRef(pipelineName, currentStep, refName string, currentIdx int, stepNames map[string]int, result *templateValidationResult) {
+// current step in the pipeline execution order.  When fieldPath is non-empty it
+// also validates the first output field name against the step's known outputs, and
+// for db_query steps it performs best-effort SQL alias checking for "row.<col>" paths.
+func validateStepRef(pipelineName, currentStep, refName, fieldPath string, currentIdx int, stepNames map[string]int, stepInfos map[string]stepBuildInfo, reg *schema.StepSchemaRegistry, result *templateValidationResult) {
 	refIdx, exists := stepNames[refName]
 	switch {
 	case !exists:
 		result.Warnings = append(result.Warnings,
 			fmt.Sprintf("pipeline %q step %q: references step %q which does not exist in this pipeline", pipelineName, currentStep, refName))
+		return
 	case refIdx == currentIdx:
 		result.Warnings = append(result.Warnings,
 			fmt.Sprintf("pipeline %q step %q: references itself; a step cannot use its own outputs because they are not available until after execution", pipelineName, currentStep))
+		return
 	case refIdx > currentIdx:
 		result.Warnings = append(result.Warnings,
 			fmt.Sprintf("pipeline %q step %q: references step %q which has not executed yet (appears later in pipeline)", pipelineName, currentStep, refName))
+		return
+	}
+
+	// Step exists and precedes the current step — validate the output field path.
+	if fieldPath == "" {
+		return
+	}
+
+	info, ok := stepInfos[refName]
+	if !ok || info.stepType == "" {
+		return
+	}
+
+	outputs := reg.InferStepOutputs(info.stepType, info.stepConfig)
+	if len(outputs) == 0 {
+		return // no schema information available; skip
+	}
+
+	// If any output key is a placeholder (e.g. "(key)", "(dynamic)", "(nested)"),
+	// the step emits dynamic fields whose names cannot be statically determined.
+	// Skip field-path validation for such steps to avoid false positives.
+	if hasDynamicOutputs(outputs) {
+		return
+	}
+
+	// Split ".row.auth_token" → ["row", "auth_token"]
+	parts := strings.Split(strings.TrimPrefix(fieldPath, "."), ".")
+	if len(parts) == 0 || parts[0] == "" {
+		return
+	}
+	firstField := parts[0]
+
+	// Check the first field against known output keys.
+	var matchedOutput *schema.InferredOutput
+	for i := range outputs {
+		if outputs[i].Key == firstField {
+			matchedOutput = &outputs[i]
+			break
+		}
+	}
+	if matchedOutput == nil {
+		result.Warnings = append(result.Warnings,
+			fmt.Sprintf("pipeline %q step %q: references step %q output field %q which is not a known output of step type %q (known outputs: %s)",
+				pipelineName, currentStep, refName, firstField, info.stepType, joinOutputKeys(outputs)))
+		return
+	}
+
+	// For db_query/db_query_cached steps, try SQL alias validation on "row.<col>" paths.
+	if firstField == "row" && len(parts) > 1 && isDBQueryStep(info.stepType) {
+		columnName := parts[1]
+		query, _ := info.stepConfig["query"].(string)
+		if query != "" {
+			sqlCols := extractSQLColumns(query)
+			if len(sqlCols) > 0 {
+				found := false
+				for _, col := range sqlCols {
+					if col == columnName {
+						found = true
+						break
+					}
+				}
+				if !found {
+					result.Warnings = append(result.Warnings,
+						fmt.Sprintf("pipeline %q step %q: references step %q output field \"row.%s\" but the SQL query does not select column %q (available: %s)",
+							pipelineName, currentStep, refName, columnName, columnName, strings.Join(sqlCols, ", ")))
+				}
+			}
+		}
+	}
+}
+
+// validatePlainStepRefs checks plain-string config values that contain bare step
+// context-key references (e.g. "steps.STEP_NAME.field") in config fields known to
+// accept such paths: secret_from, backend_url_key, and field (conditional/branch).
+func validatePlainStepRefs(pipelineName, stepName string, stepIdx int, stepCfg map[string]any, stepNames map[string]int, stepInfos map[string]stepBuildInfo, reg *schema.StepSchemaRegistry, result *templateValidationResult) {
+	// Config keys that are documented to accept a bare "steps.X.y" context path.
+	plainRefKeys := []string{"secret_from", "backend_url_key", "field"}
+	for _, key := range plainRefKeys {
+		val, ok := stepCfg[key].(string)
+		if !ok || val == "" {
+			continue
+		}
+		m := plainStepPathRe.FindStringSubmatch(val)
+		if m == nil {
+			continue
+		}
+		refName := m[1]
+		fieldPath := m[2] // already in ".field.subfield" form
+		validateStepRef(pipelineName, stepName, refName, fieldPath, stepIdx, stepNames, stepInfos, reg, result)
 	}
 }