Classify framework process diagnostics

internal/framework/pipeline/diagnostics.go

@@ -22,6 +22,17 @@ func terminalDiagnosticGroups(terminal producerAttemptTerminal, origin contracts
 		}
 		groups = append(groups, promoted...)
 	}
+	if terminal.Action == producerTerminalIncompleteAccepted {
+		for _, record := range incompleteValidationDiagnostics(terminal.Validation) {
+			validatorOrigin := origin
+			validatorOrigin.ValidatorKey = record.validatorName
+			promoted, err := promoteProducerDiagnostics([]contracts.ProducerDiagnostic{record.diagnostic}, validatorOrigin, chunk)
+			if err != nil {
+				return nil, fmt.Errorf("validator %q incomplete diagnostic: %w", record.validatorName, err)
+			}
+			groups = append(groups, promoted...)
+		}
+	}
 	return groups, nil
 }
 

internal/framework/pipeline/producer_attempts.go

@@ -260,7 +260,6 @@ func runProducerAttempts(ctx context.Context, config producerAttemptConfig, prod
 			provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptIncompleteAccepted, Validation: report})
 			if config.Policy.ValidatorFailure == ValidatorFailureWarnContinue {
 				warnings := terminalWarnings(output, report)
-				warnings = append(warnings, incompleteValidationWarnings(report)...)
 				return producerAttemptTerminal{Action: producerTerminalIncompleteAccepted, Value: output.Value, Warnings: warnings, Diagnostics: cloneProducerDiagnostics(output.Diagnostics), Validation: report, ValidationIncomplete: true, Provenance: cloneProducerAttemptProvenance(provenance)}, nil
 			}
 			return failedProducerAttempt(provenance), validatorFailureError(*incomplete)
@@ -324,22 +323,27 @@ func cloneProducerDiagnostics(diagnostics []contracts.ProducerDiagnostic) []cont
 	return contracts.CloneProducerDiagnostics(diagnostics)
 }
 
-// incompleteValidationWarnings reports only validators that exhausted their
-// execution budget. It never reports rejected candidates, and it uses fixed
-// text so provider errors and correction content cannot cross this boundary.
-func incompleteValidationWarnings(report validationReport) []contracts.Warning {
-	warnings := make([]contracts.Warning, 0)
+// incompleteValidationDiagnostics reports every applicable validator that
+// could not complete under warn_continue. It uses fixed text so provider
+// errors and arbitrary validator prose cannot cross this boundary.
+func incompleteValidationDiagnostics(report validationReport) []validationDiagnosticRecord {
+	diagnostics := make([]validationDiagnosticRecord, 0)
 	for _, record := range report.records {
-		if record.outcome != validationFailed {
+		if record.outcome != validationFailed && record.outcome != validationSkipped {
 			continue
 		}
-		warnings = append(warnings, contracts.Warning{
-			Scope:      record.validatorName,
-			ReasonCode: "validator_execution_incomplete",
-			Message:    "Validator execution did not complete within its configured budget.",
-		})
+		diagnostics = append(diagnostics, validationDiagnosticRecord{validatorName: record.validatorName, diagnostic: contracts.ProducerDiagnostic{
+			Disposition:     contracts.DiagnosticDispositionWarning,
+			Category:        contracts.DiagnosticCategoryValidationIncomplete,
+			ReasonCode:      "validator_execution_incomplete",
+			OccurrenceCount: 1,
+			Samples: []contracts.DiagnosticSample{{
+				Scope:   record.validatorName,
+				Message: "Validator execution did not complete within its configured budget.",
+			}},
+		}})
 	}
-	return warnings
+	return diagnostics
 }
 
 // validationSummary projects a terminal state-machine result into the durable

internal/framework/pipeline/producer_attempts_test.go

@@ -324,11 +324,20 @@ func TestWarnContinueRecordsOneWarningForEachExhaustedValidator(t *testing.T) {
 	if terminal.Action != producerTerminalIncompleteAccepted {
 		t.Fatalf("terminal action = %q", terminal.Action)
 	}
-	if got, want := terminal.Warnings, []contracts.Warning{
-		{Scope: "first", ReasonCode: "validator_execution_incomplete", Message: "Validator execution did not complete within its configured budget."},
-		{Scope: "third", ReasonCode: "validator_execution_incomplete", Message: "Validator execution did not complete within its configured budget."},
-	}; !reflect.DeepEqual(got, want) {
-		t.Fatalf("warnings = %#v, want %#v", got, want)
+	if len(terminal.Warnings) != 0 {
+		t.Fatalf("warnings = %#v, want structured diagnostics only", terminal.Warnings)
+	}
+	groups, groupErr := terminalDiagnosticGroups(terminal, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageNormalize, StepID: "step", LaneID: "lane", ModuleKey: "module"}, nil)
+	if groupErr != nil {
+		t.Fatalf("terminalDiagnosticGroups() error = %v", groupErr)
+	}
+	if len(groups) != 3 {
+		t.Fatalf("diagnostic groups = %#v, want every failed and skipped validator", groups)
+	}
+	for index, validator := range []string{"first", "second", "third"} {
+		if group := groups[index]; group.Origin.ValidatorKey != validator || group.Disposition != contracts.DiagnosticDispositionWarning || group.Category != contracts.DiagnosticCategoryValidationIncomplete || group.ReasonCode != "validator_execution_incomplete" {
+			t.Fatalf("diagnostic group %d = %#v, want incomplete warning for %q", index, group, validator)
+		}
 	}
 	summary := validationSummary(terminal, StageNormalize, "step", "lane", "module", "", 0)
 	if summary.Status != "incomplete" || !reflect.DeepEqual(summary.IncompleteValidators, []string{"first", "second", "third"}) || !reflect.DeepEqual(summary.ReasonCodes, []string{"missing_prerequisite"}) {

internal/framework/pipeline/references.go

@@ -88,6 +88,26 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt
 	return out, warnings, nil
 }
 
+// ReferenceDiagnostics converts framework-owned empty-reference findings into
+// structured process diagnostics. Legacy callers may continue to use the
+// materializer's warning return while the framework migration is in progress.
+func ReferenceDiagnostics(warnings []contracts.Warning) []contracts.ProducerDiagnostic {
+	diagnostics := make([]contracts.ProducerDiagnostic, 0, len(warnings))
+	for _, warning := range warnings {
+		if warning.ReasonCode != "empty_reference" {
+			continue
+		}
+		diagnostics = append(diagnostics, contracts.ProducerDiagnostic{
+			Disposition:     contracts.DiagnosticDispositionWarning,
+			Category:        contracts.DiagnosticCategoryConfiguration,
+			ReasonCode:      "empty_reference",
+			OccurrenceCount: 1,
+			Samples:         []contracts.DiagnosticSample{{Scope: warning.Scope, Message: warning.Message}},
+		})
+	}
+	return contracts.CloneProducerDiagnostics(diagnostics)
+}
+
 func materializeReferenceTarget(
 	pipelineID string,
 	target ResolvedReferenceTarget,

internal/framework/pipeline/references_test.go

@@ -454,6 +454,10 @@ func TestMaterializeReferencesWarnsForEmptyFiles(t *testing.T) {
 	if len(warnings) != 1 || warnings[0].ReasonCode != "empty_reference" {
 		t.Fatalf("warnings = %#v, want empty reference warning", warnings)
 	}
+	diagnostics := ReferenceDiagnostics(warnings)
+	if len(diagnostics) != 1 || diagnostics[0].Disposition != contracts.DiagnosticDispositionWarning || diagnostics[0].Category != contracts.DiagnosticCategoryConfiguration || diagnostics[0].ReasonCode != "empty_reference" || diagnostics[0].OccurrenceCount != 1 || !reflect.DeepEqual(diagnostics[0].Samples, []contracts.DiagnosticSample{{Scope: warnings[0].Scope, Message: warnings[0].Message}}) {
+		t.Fatalf("diagnostics = %#v, want structured empty-reference signal", diagnostics)
+	}
 	item := materialized.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
 	if item.SizeBytes != 0 || item.Digest != referenceDigest(nil) {
 		t.Fatalf("empty item = %#v, want zero size and empty digest", item)

internal/framework/pipeline/runner.go

@@ -408,8 +408,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 		ModuleKey: encoder.Key(),
 		StartedAt: outputStarted,
 		Payload: map[string]any{
-			"files":    debugOutputFiles(files),
-			"warnings": encoded.Warnings,
+			"files": debugOutputFiles(files),
 		},
 	}); err != nil {
 		return failOutput(output), fmt.Errorf("write output debug artifact: %w", err)
@@ -417,7 +416,6 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 	if err := ctx.Err(); err != nil {
 		return failOutput(output), err
 	}
-	output.Warnings = append(output.Warnings, encoded.Warnings...)
 	output.OutputFiles = files
 
 	return output, nil

internal/framework/pipeline/runner_concurrency_test.go

@@ -146,8 +146,7 @@ func (e *cancelingOutputEncoder) Encode(context.Context, contracts.OutputRequest
 	e.calls.Add(1)
 	e.cancel()
 	return contracts.OutputResult{
-		Files:    []contracts.OutputFile{{Name: "result.txt", ContentType: "text/plain", Bytes: []byte("result")}},
-		Warnings: []contracts.Warning{{ReasonCode: "returned-after-cancel"}},
+		Files: []contracts.OutputFile{{Name: "result.txt", ContentType: "text/plain", Bytes: []byte("result")}},
 	}, nil
 }