From 582c5dceede1a87716bf9df6570b4c0b38f1c54e Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 7 Jul 2026 15:32:35 -0500 Subject: [PATCH] Cleanup and complete the pipeline refactor --- docs/cli.md | 3 +- docs/config.md | 9 +-- docs/internal/modules.md | 7 ++- docs/internal/pipeline.md | 14 +++-- docs/troubleshooting.md | 2 +- internal/core/config/effective_config_test.go | 5 +- internal/core/config/validation.go | 6 +- internal/core/config/validation_test.go | 37 ++++++------ internal/framework/pipeline/profile.go | 19 ++---- internal/framework/pipeline/profile_test.go | 45 ++++++-------- internal/framework/pipeline/runner.go | 53 ++++------------ internal/framework/pipeline/runner_test.go | 60 +++++++++---------- .../modules/extract/dnd/spells/extractor.go | 27 ++++++++- .../extract/dnd/spells/extractor_test.go | 37 ++++++++---- 14 files changed, 164 insertions(+), 160 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index dd9ef45..c9f88e5 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -207,7 +207,8 @@ The production CLI currently registers these module keys: - normalize: `noop` - output: `json` -The production CLI does not currently register validator modules. +Configured validator module lists are reserved for a future validator-chain +feature and are rejected by current configuration validation. For YAML structure, Scriptorium profile sources, environment overrides, and module binding syntax, see [Configuration](config.md). diff --git a/docs/config.md b/docs/config.md index 608dc28..e2f47d5 100644 --- a/docs/config.md +++ b/docs/config.md @@ -131,8 +131,8 @@ Artifact lane fields: - `extract`: required module binding. - `merge`: optional module binding. Default module is `appendorder`. - `normalize`: optional module binding. Default module is `noop`. -- `validators`: optional list of module bindings. The production CLI currently - does not register validator modules. +- `validators`: reserved for future configurable validator chains. Non-empty + lists are rejected by current configuration validation. - `references`: optional compatibility alias for extractor reference bindings. Lane bindings override pipeline-level bindings for the same slot. @@ -233,8 +233,9 @@ Binding fields: `normalize` bindings. - `options`: optional module-specific settings. - `references`: optional reference bindings. Supported only for `chunk`, - `extract`, `merge`, and `normalize` bindings. `input`, validator, and - `output` bindings reject this field during validation. + `extract`, `merge`, and `normalize` bindings. `input` and `output` bindings + reject this field during validation. Validator bindings are reserved for a + future validator-chain feature and are rejected when configured. The `--llm-profile` run flag overrides every effective LLM-capable module binding to use one Scriptorium profile ID: chunk, every selected lane extract, diff --git a/docs/internal/modules.md b/docs/internal/modules.md index 292789c..71ad88c 100644 --- a/docs/internal/modules.md +++ b/docs/internal/modules.md @@ -149,9 +149,10 @@ text, or secrets. Package: `internal/modules/extract/dnd/spells` The `dnd/spells` extractor owns D&D spell-cast extraction semantics. It -supplies the embedded Scriptorium prompt ID, prompt version, transcript and -reference input materials, response schema, and session ID to the runtime; then -returns the structured LLM `spell_casts` response as raw JSON. +supplies the embedded Scriptorium prompt ID, prompt version, chunk-scoped +transcript input material, reference input materials, response schema, and +session ID to the runtime; then returns the structured LLM `spell_casts` +response as raw JSON. Its LLM-facing source-reference schema uses integer `start_unit_id` and `end_unit_id` values matching source-unit IDs. diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 2da2277..63c90c2 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -60,10 +60,11 @@ extractor, merger, or normalizer request. LLM-backed modules pass that material onward as named Scriptorium prompt inputs. The CLI carries raw input bytes into `pipeline.RunInput`. Input adapters parse -those bytes into the source document, while LLM-backed modules that need the -original transcript material can pass the same bytes as a prompt input with -origin metadata. The raw input payload is not written to manifests or default -diagnostics. +those bytes into the source document. Chunk, merge, and normalize requests +receive the original source material as `SourceInput`; extraction requests +receive chunk-scoped source material built from the current `SourceChunk` +content, media type, and origin metadata. The raw input payload is not written +to manifests or default diagnostics. The CLI also carries an optional run `session_id`. The runner makes it available to chunk, extract, merge, and normalize requests; LLM-backed modules forward it @@ -186,6 +187,11 @@ Runner-side raw validation chains receive the raw module output plus stage, lane, module, source, and chunk provenance. Empty raw validation chains approve output by default. +Pipeline-configured validator lists are not part of the current runner +contract. Non-empty configured validator lists are rejected during configuration +validation or resolved-run validation so they cannot appear in manifests without +executing. + Validator rejection is a non-fatal run outcome: the rejected output is recorded in `RunOutput.Rejected` and does not pass to the next stage. Validator execution errors are framework-level errors and retry according to the relevant binding. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 6198554..ea0374e 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -320,7 +320,7 @@ Explanation and fixes: - Increase a module binding's `retries` only when re-running the same module input can reasonably produce an acceptable output. - If rejection is deterministic, fix the source input, module configuration, or - validator configuration rather than adding retries. + validation policy rather than adding retries. ## Retry Exhaustion diff --git a/internal/core/config/effective_config_test.go b/internal/core/config/effective_config_test.go index d0eedb9..a127086 100644 --- a/internal/core/config/effective_config_test.go +++ b/internal/core/config/effective_config_test.go @@ -161,7 +161,6 @@ func TestResolveLLMProfileOverrideAppliesBeforeDigest(t *testing.T) { profile.Output.LLMProfile = "output-profile" lane := profile.Artifacts["events"] lane.Merge.LLMProfile = "merge-profile" - lane.Validators[0].LLMProfile = "validator-profile" profile.Artifacts["events"] = lane cfg.Pipelines["example"] = profile @@ -196,8 +195,8 @@ func TestResolveLLMProfileOverrideAppliesBeforeDigest(t *testing.T) { if eventLane.Merge.LLMProfile != "runtime" { t.Fatalf("merge profile = %q, want runtime", eventLane.Merge.LLMProfile) } - if len(eventLane.Validators) != 1 || eventLane.Validators[0].LLMProfile != "validator-profile" { - t.Fatalf("validator profiles = %#v, want original validator-profile", eventLane.Validators) + if len(eventLane.Validators) != 0 { + t.Fatalf("validator profiles = %#v, want none", eventLane.Validators) } } diff --git a/internal/core/config/validation.go b/internal/core/config/validation.go index deac65a..51aaeae 100644 --- a/internal/core/config/validation.go +++ b/internal/core/config/validation.go @@ -84,10 +84,8 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile) erro if err := validateBinding(id, laneID, "normalize", lane.Normalize, true); err != nil { return err } - for i, validator := range lane.Validators { - if err := validateBinding(id, laneID, fmt.Sprintf("validator[%d]", i), validator, false); err != nil { - return err - } + if len(lane.Validators) > 0 { + return fmt.Errorf("pipeline %q lane %q validators are not supported by the current raw validation runner", id, laneID) } } } diff --git a/internal/core/config/validation_test.go b/internal/core/config/validation_test.go index b674a54..cd300ce 100644 --- a/internal/core/config/validation_test.go +++ b/internal/core/config/validation_test.go @@ -250,21 +250,6 @@ func TestValidateRejectsReferencesOnUnsupportedBindings(t *testing.T) { }, want: []string{"example", "input", "references", "not supported"}, }, - { - name: "validator", - mutate: func(cfg Config) Config { - profile := cfg.Pipelines["example"] - lane := profile.Artifacts["events"] - lane.Validators = []pipeline.ModuleBinding{{ - Module: "fake/validator", - References: map[string]string{"roster": "./roster.yml"}, - }} - profile.Artifacts["events"] = lane - cfg.Pipelines["example"] = profile - return cfg - }, - want: []string{"example", "events", "validator[0]", "references", "not supported"}, - }, { name: "output", mutate: func(cfg Config) Config { @@ -344,14 +329,32 @@ func TestValidateRejectsIDsDuplicatedAfterTrimming(t *testing.T) { } } +func TestValidateRejectsConfiguredValidators(t *testing.T) { + cfg := validConfig() + profile := cfg.Pipelines["example"] + lane := profile.Artifacts["events"] + lane.Validators = []pipeline.ModuleBinding{pipeline.Binding("fake/validator")} + profile.Artifacts["events"] = lane + cfg.Pipelines["example"] = profile + + err := cfg.Validate() + if err == nil { + t.Fatal("Validate() error = nil, want configured validators error") + } + for _, want := range []string{"example", "events", "validators", "not supported"} { + if !strings.Contains(err.Error(), want) { + t.Fatalf("Validate() error = %q, want substring %q", err.Error(), want) + } + } +} + func validConfig() Config { cfg := Default() cfg.Pipelines["example"] = pipeline.PipelineProfile{ Input: pipeline.Binding("fake/input"), Artifacts: map[string]pipeline.ArtifactLaneProfile{ "events": { - Extract: pipeline.Binding("fake/extract"), - Validators: []pipeline.ModuleBinding{pipeline.Binding("fake/validator")}, + Extract: pipeline.Binding("fake/extract"), }, "notes": { Extract: pipeline.Binding("fake/extract"), diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index 61a11ef..0e413b5 100644 --- 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) } diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go index 08f0077..b857739 100644 --- 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 { diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 95a671a..740b4ca 100644 --- 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 -} diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index 6fecc02..57475e0 100644 --- 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) } } diff --git a/internal/modules/extract/dnd/spells/extractor.go b/internal/modules/extract/dnd/spells/extractor.go index 135b599..4c3fb2d 100644 --- a/internal/modules/extract/dnd/spells/extractor.go +++ b/internal/modules/extract/dnd/spells/extractor.go @@ -1,6 +1,7 @@ package spells import ( + "bytes" "context" "encoding/json" "fmt" @@ -89,6 +90,10 @@ func (e *Extractor) Extract(ctx context.Context, req contracts.ExtractionRequest if req.LLMClient == nil { return contracts.ExtractionResult{}, extractorErrorf("LLM client must not be nil") } + sourceInput, err := chunkSourceInput(req) + if err != nil { + return contracts.ExtractionResult{}, err + } var response extractionResponse completion, err := req.LLMClient.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ @@ -97,7 +102,7 @@ func (e *Extractor) Extract(ctx context.Context, req contracts.ExtractionRequest PromptVersion: SchemaVersion, ProfileID: req.LLMProfile, SessionID: req.SessionID, - Inputs: dnd.PromptInputs(req.SourceInput, req.References), + Inputs: dnd.PromptInputs(sourceInput, req.References), }, &response) if err != nil { return contracts.ExtractionResult{}, extractorErrorf("complete structured output: %w", err) @@ -128,6 +133,26 @@ func (e *Extractor) Extract(ctx context.Context, req contracts.ExtractionRequest }, nil } +func chunkSourceInput(req contracts.ExtractionRequest) (contracts.LLMInputMaterial, error) { + material := req.SourceInput.Clone() + if len(material.Content) == 0 { + material = contracts.NewLLMInputMaterial("source", req.Chunk.MediaType, req.Chunk.Content, "", "") + } + if !bytes.Equal(material.Content, req.Chunk.Content) { + return contracts.LLMInputMaterial{}, extractorErrorf("source input must match chunk %q content", req.Chunk.ID) + } + if material.Name == "" { + material.Name = "source" + } + if material.MediaType == "" { + material.MediaType = req.Chunk.MediaType + } + if material.SizeBytes == 0 { + material.SizeBytes = int64(len(material.Content)) + } + return material, nil +} + func ModuleSpec() pipeline.ModuleSpec { return pipeline.ModuleSpec{ Key: Key, diff --git a/internal/modules/extract/dnd/spells/extractor_test.go b/internal/modules/extract/dnd/spells/extractor_test.go index 4f2a0d6..893d76d 100644 --- a/internal/modules/extract/dnd/spells/extractor_test.go +++ b/internal/modules/extract/dnd/spells/extractor_test.go @@ -27,7 +27,8 @@ func TestExtractReturnsRawOutputFromStructuredResponse(t *testing.T) { content: []byte(`{"spell_casts":[{"caster":" Aria ","spell":" Cure Wounds ","effect":" Heals an injured ally. ","narrative_description":" Aria restores the fighter after the fight. ","source_refs":[{"source_id":"session-alpha","start_unit_id":1,"end_unit_id":2}]}],"raw_marker":true}`), } - result, err := New().Extract(context.Background(), extractionRequestWithClient(client)) + extractReq := extractionRequestWithClient(client) + result, err := New().Extract(context.Background(), extractReq) if err != nil { t.Fatalf("Extract() error = %v, want nil", err) } @@ -35,22 +36,22 @@ func TestExtractReturnsRawOutputFromStructuredResponse(t *testing.T) { if len(client.requests) != 1 { t.Fatalf("LLM calls = %d, want 1", len(client.requests)) } - req := client.requests[0] - if req.StageName != Key { - t.Fatalf("StageName = %q, want %q", req.StageName, Key) + llmReq := client.requests[0] + if llmReq.StageName != Key { + t.Fatalf("StageName = %q, want %q", llmReq.StageName, Key) } - if req.PromptID != PromptID || req.PromptVersion != SchemaVersion { - t.Fatalf("prompt = %q/%q, want %q/%q", req.PromptID, req.PromptVersion, PromptID, SchemaVersion) + if llmReq.PromptID != PromptID || llmReq.PromptVersion != SchemaVersion { + t.Fatalf("prompt = %q/%q, want %q/%q", llmReq.PromptID, llmReq.PromptVersion, PromptID, SchemaVersion) } - if req.SessionID != "session-123" || req.ProfileID != "profile-spells" { - t.Fatalf("session/profile = %q/%q, want session-123/profile-spells", req.SessionID, req.ProfileID) + if llmReq.SessionID != "session-123" || llmReq.ProfileID != "profile-spells" { + t.Fatalf("session/profile = %q/%q, want session-123/profile-spells", llmReq.SessionID, llmReq.ProfileID) } - transcript := req.Inputs["transcript"] - if transcript.Name != "transcript" || transcript.MediaType != "application/json" || transcript.Digest != "sha256:transcript" || transcript.OriginURI != "file:///session-alpha.json" { + transcript := llmReq.Inputs["transcript"] + if transcript.Name != "transcript" || transcript.MediaType != "application/json" || transcript.Digest != "sha256:chunk" || transcript.OriginURI != "file:///session-alpha.json" { t.Fatalf("transcript metadata = %#v", transcript) } - if got := string(transcript.Content); got != spellTranscriptJSON { - t.Fatalf("transcript content = %q, want original source input", got) + if got := string(transcript.Content); got != string(extractReq.Chunk.Content) { + t.Fatalf("transcript content = %q, want chunk content %q", got, extractReq.Chunk.Content) } if result.Output.Payload.MediaType != "application/json" { @@ -227,6 +228,7 @@ func TestExtractRejectsInvalidRequests(t *testing.T) { {name: "nil chunk", extractor: New(), ctx: context.Background(), req: contracts.ExtractionRequest{Source: validReq.Source, LLMClient: validReq.LLMClient}, want: "chunk"}, {name: "empty chunk units", extractor: New(), ctx: context.Background(), req: emptyChunkRequest(validReq), want: "units"}, {name: "nil LLM client", extractor: New(), ctx: context.Background(), req: contracts.ExtractionRequest{Source: validReq.Source, Chunk: validReq.Chunk}, want: "LLM client"}, + {name: "source input mismatches chunk", extractor: New(), ctx: context.Background(), req: mismatchedSourceInputRequest(validReq), want: "must match chunk"}, } for _, tt := range tests { @@ -310,7 +312,7 @@ func TestExtractDefensivelyCopiesRawContent(t *testing.T) { func extractionRequestWithClient(client contracts.StructuredLLMClient) contracts.ExtractionRequest { req := promptExtractionRequest() req.LLMClient = client - req.SourceInput = spellSourceInput() + req.SourceInput = spellChunkInput(req.Chunk) req.SessionID = "session-123" req.LLMProfile = "profile-spells" return req @@ -322,6 +324,10 @@ func spellSourceInput() contracts.LLMInputMaterial { return contracts.NewLLMInputMaterial("source", "application/json", []byte(spellTranscriptJSON), "sha256:transcript", "file:///session-alpha.json") } +func spellChunkInput(chunk *contracts.SourceChunk) contracts.LLMInputMaterial { + return contracts.NewLLMInputMaterial("source", chunk.MediaType, chunk.Content, "sha256:chunk", "file:///session-alpha.json") +} + func emptyChunkRequest(req contracts.ExtractionRequest) contracts.ExtractionRequest { req.Chunk = &contracts.SourceChunk{ ID: req.Chunk.ID, @@ -331,6 +337,11 @@ func emptyChunkRequest(req contracts.ExtractionRequest) contracts.ExtractionRequ return req } +func mismatchedSourceInputRequest(req contracts.ExtractionRequest) contracts.ExtractionRequest { + req.SourceInput = spellSourceInput() + return req +} + type fakeSpellsLLMClient struct { response extractionResponse content []byte