diff --git a/docs/internal/llm.md b/docs/internal/llm.md index b0735b1..000eb73 100644 --- a/docs/internal/llm.md +++ b/docs/internal/llm.md @@ -33,7 +33,9 @@ without exposing Scriptorium types through stage contracts. 2. creating a `ScriptoriumClient` from the effective profile source; 3. attaching an `LLMProfileRecorder`; 4. creating a scheduler from the effective concurrency limit; -5. returning a `ScheduledClient` wrapper. +5. returning a `ScheduledClient` wrapper; and +6. injecting that one shared client into complete pipeline preparation before + the source file is read or the runner is invoked. The CLI separately gathers explicit profile IDs from resolved LLM-capable stage and validator bindings. It prepares a small internal check prompt for each ID so diff --git a/docs/internal/modules.md b/docs/internal/modules.md index 74511f7..87ee151 100644 --- a/docs/internal/modules.md +++ b/docs/internal/modules.md @@ -19,10 +19,15 @@ validator chains and prompt asset collection. Production extract, merge, normalize, and validator packages currently use the explicit legacy raw registration APIs. Typed registration is framework-ready, -but no production artifact kind or codec is registered yet. +but no production artifact kind or codec is registered yet. All selected +production implementations are constructed during pipeline preparation through +temporary adapters around their existing zero-argument constructors. Their raw +option maps and LLM clients remain operation inputs until each implementation +migrates to its construction-owned decoder and injected dependencies. Specs expose capability and execution metadata without constructing an -implementation. Chunk, extract, merge, and normalize modules that accept +implementation. Registry entries separately expose option validation and +run-local construction. Chunk, extract, merge, and normalize modules that accept auxiliary material declare identical reference slots from both `ReferenceSlots()` and `ModuleSpec().ReferenceSlots`; registration tests enforce that agreement. Runtime delivery uses the corresponding stage request's diff --git a/docs/internal/overview.md b/docs/internal/overview.md index 52029dc..35441bc 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -15,7 +15,8 @@ output files returned by the runner. Diagnostics, checkpoints, and debug recorders are optional side-channel collaborators supplied at this boundary. Pipeline execution is serial. Resolution produces a fixed ordered workflow and -a sorted set of artifact lanes before the runner constructs any stage module. +a sorted set of artifact lanes. Preparation constructs the complete module and +validator set before the runner receives source bytes. ## Application Boundary @@ -39,7 +40,7 @@ a sorted set of artifact lanes before the runner constructs any stage module. | Package | Implemented responsibility | | --- | --- | | `internal/framework/contracts` | Source-stage contracts plus artifact identity, schema, serialized representation, codec, validator, reference, output, and structured-completion interfaces and data types. | -| `internal/framework/pipeline` | Module and artifact-codec registries, profile resolution, capability checks, reference materialization, validator-chain resolution, retries, orchestration, warnings, and manifest population. | +| `internal/framework/pipeline` | Module and artifact-codec registries, option validation, profile resolution, capability checks, reference materialization, complete pipeline preparation, retries, orchestration, warnings, and manifest population. | | `internal/framework/validate` | Shared validator decision and cardinality helpers. | | `internal/framework/llm` | Scriptorium-backed structured completions, prompt/schema registration, scheduling, profile recording, and secret redaction. | | `internal/framework/promptfs` | Builds module prompt filesystems from module-owned and caller-provided shared prompt assets. | @@ -56,9 +57,13 @@ Artifact registries support heterogeneous typed extraction entries and kind-specific merger, normalizer, and validator variants. Resolution derives a lane's kind from its extractor, requires the matching codec, verifies exact Go type equality across the lane, and records schema identity in the resolved lane -and pipeline digest. Production module families do not register typed variants -yet and continue through explicitly named legacy raw registrations. The current -runner rejects a typed resolved lane instead of routing it through raw execution. +and pipeline digest. Registry entries carry separate option-validation and +run-local construction closures. Preparation injects shared dependencies and +constructs input, chunk, validators, ordered lanes, and output before source +parsing. Production module families do not register typed variants yet and +continue through explicitly named legacy raw registrations and temporary +zero-argument constructor adapters. The current runner rejects a typed prepared +lane instead of routing it through raw execution. ## Production Extensions diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 328316a..c2efe81 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -7,7 +7,8 @@ defaults, and selectable keys are defined in [Configuration](../config.md#pipelines). Pipeline execution is serial. Resolution fixes the selected lanes and all -stage bindings before the runner constructs stage implementations. +stage bindings; preparation constructs every selected implementation before the +runner begins source work. ## Resolution @@ -24,7 +25,9 @@ calls `pipeline.ResolvePipeline`. selects exact-type merger, normalizer, and validator variants; 5. checks required and provided capabilities in workflow order; 6. resolves target-aware reference bindings and validator chains; -7. calculates a digest over the resolved structure, including typed artifact +7. validates each selected module and validator option set through its registry + entry; and +8. calculates a digest over the resolved structure, including typed artifact kind and schema identity. Resolution returns a `ResolvedPipeline` containing ordered lanes, concrete @@ -53,7 +56,8 @@ runtime sensitive-data handling belongs in [Operations](../operations.md). ## Registries And Specs -`pipeline.Registries` holds constructors used during execution. +`pipeline.Registries` holds option validators and run-local builders used during +resolution and preparation. `pipeline.ModuleCatalog` exposes their specs during configuration validation and resolution. Separate registries exist for every stage and for validators; `ValidatorChainRegistry` stores production default-chain mappings. Both @@ -77,24 +81,37 @@ A `ModuleSpec` declares its stage plus required and provided capabilities. Chunk, extract, merge, and normalize specs may also declare reference slots. Registry implementations defensively copy spec metadata, reject duplicate keys, and verify that a constructed implementation reports the registered key. +Builder registrations accept `ModuleDependencies` and cloned raw options through +one `BuildRequest`. Existing production registrations are adapted from their +zero-argument constructors while their implementation-owned option decoders are +migrated separately. A `ValidatorSpec` declares a validator key and execution class. Resolution uses the execution class to reject incompatible profile bindings before execution. The current production catalog and default chain are listed only in [Configuration](../config.md#implemented-production-validators). -## Runner Boundary +## Preparation And Runner Boundary -`pipeline.RunInput` carries the resolved pipeline, raw source input, structured -LLM client, run identity and timing, optional session and profile metadata, and -checkpoint/debug collaborators. The runner parses source bytes through the -selected input adapter. Later stage requests receive the generic source model; -extract requests receive chunk-scoped input material, while chunk, merge, and -normalize requests retain access to the original source material. +`pipeline.Prepare` receives a resolved pipeline, the registries, and shared +module dependencies. It constructs input; chunk and its validators; each lane's +extract, merge, and normalize modules and validator chains in resolved order; +then output. It stops at the first error with pipeline, stage, lane, module, and +validator context as applicable. It never invokes an operation method. -Typed lanes can be composed and resolved but are not passed to the current raw -runner. The runner rejects such input before source work; production lanes are -still resolved and executed exclusively through the legacy raw path. +`PreparedPipeline` keeps private constructed executors and exposes cloned +resolved input, chunk, lane, and output identities. `pipeline.RunInput` carries +that prepared pipeline, raw source input, run identity and timing, optional +session and profile metadata, and checkpoint/debug collaborators. The runner +parses source bytes through the already constructed input adapter. Later stage +requests receive the generic source model; extract requests receive +chunk-scoped input material, while chunk, merge, and normalize requests retain +access to the original source material. + +Typed lanes can be composed, resolved, and prepared but are not executed by the +current raw runner. The runner rejects such input before source work; +production lanes are still resolved and executed exclusively through the +legacy raw path. Source validation requires every unit to carry a canonical self-reference to its containing document and its own unit ID. Explicit clone, checkpoint, and @@ -111,17 +128,17 @@ runner returns. The runner: -1. validates its input and registries; -2. builds the input adapter, parses the raw input, and validates the generic +1. validates its prepared input; +2. parses the raw input with the prepared adapter and validates the generic source document; 3. obtains or executes the chunk result; 4. validates and canonicalizes chunks; 5. executes each resolved artifact lane in order; -6. builds the output encoder and validates its logical file results; +6. invokes the prepared output encoder and validates its logical file results; 7. returns the assembled manifest, outcomes, warnings, and files. -Within each artifact lane, it builds the extractor, merger, and normalizer, -then performs these transitions: +Within each artifact lane, it reuses the prepared extractor, merger, normalizer, +and validators while performing these transitions: 1. extract once per accepted chunk and add runner-owned lane, source, and chunk provenance; @@ -208,8 +225,10 @@ durable manifest and logical file schemas are defined in the - `internal/framework/pipeline/artifact_codec_registry_test.go`: typed codec metadata, registration, erasure safety, strict decoding, and cloning. - `internal/framework/pipeline/typed_resolution_test.go`: heterogeneous typed - lane resolution, target-specific validators, incompatibilities, ordering, and - schema-sensitive pipeline identity. + lane resolution and preparation, target-specific validators, + incompatibilities, ordering, and schema-sensitive pipeline identity. +- `internal/framework/pipeline/preparation_test.go`: option validation, + construction order, dependency failures, and the before-source-work boundary. - `internal/framework/pipeline/references_test.go`: target resolution and materialization. - `internal/framework/pipeline/runner_test.go`: stage transitions, retries, diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index 10a6cbe..25d413f 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -90,6 +90,12 @@ and verifies module availability and capabilities before execution. Structural pipeline choices must not be scattered through conditionals or hidden behind ad hoc command flags. +Resolution validates every selected module and validator option set. A separate +preparation boundary then constructs the complete input, chunk, lane, +validation, and output implementation set in pipeline order. The runner accepts +only that prepared set, so construction and dependency failures occur before +source parsing or any other module operation. + Stage ownership is explicit: - input modules convert external material into the generic source model; diff --git a/internal/cli/run.go b/internal/cli/run.go index 8589283..c47260f 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -263,11 +263,6 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics resolved references: %w", err)) } - rawInput, err := os.ReadFile(strings.TrimSpace(*inputPath)) - if err != nil { - return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("read input %q: %w", strings.TrimSpace(*inputPath), err)) - } - registries, err := effectiveRegistries(opts) if err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err) @@ -282,16 +277,23 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i if err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("create LLM client for profile %q: %w", factoryProfileID, err)) } + prepared, err := pipeline.Prepare(effective.ResolvedPipeline, registries, pipeline.ModuleDependencies{LLM: llmClient}) + if err != nil { + return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("prepare pipeline %q: %w", pipelineID, err)) + } + rawInput, err := os.ReadFile(strings.TrimSpace(*inputPath)) + if err != nil { + return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("read input %q: %w", strings.TrimSpace(*inputPath), err)) + } checkpointRecorder, checkpointLoader, err := checkpointHandlersForRun(workspaceSettings, effective.ResolvedPipeline, rawInput, only, llmProfiles, strings.TrimSpace(*llmProfile), strings.TrimSpace(sessionID.value), *resume) if err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err) } - output, err := pipeline.New(registries).Run(ctx, pipeline.RunInput{ - Pipeline: effective.ResolvedPipeline, + output, err := pipeline.New().Run(ctx, pipeline.RunInput{ + Prepared: prepared, Path: strings.TrimSpace(*inputPath), RawInput: rawInput, - LLMClient: llmClient, SessionID: strings.TrimSpace(sessionID.value), RunID: runID, StartedAt: startedAt, diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index 7ebf243..ad9bc3b 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -3989,42 +3989,42 @@ func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.Module func mustRegisterInput(t *testing.T, registry *pipeline.InputAdapterRegistry, spec pipeline.ModuleSpec) { t.Helper() - if err := registry.RegisterWithSpec(spec, func() (contracts.InputAdapter, error) { return nil, nil }); err != nil { + if err := registry.RegisterWithSpec(spec, func() (contracts.InputAdapter, error) { return fakeRunInputAdapter{}, nil }); err != nil { t.Fatalf("register input: %v", err) } } func mustRegisterChunker(t *testing.T, registry *pipeline.ChunkerRegistry, spec pipeline.ModuleSpec) { t.Helper() - if err := registry.RegisterWithSpec(spec, func() (contracts.Chunker, error) { return nil, nil }); err != nil { + if err := registry.RegisterWithSpec(spec, func() (contracts.Chunker, error) { return fakeRunChunker{}, nil }); err != nil { t.Fatalf("register chunker: %v", err) } } func mustRegisterExtractor(t *testing.T, registry *pipeline.ExtractorRegistry, spec pipeline.ModuleSpec) { t.Helper() - if err := registry.RegisterLegacyRawWithSpec(spec, func() (contracts.LegacyRawExtractor, error) { return nil, nil }); err != nil { + if err := registry.RegisterLegacyRawWithSpec(spec, func() (contracts.LegacyRawExtractor, error) { return fakeRunExtractor{}, nil }); err != nil { t.Fatalf("register extractor: %v", err) } } func mustRegisterMerger(t *testing.T, registry *pipeline.MergerRegistry, spec pipeline.ModuleSpec) { t.Helper() - if err := registry.RegisterLegacyRawWithSpec(spec, func() (contracts.LegacyRawMerger, error) { return nil, nil }); err != nil { + if err := registry.RegisterLegacyRawWithSpec(spec, func() (contracts.LegacyRawMerger, error) { return fakeRunMerger{}, nil }); err != nil { t.Fatalf("register merger: %v", err) } } func mustRegisterNormalizer(t *testing.T, registry *pipeline.NormalizerRegistry, spec pipeline.ModuleSpec) { t.Helper() - if err := registry.RegisterLegacyRawWithSpec(spec, func() (contracts.LegacyRawNormalizer, error) { return nil, nil }); err != nil { + if err := registry.RegisterLegacyRawWithSpec(spec, func() (contracts.LegacyRawNormalizer, error) { return fakeRunNormalizer{}, nil }); err != nil { t.Fatalf("register normalizer: %v", err) } } func mustRegisterOutput(t *testing.T, registry *pipeline.OutputEncoderRegistry, spec pipeline.ModuleSpec) { t.Helper() - if err := registry.RegisterWithSpec(spec, func() (contracts.OutputEncoder, error) { return nil, nil }); err != nil { + if err := registry.RegisterWithSpec(spec, func() (contracts.OutputEncoder, error) { return jsonoutput.New(), nil }); err != nil { t.Fatalf("register output: %v", err) } } diff --git a/internal/framework/pipeline/chunker_registry.go b/internal/framework/pipeline/chunker_registry.go index 3f133b6..3d8a4f5 100644 --- a/internal/framework/pipeline/chunker_registry.go +++ b/internal/framework/pipeline/chunker_registry.go @@ -8,16 +8,19 @@ import ( ) type ChunkerConstructor func() (contracts.Chunker, error) +type ChunkerBuilder func(BuildRequest) (contracts.Chunker, error) type ChunkerRegistry struct { - constructors map[string]ChunkerConstructor - specs map[string]ModuleSpec + builders map[string]ChunkerBuilder + optionValidators map[string]OptionValidator + specs map[string]ModuleSpec } func NewChunkerRegistry() *ChunkerRegistry { return &ChunkerRegistry{ - constructors: make(map[string]ChunkerConstructor), - specs: make(map[string]ModuleSpec), + builders: make(map[string]ChunkerBuilder), + optionValidators: make(map[string]OptionValidator), + specs: make(map[string]ModuleSpec), } } @@ -26,6 +29,15 @@ func (r *ChunkerRegistry) Register(key string, constructor ChunkerConstructor) e } func (r *ChunkerRegistry) RegisterWithSpec(spec ModuleSpec, constructor ChunkerConstructor) error { + if constructor == nil { + return fmt.Errorf("chunker constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return r.RegisterBuilderWithSpec(spec, allowLegacyOptions, func(BuildRequest) (contracts.Chunker, error) { + return constructor() + }) +} + +func (r *ChunkerRegistry) RegisterBuilderWithSpec(spec ModuleSpec, validateOptions OptionValidator, builder ChunkerBuilder) error { if r == nil { return fmt.Errorf("chunker registry must not be nil") } @@ -34,25 +46,36 @@ func (r *ChunkerRegistry) RegisterWithSpec(spec ModuleSpec, constructor ChunkerC if err := validateModuleSpec("chunker", StageChunk, normalizedSpec); err != nil { return err } - if constructor == nil { - return fmt.Errorf("chunker constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("chunker option validator for %q must not be nil", normalizedSpec.Key) } - if _, ok := r.constructors[normalizedSpec.Key]; ok { + if builder == nil { + return fmt.Errorf("chunker builder for %q must not be nil", normalizedSpec.Key) + } + if _, ok := r.builders[normalizedSpec.Key]; ok { return fmt.Errorf("chunker %q is already registered", normalizedSpec.Key) } - if r.constructors == nil { - r.constructors = make(map[string]ChunkerConstructor) + if r.builders == nil { + r.builders = make(map[string]ChunkerBuilder) + } + if r.optionValidators == nil { + r.optionValidators = make(map[string]OptionValidator) } if r.specs == nil { r.specs = make(map[string]ModuleSpec) } - r.constructors[normalizedSpec.Key] = constructor + r.builders[normalizedSpec.Key] = builder + r.optionValidators[normalizedSpec.Key] = validateOptions r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec) return nil } func (r *ChunkerRegistry) Build(key string) (contracts.Chunker, error) { + return r.BuildWithRequest(key, BuildRequest{}) +} + +func (r *ChunkerRegistry) BuildWithRequest(key string, request BuildRequest) (contracts.Chunker, error) { if r == nil { return nil, fmt.Errorf("chunker registry must not be nil") } @@ -62,12 +85,12 @@ func (r *ChunkerRegistry) Build(key string) (contracts.Chunker, error) { return nil, fmt.Errorf("chunker key must not be empty") } - constructor, ok := r.constructors[normalizedKey] + builder, ok := r.builders[normalizedKey] if !ok { return nil, fmt.Errorf("chunker %q is not registered", normalizedKey) } - chunker, err := constructor() + chunker, err := builder(cloneBuildRequest(request)) if err != nil { return nil, fmt.Errorf("build chunker %q: %w", normalizedKey, err) } @@ -81,6 +104,18 @@ func (r *ChunkerRegistry) Build(key string) (contracts.Chunker, error) { return chunker, nil } +func (r *ChunkerRegistry) ValidateOptions(key string, options map[string]any) error { + if r == nil { + return fmt.Errorf("chunker registry must not be nil") + } + normalizedKey := strings.TrimSpace(key) + validator, ok := r.optionValidators[normalizedKey] + if !ok { + return fmt.Errorf("chunker %q is not registered", normalizedKey) + } + return validateRegisteredOptions(validator, options) +} + func (r *ChunkerRegistry) Spec(key string) (ModuleSpec, bool) { if r == nil { return ModuleSpec{}, false @@ -98,5 +133,5 @@ func (r *ChunkerRegistry) RegisteredKeys() []string { return nil } - return sortedRegistryKeys(r.constructors) + return sortedRegistryKeys(r.builders) } diff --git a/internal/framework/pipeline/construction.go b/internal/framework/pipeline/construction.go new file mode 100644 index 0000000..2021ea8 --- /dev/null +++ b/internal/framework/pipeline/construction.go @@ -0,0 +1,67 @@ +package pipeline + +import ( + "fmt" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +// ModuleDependencies contains run-scoped collaborators shared by constructed +// modules. Implementations retain only the dependencies they use. +type ModuleDependencies struct { + LLM contracts.StructuredLLMClient +} + +// BuildRequest contains the stable dependencies and configured options used to +// construct one module or validator for a run. +type BuildRequest struct { + Dependencies ModuleDependencies + Options map[string]any +} + +// OptionValidator validates one module binding without constructing it. +type OptionValidator func(map[string]any) error + +func allowLegacyOptions(map[string]any) error { return nil } + +func validateRegisteredOptions(validator OptionValidator, options map[string]any) error { + if validator == nil { + return fmt.Errorf("option validator must not be nil") + } + return validator(cloneOptions(options)) +} + +func cloneBuildRequest(request BuildRequest) BuildRequest { + return BuildRequest{ + Dependencies: request.Dependencies, + Options: cloneOptions(request.Options), + } +} + +// RejectUnknownOptions provides the common strict-map check used by module- +// owned option decoders. Values remain the implementation's responsibility. +func RejectUnknownOptions(options map[string]any, allowed ...string) error { + known := make(map[string]struct{}, len(allowed)) + for _, key := range allowed { + key = strings.TrimSpace(key) + if key != "" { + known[key] = struct{}{} + } + } + unknown := make([]string, 0) + for key := range options { + if _, ok := known[key]; !ok { + unknown = append(unknown, key) + } + } + if len(unknown) == 0 { + return nil + } + sort.Strings(unknown) + if len(unknown) == 1 { + return fmt.Errorf("unknown option %q", unknown[0]) + } + return fmt.Errorf("unknown options %q", unknown) +} diff --git a/internal/framework/pipeline/extractor_registry.go b/internal/framework/pipeline/extractor_registry.go index eecf519..2c9011c 100644 --- a/internal/framework/pipeline/extractor_registry.go +++ b/internal/framework/pipeline/extractor_registry.go @@ -9,24 +9,28 @@ import ( ) type LegacyRawExtractorConstructor func() (contracts.LegacyRawExtractor, error) +type LegacyRawExtractorBuilder func(BuildRequest) (contracts.LegacyRawExtractor, error) type ExtractorRegistry struct { - legacyConstructors map[string]LegacyRawExtractorConstructor - typedEntries map[string]typedExtractorEntry - specs map[string]ModuleSpec + legacyBuilders map[string]LegacyRawExtractorBuilder + legacyValidators map[string]OptionValidator + typedEntries map[string]typedExtractorEntry + specs map[string]ModuleSpec } type typedExtractorEntry struct { - spec ModuleSpec - valueType reflect.Type - constructor func() (any, error) + spec ModuleSpec + valueType reflect.Type + validateOptions OptionValidator + builder func(BuildRequest) (any, error) } func NewExtractorRegistry() *ExtractorRegistry { return &ExtractorRegistry{ - legacyConstructors: make(map[string]LegacyRawExtractorConstructor), - typedEntries: make(map[string]typedExtractorEntry), - specs: make(map[string]ModuleSpec), + legacyBuilders: make(map[string]LegacyRawExtractorBuilder), + legacyValidators: make(map[string]OptionValidator), + typedEntries: make(map[string]typedExtractorEntry), + specs: make(map[string]ModuleSpec), } } @@ -35,6 +39,15 @@ func (r *ExtractorRegistry) RegisterLegacyRaw(key string, constructor LegacyRawE } func (r *ExtractorRegistry) RegisterLegacyRawWithSpec(spec ModuleSpec, constructor LegacyRawExtractorConstructor) error { + if constructor == nil { + return fmt.Errorf("extractor constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return r.RegisterLegacyRawBuilderWithSpec(spec, allowLegacyOptions, func(BuildRequest) (contracts.LegacyRawExtractor, error) { + return constructor() + }) +} + +func (r *ExtractorRegistry) RegisterLegacyRawBuilderWithSpec(spec ModuleSpec, validateOptions OptionValidator, builder LegacyRawExtractorBuilder) error { if r == nil { return fmt.Errorf("extractor registry must not be nil") } @@ -45,24 +58,40 @@ func (r *ExtractorRegistry) RegisterLegacyRawWithSpec(spec ModuleSpec, construct if normalizedSpec.ArtifactKind != "" { return fmt.Errorf("legacy raw extractor %q must not declare an artifact kind", normalizedSpec.Key) } - if constructor == nil { - return fmt.Errorf("extractor constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("extractor option validator for %q must not be nil", normalizedSpec.Key) + } + if builder == nil { + return fmt.Errorf("extractor builder for %q must not be nil", normalizedSpec.Key) } if _, ok := r.specs[normalizedSpec.Key]; ok { return fmt.Errorf("extractor %q is already registered", normalizedSpec.Key) } - if r.legacyConstructors == nil { - r.legacyConstructors = make(map[string]LegacyRawExtractorConstructor) + if r.legacyBuilders == nil { + r.legacyBuilders = make(map[string]LegacyRawExtractorBuilder) + } + if r.legacyValidators == nil { + r.legacyValidators = make(map[string]OptionValidator) } if r.specs == nil { r.specs = make(map[string]ModuleSpec) } - r.legacyConstructors[normalizedSpec.Key] = constructor + r.legacyBuilders[normalizedSpec.Key] = builder + r.legacyValidators[normalizedSpec.Key] = validateOptions r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec) return nil } func RegisterExtractor[T any](registry *ExtractorRegistry, spec ModuleSpec, constructor func() (contracts.Extractor[T], error)) error { + if constructor == nil { + return fmt.Errorf("extractor constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return RegisterExtractorBuilder(registry, spec, allowLegacyOptions, func(BuildRequest) (contracts.Extractor[T], error) { + return constructor() + }) +} + +func RegisterExtractorBuilder[T any](registry *ExtractorRegistry, spec ModuleSpec, validateOptions OptionValidator, builder func(BuildRequest) (contracts.Extractor[T], error)) error { if registry == nil { return fmt.Errorf("extractor registry must not be nil") } @@ -73,17 +102,21 @@ func RegisterExtractor[T any](registry *ExtractorRegistry, spec ModuleSpec, cons if normalizedSpec.ArtifactKind == "" { return fmt.Errorf("typed extractor %q artifact kind must not be empty", normalizedSpec.Key) } - if constructor == nil { - return fmt.Errorf("extractor constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("extractor option validator for %q must not be nil", normalizedSpec.Key) + } + if builder == nil { + return fmt.Errorf("extractor builder for %q must not be nil", normalizedSpec.Key) } if _, ok := registry.specs[normalizedSpec.Key]; ok { return fmt.Errorf("extractor %q is already registered", normalizedSpec.Key) } entry := typedExtractorEntry{ - spec: cloneModuleSpec(normalizedSpec), - valueType: reflect.TypeFor[T](), - constructor: func() (any, error) { - return constructor() + spec: cloneModuleSpec(normalizedSpec), + valueType: reflect.TypeFor[T](), + validateOptions: validateOptions, + builder: func(request BuildRequest) (any, error) { + return builder(cloneBuildRequest(request)) }, } if registry.typedEntries == nil { @@ -98,6 +131,10 @@ func RegisterExtractor[T any](registry *ExtractorRegistry, spec ModuleSpec, cons } func (r *ExtractorRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawExtractor, error) { + return r.BuildLegacyRawWithRequest(key, BuildRequest{}) +} + +func (r *ExtractorRegistry) BuildLegacyRawWithRequest(key string, request BuildRequest) (contracts.LegacyRawExtractor, error) { if r == nil { return nil, fmt.Errorf("extractor registry must not be nil") } @@ -105,11 +142,11 @@ func (r *ExtractorRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawExtra if normalizedKey == "" { return nil, fmt.Errorf("extractor key must not be empty") } - constructor, ok := r.legacyConstructors[normalizedKey] + builder, ok := r.legacyBuilders[normalizedKey] if !ok { return nil, fmt.Errorf("legacy raw extractor %q is not registered", normalizedKey) } - extractor, err := constructor() + extractor, err := builder(cloneBuildRequest(request)) if err != nil { return nil, fmt.Errorf("build extractor %q: %w", normalizedKey, err) } @@ -122,6 +159,21 @@ func (r *ExtractorRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawExtra return extractor, nil } +func (r *ExtractorRegistry) validateOptions(key string, options map[string]any) error { + if r == nil { + return fmt.Errorf("extractor registry must not be nil") + } + normalizedKey := strings.TrimSpace(key) + if entry, ok := r.typedEntries[normalizedKey]; ok { + return validateRegisteredOptions(entry.validateOptions, options) + } + validator, ok := r.legacyValidators[normalizedKey] + if !ok { + return fmt.Errorf("extractor %q is not registered", normalizedKey) + } + return validateRegisteredOptions(validator, options) +} + func (r *ExtractorRegistry) Spec(key string) (ModuleSpec, bool) { if r == nil { return ModuleSpec{}, false diff --git a/internal/framework/pipeline/input_registry.go b/internal/framework/pipeline/input_registry.go index 9a6d5a7..14a79e7 100644 --- a/internal/framework/pipeline/input_registry.go +++ b/internal/framework/pipeline/input_registry.go @@ -8,16 +8,19 @@ import ( ) type InputAdapterConstructor func() (contracts.InputAdapter, error) +type InputAdapterBuilder func(BuildRequest) (contracts.InputAdapter, error) type InputAdapterRegistry struct { - constructors map[string]InputAdapterConstructor - specs map[string]ModuleSpec + builders map[string]InputAdapterBuilder + optionValidators map[string]OptionValidator + specs map[string]ModuleSpec } func NewInputAdapterRegistry() *InputAdapterRegistry { return &InputAdapterRegistry{ - constructors: make(map[string]InputAdapterConstructor), - specs: make(map[string]ModuleSpec), + builders: make(map[string]InputAdapterBuilder), + optionValidators: make(map[string]OptionValidator), + specs: make(map[string]ModuleSpec), } } @@ -26,6 +29,15 @@ func (r *InputAdapterRegistry) Register(key string, constructor InputAdapterCons } func (r *InputAdapterRegistry) RegisterWithSpec(spec ModuleSpec, constructor InputAdapterConstructor) error { + if constructor == nil { + return fmt.Errorf("input adapter constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return r.RegisterBuilderWithSpec(spec, allowLegacyOptions, func(BuildRequest) (contracts.InputAdapter, error) { + return constructor() + }) +} + +func (r *InputAdapterRegistry) RegisterBuilderWithSpec(spec ModuleSpec, validateOptions OptionValidator, builder InputAdapterBuilder) error { if r == nil { return fmt.Errorf("input adapter registry must not be nil") } @@ -34,25 +46,36 @@ func (r *InputAdapterRegistry) RegisterWithSpec(spec ModuleSpec, constructor Inp if err := validateModuleSpec("input adapter", StageInput, normalizedSpec); err != nil { return err } - if constructor == nil { - return fmt.Errorf("input adapter constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("input adapter option validator for %q must not be nil", normalizedSpec.Key) } - if _, ok := r.constructors[normalizedSpec.Key]; ok { + if builder == nil { + return fmt.Errorf("input adapter builder for %q must not be nil", normalizedSpec.Key) + } + if _, ok := r.builders[normalizedSpec.Key]; ok { return fmt.Errorf("input adapter %q is already registered", normalizedSpec.Key) } - if r.constructors == nil { - r.constructors = make(map[string]InputAdapterConstructor) + if r.builders == nil { + r.builders = make(map[string]InputAdapterBuilder) + } + if r.optionValidators == nil { + r.optionValidators = make(map[string]OptionValidator) } if r.specs == nil { r.specs = make(map[string]ModuleSpec) } - r.constructors[normalizedSpec.Key] = constructor + r.builders[normalizedSpec.Key] = builder + r.optionValidators[normalizedSpec.Key] = validateOptions r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec) return nil } func (r *InputAdapterRegistry) Build(key string) (contracts.InputAdapter, error) { + return r.BuildWithRequest(key, BuildRequest{}) +} + +func (r *InputAdapterRegistry) BuildWithRequest(key string, request BuildRequest) (contracts.InputAdapter, error) { if r == nil { return nil, fmt.Errorf("input adapter registry must not be nil") } @@ -62,12 +85,12 @@ func (r *InputAdapterRegistry) Build(key string) (contracts.InputAdapter, error) return nil, fmt.Errorf("input adapter key must not be empty") } - constructor, ok := r.constructors[normalizedKey] + builder, ok := r.builders[normalizedKey] if !ok { return nil, fmt.Errorf("input adapter %q is not registered", normalizedKey) } - adapter, err := constructor() + adapter, err := builder(cloneBuildRequest(request)) if err != nil { return nil, fmt.Errorf("build input adapter %q: %w", normalizedKey, err) } @@ -81,6 +104,18 @@ func (r *InputAdapterRegistry) Build(key string) (contracts.InputAdapter, error) return adapter, nil } +func (r *InputAdapterRegistry) ValidateOptions(key string, options map[string]any) error { + if r == nil { + return fmt.Errorf("input adapter registry must not be nil") + } + normalizedKey := strings.TrimSpace(key) + validator, ok := r.optionValidators[normalizedKey] + if !ok { + return fmt.Errorf("input adapter %q is not registered", normalizedKey) + } + return validateRegisteredOptions(validator, options) +} + func (r *InputAdapterRegistry) Spec(key string) (ModuleSpec, bool) { if r == nil { return ModuleSpec{}, false @@ -98,5 +133,5 @@ func (r *InputAdapterRegistry) RegisteredKeys() []string { return nil } - return sortedRegistryKeys(r.constructors) + return sortedRegistryKeys(r.builders) } diff --git a/internal/framework/pipeline/merger_registry.go b/internal/framework/pipeline/merger_registry.go index 0b4512b..8435455 100644 --- a/internal/framework/pipeline/merger_registry.go +++ b/internal/framework/pipeline/merger_registry.go @@ -9,6 +9,7 @@ import ( ) type LegacyRawMergerConstructor func() (contracts.LegacyRawMerger, error) +type LegacyRawMergerBuilder func(BuildRequest) (contracts.LegacyRawMerger, error) type artifactVariantKey struct { module string @@ -16,22 +17,25 @@ type artifactVariantKey struct { } type MergerRegistry struct { - legacyConstructors map[string]LegacyRawMergerConstructor - legacySpecs map[string]ModuleSpec - typedEntries map[artifactVariantKey]typedMergerEntry + legacyBuilders map[string]LegacyRawMergerBuilder + legacyValidators map[string]OptionValidator + legacySpecs map[string]ModuleSpec + typedEntries map[artifactVariantKey]typedMergerEntry } type typedMergerEntry struct { - spec ModuleSpec - valueType reflect.Type - constructor func() (any, error) + spec ModuleSpec + valueType reflect.Type + validateOptions OptionValidator + builder func(BuildRequest) (any, error) } func NewMergerRegistry() *MergerRegistry { return &MergerRegistry{ - legacyConstructors: make(map[string]LegacyRawMergerConstructor), - legacySpecs: make(map[string]ModuleSpec), - typedEntries: make(map[artifactVariantKey]typedMergerEntry), + legacyBuilders: make(map[string]LegacyRawMergerBuilder), + legacyValidators: make(map[string]OptionValidator), + legacySpecs: make(map[string]ModuleSpec), + typedEntries: make(map[artifactVariantKey]typedMergerEntry), } } @@ -40,6 +44,15 @@ func (r *MergerRegistry) RegisterLegacyRaw(key string, constructor LegacyRawMerg } func (r *MergerRegistry) RegisterLegacyRawWithSpec(spec ModuleSpec, constructor LegacyRawMergerConstructor) error { + if constructor == nil { + return fmt.Errorf("merger constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return r.RegisterLegacyRawBuilderWithSpec(spec, allowLegacyOptions, func(BuildRequest) (contracts.LegacyRawMerger, error) { + return constructor() + }) +} + +func (r *MergerRegistry) RegisterLegacyRawBuilderWithSpec(spec ModuleSpec, validateOptions OptionValidator, builder LegacyRawMergerBuilder) error { if r == nil { return fmt.Errorf("merger registry must not be nil") } @@ -50,24 +63,40 @@ func (r *MergerRegistry) RegisterLegacyRawWithSpec(spec ModuleSpec, constructor if normalizedSpec.ArtifactKind != "" { return fmt.Errorf("legacy raw merger %q must not declare an artifact kind", normalizedSpec.Key) } - if constructor == nil { - return fmt.Errorf("merger constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("merger option validator for %q must not be nil", normalizedSpec.Key) } - if _, ok := r.legacyConstructors[normalizedSpec.Key]; ok { + if builder == nil { + return fmt.Errorf("merger builder for %q must not be nil", normalizedSpec.Key) + } + if _, ok := r.legacyBuilders[normalizedSpec.Key]; ok { return fmt.Errorf("legacy raw merger %q is already registered", normalizedSpec.Key) } - if r.legacyConstructors == nil { - r.legacyConstructors = make(map[string]LegacyRawMergerConstructor) + if r.legacyBuilders == nil { + r.legacyBuilders = make(map[string]LegacyRawMergerBuilder) + } + if r.legacyValidators == nil { + r.legacyValidators = make(map[string]OptionValidator) } if r.legacySpecs == nil { r.legacySpecs = make(map[string]ModuleSpec) } - r.legacyConstructors[normalizedSpec.Key] = constructor + r.legacyBuilders[normalizedSpec.Key] = builder + r.legacyValidators[normalizedSpec.Key] = validateOptions r.legacySpecs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec) return nil } func RegisterMerger[T any](registry *MergerRegistry, spec ModuleSpec, constructor func() (contracts.Merger[T], error)) error { + if constructor == nil { + return fmt.Errorf("merger constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return RegisterMergerBuilder(registry, spec, allowLegacyOptions, func(BuildRequest) (contracts.Merger[T], error) { + return constructor() + }) +} + +func RegisterMergerBuilder[T any](registry *MergerRegistry, spec ModuleSpec, validateOptions OptionValidator, builder func(BuildRequest) (contracts.Merger[T], error)) error { if registry == nil { return fmt.Errorf("merger registry must not be nil") } @@ -78,8 +107,11 @@ func RegisterMerger[T any](registry *MergerRegistry, spec ModuleSpec, constructo if normalizedSpec.ArtifactKind == "" { return fmt.Errorf("typed merger %q artifact kind must not be empty", normalizedSpec.Key) } - if constructor == nil { - return fmt.Errorf("merger constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("merger option validator for %q must not be nil", normalizedSpec.Key) + } + if builder == nil { + return fmt.Errorf("merger builder for %q must not be nil", normalizedSpec.Key) } key := artifactVariantKey{module: normalizedSpec.Key, kind: normalizedSpec.ArtifactKind} if _, ok := registry.typedEntries[key]; ok { @@ -89,16 +121,21 @@ func RegisterMerger[T any](registry *MergerRegistry, spec ModuleSpec, constructo registry.typedEntries = make(map[artifactVariantKey]typedMergerEntry) } registry.typedEntries[key] = typedMergerEntry{ - spec: cloneModuleSpec(normalizedSpec), - valueType: reflect.TypeFor[T](), - constructor: func() (any, error) { - return constructor() + spec: cloneModuleSpec(normalizedSpec), + valueType: reflect.TypeFor[T](), + validateOptions: validateOptions, + builder: func(request BuildRequest) (any, error) { + return builder(cloneBuildRequest(request)) }, } return nil } func (r *MergerRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawMerger, error) { + return r.BuildLegacyRawWithRequest(key, BuildRequest{}) +} + +func (r *MergerRegistry) BuildLegacyRawWithRequest(key string, request BuildRequest) (contracts.LegacyRawMerger, error) { if r == nil { return nil, fmt.Errorf("merger registry must not be nil") } @@ -106,11 +143,11 @@ func (r *MergerRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawMerger, if normalizedKey == "" { return nil, fmt.Errorf("merger key must not be empty") } - constructor, ok := r.legacyConstructors[normalizedKey] + builder, ok := r.legacyBuilders[normalizedKey] if !ok { return nil, fmt.Errorf("legacy raw merger %q is not registered", normalizedKey) } - merger, err := constructor() + merger, err := builder(cloneBuildRequest(request)) if err != nil { return nil, fmt.Errorf("build merger %q: %w", normalizedKey, err) } @@ -123,6 +160,25 @@ func (r *MergerRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawMerger, return merger, nil } +func (r *MergerRegistry) validateOptions(key string, kind contracts.ArtifactKind, options map[string]any) error { + if r == nil { + return fmt.Errorf("merger registry must not be nil") + } + normalizedKey := strings.TrimSpace(key) + if kind != "" { + entry, ok := r.typedEntry(normalizedKey, kind) + if !ok { + return fmt.Errorf("merger %q variant for artifact kind %q is not registered", normalizedKey, kind) + } + return validateRegisteredOptions(entry.validateOptions, options) + } + validator, ok := r.legacyValidators[normalizedKey] + if !ok { + return fmt.Errorf("legacy raw merger %q is not registered", normalizedKey) + } + return validateRegisteredOptions(validator, options) +} + func (r *MergerRegistry) Spec(key string) (ModuleSpec, bool) { if r == nil { return ModuleSpec{}, false diff --git a/internal/framework/pipeline/normalizer_registry.go b/internal/framework/pipeline/normalizer_registry.go index 05dc9e0..70113c4 100644 --- a/internal/framework/pipeline/normalizer_registry.go +++ b/internal/framework/pipeline/normalizer_registry.go @@ -9,24 +9,28 @@ import ( ) type LegacyRawNormalizerConstructor func() (contracts.LegacyRawNormalizer, error) +type LegacyRawNormalizerBuilder func(BuildRequest) (contracts.LegacyRawNormalizer, error) type NormalizerRegistry struct { - legacyConstructors map[string]LegacyRawNormalizerConstructor - legacySpecs map[string]ModuleSpec - typedEntries map[artifactVariantKey]typedNormalizerEntry + legacyBuilders map[string]LegacyRawNormalizerBuilder + legacyValidators map[string]OptionValidator + legacySpecs map[string]ModuleSpec + typedEntries map[artifactVariantKey]typedNormalizerEntry } type typedNormalizerEntry struct { - spec ModuleSpec - valueType reflect.Type - constructor func() (any, error) + spec ModuleSpec + valueType reflect.Type + validateOptions OptionValidator + builder func(BuildRequest) (any, error) } func NewNormalizerRegistry() *NormalizerRegistry { return &NormalizerRegistry{ - legacyConstructors: make(map[string]LegacyRawNormalizerConstructor), - legacySpecs: make(map[string]ModuleSpec), - typedEntries: make(map[artifactVariantKey]typedNormalizerEntry), + legacyBuilders: make(map[string]LegacyRawNormalizerBuilder), + legacyValidators: make(map[string]OptionValidator), + legacySpecs: make(map[string]ModuleSpec), + typedEntries: make(map[artifactVariantKey]typedNormalizerEntry), } } @@ -35,6 +39,15 @@ func (r *NormalizerRegistry) RegisterLegacyRaw(key string, constructor LegacyRaw } func (r *NormalizerRegistry) RegisterLegacyRawWithSpec(spec ModuleSpec, constructor LegacyRawNormalizerConstructor) error { + if constructor == nil { + return fmt.Errorf("normalizer constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return r.RegisterLegacyRawBuilderWithSpec(spec, allowLegacyOptions, func(BuildRequest) (contracts.LegacyRawNormalizer, error) { + return constructor() + }) +} + +func (r *NormalizerRegistry) RegisterLegacyRawBuilderWithSpec(spec ModuleSpec, validateOptions OptionValidator, builder LegacyRawNormalizerBuilder) error { if r == nil { return fmt.Errorf("normalizer registry must not be nil") } @@ -45,24 +58,40 @@ func (r *NormalizerRegistry) RegisterLegacyRawWithSpec(spec ModuleSpec, construc if normalizedSpec.ArtifactKind != "" { return fmt.Errorf("legacy raw normalizer %q must not declare an artifact kind", normalizedSpec.Key) } - if constructor == nil { - return fmt.Errorf("normalizer constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("normalizer option validator for %q must not be nil", normalizedSpec.Key) } - if _, ok := r.legacyConstructors[normalizedSpec.Key]; ok { + if builder == nil { + return fmt.Errorf("normalizer builder for %q must not be nil", normalizedSpec.Key) + } + if _, ok := r.legacyBuilders[normalizedSpec.Key]; ok { return fmt.Errorf("legacy raw normalizer %q is already registered", normalizedSpec.Key) } - if r.legacyConstructors == nil { - r.legacyConstructors = make(map[string]LegacyRawNormalizerConstructor) + if r.legacyBuilders == nil { + r.legacyBuilders = make(map[string]LegacyRawNormalizerBuilder) + } + if r.legacyValidators == nil { + r.legacyValidators = make(map[string]OptionValidator) } if r.legacySpecs == nil { r.legacySpecs = make(map[string]ModuleSpec) } - r.legacyConstructors[normalizedSpec.Key] = constructor + r.legacyBuilders[normalizedSpec.Key] = builder + r.legacyValidators[normalizedSpec.Key] = validateOptions r.legacySpecs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec) return nil } func RegisterNormalizer[T any](registry *NormalizerRegistry, spec ModuleSpec, constructor func() (contracts.Normalizer[T], error)) error { + if constructor == nil { + return fmt.Errorf("normalizer constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return RegisterNormalizerBuilder(registry, spec, allowLegacyOptions, func(BuildRequest) (contracts.Normalizer[T], error) { + return constructor() + }) +} + +func RegisterNormalizerBuilder[T any](registry *NormalizerRegistry, spec ModuleSpec, validateOptions OptionValidator, builder func(BuildRequest) (contracts.Normalizer[T], error)) error { if registry == nil { return fmt.Errorf("normalizer registry must not be nil") } @@ -73,8 +102,11 @@ func RegisterNormalizer[T any](registry *NormalizerRegistry, spec ModuleSpec, co if normalizedSpec.ArtifactKind == "" { return fmt.Errorf("typed normalizer %q artifact kind must not be empty", normalizedSpec.Key) } - if constructor == nil { - return fmt.Errorf("normalizer constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("normalizer option validator for %q must not be nil", normalizedSpec.Key) + } + if builder == nil { + return fmt.Errorf("normalizer builder for %q must not be nil", normalizedSpec.Key) } key := artifactVariantKey{module: normalizedSpec.Key, kind: normalizedSpec.ArtifactKind} if _, ok := registry.typedEntries[key]; ok { @@ -84,16 +116,21 @@ func RegisterNormalizer[T any](registry *NormalizerRegistry, spec ModuleSpec, co registry.typedEntries = make(map[artifactVariantKey]typedNormalizerEntry) } registry.typedEntries[key] = typedNormalizerEntry{ - spec: cloneModuleSpec(normalizedSpec), - valueType: reflect.TypeFor[T](), - constructor: func() (any, error) { - return constructor() + spec: cloneModuleSpec(normalizedSpec), + valueType: reflect.TypeFor[T](), + validateOptions: validateOptions, + builder: func(request BuildRequest) (any, error) { + return builder(cloneBuildRequest(request)) }, } return nil } func (r *NormalizerRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawNormalizer, error) { + return r.BuildLegacyRawWithRequest(key, BuildRequest{}) +} + +func (r *NormalizerRegistry) BuildLegacyRawWithRequest(key string, request BuildRequest) (contracts.LegacyRawNormalizer, error) { if r == nil { return nil, fmt.Errorf("normalizer registry must not be nil") } @@ -101,11 +138,11 @@ func (r *NormalizerRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawNorm if normalizedKey == "" { return nil, fmt.Errorf("normalizer key must not be empty") } - constructor, ok := r.legacyConstructors[normalizedKey] + builder, ok := r.legacyBuilders[normalizedKey] if !ok { return nil, fmt.Errorf("legacy raw normalizer %q is not registered", normalizedKey) } - normalizer, err := constructor() + normalizer, err := builder(cloneBuildRequest(request)) if err != nil { return nil, fmt.Errorf("build normalizer %q: %w", normalizedKey, err) } @@ -118,6 +155,25 @@ func (r *NormalizerRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawNorm return normalizer, nil } +func (r *NormalizerRegistry) validateOptions(key string, kind contracts.ArtifactKind, options map[string]any) error { + if r == nil { + return fmt.Errorf("normalizer registry must not be nil") + } + normalizedKey := strings.TrimSpace(key) + if kind != "" { + entry, ok := r.typedEntry(normalizedKey, kind) + if !ok { + return fmt.Errorf("normalizer %q variant for artifact kind %q is not registered", normalizedKey, kind) + } + return validateRegisteredOptions(entry.validateOptions, options) + } + validator, ok := r.legacyValidators[normalizedKey] + if !ok { + return fmt.Errorf("legacy raw normalizer %q is not registered", normalizedKey) + } + return validateRegisteredOptions(validator, options) +} + func (r *NormalizerRegistry) Spec(key string) (ModuleSpec, bool) { if r == nil { return ModuleSpec{}, false diff --git a/internal/framework/pipeline/options.go b/internal/framework/pipeline/options.go new file mode 100644 index 0000000..d140a9a --- /dev/null +++ b/internal/framework/pipeline/options.go @@ -0,0 +1,65 @@ +package pipeline + +import "fmt" + +func validateResolvedOptions(resolved ResolvedPipeline, catalog ModuleCatalog) error { + if err := catalog.Inputs.ValidateOptions(resolved.Input.Module, resolved.Input.Options); err != nil { + return moduleOptionsError(resolved.ID, "", StageInput, resolved.Input.Module, err) + } + if err := catalog.Chunkers.ValidateOptions(resolved.Chunk.Module, resolved.Chunk.Options); err != nil { + return moduleOptionsError(resolved.ID, "", StageChunk, resolved.Chunk.Module, err) + } + if err := validateChainOptions(resolved, catalog, StageChunk, "", resolved.Chunk.Module); err != nil { + return err + } + + for _, lane := range resolved.ArtifactLanes { + if err := catalog.Extractors.validateOptions(lane.Extract.Module, lane.Extract.Options); err != nil { + return moduleOptionsError(resolved.ID, lane.ID, StageExtract, lane.Extract.Module, err) + } + if err := validateChainOptions(resolved, catalog, StageExtract, lane.ID, lane.Extract.Module); err != nil { + return err + } + if err := catalog.Mergers.validateOptions(lane.Merge.Module, lane.ArtifactKind, lane.Merge.Options); err != nil { + return moduleOptionsError(resolved.ID, lane.ID, StageMerge, lane.Merge.Module, err) + } + if err := validateChainOptions(resolved, catalog, StageMerge, lane.ID, lane.Merge.Module); err != nil { + return err + } + if err := catalog.Normalizers.validateOptions(lane.Normalize.Module, lane.ArtifactKind, lane.Normalize.Options); err != nil { + return moduleOptionsError(resolved.ID, lane.ID, StageNormalize, lane.Normalize.Module, err) + } + if err := validateChainOptions(resolved, catalog, StageNormalize, lane.ID, lane.Normalize.Module); err != nil { + return err + } + } + + if err := catalog.Outputs.ValidateOptions(resolved.Output.Module, resolved.Output.Options); err != nil { + return moduleOptionsError(resolved.ID, "", StageOutput, resolved.Output.Module, err) + } + return nil +} + +func validateChainOptions(resolved ResolvedPipeline, catalog ModuleCatalog, stage ModuleStage, laneID, moduleKey string) error { + chain := resolvedValidatorChain(stage, laneID, moduleKey, resolved.ValidatorChains) + for _, validator := range chain.Validators { + if err := catalog.Validators.validateOptions(validator); err != nil { + return validatorOptionsError(resolved.ID, laneID, stage, moduleKey, validator.Binding.Module, err) + } + } + return nil +} + +func moduleOptionsError(pipelineID, laneID string, stage ModuleStage, moduleKey string, cause error) error { + if laneID == "" { + return fmt.Errorf("pipeline %q %s module %q options: %w", pipelineID, stage, moduleKey, cause) + } + return fmt.Errorf("pipeline %q lane %q %s module %q options: %w", pipelineID, laneID, stage, moduleKey, cause) +} + +func validatorOptionsError(pipelineID, laneID string, stage ModuleStage, moduleKey, validatorKey string, cause error) error { + if laneID == "" { + return fmt.Errorf("pipeline %q %s module %q validator %q options: %w", pipelineID, stage, moduleKey, validatorKey, cause) + } + return fmt.Errorf("pipeline %q lane %q %s module %q validator %q options: %w", pipelineID, laneID, stage, moduleKey, validatorKey, cause) +} diff --git a/internal/framework/pipeline/output_registry.go b/internal/framework/pipeline/output_registry.go index d63126f..a7ff3cd 100644 --- a/internal/framework/pipeline/output_registry.go +++ b/internal/framework/pipeline/output_registry.go @@ -8,16 +8,19 @@ import ( ) type OutputEncoderConstructor func() (contracts.OutputEncoder, error) +type OutputEncoderBuilder func(BuildRequest) (contracts.OutputEncoder, error) type OutputEncoderRegistry struct { - constructors map[string]OutputEncoderConstructor - specs map[string]ModuleSpec + builders map[string]OutputEncoderBuilder + optionValidators map[string]OptionValidator + specs map[string]ModuleSpec } func NewOutputEncoderRegistry() *OutputEncoderRegistry { return &OutputEncoderRegistry{ - constructors: make(map[string]OutputEncoderConstructor), - specs: make(map[string]ModuleSpec), + builders: make(map[string]OutputEncoderBuilder), + optionValidators: make(map[string]OptionValidator), + specs: make(map[string]ModuleSpec), } } @@ -26,6 +29,15 @@ func (r *OutputEncoderRegistry) Register(key string, constructor OutputEncoderCo } func (r *OutputEncoderRegistry) RegisterWithSpec(spec ModuleSpec, constructor OutputEncoderConstructor) error { + if constructor == nil { + return fmt.Errorf("output encoder constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return r.RegisterBuilderWithSpec(spec, allowLegacyOptions, func(BuildRequest) (contracts.OutputEncoder, error) { + return constructor() + }) +} + +func (r *OutputEncoderRegistry) RegisterBuilderWithSpec(spec ModuleSpec, validateOptions OptionValidator, builder OutputEncoderBuilder) error { if r == nil { return fmt.Errorf("output encoder registry must not be nil") } @@ -34,25 +46,36 @@ func (r *OutputEncoderRegistry) RegisterWithSpec(spec ModuleSpec, constructor Ou if err := validateModuleSpec("output encoder", StageOutput, normalizedSpec); err != nil { return err } - if constructor == nil { - return fmt.Errorf("output encoder constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("output encoder option validator for %q must not be nil", normalizedSpec.Key) } - if _, ok := r.constructors[normalizedSpec.Key]; ok { + if builder == nil { + return fmt.Errorf("output encoder builder for %q must not be nil", normalizedSpec.Key) + } + if _, ok := r.builders[normalizedSpec.Key]; ok { return fmt.Errorf("output encoder %q is already registered", normalizedSpec.Key) } - if r.constructors == nil { - r.constructors = make(map[string]OutputEncoderConstructor) + if r.builders == nil { + r.builders = make(map[string]OutputEncoderBuilder) + } + if r.optionValidators == nil { + r.optionValidators = make(map[string]OptionValidator) } if r.specs == nil { r.specs = make(map[string]ModuleSpec) } - r.constructors[normalizedSpec.Key] = constructor + r.builders[normalizedSpec.Key] = builder + r.optionValidators[normalizedSpec.Key] = validateOptions r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec) return nil } func (r *OutputEncoderRegistry) Build(key string) (contracts.OutputEncoder, error) { + return r.BuildWithRequest(key, BuildRequest{}) +} + +func (r *OutputEncoderRegistry) BuildWithRequest(key string, request BuildRequest) (contracts.OutputEncoder, error) { if r == nil { return nil, fmt.Errorf("output encoder registry must not be nil") } @@ -62,12 +85,12 @@ func (r *OutputEncoderRegistry) Build(key string) (contracts.OutputEncoder, erro return nil, fmt.Errorf("output encoder key must not be empty") } - constructor, ok := r.constructors[normalizedKey] + builder, ok := r.builders[normalizedKey] if !ok { return nil, fmt.Errorf("output encoder %q is not registered", normalizedKey) } - encoder, err := constructor() + encoder, err := builder(cloneBuildRequest(request)) if err != nil { return nil, fmt.Errorf("build output encoder %q: %w", normalizedKey, err) } @@ -81,6 +104,18 @@ func (r *OutputEncoderRegistry) Build(key string) (contracts.OutputEncoder, erro return encoder, nil } +func (r *OutputEncoderRegistry) ValidateOptions(key string, options map[string]any) error { + if r == nil { + return fmt.Errorf("output encoder registry must not be nil") + } + normalizedKey := strings.TrimSpace(key) + validator, ok := r.optionValidators[normalizedKey] + if !ok { + return fmt.Errorf("output encoder %q is not registered", normalizedKey) + } + return validateRegisteredOptions(validator, options) +} + func (r *OutputEncoderRegistry) Spec(key string) (ModuleSpec, bool) { if r == nil { return ModuleSpec{}, false @@ -98,5 +133,5 @@ func (r *OutputEncoderRegistry) RegisteredKeys() []string { return nil } - return sortedRegistryKeys(r.constructors) + return sortedRegistryKeys(r.builders) } diff --git a/internal/framework/pipeline/preparation_test.go b/internal/framework/pipeline/preparation_test.go new file mode 100644 index 0000000..87ead8f --- /dev/null +++ b/internal/framework/pipeline/preparation_test.go @@ -0,0 +1,211 @@ +package pipeline + +import ( + "errors" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +func TestResolvePipelineValidatesModuleAndValidatorOptions(t *testing.T) { + tests := []struct { + name string + mutate func(*PipelineProfile) + want []string + }{ + { + name: "module", + mutate: func(profile *PipelineProfile) { + profile.Input.Options = map[string]any{"surprise": true} + }, + want: []string{`pipeline "construction" input module "input" options`, `unknown option "surprise"`}, + }, + { + name: "validator", + mutate: func(profile *PipelineProfile) { + profile.Chunk.Validators = ValidatorOverride{Set: true, Validators: []ModuleBinding{{Module: "configured", Options: map[string]any{"surprise": true}}}} + }, + want: []string{`pipeline "construction" chunk module "chunk" validator "configured" options`, `unknown option "surprise"`}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + registries, _ := constructionRegistries(t, nil, nil) + profile := constructionProfile() + test.mutate(&profile) + _, err := ResolvePipeline(profile, ResolveOptions{}, registries.catalog()) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want option validation error") + } + for _, want := range test.want { + if !strings.Contains(err.Error(), want) { + t.Fatalf("ResolvePipeline() error = %q, want substring %q", err, want) + } + } + }) + } +} + +func TestPrepareConstructsEverythingInStableOrder(t *testing.T) { + var built []string + registries, _ := constructionRegistries(t, &built, nil) + resolved, err := ResolvePipeline(constructionProfile(), ResolveOptions{}, registries.catalog()) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + prepared, err := Prepare(resolved, registries, ModuleDependencies{}) + if err != nil { + t.Fatalf("Prepare() error = %v, want nil", err) + } + want := []string{"input", "chunk", "validator", "extract", "validator", "merge", "validator", "normalize", "validator", "output"} + if !reflect.DeepEqual(built, want) { + t.Fatalf("construction order = %#v, want %#v", built, want) + } + if prepared.Input.Module != "input" || prepared.Chunk.Module != "chunk" || prepared.Output.Module != "output" || len(prepared.ArtifactLanes) != 1 { + t.Fatalf("PreparedPipeline = %#v, want explicit resolved components", prepared) + } +} + +func TestPrepareFailuresOccurBeforeInputParse(t *testing.T) { + tests := []struct { + name string + deps ModuleDependencies + configure func(*constructionFailure) + want string + wantBuilt []string + }{ + { + name: "missing required llm dependency", + configure: func(failure *constructionFailure) { + failure.requireExtractorLLM = true + }, + want: `lane "artifact" extract module "extract"`, + wantBuilt: []string{"input", "chunk", "validator", "extract"}, + }, + { + name: "late output construction", + configure: func(failure *constructionFailure) { + failure.output = errors.New("output unavailable") + }, + want: `output module "output"`, + wantBuilt: []string{"input", "chunk", "validator", "extract", "validator", "merge", "validator", "normalize", "validator", "output"}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + failure := &constructionFailure{} + test.configure(failure) + var built []string + registries, input := constructionRegistries(t, &built, failure) + resolved, err := ResolvePipeline(constructionProfile(), ResolveOptions{}, registries.catalog()) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + _, err = Prepare(resolved, registries, test.deps) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("Prepare() error = %v, want substring %q", err, test.want) + } + if len(input.requests) != 0 { + t.Fatalf("input Parse calls = %d, want zero", len(input.requests)) + } + if !reflect.DeepEqual(built, test.wantBuilt) { + t.Fatalf("construction order = %#v, want %#v", built, test.wantBuilt) + } + }) + } +} + +type constructionFailure struct { + requireExtractorLLM bool + output error +} + +func constructionProfile() PipelineProfile { + validators := ValidatorOverride{Set: true, Validators: []ModuleBinding{{Module: "configured"}}} + return PipelineProfile{ + ID: "construction", + Input: Binding("input"), + Chunk: ModuleBinding{Module: "chunk", Validators: validators}, + Artifacts: map[string]ArtifactLaneProfile{ + "artifact": { + Extract: ModuleBinding{Module: "extract", Validators: validators}, + Merge: ModuleBinding{Module: "merge", Validators: validators}, + Normalize: ModuleBinding{Module: "normalize", Validators: validators}, + }, + }, + Output: Binding("output"), + } +} + +func constructionRegistries(t *testing.T, built *[]string, failure *constructionFailure) (Registries, *runnerInputAdapter) { + t.Helper() + if built == nil { + built = &[]string{} + } + if failure == nil { + failure = &constructionFailure{} + } + record := func(name string) { *built = append(*built, name) } + strict := func(options map[string]any) error { return RejectUnknownOptions(options, "known") } + modules := defaultRunnerModules() + registries := Registries{ + Inputs: NewInputAdapterRegistry(), Chunkers: NewChunkerRegistry(), ArtifactCodecs: NewArtifactCodecRegistry(), + Extractors: NewExtractorRegistry(), Mergers: NewMergerRegistry(), Normalizers: NewNormalizerRegistry(), + Validators: NewValidatorRegistry(), ValidatorChains: NewValidatorChainRegistry(), Outputs: NewOutputEncoderRegistry(), + } + if err := registries.Inputs.RegisterBuilderWithSpec(defaultModuleSpec("input", StageInput), strict, func(BuildRequest) (contracts.InputAdapter, error) { + record("input") + return modules.input, nil + }); err != nil { + t.Fatal(err) + } + if err := registries.Chunkers.RegisterBuilderWithSpec(defaultModuleSpec("chunk", StageChunk), strict, func(BuildRequest) (contracts.Chunker, error) { + record("chunk") + return modules.chunker, nil + }); err != nil { + t.Fatal(err) + } + if err := registries.Extractors.RegisterLegacyRawBuilderWithSpec(defaultModuleSpec("extract", StageExtract), strict, func(request BuildRequest) (contracts.LegacyRawExtractor, error) { + record("extract") + if failure.requireExtractorLLM && request.Dependencies.LLM == nil { + return nil, errors.New("structured LLM client is required") + } + return &runnerExtractor{key: "extract"}, nil + }); err != nil { + t.Fatal(err) + } + if err := registries.Mergers.RegisterLegacyRawBuilderWithSpec(defaultModuleSpec("merge", StageMerge), strict, func(BuildRequest) (contracts.LegacyRawMerger, error) { + record("merge") + return modules.mergers["merge"], nil + }); err != nil { + t.Fatal(err) + } + if err := registries.Normalizers.RegisterLegacyRawBuilderWithSpec(defaultModuleSpec("normalize", StageNormalize), strict, func(BuildRequest) (contracts.LegacyRawNormalizer, error) { + record("normalize") + return modules.normalizers["normalize"], nil + }); err != nil { + t.Fatal(err) + } + if err := registries.Validators.RegisterLegacyRawBuilderWithSpec(ValidatorSpec{Key: "configured", ExecutionClass: contracts.ExecutionClassDeterministic}, strict, func(BuildRequest) (contracts.LegacyRawValidator, error) { + record("validator") + return modules.validators["configured"], nil + }); err != nil { + t.Fatal(err) + } + if err := registries.Outputs.RegisterBuilderWithSpec(defaultModuleSpec("output", StageOutput), strict, func(BuildRequest) (contracts.OutputEncoder, error) { + record("output") + if failure.output != nil { + return nil, failure.output + } + return modules.output, nil + }); err != nil { + t.Fatal(err) + } + return registries, modules.input +} diff --git a/internal/framework/pipeline/prepare.go b/internal/framework/pipeline/prepare.go new file mode 100644 index 0000000..2359a4a --- /dev/null +++ b/internal/framework/pipeline/prepare.go @@ -0,0 +1,370 @@ +package pipeline + +import ( + "fmt" + "reflect" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +// PreparedPipeline owns the constructed, run-local implementation set for one +// resolved pipeline. Its implementation values are private so execution cannot +// replace or reconfigure them after preparation. +type PreparedPipeline struct { + Input ModuleBinding + Chunk ModuleBinding + ArtifactLanes []PreparedArtifactLane + Output ModuleBinding + + resolved ResolvedPipeline + dependencies ModuleDependencies + input contracts.InputAdapter + chunker contracts.Chunker + chunkValidators preparedValidatorChain + lanes []preparedLaneExecutor + output contracts.OutputEncoder +} + +type PreparedArtifactLane struct { + Resolved ResolvedArtifactLane +} + +type preparedLaneExecutor struct { + resolved ResolvedArtifactLane + legacy *preparedLegacyLane + typed *preparedTypedLane + extractValidators preparedValidatorChain + mergeValidators preparedValidatorChain + normalizeValidators preparedValidatorChain +} + +type preparedLegacyLane struct { + extractor contracts.LegacyRawExtractor + merger contracts.LegacyRawMerger + normalizer contracts.LegacyRawNormalizer +} + +type preparedTypedLane struct { + extractor any + merger any + normalizer any +} + +type preparedValidatorChain struct { + resolved ResolvedValidatorChain + validators []preparedValidator +} + +type preparedValidator struct { + resolved ResolvedValidator + legacy contracts.LegacyRawValidator + typed any + chunk contracts.ChunkValidator + serialized contracts.SerializedValidator +} + +// Prepare validates all configured options and constructs every selected +// module and validator before any operation method can run. +func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDependencies) (*PreparedPipeline, error) { + if err := validateResolvedPipeline(resolved, false); err != nil { + return nil, err + } + if err := validateRegistrySet(resolved, registries); err != nil { + return nil, err + } + stable := cloneResolvedPipeline(resolved) + prepared := &PreparedPipeline{ + Input: cloneModuleBinding(stable.Input), + Chunk: cloneModuleBinding(stable.Chunk), + Output: cloneModuleBinding(stable.Output), + resolved: stable, + dependencies: deps, + } + request := func(binding ModuleBinding) BuildRequest { + return BuildRequest{Dependencies: deps, Options: cloneOptions(binding.Options)} + } + + input, err := registries.Inputs.BuildWithRequest(stable.Input.Module, request(stable.Input)) + if err != nil { + return nil, constructionError(stable.ID, "", StageInput, stable.Input.Module, "", err) + } + prepared.input = input + + chunker, err := registries.Chunkers.BuildWithRequest(stable.Chunk.Module, request(stable.Chunk)) + if err != nil { + return nil, constructionError(stable.ID, "", StageChunk, stable.Chunk.Module, "", err) + } + prepared.chunker = chunker + prepared.chunkValidators, err = prepareValidatorChain(stable, registries, deps, StageChunk, "", stable.Chunk.Module) + if err != nil { + return nil, err + } + + prepared.ArtifactLanes = make([]PreparedArtifactLane, 0, len(stable.ArtifactLanes)) + prepared.lanes = make([]preparedLaneExecutor, 0, len(stable.ArtifactLanes)) + for _, lane := range stable.ArtifactLanes { + executor, err := prepareLane(stable, lane, registries, deps) + if err != nil { + return nil, err + } + prepared.ArtifactLanes = append(prepared.ArtifactLanes, PreparedArtifactLane{Resolved: cloneResolvedArtifactLane(lane)}) + prepared.lanes = append(prepared.lanes, executor) + } + + output, err := registries.Outputs.BuildWithRequest(stable.Output.Module, request(stable.Output)) + if err != nil { + return nil, constructionError(stable.ID, "", StageOutput, stable.Output.Module, "", err) + } + prepared.output = output + return prepared, nil +} + +func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registries Registries, deps ModuleDependencies) (preparedLaneExecutor, error) { + executor := preparedLaneExecutor{resolved: cloneResolvedArtifactLane(lane)} + request := func(binding ModuleBinding) BuildRequest { + return BuildRequest{Dependencies: deps, Options: cloneOptions(binding.Options)} + } + if lane.ArtifactKind == "" { + extractor, err := registries.Extractors.BuildLegacyRawWithRequest(lane.Extract.Module, request(lane.Extract)) + if err != nil { + return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageExtract, lane.Extract.Module, "", err) + } + executor.legacy = &preparedLegacyLane{extractor: extractor} + } else { + entry, ok := registries.Extractors.typedEntry(lane.Extract.Module) + if !ok { + return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageExtract, lane.Extract.Module, "", fmt.Errorf("typed construction entry is not registered")) + } + module, err := buildErasedModule(entry.builder, request(lane.Extract), lane.Extract.Module, "extractor") + if err != nil { + return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageExtract, lane.Extract.Module, "", err) + } + executor.typed = &preparedTypedLane{extractor: module} + } + + var err error + executor.extractValidators, err = prepareValidatorChain(pipeline, registries, deps, StageExtract, lane.ID, lane.Extract.Module) + if err != nil { + return preparedLaneExecutor{}, err + } + + if lane.ArtifactKind == "" { + module, err := registries.Mergers.BuildLegacyRawWithRequest(lane.Merge.Module, request(lane.Merge)) + if err != nil { + return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageMerge, lane.Merge.Module, "", err) + } + executor.legacy.merger = module + } else { + entry, ok := registries.Mergers.typedEntry(lane.Merge.Module, lane.ArtifactKind) + if !ok { + return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageMerge, lane.Merge.Module, "", fmt.Errorf("typed construction entry is not registered")) + } + module, err := buildErasedModule(entry.builder, request(lane.Merge), lane.Merge.Module, "merger") + if err != nil { + return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageMerge, lane.Merge.Module, "", err) + } + executor.typed.merger = module + } + executor.mergeValidators, err = prepareValidatorChain(pipeline, registries, deps, StageMerge, lane.ID, lane.Merge.Module) + if err != nil { + return preparedLaneExecutor{}, err + } + + if lane.ArtifactKind == "" { + module, err := registries.Normalizers.BuildLegacyRawWithRequest(lane.Normalize.Module, request(lane.Normalize)) + if err != nil { + return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageNormalize, lane.Normalize.Module, "", err) + } + executor.legacy.normalizer = module + } else { + entry, ok := registries.Normalizers.typedEntry(lane.Normalize.Module, lane.ArtifactKind) + if !ok { + return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageNormalize, lane.Normalize.Module, "", fmt.Errorf("typed construction entry is not registered")) + } + module, err := buildErasedModule(entry.builder, request(lane.Normalize), lane.Normalize.Module, "normalizer") + if err != nil { + return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageNormalize, lane.Normalize.Module, "", err) + } + executor.typed.normalizer = module + } + executor.normalizeValidators, err = prepareValidatorChain(pipeline, registries, deps, StageNormalize, lane.ID, lane.Normalize.Module) + if err != nil { + return preparedLaneExecutor{}, err + } + return executor, nil +} + +func prepareValidatorChain(pipeline ResolvedPipeline, registries Registries, deps ModuleDependencies, stage ModuleStage, laneID, moduleKey string) (preparedValidatorChain, error) { + resolved := resolvedValidatorChain(stage, laneID, moduleKey, pipeline.ValidatorChains) + prepared := preparedValidatorChain{resolved: resolved} + for _, validator := range resolved.Validators { + request := BuildRequest{Dependencies: deps, Options: cloneOptions(validator.Binding.Options)} + built, err := buildPreparedValidator(registries.Validators, validator, request) + if err != nil { + return preparedValidatorChain{}, constructionError(pipeline.ID, laneID, stage, moduleKey, validator.Binding.Module, err) + } + prepared.validators = append(prepared.validators, built) + } + return prepared, nil +} + +func buildPreparedValidator(registry *ValidatorRegistry, resolved ResolvedValidator, request BuildRequest) (preparedValidator, error) { + prepared := preparedValidator{resolved: resolved} + key := resolved.Binding.Module + var implementation any + var err error + switch resolved.Target { + case ValidatorTargetTyped: + entry, ok := registry.typedEntry(key, resolved.ArtifactKind) + if !ok { + return preparedValidator{}, fmt.Errorf("typed construction entry is not registered") + } + implementation, err = entry.builder(cloneBuildRequest(request)) + prepared.typed = implementation + case ValidatorTargetChunk: + entry, ok := registry.chunkEntry(key) + if !ok { + return preparedValidator{}, fmt.Errorf("chunk construction entry is not registered") + } + prepared.chunk, err = entry.builder(cloneBuildRequest(request)) + implementation = prepared.chunk + case ValidatorTargetSerialized: + entry, ok := registry.serializedEntry(key) + if !ok { + return preparedValidator{}, fmt.Errorf("serialized construction entry is not registered") + } + prepared.serialized, err = entry.builder(cloneBuildRequest(request)) + implementation = prepared.serialized + default: + prepared.legacy, err = registry.BuildLegacyRawWithRequest(key, request) + implementation = prepared.legacy + } + if err != nil { + return preparedValidator{}, err + } + if isNilImplementation(implementation) { + return preparedValidator{}, fmt.Errorf("validator %q builder returned nil", key) + } + identity, ok := implementation.(interface { + Name() string + ExecutionClass() contracts.ExecutionClass + }) + if !ok { + return preparedValidator{}, fmt.Errorf("validator %q builder returned incompatible implementation %T", key, implementation) + } + if identity.Name() != key { + return preparedValidator{}, fmt.Errorf("validator %q returned name %q", key, identity.Name()) + } + if identity.ExecutionClass() != resolved.ExecutionClass { + return preparedValidator{}, fmt.Errorf("validator %q returned execution class %q, want %q", key, identity.ExecutionClass(), resolved.ExecutionClass) + } + return prepared, nil +} + +func buildErasedModule(builder func(BuildRequest) (any, error), request BuildRequest, key, kind string) (any, error) { + implementation, err := builder(cloneBuildRequest(request)) + if err != nil { + return nil, err + } + if isNilImplementation(implementation) { + return nil, fmt.Errorf("%s %q builder returned nil", kind, key) + } + identity, ok := implementation.(interface{ Key() string }) + if !ok { + return nil, fmt.Errorf("%s %q builder returned incompatible implementation %T", kind, key, implementation) + } + if identity.Key() != key { + return nil, fmt.Errorf("%s %q returned key %q", kind, key, identity.Key()) + } + return implementation, nil +} + +func isNilImplementation(value any) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} + +func constructionError(pipelineID, laneID string, stage ModuleStage, moduleKey, validatorKey string, cause error) error { + scope := fmt.Sprintf("pipeline %q %s module %q", pipelineID, stage, moduleKey) + if laneID != "" { + scope = fmt.Sprintf("pipeline %q lane %q %s module %q", pipelineID, laneID, stage, moduleKey) + } + if strings.TrimSpace(validatorKey) != "" { + scope += fmt.Sprintf(" validator %q", validatorKey) + } + return fmt.Errorf("prepare %s: %w", scope, cause) +} + +func (registries Registries) catalog() ModuleCatalog { + return ModuleCatalog{ + Inputs: registries.Inputs, Chunkers: registries.Chunkers, ArtifactCodecs: registries.ArtifactCodecs, + Extractors: registries.Extractors, Mergers: registries.Mergers, Normalizers: registries.Normalizers, + Validators: registries.Validators, ValidatorChains: registries.ValidatorChains, Outputs: registries.Outputs, + } +} + +func validateRegistrySet(resolved ResolvedPipeline, registries Registries) error { + if registries.Inputs == nil { + return fmt.Errorf("input registry must not be nil") + } + if registries.Chunkers == nil { + return fmt.Errorf("chunker registry must not be nil") + } + if registries.Extractors == nil { + return fmt.Errorf("extractor registry must not be nil") + } + if registries.Mergers == nil { + return fmt.Errorf("merger registry must not be nil") + } + if registries.Normalizers == nil { + return fmt.Errorf("normalizer registry must not be nil") + } + if registries.Validators == nil { + for _, chain := range resolved.ValidatorChains { + if len(chain.Validators) > 0 { + return fmt.Errorf("validator registry must not be nil") + } + } + } + if registries.Outputs == nil { + return fmt.Errorf("output encoder registry must not be nil") + } + return nil +} + +func cloneResolvedPipeline(in ResolvedPipeline) ResolvedPipeline { + out := in + out.Input = cloneModuleBinding(in.Input) + out.Chunk = cloneModuleBinding(in.Chunk) + out.Output = cloneModuleBinding(in.Output) + out.ChunkReferences = CloneReferenceTarget(in.ChunkReferences) + out.ValidatorChains = cloneResolvedValidatorChains(in.ValidatorChains) + if len(in.ArtifactLanes) > 0 { + out.ArtifactLanes = make([]ResolvedArtifactLane, len(in.ArtifactLanes)) + for i, lane := range in.ArtifactLanes { + out.ArtifactLanes[i] = cloneResolvedArtifactLane(lane) + } + } + return out +} + +func cloneResolvedArtifactLane(in ResolvedArtifactLane) ResolvedArtifactLane { + out := in + out.Extract = cloneModuleBinding(in.Extract) + out.Merge = cloneModuleBinding(in.Merge) + out.Normalize = cloneModuleBinding(in.Normalize) + out.Validators = cloneModuleBindings(in.Validators) + out.ExtractReferences = CloneReferenceTarget(in.ExtractReferences) + out.MergeReferences = CloneReferenceTarget(in.MergeReferences) + out.NormalizeReferences = CloneReferenceTarget(in.NormalizeReferences) + return out +} diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index 980718a..0084e73 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -249,6 +249,9 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo if missing, ok := outputCapabilities.missing(outputSpec.Requires); ok { return ResolvedPipeline{}, capabilityError(pipelineID, "", StageOutput, resolved.Output.Module, missing) } + if err := validateResolvedOptions(resolved, catalog); err != nil { + return ResolvedPipeline{}, err + } digest, err := resolvedPipelineDigest(resolved) if err != nil { diff --git a/internal/framework/pipeline/registry_integration_test.go b/internal/framework/pipeline/registry_integration_test.go index 6651d1c..f73a385 100644 --- a/internal/framework/pipeline/registry_integration_test.go +++ b/internal/framework/pipeline/registry_integration_test.go @@ -14,8 +14,8 @@ func TestRunnerUsesRegistries(t *testing.T) { var executed []string registries := integrationRegistries(t, &built, &executed) - output, err := New(registries).Run(context.Background(), RunInput{ - Pipeline: integrationPipeline(), + output, err := newPreparedRunner(t, registries).Run(context.Background(), RunInput{ + pipeline: integrationPipeline(), SourceID: "source-1", RawInput: []byte("source text"), }) diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 431f570..0695f72 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -29,20 +29,17 @@ type Registries struct { Outputs *OutputEncoderRegistry } -type Runner struct { - registries Registries -} +type Runner struct{} -func New(registries Registries) *Runner { - return &Runner{registries: registries} +func New() *Runner { + return &Runner{} } type RunInput struct { - Pipeline ResolvedPipeline + Prepared *PreparedPipeline SourceID string Path string RawInput []byte - LLMClient contracts.StructuredLLMClient SessionID string RunID string StartedAt time.Time @@ -52,6 +49,9 @@ type RunInput struct { Checkpoints CheckpointRecorder Checkpoint CheckpointLoader Debug DebugRecorder + + pipeline ResolvedPipeline + llmClient contracts.StructuredLLMClient } type RunOutput struct { @@ -70,9 +70,8 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err if err := validateRunInput(input); err != nil { return output, err } - if err := r.validateRegistries(input.Pipeline); err != nil { - return output, err - } + input.pipeline = input.Prepared.resolved + input.llmClient = input.Prepared.dependencies.LLM output.Manifest = manifestFromPipeline(input) checkpoints := input.Checkpoints @@ -88,27 +87,24 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err debugRecorder = NoopDebugRecorder() } input.Debug = debugRecorder - input.LLMClient = wrapDebugLLMClient(input.LLMClient, debugRecorder) + input.llmClient = wrapDebugLLMClient(input.llmClient, debugRecorder) defer func() { - output.Manifest.LLMProfiles = mergeLLMProfileManifests(input.LLMProfiles, llmProfileManifests(input.LLMClient)) + output.Manifest.LLMProfiles = mergeLLMProfileManifests(input.LLMProfiles, llmProfileManifests(input.llmClient)) }() output.Warnings = append(output.Warnings, cloneWarnings(input.Warnings)...) if err := writeDebugTimed(debugRecorder, "run.json", debugTimedEnvelope{ Stage: "run", StartedAt: startedTime(input.StartedAt), Payload: map[string]any{ - "pipeline_id": input.Pipeline.ID, - "pipeline_digest": input.Pipeline.Digest, + "pipeline_id": input.pipeline.ID, + "pipeline_digest": input.pipeline.Digest, "run_id": output.Manifest.RunID, }, }); err != nil { return failOutput(output), fmt.Errorf("write debug run artifact: %w", err) } - adapter, err := r.registries.Inputs.Build(input.Pipeline.Input.Module) - if err != nil { - return failOutput(output), fmt.Errorf("build input adapter %q: %w", input.Pipeline.Input.Module, err) - } + adapter := input.Prepared.input attachModuleManifestMetadata(&output, "input", adapter) sourceCheckpoint, sourceDecision := checkpointLoader.Source(adapter.Key()) recordCheckpointEvent(&output, checkpointLoader, "source", "", adapter.Key(), sourceDecision) @@ -122,7 +118,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err SourceID: input.SourceID, Path: input.Path, Raw: debugContentEnvelope(input.RawInput, sourceInputMediaType(input.Path), nil, nil), - Options: redactSensitiveMap(input.Pipeline.Input.Options), + Options: redactSensitiveMap(input.pipeline.Input.Options), Metadata: redactSensitiveMap(input.Metadata), }, }); err != nil { @@ -136,8 +132,8 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err SourceID: input.SourceID, Path: input.Path, Raw: input.RawInput, - LLMProfile: input.Pipeline.Input.LLMProfile, - Options: cloneOptions(input.Pipeline.Input.Options), + LLMProfile: input.pipeline.Input.LLMProfile, + Options: cloneOptions(input.pipeline.Input.Options), Metadata: input.Metadata, }) if err != nil { @@ -169,10 +165,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err output.Manifest.Metadata = manifestMetadataWithSessionID(output.Manifest.Metadata, sessionID) output.Manifest.SourceDigests = []string{doc.Digest} - chunker, err := r.registries.Chunkers.Build(input.Pipeline.Chunk.Module) - if err != nil { - return failOutput(output), fmt.Errorf("build chunker %q: %w", input.Pipeline.Chunk.Module, err) - } + chunker := input.Prepared.chunker attachModuleManifestMetadata(&output, "chunker", chunker) var canonicalChunks []source.Chunk var chunkWarnings []contracts.Warning @@ -188,7 +181,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err "decision": chunkDecision, "source": debugSourceDocumentEnvelope(doc), "source_input": debugContentEnvelope(sourceInput.Content, sourceInput.MediaType, nil, nil), - "options": redactSensitiveMap(input.Pipeline.Chunk.Options), + "options": redactSensitiveMap(input.pipeline.Chunk.Options), "metadata": redactSensitiveMap(input.Metadata), }, }); err != nil { @@ -204,7 +197,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err if err := checkpoints.ChunkRunning(chunker.Key(), doc.Digest); err != nil { return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err) } - chunksAccepted, chunkRejection, err = runWithRetry(ctx, input.Pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { + chunksAccepted, chunkRejection, err = runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { attemptStarted := time.Now().UTC() attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) @@ -212,10 +205,10 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err Source: doc, SourceInput: sourceInput.Clone(), SessionID: sessionID, - References: CloneReferenceSet(input.Pipeline.ChunkReferences.ReferenceSet), - LLMClient: input.LLMClient, - LLMProfile: input.Pipeline.Chunk.LLMProfile, - Options: cloneOptions(input.Pipeline.Chunk.Options), + References: CloneReferenceSet(input.pipeline.ChunkReferences.ReferenceSet), + LLMClient: input.llmClient, + LLMProfile: input.pipeline.Chunk.LLMProfile, + Options: cloneOptions(input.pipeline.Chunk.Options), Metadata: input.Metadata, }) if err != nil { @@ -254,7 +247,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err }, llmScope)) return false, nil, err } - validationWarnings, rejection, err := r.validateChunksRaw(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.Pipeline.ChunkReferences.ReferenceSet, input.LLMClient, input.Metadata, input.Pipeline.ValidatorChains, attempt, input.Debug) + validationWarnings, rejection, err := r.validateChunksRaw(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.llmClient, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug) if err != nil || rejection != nil { _ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ Stage: string(StageChunk), @@ -320,7 +313,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err } if chunksAccepted { - for _, lane := range input.Pipeline.ArtifactLanes { + for _, lane := range input.Prepared.lanes { if err := r.runLane(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, canonicalChunks, lane, &output); err != nil { return failOutput(output), err } @@ -335,10 +328,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err populateRawOutputManifest(&output) output.Manifest.CompletedAt = timePtr(time.Now().UTC()) - encoder, err := r.registries.Outputs.Build(input.Pipeline.Output.Module) - if err != nil { - return failOutput(output), fmt.Errorf("build output encoder %q: %w", input.Pipeline.Output.Module, err) - } + encoder := input.Prepared.output attachModuleManifestMetadata(&output, "output", encoder) outputStarted := time.Now().UTC() if err := writeDebugTimed(debugRecorder, "output/input.json", debugTimedEnvelope{ @@ -350,7 +340,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err "normalize_outputs": debugNormalizeOutputEnvelopes(output.NormalizeOutputs), "rejected": debugRejectedOutputEnvelopes(output.Rejected), "warnings": output.Warnings, - "options": redactSensitiveMap(input.Pipeline.Output.Options), + "options": redactSensitiveMap(input.pipeline.Output.Options), "metadata": redactSensitiveMap(input.Metadata), }, }); err != nil { @@ -361,8 +351,8 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err NormalizeOutputs: cloneNormalizeOutputs(output.NormalizeOutputs), Rejected: cloneRejectedOutputs(output.Rejected), Warnings: output.Warnings, - LLMProfile: input.Pipeline.Output.LLMProfile, - Options: cloneOptions(input.Pipeline.Output.Options), + LLMProfile: input.pipeline.Output.LLMProfile, + Options: cloneOptions(input.pipeline.Output.Options), Metadata: input.Metadata, }) output.Warnings = append(output.Warnings, encoded.Warnings...) @@ -389,19 +379,14 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return output, nil } -func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, checkpointLoader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, lane ResolvedArtifactLane, output *RunOutput) error { - extractor, err := r.registries.Extractors.BuildLegacyRaw(lane.Extract.Module) - if err != nil { - return fmt.Errorf("build extractor %q for lane %q: %w", lane.Extract.Module, lane.ID, err) - } - merger, err := r.registries.Mergers.BuildLegacyRaw(lane.Merge.Module) - if err != nil { - return fmt.Errorf("build merger %q for lane %q: %w", lane.Merge.Module, lane.ID, err) - } - normalizer, err := r.registries.Normalizers.BuildLegacyRaw(lane.Normalize.Module) - if err != nil { - return fmt.Errorf("build normalizer %q for lane %q: %w", lane.Normalize.Module, lane.ID, err) +func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, checkpointLoader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, prepared preparedLaneExecutor, output *RunOutput) error { + lane := prepared.resolved + if prepared.legacy == nil { + return fmt.Errorf("resolved pipeline lane %q uses typed artifact kind %q, which the legacy raw runner cannot execute", lane.ID, lane.ArtifactKind) } + extractor := prepared.legacy.extractor + merger := prepared.legacy.merger + normalizer := prepared.legacy.normalizer setLaneManifestMetadata(output, lane.ID, extractor, merger, normalizer) extractOutputs := make([]contracts.ExtractOutput, 0, len(chunks)) @@ -454,7 +439,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), - LLMClient: input.LLMClient, + LLMClient: input.llmClient, LLMProfile: lane.Extract.LLMProfile, Options: cloneOptions(lane.Extract.Options), Metadata: input.Metadata, @@ -489,11 +474,11 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: lane.ExtractReferences.ReferenceSet, - llmClient: input.LLMClient, + llmClient: input.llmClient, schema: extractOutput.Schema, payload: extractOutput.Payload, metadata: input.Metadata, - chains: input.Pipeline.ValidatorChains, + prepared: prepared.extractValidators, attempt: attempt, debug: input.Debug, }) @@ -606,7 +591,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.MergeReferences.ReferenceSet), - LLMClient: input.LLMClient, + LLMClient: input.llmClient, LLMProfile: lane.Merge.LLMProfile, Options: cloneOptions(lane.Merge.Options), Metadata: input.Metadata, @@ -636,12 +621,12 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, - llmClient: input.LLMClient, + llmClient: input.llmClient, schema: mergeOutput.Schema, payload: mergeOutput.Payload, extractOutputs: extractOutputs, metadata: input.Metadata, - chains: input.Pipeline.ValidatorChains, + prepared: prepared.mergeValidators, attempt: attempt, debug: input.Debug, }) @@ -762,7 +747,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), - LLMClient: input.LLMClient, + LLMClient: input.llmClient, LLMProfile: lane.Normalize.LLMProfile, Options: cloneOptions(lane.Normalize.Options), Metadata: input.Metadata, @@ -792,12 +777,12 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, - llmClient: input.LLMClient, + llmClient: input.llmClient, schema: normalizeOutput.Schema, payload: normalizeOutput.Payload, mergeOutput: acceptedMerge, metadata: input.Metadata, - chains: input.Pipeline.ValidatorChains, + prepared: prepared.normalizeValidators, attempt: attempt, debug: input.Debug, }) @@ -899,7 +884,7 @@ type rawValidationTarget struct { extractOutputs []contracts.ExtractOutput mergeOutput contracts.MergeOutput metadata map[string]any - chains []ResolvedValidatorChain + prepared preparedValidatorChain attempt int debug DebugRecorder } @@ -951,7 +936,7 @@ func runWithRetry(ctx context.Context, retries int, run func(attempt int) (bool, return false, lastRejection, nil } -func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []source.Chunk, sourceInput contracts.LLMInputMaterial, sessionID string, references contracts.ReferenceSet, llmClient contracts.StructuredLLMClient, metadata map[string]any, chains []ResolvedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) { +func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []source.Chunk, sourceInput contracts.LLMInputMaterial, sessionID string, references contracts.ReferenceSet, llmClient contracts.StructuredLLMClient, metadata map[string]any, prepared preparedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) { return r.validateRaw(ctx, rawValidationTarget{ stage: StageChunk, moduleKey: moduleKey, @@ -963,26 +948,23 @@ func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocume llmClient: llmClient, chunks: chunks, metadata: metadata, - chains: chains, + prepared: prepared, attempt: attempt, debug: debug, }) } func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([]contracts.Warning, *contracts.RejectedOutput, error) { - chain := resolvedValidatorChain(target.stage, target.laneID, target.moduleKey, target.chains) - if len(chain.Validators) == 0 { + if len(target.prepared.validators) == 0 { return nil, nil, nil } - if r.registries.Validators == nil { - return nil, nil, fmt.Errorf("validator registry must not be nil") - } var warnings []contracts.Warning - for index, validatorBinding := range chain.Validators { - validator, err := r.registries.Validators.BuildLegacyRaw(validatorBinding.Binding.Module) - if err != nil { - return nil, nil, fmt.Errorf("build validator %q: %w", validatorBinding.Binding.Module, err) + for index, preparedValidator := range target.prepared.validators { + validator := preparedValidator.legacy + validatorBinding := preparedValidator.resolved + if validator == nil { + return nil, nil, fmt.Errorf("validator %q is not available on the legacy raw path", validatorBinding.Binding.Module) } request := target.validationRequest(validatorBinding.Binding) started := time.Now().UTC() @@ -1088,52 +1070,37 @@ func resolvedValidatorChain(stage ModuleStage, laneID string, moduleKey string, } } -func (r *Runner) validateRegistries(pipeline ResolvedPipeline) error { - if r.registries.Inputs == nil { - return fmt.Errorf("input registry must not be nil") +func validateRunInput(input RunInput) error { + if input.Prepared == nil { + return fmt.Errorf("prepared pipeline must not be nil") } - if r.registries.Chunkers == nil { - return fmt.Errorf("chunker registry must not be nil") - } - if r.registries.Extractors == nil { - return fmt.Errorf("extractor registry must not be nil") - } - if r.registries.Mergers == nil { - return fmt.Errorf("merger registry must not be nil") - } - if r.registries.Normalizers == nil { - return fmt.Errorf("normalizer registry must not be nil") - } - if r.registries.Outputs == nil { - return fmt.Errorf("output encoder registry must not be nil") - } - return nil + return validateResolvedPipeline(input.Prepared.resolved, true) } -func validateRunInput(input RunInput) error { - if input.Pipeline.ID == "" { +func validateResolvedPipeline(pipeline ResolvedPipeline, rejectTyped bool) error { + if pipeline.ID == "" { return fmt.Errorf("resolved pipeline id must not be empty") } - if input.Pipeline.Digest == "" { + if pipeline.Digest == "" { return fmt.Errorf("resolved pipeline digest must not be empty") } - if input.Pipeline.Input.Module == "" { + if pipeline.Input.Module == "" { return fmt.Errorf("resolved pipeline input module must not be empty") } - if input.Pipeline.Chunk.Module == "" { + if pipeline.Chunk.Module == "" { return fmt.Errorf("resolved pipeline chunk module must not be empty") } - if input.Pipeline.Output.Module == "" { + if pipeline.Output.Module == "" { return fmt.Errorf("resolved pipeline output module must not be empty") } - if len(input.Pipeline.ArtifactLanes) == 0 { + if len(pipeline.ArtifactLanes) == 0 { return fmt.Errorf("resolved pipeline artifact lanes must not be empty") } - for _, lane := range input.Pipeline.ArtifactLanes { + for _, lane := range pipeline.ArtifactLanes { if lane.ID == "" { return fmt.Errorf("resolved pipeline artifact lane id must not be empty") } - if lane.ArtifactKind != "" { + if rejectTyped && lane.ArtifactKind != "" { return fmt.Errorf("resolved pipeline lane %q uses typed artifact kind %q, which the legacy raw runner cannot execute", lane.ID, lane.ArtifactKind) } if lane.Extract.Module == "" { @@ -1169,7 +1136,7 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest { runID = fmt.Sprintf("run-%d", startedAt.UnixNano()) } - pipeline := input.Pipeline + pipeline := input.pipeline manifest := artifacts.RunManifest{ PipelineID: pipeline.ID, PipelineDigest: pipeline.Digest, diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index 00a9f50..ccdee98 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -14,17 +14,17 @@ import ( ) func TestNewAndDataTypes(t *testing.T) { - runner := New(Registries{}) + runner := New() if runner == nil { t.Fatal("New() = nil, want runner") } input := RunInput{ - Pipeline: resolvedPipeline(), + pipeline: resolvedPipeline(), SourceID: "source-1", Path: "input.txt", RawInput: []byte("source text"), - LLMClient: fakeLLMClient{}, + llmClient: fakeLLMClient{}, Metadata: map[string]any{"request": "test"}, } output := RunOutput{ @@ -35,7 +35,7 @@ func TestNewAndDataTypes(t *testing.T) { OutputFiles: []contracts.OutputFile{{Name: "outputs/generic.json", ContentType: "application/json", Bytes: []byte(`{}`)}}, } - if input.Pipeline.ID != "pipeline-1" || input.SourceID != "source-1" { + if input.pipeline.ID != "pipeline-1" || input.SourceID != "source-1" { t.Fatalf("RunInput = %#v, want constructed fields", input) } if output.Manifest.PipelineID != "pipeline-1" || len(output.NormalizeOutputs) != 1 || len(output.Rejected) != 1 || len(output.Warnings) != 1 || len(output.OutputFiles) != 1 { @@ -43,6 +43,26 @@ func TestNewAndDataTypes(t *testing.T) { } } +type preparedRunnerHarness struct { + t *testing.T + registries Registries +} + +func newPreparedRunner(t *testing.T, registries Registries) preparedRunnerHarness { + t.Helper() + return preparedRunnerHarness{t: t, registries: registries} +} + +func (h preparedRunnerHarness) Run(ctx context.Context, input RunInput) (RunOutput, error) { + h.t.Helper() + prepared, err := Prepare(input.pipeline, h.registries, ModuleDependencies{LLM: input.llmClient}) + if err != nil { + return RunOutput{}, err + } + input.Prepared = prepared + return New().Run(ctx, input) +} + func TestRunRejectsInvalidSetup(t *testing.T) { tests := []struct { name string @@ -54,10 +74,15 @@ func TestRunRejectsInvalidSetup(t *testing.T) { run: func() (RunOutput, error) { return (*Runner)(nil).Run(context.Background(), RunInput{}) }, error: "runner must not be nil", }, + { + name: "nil prepared pipeline", + run: func() (RunOutput, error) { return New().Run(context.Background(), RunInput{}) }, + error: "prepared pipeline must not be nil", + }, { name: "empty pipeline id", run: func() (RunOutput, error) { - return New(newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{Pipeline: ResolvedPipeline{Digest: "sha256:pipeline"}}) + return newPreparedRunner(t, newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{pipeline: ResolvedPipeline{Digest: "sha256:pipeline"}}) }, error: "pipeline id", }, @@ -66,7 +91,7 @@ func TestRunRejectsInvalidSetup(t *testing.T) { run: func() (RunOutput, error) { pipeline := resolvedPipeline() pipeline.Digest = "" - return New(newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{Pipeline: pipeline}) + return newPreparedRunner(t, newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{pipeline: pipeline}) }, error: "pipeline digest", }, @@ -75,7 +100,7 @@ func TestRunRejectsInvalidSetup(t *testing.T) { run: func() (RunOutput, error) { pipeline := resolvedPipeline() pipeline.ArtifactLanes = nil - return New(newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{Pipeline: pipeline}) + return newPreparedRunner(t, newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{pipeline: pipeline}) }, error: "artifact lanes", }, @@ -84,7 +109,7 @@ func TestRunRejectsInvalidSetup(t *testing.T) { run: func() (RunOutput, error) { registries := newRunnerRegistries(t, nil) registries.Inputs = nil - return New(registries).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + return newPreparedRunner(t, registries).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) }, error: "input registry", }, @@ -93,7 +118,7 @@ func TestRunRejectsInvalidSetup(t *testing.T) { run: func() (RunOutput, error) { registries := newRunnerRegistries(t, nil) registries.Chunkers = nil - return New(registries).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + return newPreparedRunner(t, registries).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) }, error: "chunker registry", }, @@ -102,7 +127,7 @@ func TestRunRejectsInvalidSetup(t *testing.T) { run: func() (RunOutput, error) { registries := newRunnerRegistries(t, nil) registries.Extractors = nil - return New(registries).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + return newPreparedRunner(t, registries).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) }, error: "extractor registry", }, @@ -111,7 +136,7 @@ func TestRunRejectsInvalidSetup(t *testing.T) { run: func() (RunOutput, error) { registries := newRunnerRegistries(t, nil) registries.Mergers = nil - return New(registries).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + return newPreparedRunner(t, registries).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) }, error: "merger registry", }, @@ -120,7 +145,7 @@ func TestRunRejectsInvalidSetup(t *testing.T) { run: func() (RunOutput, error) { registries := newRunnerRegistries(t, nil) registries.Normalizers = nil - return New(registries).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + return newPreparedRunner(t, registries).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) }, error: "normalizer registry", }, @@ -129,7 +154,7 @@ func TestRunRejectsInvalidSetup(t *testing.T) { run: func() (RunOutput, error) { registries := newRunnerRegistries(t, nil) registries.Outputs = nil - return New(registries).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + return newPreparedRunner(t, registries).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) }, error: "output encoder registry", }, @@ -147,7 +172,7 @@ func TestRunAllowsNilValidatorRegistryWithoutConfiguredValidators(t *testing.T) registries := newRunnerRegistries(t, nil) registries.Validators = nil - _, err := New(registries).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + _, err := newPreparedRunner(t, registries).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -190,9 +215,15 @@ func TestRunRejectsInputBuildParseAndInvalidSourceErrors(t *testing.T) { modules := defaultRunnerModules() test.configure(modules) - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) assertRunError(t, err, test.want) + if test.name == "input build" { + if output.Manifest.PipelineID != "" { + t.Fatalf("PipelineID = %q, want no run manifest for preparation failure", output.Manifest.PipelineID) + } + return + } if output.Manifest.ValidationStatus != "failed" { t.Fatalf("ValidationStatus = %q, want failed", output.Manifest.ValidationStatus) } @@ -236,9 +267,15 @@ func TestRunRejectsChunkerBuildChunkAndEmptyChunkErrors(t *testing.T) { modules := defaultRunnerModules() test.configure(modules) - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) assertRunError(t, err, test.want) + if test.name == "chunker build" { + if output.Manifest.PipelineID != "" { + t.Fatalf("PipelineID = %q, want no run manifest for preparation failure", output.Manifest.PipelineID) + } + return + } if output.Manifest.ValidationStatus != "failed" { t.Fatalf("ValidationStatus = %q, want failed", output.Manifest.ValidationStatus) } @@ -364,7 +401,7 @@ func TestRunRejectsInvalidChunks(t *testing.T) { modules := defaultRunnerModules() modules.chunker.chunks = test.chunks - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) assertRunError(t, err, test.want) if output.Manifest.ValidationStatus != "failed" { @@ -384,7 +421,7 @@ func TestRunAllowsPartialCoverageAndOverlappingChunks(t *testing.T) { chunkWithUnits("chunk-1", "source-1", 1, unitWithID("u2")), } - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -419,7 +456,7 @@ func TestRunCanonicalizesChunkUnitsBeforeExtraction(t *testing.T) { }, } - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -478,7 +515,7 @@ func TestRunPreservesChunkMetadataDuringCanonicalization(t *testing.T) { }, } - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -508,9 +545,9 @@ func TestRunExecutesChunksAndPassesChunkAndLLMClient(t *testing.T) { modules := defaultRunnerModules() llmClient := fakeLLMClient{} - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), - LLMClient: llmClient, + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), + llmClient: llmClient, Metadata: map[string]any{"request": "test"}, }) if err != nil { @@ -546,8 +583,8 @@ func TestRunPassesSourceInputAndSessionIDToPromptCapableStages(t *testing.T) { modules := defaultRunnerModules() rawInput := []byte("{\"source\":\"exact bytes\"}") - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), Path: "session.json", RawInput: rawInput, SessionID: " explicit-session ", @@ -619,8 +656,8 @@ func TestRunPassesSourceInputAndSessionIDToPromptCapableStages(t *testing.T) { func TestRunDefaultsSessionIDFromParsedSourceDocumentID(t *testing.T) { modules := defaultRunnerModules() - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), Path: "notes.unknown", RawInput: []byte("notes"), }) @@ -643,8 +680,8 @@ func TestRunPassesInputRequestFields(t *testing.T) { modules := defaultRunnerModules() metadata := map[string]any{"request": "test"} - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), SourceID: "source-1", Path: "input.txt", RawInput: []byte("source text"), @@ -676,7 +713,7 @@ func TestRunPassesModuleBindingConfigToStageRequests(t *testing.T) { 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"}} - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -724,7 +761,7 @@ func TestRunPassesLaneReferencesToExtractorRequests(t *testing.T) { pipeline := resolvedPipeline() pipeline.ArtifactLanes[0].ExtractReferences.ReferenceSet = testReferenceSet("roster", "reference text") - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -745,7 +782,7 @@ func TestRunPassesMergeReferencesToMergerRequest(t *testing.T) { pipeline := resolvedPipeline() pipeline.ArtifactLanes[0].MergeReferences.ReferenceSet = testReferenceSet("merge_notes", "merge reference text") - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -766,7 +803,7 @@ func TestRunPassesChunkReferencesToChunkerRequest(t *testing.T) { pipeline := resolvedPipeline() pipeline.ChunkReferences.ReferenceSet = testReferenceSet("scene_guide", "chunk reference text") - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -787,7 +824,7 @@ func TestRunPassesNormalizeReferencesToNormalizerRequest(t *testing.T) { pipeline := resolvedPipeline() pipeline.ArtifactLanes[0].NormalizeReferences.ReferenceSet = testReferenceSet("normalization_notes", "normalize reference text") - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -829,11 +866,11 @@ func TestRunPassesValidationRequestContextToValidators(t *testing.T) { rawInput := []byte("{\"source\":\"exact bytes\"}") llmClient := fakeLLMClient{} - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: pipeline, + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: pipeline, Path: "session.json", RawInput: rawInput, - LLMClient: llmClient, + llmClient: llmClient, SessionID: "session-123", Metadata: map[string]any{"request": "test"}, }) @@ -911,7 +948,7 @@ func TestRunPassesValidationRequestContextToValidators(t *testing.T) { } func TestRunAllowsNilLLMClientWhenModulesDoNotUseIt(t *testing.T) { - _, err := New(newRunnerRegistries(t, defaultRunnerModules())).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + _, err := newPreparedRunner(t, newRunnerRegistries(t, defaultRunnerModules())).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil with nil LLM client when modules do not use it", err) } @@ -921,8 +958,8 @@ func TestRunIncludesInputWarnings(t *testing.T) { modules := defaultRunnerModules() warning := contracts.Warning{Scope: "reference", ReasonCode: "empty_reference", Message: "empty reference"} - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), Warnings: []contracts.Warning{warning}, }) if err != nil { @@ -950,7 +987,7 @@ func TestRunRecordsTopLevelModuleMetadataForSingletonModules(t *testing.T) { "output_profile": "output-metadata", } - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -977,7 +1014,7 @@ func TestRunRecordsTopLevelModuleMetadataForSingletonModules(t *testing.T) { func TestRunPassesPerChunkRawOutputsToMergeAndNormalize(t *testing.T) { modules := defaultRunnerModules() - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1015,7 +1052,7 @@ func TestRunPassesChunkContentAndMediaTypeToExtractors(t *testing.T) { sourceChunkWithContent("chunk-0", 0, []byte(`{"chunk":0}`), "application/vnd.test+json"), } - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1035,8 +1072,8 @@ func TestRunPassesChunkContentAndMediaTypeToExtractors(t *testing.T) { func TestRunDoesNotPassCheckpointPathsToModules(t *testing.T) { modules := defaultRunnerModules() - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), Checkpoints: NoopCheckpointRecorder(), }) if err != nil { @@ -1128,8 +1165,8 @@ func TestRunReusesCheckpointedWorkflowOutputs(t *testing.T) { }, } - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), Checkpoint: loader, }) if err != nil { @@ -1181,8 +1218,8 @@ func TestRunPreservesCheckpointedExtractRejections(t *testing.T) { reuse: map[string]bool{"extract": true}, } - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), Checkpoint: loader, }) if err != nil { @@ -1208,7 +1245,7 @@ func TestRunOmitsRejectedExtractOutputsFromMerge(t *testing.T) { pipeline := resolvedPipeline() setResolvedValidatorChain(t, &pipeline, StageExtract, "alpha", "extract-alpha", resolvedValidatorForTest(validator)) - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1232,7 +1269,7 @@ func TestRunOmitsLaneWithNoAcceptedExtractOutputs(t *testing.T) { pipeline := resolvedPipeline() setResolvedValidatorChain(t, &pipeline, StageExtract, "alpha", "extract-alpha", resolvedValidatorForTest(validator)) - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1258,7 +1295,7 @@ func TestRunRejectedMergePreventsNormalizeForLane(t *testing.T) { pipeline := resolvedPipeline() setResolvedValidatorChain(t, &pipeline, StageMerge, "alpha", "merge", resolvedValidatorForTest(validator)) - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1281,7 +1318,7 @@ func TestRunRejectedNormalizePreventsOutputForLane(t *testing.T) { pipeline := resolvedPipeline() setResolvedValidatorChain(t, &pipeline, StageNormalize, "alpha", "normalize", resolvedValidatorForTest(validator)) - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1304,7 +1341,7 @@ func TestRunRetriesSameModuleInputAfterFrameworkError(t *testing.T) { pipeline := resolvedPipeline() pipeline.ArtifactLanes[0].Extract.Retries = 1 - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1329,7 +1366,7 @@ func TestRunRetriesSameModuleInputAfterValidatorRejection(t *testing.T) { setResolvedValidatorChain(t, &pipeline, StageExtract, "alpha", "extract-alpha", resolvedValidatorForTest(validator)) pipeline.ArtifactLanes[0].Extract.Retries = 1 - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1353,9 +1390,9 @@ func TestRunDebugFailedChunkAttemptReferencesScopedLLMOutput(t *testing.T) { modules.chunker.err = errors.New("malformed structured output") recorder := newMemoryDebugRecorder() - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), - LLMClient: debugResponseLLMClient{content: []byte(`{"raw":true}`), profileID: "debug-profile"}, + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), + llmClient: debugResponseLLMClient{content: []byte(`{"raw":true}`), profileID: "debug-profile"}, Debug: recorder, }) if err == nil || !strings.Contains(err.Error(), "malformed structured output") { @@ -1412,9 +1449,9 @@ func TestRunDebugWritesNonJSONLLMResponseContentAsText(t *testing.T) { modules.chunker.err = errors.New("malformed structured output") recorder := newMemoryDebugRecorder() - _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), - LLMClient: debugResponseLLMClient{content: []byte("plain text response"), profileID: "debug-profile"}, + _, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), + llmClient: debugResponseLLMClient{content: []byte("plain text response"), profileID: "debug-profile"}, Debug: recorder, }) if err == nil || !strings.Contains(err.Error(), "malformed structured output") { @@ -1450,7 +1487,7 @@ func TestRunStopsRetryAfterConfiguredAttemptsAndRecordsAttemptCount(t *testing.T setResolvedValidatorChain(t, &pipeline, StageExtract, "alpha", "extract-alpha", resolvedValidatorForTest(validator)) pipeline.ArtifactLanes[0].Extract.Retries = 1 - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1480,7 +1517,7 @@ func TestRunContextCancellationStopsRetries(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - output, err := New(newRunnerRegistries(t, modules)).Run(ctx, RunInput{Pipeline: pipeline}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(ctx, RunInput{pipeline: pipeline}) if !errors.Is(err, context.Canceled) { t.Fatalf("Run() error = %v, want context.Canceled", err) @@ -1494,8 +1531,8 @@ func TestRunContextCancellationStopsRetries(t *testing.T) { } func TestRunRejectsConfiguredValidators(t *testing.T) { - _, err := New(newRunnerRegistries(t, defaultRunnerModules())).Run(context.Background(), RunInput{ - Pipeline: resolvedPipelineWithValidators("configured", "second-validator"), + _, err := newPreparedRunner(t, newRunnerRegistries(t, defaultRunnerModules())).Run(context.Background(), RunInput{ + pipeline: resolvedPipelineWithValidators("configured", "second-validator"), }) assertRunError(t, err, "extract.validators") } @@ -1508,7 +1545,7 @@ func TestRunCollectsStageWarnings(t *testing.T) { modules.normalizers["normalize"].warnings = []contracts.Warning{{ReasonCode: "normalize-warning", Message: "normalize warning"}} modules.output.warnings = []contracts.Warning{{ReasonCode: "output-warning", Message: "output warning"}} - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1530,7 +1567,7 @@ func TestRunCollectsChunkValidatorWarnings(t *testing.T) { pipeline := resolvedPipeline() setResolvedValidatorChain(t, &pipeline, StageChunk, "", "chunk", resolvedValidatorForTest(validator)) - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1543,7 +1580,7 @@ func TestRunCollectsChunkValidatorWarnings(t *testing.T) { func TestRunOutputEncoderReceivesManifestAndRawOutputs(t *testing.T) { modules := defaultRunnerModules() - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1595,7 +1632,7 @@ func TestRunRejectsUnsafeOutputFileNames(t *testing.T) { {Name: test.fileName, ContentType: "application/json", Bytes: []byte(`{}`)}, } - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) assertRunError(t, err, "output file name") if output.Manifest.ValidationStatus != "failed" { @@ -1609,7 +1646,7 @@ func TestRunReturnsFailedManifestWhenOutputEncoderFails(t *testing.T) { modules := defaultRunnerModules() modules.output.err = errors.New("encode failed") - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) assertRunError(t, err, "encode failed") if output.Manifest.ValidationStatus != "failed" { @@ -1698,7 +1735,7 @@ func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) { }, } - output, err := New(newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{Pipeline: resolved}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{pipeline: resolved}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1769,8 +1806,8 @@ func TestRunManifestIncludesRunTimingAndLLMProfiles(t *testing.T) { {ID: "default", Provider: "scriptorium", Model: "model-a"}, } - output, err := New(newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), + output, err := newPreparedRunner(t, newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), RunID: "run-test", StartedAt: startedAt, LLMProfiles: profiles, @@ -1795,9 +1832,9 @@ func TestRunManifestIncludesRunTimingAndLLMProfiles(t *testing.T) { } func TestRunManifestIncludesProfilesReportedByLLMClient(t *testing.T) { - output, err := New(newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{ - Pipeline: resolvedPipeline(), - LLMClient: manifestReportingLLMClient{profiles: []artifacts.LLMProfileManifest{ + output, err := newPreparedRunner(t, newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{ + pipeline: resolvedPipeline(), + llmClient: manifestReportingLLMClient{profiles: []artifacts.LLMProfileManifest{ {ID: "profile-b", Provider: "scriptorium", Model: "model-b"}, {ID: "profile-a", Provider: "scriptorium", Model: "model-a"}, {ID: "profile-b", Provider: "scriptorium", Model: "model-b"}, @@ -1817,7 +1854,7 @@ func TestRunManifestIncludesProfilesReportedByLLMClient(t *testing.T) { } func TestRunManifestGeneratesRunIDAndTimestamps(t *testing.T) { - output, err := New(newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1840,7 +1877,7 @@ func TestRunManifestIncludesExtractorMetadata(t *testing.T) { "response_schema_name": "test_schema", } - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: resolvedPipeline()}) if err != nil { t.Fatalf("Run() error = %v, want nil", err) } @@ -1871,7 +1908,7 @@ func TestRunReturnsPartialOutputWhenLaterLaneFails(t *testing.T) { Normalize: Binding("normalize"), }) - output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline}) + output, err := newPreparedRunner(t, newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{pipeline: pipeline}) assertRunError(t, err, "extract failed") if output.Manifest.ValidationStatus != "failed" { diff --git a/internal/framework/pipeline/typed_resolution_test.go b/internal/framework/pipeline/typed_resolution_test.go index 3d7485f..ec5c731 100644 --- a/internal/framework/pipeline/typed_resolution_test.go +++ b/internal/framework/pipeline/typed_resolution_test.go @@ -115,6 +115,21 @@ func TestResolveTypedHeterogeneousLanes(t *testing.T) { } } +func TestPrepareConstructsHeterogeneousTypedLanes(t *testing.T) { + catalog := typedResolutionCatalog(t, completeTypedCatalogOptions()) + resolved, err := ResolvePipeline(typedResolutionProfile(), ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + prepared, err := Prepare(resolved, registriesFromModuleCatalog(catalog), ModuleDependencies{}) + if err != nil { + t.Fatalf("Prepare() error = %v, want nil", err) + } + if len(prepared.ArtifactLanes) != 2 || prepared.lanes[0].typed == nil || prepared.lanes[1].typed == nil { + t.Fatalf("PreparedPipeline lanes = %#v, want two typed executors", prepared.ArtifactLanes) + } +} + func TestResolveTypedLaneRejectsIncompatibleComposition(t *testing.T) { tests := []struct { name string @@ -263,17 +278,31 @@ func typedResolutionCatalog(t *testing.T, options typedCatalogOptions) ModuleCat func mustRegisterTypedTestBase(t *testing.T, catalog ModuleCatalog) { t.Helper() - if err := catalog.Inputs.RegisterWithSpec(ModuleSpec{Key: "typed/input", Stage: StageInput}, func() (contracts.InputAdapter, error) { return nil, nil }); err != nil { + if err := catalog.Inputs.RegisterWithSpec(ModuleSpec{Key: "typed/input", Stage: StageInput}, func() (contracts.InputAdapter, error) { + return &runnerInputAdapter{key: "typed/input", doc: validSourceDocument()}, nil + }); err != nil { t.Fatalf("register input: %v", err) } - if err := catalog.Chunkers.RegisterWithSpec(ModuleSpec{Key: "typed/chunk", Stage: StageChunk}, func() (contracts.Chunker, error) { return nil, nil }); err != nil { + if err := catalog.Chunkers.RegisterWithSpec(ModuleSpec{Key: "typed/chunk", Stage: StageChunk}, func() (contracts.Chunker, error) { + return &runnerChunker{key: "typed/chunk"}, nil + }); err != nil { t.Fatalf("register chunker: %v", err) } - if err := catalog.Outputs.RegisterWithSpec(ModuleSpec{Key: "typed/output", Stage: StageOutput}, func() (contracts.OutputEncoder, error) { return nil, nil }); err != nil { + if err := catalog.Outputs.RegisterWithSpec(ModuleSpec{Key: "typed/output", Stage: StageOutput}, func() (contracts.OutputEncoder, error) { + return &runnerOutputEncoder{key: "typed/output"}, nil + }); err != nil { t.Fatalf("register output: %v", err) } } +func registriesFromModuleCatalog(catalog ModuleCatalog) Registries { + return Registries{ + Inputs: catalog.Inputs, Chunkers: catalog.Chunkers, ArtifactCodecs: catalog.ArtifactCodecs, + Extractors: catalog.Extractors, Mergers: catalog.Mergers, Normalizers: catalog.Normalizers, + Validators: catalog.Validators, ValidatorChains: catalog.ValidatorChains, Outputs: catalog.Outputs, + } +} + func mustRegisterArtifactCodec[T any](t *testing.T, registry *ArtifactCodecRegistry, codec contracts.ArtifactCodec[T]) { t.Helper() if err := RegisterArtifactCodec(registry, codec); err != nil { diff --git a/internal/framework/pipeline/validator_registry.go b/internal/framework/pipeline/validator_registry.go index 7507ef4..4409a22 100644 --- a/internal/framework/pipeline/validator_registry.go +++ b/internal/framework/pipeline/validator_registry.go @@ -10,6 +10,7 @@ import ( ) type LegacyRawValidatorConstructor func() (contracts.LegacyRawValidator, error) +type LegacyRawValidatorBuilder func(BuildRequest) (contracts.LegacyRawValidator, error) type ValidatorSpec struct { Key string `json:"key"` @@ -32,37 +33,42 @@ const ( ) type ValidatorRegistry struct { - legacyConstructors map[string]LegacyRawValidatorConstructor - legacySpecs map[string]ValidatorSpec - typedEntries map[artifactVariantKey]typedValidatorEntry - chunkEntries map[string]chunkValidatorEntry - serializedEntries map[string]serializedValidatorEntry + legacyBuilders map[string]LegacyRawValidatorBuilder + legacyValidators map[string]OptionValidator + legacySpecs map[string]ValidatorSpec + typedEntries map[artifactVariantKey]typedValidatorEntry + chunkEntries map[string]chunkValidatorEntry + serializedEntries map[string]serializedValidatorEntry } type typedValidatorEntry struct { - spec ValidatorSpec - kind contracts.ArtifactKind - valueType reflect.Type - constructor func() (any, error) + spec ValidatorSpec + kind contracts.ArtifactKind + valueType reflect.Type + validateOptions OptionValidator + builder func(BuildRequest) (any, error) } type chunkValidatorEntry struct { - spec ValidatorSpec - constructor func() (contracts.ChunkValidator, error) + spec ValidatorSpec + validateOptions OptionValidator + builder func(BuildRequest) (contracts.ChunkValidator, error) } type serializedValidatorEntry struct { - spec SerializedValidatorSpec - constructor func() (contracts.SerializedValidator, error) + spec SerializedValidatorSpec + validateOptions OptionValidator + builder func(BuildRequest) (contracts.SerializedValidator, error) } func NewValidatorRegistry() *ValidatorRegistry { return &ValidatorRegistry{ - legacyConstructors: make(map[string]LegacyRawValidatorConstructor), - legacySpecs: make(map[string]ValidatorSpec), - typedEntries: make(map[artifactVariantKey]typedValidatorEntry), - chunkEntries: make(map[string]chunkValidatorEntry), - serializedEntries: make(map[string]serializedValidatorEntry), + legacyBuilders: make(map[string]LegacyRawValidatorBuilder), + legacyValidators: make(map[string]OptionValidator), + legacySpecs: make(map[string]ValidatorSpec), + typedEntries: make(map[artifactVariantKey]typedValidatorEntry), + chunkEntries: make(map[string]chunkValidatorEntry), + serializedEntries: make(map[string]serializedValidatorEntry), } } @@ -71,6 +77,15 @@ func (r *ValidatorRegistry) RegisterLegacyRaw(key string, constructor LegacyRawV } func (r *ValidatorRegistry) RegisterLegacyRawWithSpec(spec ValidatorSpec, constructor LegacyRawValidatorConstructor) error { + if constructor == nil { + return fmt.Errorf("validator constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return r.RegisterLegacyRawBuilderWithSpec(spec, allowLegacyOptions, func(BuildRequest) (contracts.LegacyRawValidator, error) { + return constructor() + }) +} + +func (r *ValidatorRegistry) RegisterLegacyRawBuilderWithSpec(spec ValidatorSpec, validateOptions OptionValidator, builder LegacyRawValidatorBuilder) error { if r == nil { return fmt.Errorf("validator registry must not be nil") } @@ -78,24 +93,40 @@ func (r *ValidatorRegistry) RegisterLegacyRawWithSpec(spec ValidatorSpec, constr if err != nil { return err } - if constructor == nil { - return fmt.Errorf("validator constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("validator option validator for %q must not be nil", normalizedSpec.Key) } - if _, ok := r.legacyConstructors[normalizedSpec.Key]; ok { + if builder == nil { + return fmt.Errorf("validator builder for %q must not be nil", normalizedSpec.Key) + } + if _, ok := r.legacyBuilders[normalizedSpec.Key]; ok { return fmt.Errorf("legacy raw validator %q is already registered", normalizedSpec.Key) } - if r.legacyConstructors == nil { - r.legacyConstructors = make(map[string]LegacyRawValidatorConstructor) + if r.legacyBuilders == nil { + r.legacyBuilders = make(map[string]LegacyRawValidatorBuilder) + } + if r.legacyValidators == nil { + r.legacyValidators = make(map[string]OptionValidator) } if r.legacySpecs == nil { r.legacySpecs = make(map[string]ValidatorSpec) } - r.legacyConstructors[normalizedSpec.Key] = constructor + r.legacyBuilders[normalizedSpec.Key] = builder + r.legacyValidators[normalizedSpec.Key] = validateOptions r.legacySpecs[normalizedSpec.Key] = normalizedSpec return nil } func RegisterTypedValidator[T any](registry *ValidatorRegistry, kind contracts.ArtifactKind, spec ValidatorSpec, constructor func() (contracts.TypedValidator[T], error)) error { + if constructor == nil { + return fmt.Errorf("validator constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return RegisterTypedValidatorBuilder(registry, kind, spec, allowLegacyOptions, func(BuildRequest) (contracts.TypedValidator[T], error) { + return constructor() + }) +} + +func RegisterTypedValidatorBuilder[T any](registry *ValidatorRegistry, kind contracts.ArtifactKind, spec ValidatorSpec, validateOptions OptionValidator, builder func(BuildRequest) (contracts.TypedValidator[T], error)) error { if registry == nil { return fmt.Errorf("validator registry must not be nil") } @@ -107,8 +138,11 @@ func RegisterTypedValidator[T any](registry *ValidatorRegistry, kind contracts.A if kind == "" { return fmt.Errorf("typed validator %q artifact kind must not be empty", normalizedSpec.Key) } - if constructor == nil { - return fmt.Errorf("validator constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("validator option validator for %q must not be nil", normalizedSpec.Key) + } + if builder == nil { + return fmt.Errorf("validator builder for %q must not be nil", normalizedSpec.Key) } key := artifactVariantKey{module: normalizedSpec.Key, kind: kind} if _, ok := registry.typedEntries[key]; ok { @@ -118,17 +152,27 @@ func RegisterTypedValidator[T any](registry *ValidatorRegistry, kind contracts.A registry.typedEntries = make(map[artifactVariantKey]typedValidatorEntry) } registry.typedEntries[key] = typedValidatorEntry{ - spec: normalizedSpec, - kind: kind, - valueType: reflect.TypeFor[T](), - constructor: func() (any, error) { - return constructor() + spec: normalizedSpec, + kind: kind, + valueType: reflect.TypeFor[T](), + validateOptions: validateOptions, + builder: func(request BuildRequest) (any, error) { + return builder(cloneBuildRequest(request)) }, } return nil } func RegisterChunkValidator(registry *ValidatorRegistry, spec ValidatorSpec, constructor func() (contracts.ChunkValidator, error)) error { + if constructor == nil { + return fmt.Errorf("validator constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return RegisterChunkValidatorBuilder(registry, spec, allowLegacyOptions, func(BuildRequest) (contracts.ChunkValidator, error) { + return constructor() + }) +} + +func RegisterChunkValidatorBuilder(registry *ValidatorRegistry, spec ValidatorSpec, validateOptions OptionValidator, builder func(BuildRequest) (contracts.ChunkValidator, error)) error { if registry == nil { return fmt.Errorf("validator registry must not be nil") } @@ -136,8 +180,11 @@ func RegisterChunkValidator(registry *ValidatorRegistry, spec ValidatorSpec, con if err != nil { return err } - if constructor == nil { - return fmt.Errorf("validator constructor for %q must not be nil", normalizedSpec.Key) + if validateOptions == nil { + return fmt.Errorf("validator option validator for %q must not be nil", normalizedSpec.Key) + } + if builder == nil { + return fmt.Errorf("validator builder for %q must not be nil", normalizedSpec.Key) } if _, ok := registry.chunkEntries[normalizedSpec.Key]; ok { return fmt.Errorf("chunk validator %q is already registered", normalizedSpec.Key) @@ -145,11 +192,20 @@ func RegisterChunkValidator(registry *ValidatorRegistry, spec ValidatorSpec, con if registry.chunkEntries == nil { registry.chunkEntries = make(map[string]chunkValidatorEntry) } - registry.chunkEntries[normalizedSpec.Key] = chunkValidatorEntry{spec: normalizedSpec, constructor: constructor} + registry.chunkEntries[normalizedSpec.Key] = chunkValidatorEntry{spec: normalizedSpec, validateOptions: validateOptions, builder: builder} return nil } func RegisterSerializedValidator(registry *ValidatorRegistry, spec SerializedValidatorSpec, constructor func() (contracts.SerializedValidator, error)) error { + if constructor == nil { + return fmt.Errorf("validator constructor for %q must not be nil", strings.TrimSpace(spec.Key)) + } + return RegisterSerializedValidatorBuilder(registry, spec, allowLegacyOptions, func(BuildRequest) (contracts.SerializedValidator, error) { + return constructor() + }) +} + +func RegisterSerializedValidatorBuilder(registry *ValidatorRegistry, spec SerializedValidatorSpec, validateOptions OptionValidator, builder func(BuildRequest) (contracts.SerializedValidator, error)) error { if registry == nil { return fmt.Errorf("validator registry must not be nil") } @@ -161,8 +217,11 @@ func RegisterSerializedValidator(registry *ValidatorRegistry, spec SerializedVal if !spec.SupportsChunks && !spec.SupportsArtifacts { return fmt.Errorf("serialized validator %q must support chunks, artifacts, or both", spec.Key) } - if constructor == nil { - return fmt.Errorf("validator constructor for %q must not be nil", spec.Key) + if validateOptions == nil { + return fmt.Errorf("validator option validator for %q must not be nil", spec.Key) + } + if builder == nil { + return fmt.Errorf("validator builder for %q must not be nil", spec.Key) } if _, ok := registry.serializedEntries[spec.Key]; ok { return fmt.Errorf("serialized validator %q is already registered", spec.Key) @@ -170,11 +229,15 @@ func RegisterSerializedValidator(registry *ValidatorRegistry, spec SerializedVal if registry.serializedEntries == nil { registry.serializedEntries = make(map[string]serializedValidatorEntry) } - registry.serializedEntries[spec.Key] = serializedValidatorEntry{spec: spec, constructor: constructor} + registry.serializedEntries[spec.Key] = serializedValidatorEntry{spec: spec, validateOptions: validateOptions, builder: builder} return nil } func (r *ValidatorRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawValidator, error) { + return r.BuildLegacyRawWithRequest(key, BuildRequest{}) +} + +func (r *ValidatorRegistry) BuildLegacyRawWithRequest(key string, request BuildRequest) (contracts.LegacyRawValidator, error) { if r == nil { return nil, fmt.Errorf("validator registry must not be nil") } @@ -182,11 +245,11 @@ func (r *ValidatorRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawValid if normalizedKey == "" { return nil, fmt.Errorf("validator key must not be empty") } - constructor, ok := r.legacyConstructors[normalizedKey] + builder, ok := r.legacyBuilders[normalizedKey] if !ok { return nil, fmt.Errorf("legacy raw validator %q is not registered", normalizedKey) } - validator, err := constructor() + validator, err := builder(cloneBuildRequest(request)) if err != nil { return nil, fmt.Errorf("build validator %q: %w", normalizedKey, err) } @@ -203,6 +266,37 @@ func (r *ValidatorRegistry) BuildLegacyRaw(key string) (contracts.LegacyRawValid return validator, nil } +func (r *ValidatorRegistry) validateOptions(resolved ResolvedValidator) error { + if r == nil { + return fmt.Errorf("validator registry must not be nil") + } + key := strings.TrimSpace(resolved.Binding.Module) + var validator OptionValidator + switch resolved.Target { + case ValidatorTargetTyped: + entry, ok := r.typedEntry(key, resolved.ArtifactKind) + if ok { + validator = entry.validateOptions + } + case ValidatorTargetChunk: + entry, ok := r.chunkEntry(key) + if ok { + validator = entry.validateOptions + } + case ValidatorTargetSerialized: + entry, ok := r.serializedEntry(key) + if ok { + validator = entry.validateOptions + } + default: + validator = r.legacyValidators[key] + } + if validator == nil { + return fmt.Errorf("validator %q construction entry is not registered", key) + } + return validateRegisteredOptions(validator, resolved.Binding.Options) +} + func (r *ValidatorRegistry) Spec(key string) (ValidatorSpec, bool) { if r == nil { return ValidatorSpec{}, false diff --git a/internal/framework/pipeline/walking_skeleton_test.go b/internal/framework/pipeline/walking_skeleton_test.go index 2c01534..6a3de10 100644 --- a/internal/framework/pipeline/walking_skeleton_test.go +++ b/internal/framework/pipeline/walking_skeleton_test.go @@ -26,12 +26,12 @@ func TestWalkingSkeletonFixture(t *testing.T) { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } - output, err := New(walkingSkeletonRegistries(t)).Run(context.Background(), RunInput{ - Pipeline: resolved, + output, err := newPreparedRunner(t, walkingSkeletonRegistries(t)).Run(context.Background(), RunInput{ + pipeline: resolved, SourceID: "fixture-source", Path: "walking_skeleton_input.json", RawInput: inputBytes, - LLMClient: llmClient, + llmClient: llmClient, }) if err != nil { t.Fatalf("Run() error = %v, want nil", err) diff --git a/internal/modules/integration/dnd_spells_runner_test.go b/internal/modules/integration/dnd_spells_runner_test.go index c802103..2a45643 100644 --- a/internal/modules/integration/dnd_spells_runner_test.go +++ b/internal/modules/integration/dnd_spells_runner_test.go @@ -15,6 +15,16 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/modules/seriatim/input/transcript" ) +func runPreparedPipeline(t *testing.T, registries pipeline.Registries, resolved pipeline.ResolvedPipeline, llmClient contracts.StructuredLLMClient, input pipeline.RunInput) (pipeline.RunOutput, error) { + t.Helper() + prepared, err := pipeline.Prepare(resolved, registries, pipeline.ModuleDependencies{LLM: llmClient}) + if err != nil { + return pipeline.RunOutput{}, err + } + input.Prepared = prepared + return pipeline.New().Run(context.Background(), input) +} + func TestRunnerProcessesSeriatimInputWithDNDSpellsExtractor(t *testing.T) { raw := readDNDSpellsFixture(t) expectedDoc := parseDNDSpellsFixture(t, raw) @@ -40,10 +50,8 @@ func TestRunnerProcessesSeriatimInputWithDNDSpellsExtractor(t *testing.T) { }, } - output, err := pipeline.New(dndSpellsRunnerRegistries(t)).Run(context.Background(), pipeline.RunInput{ - Pipeline: resolved.ResolvedPipeline, - RawInput: raw, - LLMClient: llmClient, + output, err := runPreparedPipeline(t, dndSpellsRunnerRegistries(t), resolved.ResolvedPipeline, llmClient, pipeline.RunInput{ + RawInput: raw, }) if err != nil { t.Fatalf("Run() error = %v, want nil", err) @@ -128,10 +136,8 @@ func TestRunnerPassesPartyAndGlossaryReferencesToDNDSpellsPrompt(t *testing.T) { }, } - output, err := pipeline.New(dndSpellsRunnerRegistries(t)).Run(context.Background(), pipeline.RunInput{ - Pipeline: resolved.ResolvedPipeline, - RawInput: raw, - LLMClient: llmClient, + output, err := runPreparedPipeline(t, dndSpellsRunnerRegistries(t), resolved.ResolvedPipeline, llmClient, pipeline.RunInput{ + RawInput: raw, }) if err != nil { t.Fatalf("Run() error = %v, want nil", err) @@ -169,10 +175,8 @@ func TestRunnerDoesNotExtractSpellMentionedOnlyInPartyReference(t *testing.T) { response: extractionResponse{SpellCasts: []spellCastResponse{}}, } - output, err := pipeline.New(dndSpellsRunnerRegistries(t)).Run(context.Background(), pipeline.RunInput{ - Pipeline: resolved.ResolvedPipeline, - RawInput: raw, - LLMClient: llmClient, + output, err := runPreparedPipeline(t, dndSpellsRunnerRegistries(t), resolved.ResolvedPipeline, llmClient, pipeline.RunInput{ + RawInput: raw, }) if err != nil { t.Fatalf("Run() error = %v, want nil", err) @@ -217,10 +221,8 @@ func TestRunnerCarriesDNDSpellCastWithInvalidSourceRefAsRawOutput(t *testing.T) }, } - output, err := pipeline.New(dndSpellsRunnerRegistries(t)).Run(context.Background(), pipeline.RunInput{ - Pipeline: resolved.ResolvedPipeline, - RawInput: raw, - LLMClient: llmClient, + output, err := runPreparedPipeline(t, dndSpellsRunnerRegistries(t), resolved.ResolvedPipeline, llmClient, pipeline.RunInput{ + RawInput: raw, }) if err != nil { t.Fatalf("Run() error = %v, want nil", err) @@ -285,10 +287,8 @@ func TestRunnerCarriesMalformedDNDSpellsExtractorOutput(t *testing.T) { resolved := resolveDNDSpellsPipeline(t) llmClient := &fakeSpellsLLMClient{response: extractionResponse{}} - output, err := pipeline.New(dndSpellsRunnerRegistries(t)).Run(context.Background(), pipeline.RunInput{ - Pipeline: resolved.ResolvedPipeline, - RawInput: raw, - LLMClient: llmClient, + output, err := runPreparedPipeline(t, dndSpellsRunnerRegistries(t), resolved.ResolvedPipeline, llmClient, pipeline.RunInput{ + RawInput: raw, }) if err != nil { t.Fatalf("Run() error = %v, want nil", err) diff --git a/internal/modules/seriatim/input/transcript/runner_test.go b/internal/modules/seriatim/input/transcript/runner_test.go index c95e9c5..eb66d2f 100644 --- a/internal/modules/seriatim/input/transcript/runner_test.go +++ b/internal/modules/seriatim/input/transcript/runner_test.go @@ -15,6 +15,16 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/normalize/noop" ) +func runPreparedPipeline(t *testing.T, registries pipeline.Registries, resolved pipeline.ResolvedPipeline, llmClient contracts.StructuredLLMClient, input pipeline.RunInput) (pipeline.RunOutput, error) { + t.Helper() + prepared, err := pipeline.Prepare(resolved, registries, pipeline.ModuleDependencies{LLM: llmClient}) + if err != nil { + return pipeline.RunOutput{}, err + } + input.Prepared = prepared + return pipeline.New().Run(context.Background(), input) +} + func TestRunnerProcessesSeriatimInputWithFakeModules(t *testing.T) { raw := readFixture(t, "testdata/valid_minimal.json") expectedDoc, err := New().Parse(context.Background(), contracts.ParseRequest{Raw: raw}) @@ -28,8 +38,7 @@ func TestRunnerProcessesSeriatimInputWithFakeModules(t *testing.T) { } extractor := &runnerSeriatimExtractor{} - output, err := pipeline.New(seriatimRunnerRegistries(t, extractor)).Run(context.Background(), pipeline.RunInput{ - Pipeline: resolved.ResolvedPipeline, + output, err := runPreparedPipeline(t, seriatimRunnerRegistries(t, extractor), resolved.ResolvedPipeline, nil, pipeline.RunInput{ RawInput: raw, }) if err != nil { @@ -83,11 +92,9 @@ func TestRunnerFailsOnInvalidSeriatimInput(t *testing.T) { t.Fatalf("Resolve() error = %v, want nil", err) } - output, err := pipeline.New(seriatimRunnerRegistries(t, &runnerSeriatimExtractor{})).Run(context.Background(), pipeline.RunInput{ - Pipeline: resolved.ResolvedPipeline, - RawInput: []byte(`{"metadata":{},"segments":[]}`), - SourceID: "invalid-source", - LLMClient: nil, + output, err := runPreparedPipeline(t, seriatimRunnerRegistries(t, &runnerSeriatimExtractor{}), resolved.ResolvedPipeline, nil, pipeline.RunInput{ + RawInput: []byte(`{"metadata":{},"segments":[]}`), + SourceID: "invalid-source", }) if err == nil { t.Fatal("Run() error = nil, want invalid input error")