feat: pipeline output schema declarations + step output field static analysis

cmd/wfctl/contract.go

@@ -26,10 +26,11 @@ type Contract struct {
 
 // EndpointContract describes an HTTP endpoint in the contract.
 type EndpointContract struct {
-	Method       string `json:"method"`
-	Path         string `json:"path"`
-	AuthRequired bool   `json:"authRequired"`
-	Pipeline     string `json:"pipeline"`
+	Method         string            `json:"method"`
+	Path           string            `json:"path"`
+	AuthRequired   bool              `json:"authRequired"`
+	Pipeline       string            `json:"pipeline"`
+	ResponseSchema map[string]string `json:"responseSchema,omitempty"` // field name → type
 }
 
 // ModuleContract describes a module in the contract.
@@ -268,6 +269,8 @@ func generateContract(cfg *config.WorkflowConfig) *Contract {
 									}
 								}
 							}
+							// Populate response schema from pipeline outputs declaration
+							ep.ResponseSchema = extractPipelineOutputSchema(pipelineMap)
 							contract.Endpoints = append(contract.Endpoints, ep)
 						}
 					}
@@ -343,6 +346,31 @@ func generateContract(cfg *config.WorkflowConfig) *Contract {
 	return contract
 }
 
+// extractPipelineOutputSchema reads the optional "outputs" block from a raw
+// pipeline map and returns a field→type map for use in EndpointContract.
+// Returns nil when the pipeline has no outputs declaration.
+func extractPipelineOutputSchema(pipelineMap map[string]any) map[string]string {
+	outputsRaw, ok := pipelineMap["outputs"]
+	if !ok || outputsRaw == nil {
+		return nil
+	}
+	outputsMap, ok := outputsRaw.(map[string]any)
+	if !ok || len(outputsMap) == 0 {
+		return nil
+	}
+	schema := make(map[string]string, len(outputsMap))
+	for field, defRaw := range outputsMap {
+		fieldType := "any"
+		if defMap, ok := defRaw.(map[string]any); ok {
+			if t, ok := defMap["type"].(string); ok && t != "" {
+				fieldType = t
+			}
+		}
+		schema[field] = fieldType
+	}
+	return schema
+}
+
 // compareContracts compares a baseline contract to the current one.
 func compareContracts(base, current *Contract) *contractComparison {
 	comp := &contractComparison{
@@ -365,18 +393,37 @@ func compareContracts(base, current *Contract) *contractComparison {
 	// Check base endpoints
 	for key, baseEP := range baseEPs {
 		if currentEP, exists := currentEPs[key]; exists {
-			// Check for breaking changes
+			// Collect all breaking changes for this endpoint
+			var breakingDetails []string
+
+			// Auth was added to a public endpoint
 			if baseEP.AuthRequired != currentEP.AuthRequired && !baseEP.AuthRequired {
-				// Auth was added to a public endpoint
-				comp.Endpoints = append(comp.Endpoints, endpointChange{
-					Method:     baseEP.Method,
-					Path:       baseEP.Path,
-					Pipeline:   currentEP.Pipeline,
-					Change:     changeChanged,
-					Detail:     "auth requirement added (clients without tokens will get 401)",
-					IsBreaking: true,
-				})
-				comp.BreakingCount++
+				breakingDetails = append(breakingDetails, "auth requirement added (clients without tokens will get 401)")
+			}
+
+			// Response schema fields were removed or changed type
+			for field, baseType := range baseEP.ResponseSchema {
+				if currentType, exists := currentEP.ResponseSchema[field]; !exists {
+					breakingDetails = append(breakingDetails,
+						fmt.Sprintf("response field %q removed (was %s)", field, baseType))
+				} else if baseType != currentType && baseType != "any" && currentType != "any" {
+					breakingDetails = append(breakingDetails,
+						fmt.Sprintf("response field %q changed type from %s to %s", field, baseType, currentType))
+				}
+			}
+
+			if len(breakingDetails) > 0 {
+				for _, detail := range breakingDetails {
+					comp.Endpoints = append(comp.Endpoints, endpointChange{
+						Method:     baseEP.Method,
+						Path:       baseEP.Path,
+						Pipeline:   currentEP.Pipeline,
+						Change:     changeChanged,
+						Detail:     detail,
+						IsBreaking: true,
+					})
+					comp.BreakingCount++
+				}
 			} else {
 				comp.Endpoints = append(comp.Endpoints, endpointChange{
 					Method:   baseEP.Method,
@@ -511,6 +558,17 @@ func printContract(c *Contract) {
 			auth = " [auth]"
 		}
 		fmt.Printf("  %-7s %s%s  (pipeline: %s)\n", ep.Method, ep.Path, auth, ep.Pipeline)
+		if len(ep.ResponseSchema) > 0 {
+			// Print response schema fields sorted for stable output
+			fields := make([]string, 0, len(ep.ResponseSchema))
+			for f := range ep.ResponseSchema {
+				fields = append(fields, f)
+			}
+			sort.Strings(fields)
+			for _, f := range fields {
+				fmt.Printf("    response.%s: %s\n", f, ep.ResponseSchema[f])
+			}
+		}
 	}
 
 	fmt.Printf("\nModules (%d):\n", len(c.Modules))

cmd/wfctl/contract_test.go

@@ -560,3 +560,235 @@ pipelines:
 		t.Errorf("expected 'breaking' in error message, got: %v", err)
 	}
 }
+
+// TestGenerateContractResponseSchema checks that the response schema is extracted
+// from the pipeline's optional outputs declaration.
+func TestGenerateContractResponseSchema(t *testing.T) {
+	cfg := &config.WorkflowConfig{
+		Pipelines: map[string]any{
+			"get-item": map[string]any{
+				"trigger": map[string]any{
+					"type": "http",
+					"config": map[string]any{
+						"path":   "/api/items/:id",
+						"method": "GET",
+					},
+				},
+				"outputs": map[string]any{
+					"id":    map[string]any{"type": "string"},
+					"name":  map[string]any{"type": "string"},
+					"found": map[string]any{"type": "boolean"},
+				},
+				"steps": []any{},
+			},
+		},
+	}
+
+	contract := generateContract(cfg)
+
+	if len(contract.Endpoints) != 1 {
+		t.Fatalf("expected 1 endpoint, got %d", len(contract.Endpoints))
+	}
+	ep := contract.Endpoints[0]
+	if len(ep.ResponseSchema) != 3 {
+		t.Fatalf("expected 3 response schema fields, got %d: %v", len(ep.ResponseSchema), ep.ResponseSchema)
+	}
+	if ep.ResponseSchema["id"] != "string" {
+		t.Errorf("expected id=string, got %q", ep.ResponseSchema["id"])
+	}
+	if ep.ResponseSchema["found"] != "boolean" {
+		t.Errorf("expected found=boolean, got %q", ep.ResponseSchema["found"])
+	}
+}
+
+// TestGenerateContractNoResponseSchema checks that endpoints without an outputs
+// block have a nil ResponseSchema (backwards-compatible).
+func TestGenerateContractNoResponseSchema(t *testing.T) {
+	cfg := &config.WorkflowConfig{
+		Pipelines: map[string]any{
+			"get-item": map[string]any{
+				"trigger": map[string]any{
+					"type": "http",
+					"config": map[string]any{
+						"path":   "/api/items",
+						"method": "GET",
+					},
+				},
+				"steps": []any{},
+			},
+		},
+	}
+
+	contract := generateContract(cfg)
+	if len(contract.Endpoints) != 1 {
+		t.Fatalf("expected 1 endpoint, got %d", len(contract.Endpoints))
+	}
+	if contract.Endpoints[0].ResponseSchema != nil {
+		t.Errorf("expected nil ResponseSchema for pipeline without outputs, got %v", contract.Endpoints[0].ResponseSchema)
+	}
+}
+
+// TestCompareContracts_ResponseSchemaFieldRemoved checks that removing a
+// declared response schema field is detected as a breaking change.
+func TestCompareContracts_ResponseSchemaFieldRemoved(t *testing.T) {
+	base := &Contract{
+		Version: "1.0",
+		Endpoints: []EndpointContract{
+			{
+				Method:   "GET",
+				Path:     "/api/items",
+				Pipeline: "get-items",
+				ResponseSchema: map[string]string{
+					"id":    "string",
+					"name":  "string",
+					"found": "boolean",
+				},
+			},
+		},
+	}
+	current := &Contract{
+		Version: "1.0",
+		Endpoints: []EndpointContract{
+			{
+				Method:   "GET",
+				Path:     "/api/items",
+				Pipeline: "get-items",
+				ResponseSchema: map[string]string{
+					"id":   "string",
+					"name": "string",
+					// "found" has been removed
+				},
+			},
+		},
+	}
+
+	comp := compareContracts(base, current)
+
+	if comp.BreakingCount == 0 {
+		t.Error("expected breaking change when response field is removed")
+	}
+	found := false
+	for _, ec := range comp.Endpoints {
+		if ec.IsBreaking && strings.Contains(ec.Detail, "found") && strings.Contains(ec.Detail, "removed") {
+			found = true
+			break
+		}
+	}
+	if !found {
+		t.Errorf("expected breaking change about removed field 'found', got: %v", comp.Endpoints)
+	}
+}
+
+// TestCompareContracts_ResponseSchemaTypeChanged checks that changing a
+// declared response schema field type is detected as a breaking change.
+func TestCompareContracts_ResponseSchemaTypeChanged(t *testing.T) {
+	base := &Contract{
+		Version: "1.0",
+		Endpoints: []EndpointContract{
+			{
+				Method:   "GET",
+				Path:     "/api/items",
+				Pipeline: "get-items",
+				ResponseSchema: map[string]string{
+					"count": "integer",
+				},
+			},
+		},
+	}
+	current := &Contract{
+		Version: "1.0",
+		Endpoints: []EndpointContract{
+			{
+				Method:   "GET",
+				Path:     "/api/items",
+				Pipeline: "get-items",
+				ResponseSchema: map[string]string{
+					"count": "string", // changed from integer to string
+				},
+			},
+		},
+	}
+
+	comp := compareContracts(base, current)
+
+	if comp.BreakingCount == 0 {
+		t.Error("expected breaking change when response field type changes")
+	}
+	found := false
+	for _, ec := range comp.Endpoints {
+		if ec.IsBreaking && strings.Contains(ec.Detail, "count") && strings.Contains(ec.Detail, "changed type") {
+			found = true
+			break
+		}
+	}
+	if !found {
+		t.Errorf("expected breaking change about type change for 'count', got: %v", comp.Endpoints)
+	}
+}
+
+// TestCompareContracts_ResponseSchemaFieldAdded checks that adding a new
+// response schema field is not a breaking change.
+func TestCompareContracts_ResponseSchemaFieldAdded(t *testing.T) {
+	base := &Contract{
+		Version: "1.0",
+		Endpoints: []EndpointContract{
+			{
+				Method:         "GET",
+				Path:           "/api/items",
+				Pipeline:       "get-items",
+				ResponseSchema: map[string]string{"id": "string"},
+			},
+		},
+	}
+	current := &Contract{
+		Version: "1.0",
+		Endpoints: []EndpointContract{
+			{
+				Method:         "GET",
+				Path:           "/api/items",
+				Pipeline:       "get-items",
+				ResponseSchema: map[string]string{"id": "string", "name": "string"},
+			},
+		},
+	}
+
+	comp := compareContracts(base, current)
+
+	if comp.BreakingCount != 0 {
+		t.Errorf("expected no breaking changes when a new field is added, got %d: %v", comp.BreakingCount, comp.Endpoints)
+	}
+}
+
+// TestExtractPipelineOutputSchema tests the helper that reads the outputs block.
+func TestExtractPipelineOutputSchema(t *testing.T) {
+	t.Run("with outputs", func(t *testing.T) {
+		pipelineMap := map[string]any{
+			"outputs": map[string]any{
+				"id":    map[string]any{"type": "string", "description": "Item ID"},
+				"found": map[string]any{"type": "boolean"},
+				"score": map[string]any{}, // missing type → defaults to "any"
+			},
+		}
+		got := extractPipelineOutputSchema(pipelineMap)
+		if got == nil {
+			t.Fatal("expected non-nil schema")
+		}
+		if got["id"] != "string" {
+			t.Errorf("id: want string, got %q", got["id"])
+		}
+		if got["found"] != "boolean" {
+			t.Errorf("found: want boolean, got %q", got["found"])
+		}
+		if got["score"] != "any" {
+			t.Errorf("score: want any, got %q", got["score"])
+		}
+	})
+
+	t.Run("without outputs", func(t *testing.T) {
+		pipelineMap := map[string]any{"steps": []any{}}
+		got := extractPipelineOutputSchema(pipelineMap)
+		if got != nil {
+			t.Errorf("expected nil for pipeline without outputs, got %v", got)
+		}
+	})
+}