Cleanup and complete the pipeline refactor

2026-07-07 15:32:35 -05:00
parent a9d8505cdb
commit 582c5dceed
14 changed files with 164 additions and 160 deletions
--- a/internal/framework/pipeline/profile.go
+++ b/internal/framework/pipeline/profile.go
@@ -292,20 +292,17 @@ func resolveArtifactLane(
 	lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, normalizeReferences)
 	capabilities.add(normalizeSpec.Provides...)

-	for _, validator := range lane.Validators {
-		validatorSpec, err := validatorSpec(catalog, validator.Module)
-		if err != nil {
-			return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageValidate, validator.Module, err)
-		}
-		if missing, ok := capabilities.missing(validatorSpec.Requires); ok {
-			return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageValidate, validator.Module, missing)
-		}
-		capabilities.add(validatorSpec.Provides...)
+	if len(lane.Validators) > 0 {
+		return ResolvedArtifactLane{}, nil, configuredValidatorsError(pipelineID, laneID)
 	}

 	return lane, capabilities, nil
 }

+func configuredValidatorsError(pipelineID string, laneID string) error {
+	return fmt.Errorf("pipeline %q lane %q configured validators are not supported by the current raw validation runner", pipelineID, laneID)
+}
+
 func referenceTarget(stage ModuleStage, laneID string, module string, bindings []ReferenceBinding) ResolvedReferenceTarget {
 	return ResolvedReferenceTarget{
 		Stage:    stage,
@@ -762,10 +759,6 @@ func normalizerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
 	return registrySpec(catalog.Normalizers, key)
 }

-func validatorSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
-	return registrySpec(catalog.Validators, key)
-}
-
 func outputSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
 	return registrySpec(catalog.Outputs, key)
 }
--- a/internal/framework/pipeline/profile_test.go
+++ b/internal/framework/pipeline/profile_test.go
@@ -19,8 +19,7 @@ func TestResolvePipelineWithExplicitModules(t *testing.T) {
 		ModuleSpec{Key: "record-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}},
 		ModuleSpec{Key: "dedupe", Stage: StageMerge, Requires: []string{"candidate"}, Provides: []string{"merged"}},
 		ModuleSpec{Key: "canonical", Stage: StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}},
-		ModuleSpec{Key: "schema-check", Stage: StageValidate, Requires: []string{"normalized"}, Provides: []string{"validated"}},
-		ModuleSpec{Key: "ndjson", Stage: StageOutput, Requires: []string{"validated"}, Provides: []string{"encoded"}},
+		ModuleSpec{Key: "ndjson", Stage: StageOutput, Requires: []string{"normalized"}, Provides: []string{"encoded"}},
 	)

 	resolved, err := ResolvePipeline(PipelineProfile{
@@ -31,10 +30,9 @@ func TestResolvePipelineWithExplicitModules(t *testing.T) {
 		}},
 		Artifacts: map[string]ArtifactLaneProfile{
 			" records ": {
-				Extract:    ModuleBinding{Module: " record-extractor ", LLMProfile: " careful "},
-				Merge:      Binding(" dedupe "),
-				Normalize:  Binding(" canonical "),
-				Validators: []ModuleBinding{Binding(" schema-check ")},
+				Extract:   ModuleBinding{Module: " record-extractor ", LLMProfile: " careful "},
+				Merge:     Binding(" dedupe "),
+				Normalize: Binding(" canonical "),
 			},
 		},
 		Output: Binding(" ndjson "),
@@ -68,8 +66,8 @@ func TestResolvePipelineWithExplicitModules(t *testing.T) {
 	if lane.Merge.Module != "dedupe" || lane.Normalize.Module != "canonical" {
 		t.Fatalf("lane merge/normalize = %#v/%#v, want explicit modules", lane.Merge, lane.Normalize)
 	}
-	if len(lane.Validators) != 1 || lane.Validators[0].Module != "schema-check" {
-		t.Fatalf("lane.Validators = %#v, want schema-check", lane.Validators)
+	if len(lane.Validators) != 0 {
+		t.Fatalf("lane.Validators = %#v, want none", lane.Validators)
 	}
 	if resolved.Output.Module != "ndjson" {
 		t.Fatalf("Output.Module = %q, want ndjson", resolved.Output.Module)
@@ -697,16 +695,6 @@ func TestResolvePipelineRejectsUnknownModuleKeys(t *testing.T) {
 			}),
 			want: []string{"baseline", "events", "normalize", "missing-normalize"},
 		},
-		{
-			name: "validate",
-			profile: withProfileChange(func(profile PipelineProfile) PipelineProfile {
-				lane := profile.Artifacts["events"]
-				lane.Validators = []ModuleBinding{Binding("missing-validator")}
-				profile.Artifacts["events"] = lane
-				return profile
-			}),
-			want: []string{"baseline", "events", "validate", "missing-validator"},
-		},
 		{
 			name: "output",
 			profile: withProfileChange(func(profile PipelineProfile) PipelineProfile {
@@ -759,11 +747,6 @@ func TestResolvePipelineRejectsMissingCapabilities(t *testing.T) {
 			spec: ModuleSpec{Key: "noop", Stage: StageNormalize, Requires: []string{"missing"}},
 			want: []string{"baseline", "events", "normalize", "noop", "missing"},
 		},
-		{
-			name: "validate",
-			spec: ModuleSpec{Key: "grounded", Stage: StageValidate, Requires: []string{"missing"}},
-			want: []string{"baseline", "events", "validate", "grounded", "missing"},
-		},
 		{
 			name: "output",
 			spec: ModuleSpec{Key: "json", Stage: StageOutput, Requires: []string{"missing"}},
@@ -775,9 +758,6 @@ func TestResolvePipelineRejectsMissingCapabilities(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			catalog := newProfileCatalogWithOverride(t, test.spec)
 			profile := baselineProfile()
-			lane := profile.Artifacts["events"]
-			lane.Validators = []ModuleBinding{Binding("grounded")}
-			profile.Artifacts["events"] = lane

 			_, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
 			if err == nil {
@@ -788,6 +768,19 @@ func TestResolvePipelineRejectsMissingCapabilities(t *testing.T) {
 	}
 }

+func TestResolvePipelineRejectsConfiguredValidators(t *testing.T) {
+	profile := baselineProfile()
+	lane := profile.Artifacts["events"]
+	lane.Validators = []ModuleBinding{Binding("grounded")}
+	profile.Artifacts["events"] = lane
+
+	_, err := ResolvePipeline(profile, ResolveOptions{}, newProfileCatalog(t))
+	if err == nil {
+		t.Fatal("ResolvePipeline() error = nil, want error")
+	}
+	assertErrorContains(t, err, "baseline", "events", "configured validators", "not supported")
+}
+
 func TestResolvePipelineOrdersLanesDeterministically(t *testing.T) {
 	resolved, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{}, newProfileCatalog(t))
 	if err != nil {
--- a/internal/framework/pipeline/runner.go
+++ b/internal/framework/pipeline/runner.go
@@ -211,7 +211,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source
 			result, err := extractor.Extract(ctx, contracts.ExtractionRequest{
 				Source:      doc,
 				Chunk:       &chunk,
-				SourceInput: sourceInput.Clone(),
+				SourceInput: chunkInputMaterial(sourceInput, chunk),
 				SessionID:   sessionID,
 				References:  CloneReferenceSet(lane.ExtractReferences.ReferenceSet),
 				LLMClient:   input.LLMClient,
@@ -367,11 +367,6 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source
 	return nil
 }

-type validatorExecution struct {
-	validator contracts.Validator
-	binding   ModuleBinding
-}
-
 type rawValidationTarget struct {
 	stage      ModuleStage
 	laneID     string
@@ -512,21 +507,6 @@ func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([
 	return warnings, nil, nil
 }

-func (r *Runner) buildConfiguredValidators(lane ResolvedArtifactLane) ([]validatorExecution, error) {
-	validators := make([]validatorExecution, 0, len(lane.Validators))
-	for _, binding := range lane.Validators {
-		validator, err := r.registries.Validators.Build(binding.Module)
-		if err != nil {
-			return nil, fmt.Errorf("build validator %q for lane %q: %w", binding.Module, lane.ID, err)
-		}
-		validators = append(validators, validatorExecution{
-			validator: validator,
-			binding:   binding,
-		})
-	}
-	return validators, nil
-}
-
 func (r *Runner) validateRegistries(pipeline ResolvedPipeline) error {
 	if r.registries.Inputs == nil {
 		return fmt.Errorf("input registry must not be nil")
@@ -546,9 +526,6 @@ func (r *Runner) validateRegistries(pipeline ResolvedPipeline) error {
 	if r.registries.Outputs == nil {
 		return fmt.Errorf("output encoder registry must not be nil")
 	}
-	if pipelineUsesConfiguredValidators(pipeline) && r.registries.Validators == nil {
-		return fmt.Errorf("validator registry must not be nil")
-	}
 	return nil
 }

@@ -584,10 +561,8 @@ func validateRunInput(input RunInput) error {
 		if lane.Normalize.Module == "" {
 			return fmt.Errorf("resolved pipeline lane %q normalize module must not be empty", lane.ID)
 		}
-		for _, validator := range lane.Validators {
-			if validator.Module == "" {
-				return fmt.Errorf("resolved pipeline lane %q validator module must not be empty", lane.ID)
-			}
+		if len(lane.Validators) > 0 {
+			return fmt.Errorf("resolved pipeline lane %q configured validators are not supported by the current raw validation runner", lane.ID)
 		}
 	}
 	return nil
@@ -627,9 +602,6 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest {
 			Merger:     lane.Merge.Module,
 			Normalizer: lane.Normalize.Module,
 		}
-		for _, validator := range lane.Validators {
-			laneManifest.Validators = append(laneManifest.Validators, validator.Module)
-		}
 		manifest.ArtifactLanes = append(manifest.ArtifactLanes, laneManifest)
 	}
 	return manifest
@@ -867,6 +839,16 @@ func sourceInputMaterial(inputPath string, content []byte) contracts.LLMInputMat
 	)
 }

+func chunkInputMaterial(sourceInput contracts.LLMInputMaterial, chunk contracts.SourceChunk) contracts.LLMInputMaterial {
+	return contracts.NewLLMInputMaterial(
+		"source",
+		chunk.MediaType,
+		chunk.Content,
+		sourceInputDigest(chunk.Content),
+		sourceInput.OriginURI,
+	)
+}
+
 func sourceInputMediaType(inputPath string) string {
 	extension := strings.ToLower(filepath.Ext(strings.TrimSpace(inputPath)))
 	if extension == ".json" {
@@ -973,12 +955,3 @@ func cloneRejectedOutputs(rejected []contracts.RejectedOutput) []contracts.Rejec
 func timePtr(t time.Time) *time.Time {
 	return &t
 }
-
-func pipelineUsesConfiguredValidators(pipeline ResolvedPipeline) bool {
-	for _, lane := range pipeline.ArtifactLanes {
-		if len(lane.Validators) > 0 {
-			return true
-		}
-	}
-	return false
-}
--- a/internal/framework/pipeline/runner_test.go
+++ b/internal/framework/pipeline/runner_test.go
@@ -124,15 +124,6 @@ func TestRunRejectsInvalidSetup(t *testing.T) {
 			},
 			error: "normalizer registry",
 		},
-		{
-			name: "missing validator registry only when configured validators are used",
-			run: func() (RunOutput, error) {
-				registries := newRunnerRegistries(t, nil)
-				registries.Validators = nil
-				return New(registries).Run(context.Background(), RunInput{Pipeline: resolvedPipelineWithValidators("configured")})
-			},
-			error: "validator registry",
-		},
 		{
 			name: "missing output registry",
 			run: func() (RunOutput, error) {
@@ -529,18 +520,16 @@ func TestRunPassesSourceInputAndSessionIDToPromptCapableStages(t *testing.T) {
 		t.Fatalf("manifest metadata = %#v, want session_id", output.Manifest.Metadata)
 	}

-	requests := []struct {
+	sourceRequests := []struct {
 		name      string
 		material  contracts.LLMInputMaterial
 		sessionID string
 	}{
 		{name: "chunk", material: modules.chunker.requests[0].SourceInput, sessionID: modules.chunker.requests[0].SessionID},
-		{name: "extract first", material: modules.extractors["extract-alpha"].requests[0].SourceInput, sessionID: modules.extractors["extract-alpha"].requests[0].SessionID},
-		{name: "extract second", material: modules.extractors["extract-alpha"].requests[1].SourceInput, sessionID: modules.extractors["extract-alpha"].requests[1].SessionID},
 		{name: "merge", material: modules.mergers["merge"].requests[0].SourceInput, sessionID: modules.mergers["merge"].requests[0].SessionID},
 		{name: "normalize", material: modules.normalizers["normalize"].requests[0].SourceInput, sessionID: modules.normalizers["normalize"].requests[0].SessionID},
 	}
-	for _, req := range requests {
+	for _, req := range sourceRequests {
 		if req.sessionID != "explicit-session" {
 			t.Fatalf("%s session ID = %q, want explicit-session", req.name, req.sessionID)
 		}
@@ -557,9 +546,29 @@ func TestRunPassesSourceInputAndSessionIDToPromptCapableStages(t *testing.T) {
 			t.Fatalf("%s origin URI = %q, want file URI ending in session.json", req.name, req.material.OriginURI)
 		}
 	}
+	for i, req := range modules.extractors["extract-alpha"].requests {
+		if req.SessionID != "explicit-session" {
+			t.Fatalf("extract %d session ID = %q, want explicit-session", i, req.SessionID)
+		}
+		if req.Chunk == nil {
+			t.Fatalf("extract %d chunk = nil, want chunk", i)
+		}
+		if got := string(req.SourceInput.Content); got != string(req.Chunk.Content) {
+			t.Fatalf("extract %d source input content = %q, want chunk content %q", i, got, req.Chunk.Content)
+		}
+		if req.SourceInput.Name != "source" || req.SourceInput.MediaType != req.Chunk.MediaType || req.SourceInput.SizeBytes != int64(len(req.Chunk.Content)) {
+			t.Fatalf("extract %d source input = %#v, want chunk metadata", i, req.SourceInput)
+		}
+		if req.SourceInput.Digest != sourceInputDigest(req.Chunk.Content) {
+			t.Fatalf("extract %d digest = %q, want %q", i, req.SourceInput.Digest, sourceInputDigest(req.Chunk.Content))
+		}
+		if !strings.HasPrefix(req.SourceInput.OriginURI, "file://") || !strings.HasSuffix(req.SourceInput.OriginURI, "/session.json") {
+			t.Fatalf("extract %d origin URI = %q, want file URI ending in session.json", i, req.SourceInput.OriginURI)
+		}
+	}

 	modules.chunker.requests[0].SourceInput.Content[0] = 'X'
-	if got := string(modules.extractors["extract-alpha"].requests[0].SourceInput.Content); got != string(rawInput) {
+	if got := string(modules.extractors["extract-alpha"].requests[0].SourceInput.Content); got != string(modules.extractors["extract-alpha"].requests[0].Chunk.Content) {
 		t.Fatalf("source input content aliased across requests: %q", got)
 	}
 	if got := string(rawInput); got != "{\"source\":\"exact bytes\"}" {
@@ -619,14 +628,13 @@ func TestRunPassesInputRequestFields(t *testing.T) {

 func TestRunPassesModuleBindingConfigToStageRequests(t *testing.T) {
 	modules := defaultRunnerModules()
-	pipeline := resolvedPipelineWithValidators("configured")
+	pipeline := resolvedPipeline()
 	pipeline.Input = ModuleBinding{Module: "input", LLMProfile: "input-profile", Options: map[string]any{"input_option": "input-value"}}
 	pipeline.Chunk = ModuleBinding{Module: "chunk", LLMProfile: "chunk-profile", Options: map[string]any{"chunk_option": "chunk-value"}}
 	pipeline.Output = ModuleBinding{Module: "output", LLMProfile: "output-profile", Options: map[string]any{"output_option": "output-value"}}
 	pipeline.ArtifactLanes[0].Extract = ModuleBinding{Module: "extract-alpha", LLMProfile: "extract-profile", Options: map[string]any{"extract_option": "extract-value"}}
 	pipeline.ArtifactLanes[0].Merge = ModuleBinding{Module: "merge", LLMProfile: "merge-profile", Options: map[string]any{"merge_option": "merge-value"}}
 	pipeline.ArtifactLanes[0].Normalize = ModuleBinding{Module: "normalize", LLMProfile: "normalize-profile", Options: map[string]any{"normalize_option": "normalize-value"}}
-	pipeline.ArtifactLanes[0].Validators[0] = ModuleBinding{Module: "configured", LLMProfile: "validator-profile", Options: map[string]any{"validator_option": "validator-value"}}

 	_, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline})
 	if err != nil {
@@ -1063,19 +1071,11 @@ func TestRunContextCancellationStopsRetries(t *testing.T) {
 	}
 }

-func TestRunRecordsConfiguredValidatorsInManifest(t *testing.T) {
-	modules := defaultRunnerModules()
-
-	output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{
+func TestRunRejectsConfiguredValidators(t *testing.T) {
+	_, err := New(newRunnerRegistries(t, defaultRunnerModules())).Run(context.Background(), RunInput{
 		Pipeline: resolvedPipelineWithValidators("configured", "second-validator"),
 	})
-	if err != nil {
-		t.Fatalf("Run() error = %v, want nil", err)
-	}
-
-	if got := output.Manifest.ArtifactLanes[0].Validators; !reflect.DeepEqual(got, []string{"configured", "second-validator"}) {
-		t.Fatalf("manifest validators = %#v, want configured validators", got)
-	}
+	assertRunError(t, err, "configured validators")
 }

 func TestRunCollectsStageWarnings(t *testing.T) {
@@ -1181,7 +1181,7 @@ func TestRunReturnsFailedManifestWhenOutputEncoderFails(t *testing.T) {
 }

 func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) {
-	resolved := resolvedPipelineWithValidators("configured")
+	resolved := resolvedPipeline()
 	resolved.ChunkReferences.ReferenceSet = contracts.ReferenceSet{
 		Slots: map[string]contracts.ResolvedReferenceSlot{
 			"scene_guide": {
@@ -1312,8 +1312,8 @@ func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) {
 	if lane.ID != "alpha" || lane.Extractor != "extract-alpha" || lane.Merger != "merge" || lane.Normalizer != "normalize" {
 		t.Fatalf("ArtifactLanes[0] = %#v, want lane details", lane)
 	}
-	if !reflect.DeepEqual(lane.Validators, []string{"configured"}) {
-		t.Fatalf("lane validators = %#v, want configured validator", lane.Validators)
+	if len(lane.Validators) != 0 {
+		t.Fatalf("lane validators = %#v, want none", lane.Validators)
 	}
 }