diff --git a/internal/cli/catalog.go b/internal/cli/catalog.go index dbaf6c7..197e573 100644 --- a/internal/cli/catalog.go +++ b/internal/cli/catalog.go @@ -21,15 +21,16 @@ type productionComponents struct { func newProductionComponents() (productionComponents, error) { registries := pipeline.Registries{ - Inputs: pipeline.NewInputAdapterRegistry(), - Chunkers: pipeline.NewChunkerRegistry(), - ArtifactCodecs: pipeline.NewArtifactCodecRegistry(), - Extractors: pipeline.NewExtractorRegistry(), - Mergers: pipeline.NewMergerRegistry(), - Normalizers: pipeline.NewNormalizerRegistry(), - Validators: pipeline.NewValidatorRegistry(), - ValidatorChains: pipeline.NewValidatorChainRegistry(), - Outputs: pipeline.NewOutputEncoderRegistry(), + Inputs: pipeline.NewInputAdapterRegistry(), + Chunkers: pipeline.NewChunkerRegistry(), + ArtifactCodecs: pipeline.NewArtifactCodecRegistry(), + ArtifactEvidence: pipeline.NewArtifactEvidenceRegistry(), + Extractors: pipeline.NewExtractorRegistry(), + Mergers: pipeline.NewMergerRegistry(), + Normalizers: pipeline.NewNormalizerRegistry(), + Validators: pipeline.NewValidatorRegistry(), + ValidatorChains: pipeline.NewValidatorChainRegistry(), + Outputs: pipeline.NewOutputEncoderRegistry(), } assets := llm.NewAssetRegistry() registrars := []struct { @@ -88,29 +89,31 @@ func effectiveRegistries(opts Options) (pipeline.Registries, error) { func catalogFromRegistries(registries pipeline.Registries) pipeline.ModuleCatalog { return pipeline.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, + Inputs: registries.Inputs, + Chunkers: registries.Chunkers, + ArtifactCodecs: registries.ArtifactCodecs, + ArtifactEvidence: registries.ArtifactEvidence, + Extractors: registries.Extractors, + Mergers: registries.Mergers, + Normalizers: registries.Normalizers, + Validators: registries.Validators, + ValidatorChains: registries.ValidatorChains, + Outputs: registries.Outputs, } } func registriesFromCatalog(catalog pipeline.ModuleCatalog) pipeline.Registries { return pipeline.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, + Inputs: catalog.Inputs, + Chunkers: catalog.Chunkers, + ArtifactCodecs: catalog.ArtifactCodecs, + ArtifactEvidence: catalog.ArtifactEvidence, + Extractors: catalog.Extractors, + Mergers: catalog.Mergers, + Normalizers: catalog.Normalizers, + Validators: catalog.Validators, + ValidatorChains: catalog.ValidatorChains, + Outputs: catalog.Outputs, } } @@ -118,6 +121,7 @@ func isEmptyCatalog(catalog pipeline.ModuleCatalog) bool { return catalog.Inputs == nil && catalog.Chunkers == nil && catalog.ArtifactCodecs == nil && + catalog.ArtifactEvidence == nil && catalog.Extractors == nil && catalog.Mergers == nil && catalog.Normalizers == nil && @@ -130,6 +134,7 @@ func isEmptyRegistries(registries pipeline.Registries) bool { return registries.Inputs == nil && registries.Chunkers == nil && registries.ArtifactCodecs == nil && + registries.ArtifactEvidence == nil && registries.Extractors == nil && registries.Mergers == nil && registries.Normalizers == nil && diff --git a/internal/cli/production_contract_test.go b/internal/cli/production_contract_test.go index 96c86ba..2b25661 100644 --- a/internal/cli/production_contract_test.go +++ b/internal/cli/production_contract_test.go @@ -161,8 +161,8 @@ func TestProductionCatalogCoversMaintainedConfigurations(t *testing.T) { catalog := catalogFromRegistries(registries) converted := registriesFromCatalog(catalog) - if converted.ArtifactCodecs != registries.ArtifactCodecs || converted.ValidatorChains != registries.ValidatorChains { - t.Fatal("catalog/registry conversion did not preserve codec and validator-chain registries") + if converted.ArtifactCodecs != registries.ArtifactCodecs || converted.ArtifactEvidence != registries.ArtifactEvidence || converted.ValidatorChains != registries.ValidatorChains { + t.Fatal("catalog/registry conversion did not preserve artifact and validator registries") } codecSpec, ok := catalog.ArtifactCodecs.Spec(dnd.SpellListKind) if !ok || codecSpec.Kind != dnd.SpellListKind || codecSpec.Schema.ID != spellcodec.SchemaID { diff --git a/internal/framework/pipeline/artifact_evidence_registry.go b/internal/framework/pipeline/artifact_evidence_registry.go new file mode 100644 index 0000000..aefe9da --- /dev/null +++ b/internal/framework/pipeline/artifact_evidence_registry.go @@ -0,0 +1,148 @@ +package pipeline + +import ( + "fmt" + "reflect" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +// ArtifactEvidenceProjector returns the direct source references represented +// by one normalized artifact value. +type ArtifactEvidenceProjector[T any] func(T) []source.SourceRef + +// ArtifactEvidenceRegistry keeps the typed projection boundary private while +// allowing preparation and execution to discover registered capabilities. +type ArtifactEvidenceRegistry struct { + entries map[contracts.ArtifactKind]artifactEvidenceEntry +} + +type artifactEvidenceEntry struct { + valueType reflect.Type + project func(any) ([]source.SourceRef, error) +} + +func NewArtifactEvidenceRegistry() *ArtifactEvidenceRegistry { + return &ArtifactEvidenceRegistry{entries: make(map[contracts.ArtifactKind]artifactEvidenceEntry)} +} + +// RegisterArtifactEvidence registers one evidence projector for an artifact +// kind. Projectors are invoked only after an exact Go-type check. +func RegisterArtifactEvidence[T any](registry *ArtifactEvidenceRegistry, kind contracts.ArtifactKind, projector ArtifactEvidenceProjector[T]) error { + if registry == nil { + return fmt.Errorf("artifact evidence registry must not be nil") + } + if projector == nil { + return fmt.Errorf("artifact evidence projector must not be nil") + } + kind = normalizeArtifactKind(kind) + if kind == "" { + return fmt.Errorf("artifact evidence kind must not be empty") + } + if _, ok := registry.entries[kind]; ok { + return fmt.Errorf("artifact evidence %q is already registered", kind) + } + valueType := reflect.TypeFor[T]() + if registry.entries == nil { + registry.entries = make(map[contracts.ArtifactKind]artifactEvidenceEntry) + } + registry.entries[kind] = artifactEvidenceEntry{ + valueType: valueType, + project: func(value any) ([]source.SourceRef, error) { + actualType := reflect.TypeOf(value) + if actualType != valueType { + return nil, newArtifactEvidenceTypeError(kind, valueType, actualType) + } + typed, ok := value.(T) + if !ok { + return nil, newArtifactEvidenceTypeError(kind, valueType, actualType) + } + return append([]source.SourceRef(nil), projector(typed)...), nil + }, + } + return nil +} + +func (r *ArtifactEvidenceRegistry) RegisteredKinds() []contracts.ArtifactKind { + if r == nil || len(r.entries) == 0 { + return nil + } + kinds := make([]contracts.ArtifactKind, 0, len(r.entries)) + for kind := range r.entries { + kinds = append(kinds, kind) + } + sort.Slice(kinds, func(i, j int) bool { return kinds[i] < kinds[j] }) + return kinds +} + +// Project returns independently owned direct references for a registered +// artifact value. +func (r *ArtifactEvidenceRegistry) Project(kind contracts.ArtifactKind, value any) ([]source.SourceRef, error) { + entry, normalizedKind, err := r.entry(kind) + if err != nil { + return nil, err + } + refs, err := entry.project(value) + if err != nil { + return nil, fmt.Errorf("project artifact evidence %q: %w", normalizedKind, err) + } + return append([]source.SourceRef(nil), refs...), nil +} + +func (r *ArtifactEvidenceRegistry) entry(kind contracts.ArtifactKind) (artifactEvidenceEntry, contracts.ArtifactKind, error) { + if r == nil { + return artifactEvidenceEntry{}, "", fmt.Errorf("artifact evidence registry must not be nil") + } + kind = normalizeArtifactKind(kind) + if kind == "" { + return artifactEvidenceEntry{}, "", fmt.Errorf("artifact evidence kind must not be empty") + } + entry, ok := r.entries[kind] + if !ok { + return artifactEvidenceEntry{}, kind, fmt.Errorf("artifact evidence %q is not registered", kind) + } + return entry, kind, nil +} + +func (r *ArtifactEvidenceRegistry) valueType(kind contracts.ArtifactKind) (reflect.Type, bool) { + if r == nil { + return nil, false + } + entry, ok := r.entries[normalizeArtifactKind(kind)] + if !ok { + return nil, false + } + return entry.valueType, true +} + +func newArtifactEvidenceTypeError(kind contracts.ArtifactKind, expected, actual reflect.Type) error { + actualName := "" + if actual != nil { + actualName = actual.String() + } + return fmt.Errorf("project artifact evidence %q: expected exact Go type %s, got %s", kind, expected, actualName) +} + +func normalizeEvidenceLaneIDs(values []string) ([]string, error) { + if len(values) == 0 { + return nil, fmt.Errorf("evidence lane ids must not be empty") + } + seen := make(map[string]struct{}, len(values)) + lanes := make([]string, 0, len(values)) + for _, raw := range values { + lane := strings.TrimSpace(raw) + if lane == "" { + return nil, fmt.Errorf("evidence lane id must not be empty") + } + if _, ok := seen[lane]; ok { + return nil, fmt.Errorf("evidence lane id %q is duplicated", lane) + } + seen[lane] = struct{}{} + lanes = append(lanes, lane) + } + sort.Strings(lanes) + return lanes, nil +} diff --git a/internal/framework/pipeline/artifact_evidence_registry_test.go b/internal/framework/pipeline/artifact_evidence_registry_test.go new file mode 100644 index 0000000..b3e9c95 --- /dev/null +++ b/internal/framework/pipeline/artifact_evidence_registry_test.go @@ -0,0 +1,55 @@ +package pipeline + +import ( + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +func TestArtifactEvidenceRegistryProjectsExactRegisteredTypesWithOwnedReferences(t *testing.T) { + registry := NewArtifactEvidenceRegistry() + input := codecNotes{Items: []string{"one"}} + refs := []source.SourceRef{{SourceID: "source", StartUnitID: 2, EndUnitID: 2}} + if err := RegisterArtifactEvidence(registry, "test/notes", func(codecNotes) []source.SourceRef { return refs }); err != nil { + t.Fatalf("RegisterArtifactEvidence() error = %v, want nil", err) + } + if err := RegisterArtifactEvidence(registry, "test/score", func(codecScore) []source.SourceRef { return nil }); err != nil { + t.Fatalf("RegisterArtifactEvidence() second kind error = %v, want nil", err) + } + if got, want := registry.RegisteredKinds(), []contracts.ArtifactKind{"test/notes", "test/score"}; !reflect.DeepEqual(got, want) { + t.Fatalf("RegisteredKinds() = %#v, want %#v", got, want) + } + projected, err := registry.Project(" test/notes ", input) + if err != nil { + t.Fatalf("Project() error = %v, want nil", err) + } + projected[0].StartUnitID = 99 + if refs[0].StartUnitID != 2 { + t.Fatal("Project() retained projector reference storage") + } + if _, err := registry.Project("test/notes", codecNotesAlias(input)); err == nil || !strings.Contains(err.Error(), "expected exact Go type") { + t.Fatalf("Project() exact type error = %v, want exact type failure", err) + } +} + +func TestArtifactEvidenceRegistryRejectsInvalidRegistration(t *testing.T) { + registry := NewArtifactEvidenceRegistry() + if err := RegisterArtifactEvidence[codecNotes](nil, "test/notes", func(codecNotes) []source.SourceRef { return nil }); err == nil || !strings.Contains(err.Error(), "must not be nil") { + t.Fatalf("nil registry error = %v, want failure", err) + } + if err := RegisterArtifactEvidence(registry, " ", func(codecNotes) []source.SourceRef { return nil }); err == nil || !strings.Contains(err.Error(), "must not be empty") { + t.Fatalf("blank kind error = %v, want failure", err) + } + if err := RegisterArtifactEvidence(registry, "test/notes", ArtifactEvidenceProjector[codecNotes](nil)); err == nil || !strings.Contains(err.Error(), "must not be nil") { + t.Fatalf("nil projector error = %v, want failure", err) + } + if err := RegisterArtifactEvidence(registry, "test/notes", func(codecNotes) []source.SourceRef { return nil }); err != nil { + t.Fatal(err) + } + if err := RegisterArtifactEvidence(registry, "test/notes", func(codecNotes) []source.SourceRef { return nil }); err == nil || !strings.Contains(err.Error(), "already registered") { + t.Fatalf("duplicate kind error = %v, want failure", err) + } +} diff --git a/internal/framework/pipeline/evidence_policy.go b/internal/framework/pipeline/evidence_policy.go new file mode 100644 index 0000000..5171276 --- /dev/null +++ b/internal/framework/pipeline/evidence_policy.go @@ -0,0 +1,20 @@ +package pipeline + +// EvidenceContextPolicy controls optional source-context publication by an +// output encoder. +type EvidenceContextPolicy struct { + Enabled bool + WindowUnits int + LaneIDs []string +} + +// EvidenceContextPolicyProvider is implemented by output encoders that opt in +// to evidence-context publication. +type EvidenceContextPolicyProvider interface { + EvidenceContextPolicy() EvidenceContextPolicy +} + +func cloneEvidenceContextPolicy(policy EvidenceContextPolicy) EvidenceContextPolicy { + policy.LaneIDs = append([]string(nil), policy.LaneIDs...) + return policy +} diff --git a/internal/framework/pipeline/evidence_preparation_test.go b/internal/framework/pipeline/evidence_preparation_test.go new file mode 100644 index 0000000..8a5f7a7 --- /dev/null +++ b/internal/framework/pipeline/evidence_preparation_test.go @@ -0,0 +1,98 @@ +package pipeline + +import ( + "context" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type testEvidenceOutput struct { + policy EvidenceContextPolicy +} + +func (output testEvidenceOutput) Key() string { return "output" } +func (output testEvidenceOutput) Encode(context.Context, contracts.OutputRequest) (contracts.OutputResult, error) { + return contracts.OutputResult{}, nil +} +func (output testEvidenceOutput) EvidenceContextPolicy() EvidenceContextPolicy { + return cloneEvidenceContextPolicy(output.policy) +} + +func TestPrepareEvidencePlanSelectsActiveLanesAndOwnsPolicy(t *testing.T) { + registries, _ := constructionRegistries(t, nil, nil) + registries.ArtifactEvidence = NewArtifactEvidenceRegistry() + registerTestEvidenceOutput(t, ®istries, EvidenceContextPolicy{Enabled: true, WindowUnits: 2, LaneIDs: []string{"artifact", "inactive"}}) + if err := RegisterArtifactEvidence(registries.ArtifactEvidence, "test/notes", func(codecNotes) []source.SourceRef { return nil }); err != nil { + t.Fatal(err) + } + profile := constructionProfile() + profile.Output.Options = map[string]any{"known": true} + resolved, err := ResolvePipeline(profile, ResolveOptions{}, registries.catalog()) + if err != nil { + t.Fatal(err) + } + prepared, err := Prepare(resolved, registries, ModuleDependencies{}) + if err != nil { + t.Fatalf("Prepare() error = %v, want nil", err) + } + if prepared.evidencePlan == nil || !reflect.DeepEqual(prepared.evidencePlan.policy.LaneIDs, []string{"artifact", "inactive"}) { + t.Fatalf("evidence plan = %#v, want complete configured policy", prepared.evidencePlan) + } + if got := prepared.evidencePlan.lanes; len(got) != 1 || got[0].laneID != "artifact" { + t.Fatalf("active evidence lanes = %#v, want artifact only", got) + } + prepared.evidencePlan.policy.LaneIDs[0] = "mutated" + if policy := prepared.output.(EvidenceContextPolicyProvider).EvidenceContextPolicy(); policy.LaneIDs[0] != "artifact" { + t.Fatalf("prepared plan mutated provider policy: %#v", policy) + } +} + +func TestPrepareEvidencePlanRejectsMissingAndMismatchedCapabilities(t *testing.T) { + for _, test := range []struct { + name string + configure func(*Registries) + want string + }{ + {name: "missing registry", configure: func(registries *Registries) { registries.ArtifactEvidence = nil }, want: "artifact evidence registry"}, + {name: "unsupported kind", configure: func(registries *Registries) {}, want: "artifact evidence \"test/notes\" is not registered"}, + {name: "mismatched type", configure: func(registries *Registries) { + if err := RegisterArtifactEvidence(registries.ArtifactEvidence, "test/notes", func(codecScore) []source.SourceRef { return nil }); err != nil { + t.Fatal(err) + } + }, want: "requires Go type"}, + } { + t.Run(test.name, func(t *testing.T) { + registries, _ := constructionRegistries(t, nil, nil) + registries.ArtifactEvidence = NewArtifactEvidenceRegistry() + registerTestEvidenceOutput(t, ®istries, EvidenceContextPolicy{Enabled: true, LaneIDs: []string{"artifact"}}) + test.configure(®istries) + profile := constructionProfile() + profile.Output.Options = map[string]any{"known": true} + resolved, err := ResolvePipeline(profile, ResolveOptions{}, registries.catalog()) + if err != nil { + t.Fatal(err) + } + _, err = Prepare(resolved, registries, ModuleDependencies{}) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("Prepare() error = %v, want %q", err, test.want) + } + }) + } +} + +func registerTestEvidenceOutput(t *testing.T, registries *Registries, policy EvidenceContextPolicy) { + t.Helper() + registry := NewOutputEncoderRegistry() + if err := registry.RegisterBuilderWithSpec(defaultModuleSpec("output", StageOutput), func(options map[string]any) error { + return RejectUnknownOptions(options, "known") + }, func(BuildRequest) (contracts.OutputEncoder, error) { + return testEvidenceOutput{policy: cloneEvidenceContextPolicy(policy)}, nil + }); err != nil { + t.Fatal(err) + } + registries.Outputs = registry +} diff --git a/internal/framework/pipeline/options.go b/internal/framework/pipeline/options.go index a567ef0..7fa5f0e 100644 --- a/internal/framework/pipeline/options.go +++ b/internal/framework/pipeline/options.go @@ -2,7 +2,7 @@ package pipeline import "fmt" -func validateResolvedOptions(resolved ResolvedPipeline, catalog ModuleCatalog) error { +func validateResolvedOptions(resolved ResolvedPipeline, catalog ModuleCatalog, configuredLaneIDs []string) error { if err := catalog.Inputs.ValidateOptions(resolved.Input.Module, resolved.Input.Options); err != nil { return moduleOptionsError(resolved.ID, "", StageInput, resolved.Input.Module, err) } @@ -42,6 +42,9 @@ func validateResolvedOptions(resolved ResolvedPipeline, catalog ModuleCatalog) e if err := catalog.Outputs.ValidateOptions(resolved.Output.Module, resolved.Output.Options); err != nil { return moduleOptionsError(resolved.ID, "", StageOutput, resolved.Output.Module, err) } + if err := catalog.Outputs.ValidateProfileOptions(resolved.Output.Module, OutputProfileOptionContext{LaneIDs: configuredLaneIDs}, resolved.Output.Options); err != nil { + return moduleOptionsError(resolved.ID, "", StageOutput, resolved.Output.Module, err) + } return nil } diff --git a/internal/framework/pipeline/output_registry.go b/internal/framework/pipeline/output_registry.go index 5da88ee..1eb4529 100644 --- a/internal/framework/pipeline/output_registry.go +++ b/internal/framework/pipeline/output_registry.go @@ -10,17 +10,25 @@ import ( type OutputEncoderConstructor func() (contracts.OutputEncoder, error) type OutputEncoderBuilder func(BuildRequest) (contracts.OutputEncoder, error) +type OutputProfileOptionContext struct { + LaneIDs []string +} + +type OutputProfileOptionValidator func(OutputProfileOptionContext, map[string]any) error + type OutputEncoderRegistry struct { - builders map[string]OutputEncoderBuilder - optionValidators map[string]OptionValidator - specs map[string]ModuleSpec + builders map[string]OutputEncoderBuilder + optionValidators map[string]OptionValidator + profileValidators map[string]OutputProfileOptionValidator + specs map[string]ModuleSpec } func NewOutputEncoderRegistry() *OutputEncoderRegistry { return &OutputEncoderRegistry{ - builders: make(map[string]OutputEncoderBuilder), - optionValidators: make(map[string]OptionValidator), - specs: make(map[string]ModuleSpec), + builders: make(map[string]OutputEncoderBuilder), + optionValidators: make(map[string]OptionValidator), + profileValidators: make(map[string]OutputProfileOptionValidator), + specs: make(map[string]ModuleSpec), } } @@ -38,6 +46,12 @@ func (r *OutputEncoderRegistry) RegisterWithSpec(spec ModuleSpec, constructor Ou } func (r *OutputEncoderRegistry) RegisterBuilderWithSpec(spec ModuleSpec, validateOptions OptionValidator, builder OutputEncoderBuilder) error { + return r.RegisterBuilderWithProfileValidation(spec, validateOptions, nil, builder) +} + +// RegisterBuilderWithProfileValidation registers an output builder with an +// optional validator that can inspect all configured lane identities. +func (r *OutputEncoderRegistry) RegisterBuilderWithProfileValidation(spec ModuleSpec, validateOptions OptionValidator, validateProfile OutputProfileOptionValidator, builder OutputEncoderBuilder) error { if r == nil { return fmt.Errorf("output encoder registry must not be nil") } @@ -62,11 +76,15 @@ func (r *OutputEncoderRegistry) RegisterBuilderWithSpec(spec ModuleSpec, validat if r.optionValidators == nil { r.optionValidators = make(map[string]OptionValidator) } + if r.profileValidators == nil { + r.profileValidators = make(map[string]OutputProfileOptionValidator) + } if r.specs == nil { r.specs = make(map[string]ModuleSpec) } r.builders[normalizedSpec.Key] = builder r.optionValidators[normalizedSpec.Key] = validateOptions + r.profileValidators[normalizedSpec.Key] = validateProfile r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec) return nil } @@ -116,6 +134,21 @@ func (r *OutputEncoderRegistry) ValidateOptions(key string, options map[string]a return validateRegisteredOptions(validator, options) } +func (r *OutputEncoderRegistry) ValidateProfileOptions(key string, context OutputProfileOptionContext, 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.profileValidators[normalizedKey] + if !ok { + return fmt.Errorf("output encoder %q is not registered", normalizedKey) + } + if validator == nil { + return nil + } + return validator(OutputProfileOptionContext{LaneIDs: append([]string(nil), context.LaneIDs...)}, cloneOptions(options)) +} + func (r *OutputEncoderRegistry) Spec(key string) (ModuleSpec, bool) { if r == nil { return ModuleSpec{}, false diff --git a/internal/framework/pipeline/output_registry_test.go b/internal/framework/pipeline/output_registry_test.go index a2fca67..533c652 100644 --- a/internal/framework/pipeline/output_registry_test.go +++ b/internal/framework/pipeline/output_registry_test.go @@ -1,6 +1,7 @@ package pipeline import ( + "reflect" "testing" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" @@ -56,3 +57,33 @@ func TestOutputEncoderRegistryBehavior(t *testing.T) { }, }) } + +func TestOutputProfileValidationReceivesOwnedOptionsAndLaneIDs(t *testing.T) { + registry := NewOutputEncoderRegistry() + if err := registry.RegisterBuilderWithProfileValidation(defaultModuleSpec("profile-output", StageOutput), func(options map[string]any) error { + options["nested"].(map[string]any)["value"] = "changed" + return nil + }, func(context OutputProfileOptionContext, options map[string]any) error { + context.LaneIDs[0] = "changed" + options["nested"].(map[string]any)["value"] = "changed-again" + return nil + }, func(BuildRequest) (contracts.OutputEncoder, error) { + return registryOutputEncoder{key: "profile-output"}, nil + }); err != nil { + t.Fatal(err) + } + options := map[string]any{"nested": map[string]any{"value": "original"}} + if err := registry.ValidateOptions("profile-output", options); err != nil { + t.Fatal(err) + } + context := OutputProfileOptionContext{LaneIDs: []string{"artifact"}} + if err := registry.ValidateProfileOptions("profile-output", context, options); err != nil { + t.Fatal(err) + } + if got := options["nested"].(map[string]any)["value"]; got != "original" { + t.Fatalf("options mutated by validator: %q", got) + } + if want := []string{"artifact"}; !reflect.DeepEqual(context.LaneIDs, want) { + t.Fatalf("lane ids mutated by validator: %#v, want %#v", context.LaneIDs, want) + } +} diff --git a/internal/framework/pipeline/prepare.go b/internal/framework/pipeline/prepare.go index 1f74ecf..e8d4ce6 100644 --- a/internal/framework/pipeline/prepare.go +++ b/internal/framework/pipeline/prepare.go @@ -5,6 +5,7 @@ import ( "reflect" "strings" + "gitea.maximumdirect.net/eric/notarius/internal/core/source" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) @@ -24,6 +25,7 @@ type PreparedPipeline struct { chunkValidators preparedValidatorChain output contracts.OutputEncoder artifactCodecs *ArtifactCodecRegistry + evidencePlan *preparedEvidencePlan checkpointFingerprints []CheckpointFingerprint } @@ -55,6 +57,17 @@ type preparedTypedLane struct { codec artifactCodecEntry } +type preparedEvidencePlan struct { + policy EvidenceContextPolicy + lanes []preparedEvidenceLane +} + +type preparedEvidenceLane struct { + laneID string + kind contracts.ArtifactKind + project func(any) ([]source.SourceRef, error) +} + type preparedValidatorChain struct { resolved ResolvedValidatorChain validators []preparedValidator @@ -129,6 +142,10 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend return nil, constructionError(stable.ID, "", StageOutput, stable.Output.Module, "", err) } prepared.output = output + prepared.evidencePlan, err = prepareEvidencePlan(stable, registries, output) + if err != nil { + return nil, err + } prepared.checkpointFingerprints, err = collectPreparedCheckpointFingerprints(prepared) if err != nil { return nil, err @@ -136,6 +153,52 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend return prepared, nil } +func prepareEvidencePlan(resolved ResolvedPipeline, registries Registries, output contracts.OutputEncoder) (*preparedEvidencePlan, error) { + provider, ok := output.(EvidenceContextPolicyProvider) + if !ok { + return nil, nil + } + policy := cloneEvidenceContextPolicy(provider.EvidenceContextPolicy()) + if !policy.Enabled { + return nil, nil + } + if policy.WindowUnits < 0 { + return nil, constructionError(resolved.ID, "", StageOutput, resolved.Output.Module, "", fmt.Errorf("evidence window units must not be negative")) + } + lanes, err := normalizeEvidenceLaneIDs(policy.LaneIDs) + if err != nil { + return nil, constructionError(resolved.ID, "", StageOutput, resolved.Output.Module, "", err) + } + policy.LaneIDs = lanes + active := make(map[string]ResolvedArtifactLane) + for _, lane := range resolved.AllArtifactLanes() { + active[lane.ID] = lane + } + plan := &preparedEvidencePlan{policy: cloneEvidenceContextPolicy(policy)} + for _, laneID := range policy.LaneIDs { + lane, ok := active[laneID] + if !ok { + continue + } + if registries.ArtifactEvidence == nil { + return nil, constructionError(resolved.ID, laneID, StageOutput, resolved.Output.Module, "", fmt.Errorf("artifact evidence registry must not be nil for active evidence lane")) + } + evidence, _, evidenceErr := registries.ArtifactEvidence.entry(lane.ArtifactKind) + if evidenceErr != nil { + return nil, constructionError(resolved.ID, laneID, StageOutput, resolved.Output.Module, "", evidenceErr) + } + codecType, ok := registries.ArtifactCodecs.valueType(lane.ArtifactKind) + if !ok { + return nil, constructionError(resolved.ID, laneID, StageOutput, resolved.Output.Module, "", fmt.Errorf("artifact codec %q has no Go type", lane.ArtifactKind)) + } + if evidence.valueType != codecType { + return nil, constructionError(resolved.ID, laneID, StageOutput, resolved.Output.Module, "", fmt.Errorf("artifact evidence kind %q requires Go type %s, but active artifact codec uses %s", lane.ArtifactKind, typeName(evidence.valueType), typeName(codecType))) + } + plan.lanes = append(plan.lanes, preparedEvidenceLane{laneID: laneID, kind: lane.ArtifactKind, project: evidence.project}) + } + return plan, nil +} + func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registries Registries, deps ModuleDependencies) (preparedLaneExecutor, error) { executor := preparedLaneExecutor{resolved: cloneResolvedArtifactLane(lane)} request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest { @@ -303,7 +366,7 @@ func constructionError(pipelineID, laneID string, stage ModuleStage, moduleKey, func (registries Registries) catalog() ModuleCatalog { return ModuleCatalog{ - Inputs: registries.Inputs, Chunkers: registries.Chunkers, ArtifactCodecs: registries.ArtifactCodecs, + Inputs: registries.Inputs, Chunkers: registries.Chunkers, ArtifactCodecs: registries.ArtifactCodecs, ArtifactEvidence: registries.ArtifactEvidence, Extractors: registries.Extractors, Mergers: registries.Mergers, Normalizers: registries.Normalizers, Validators: registries.Validators, ValidatorChains: registries.ValidatorChains, Outputs: registries.Outputs, } diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index ae94e06..e8c788d 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -213,15 +213,16 @@ func (resolved ResolvedPipeline) AllArtifactLanes() []ResolvedArtifactLane { } type ModuleCatalog struct { - Inputs *InputAdapterRegistry - Chunkers *ChunkerRegistry - ArtifactCodecs *ArtifactCodecRegistry - Extractors *ExtractorRegistry - Mergers *MergerRegistry - Normalizers *NormalizerRegistry - Validators *ValidatorRegistry - ValidatorChains *ValidatorChainRegistry - Outputs *OutputEncoderRegistry + Inputs *InputAdapterRegistry + Chunkers *ChunkerRegistry + ArtifactCodecs *ArtifactCodecRegistry + ArtifactEvidence *ArtifactEvidenceRegistry + Extractors *ExtractorRegistry + Mergers *MergerRegistry + Normalizers *NormalizerRegistry + Validators *ValidatorRegistry + ValidatorChains *ValidatorChainRegistry + Outputs *OutputEncoderRegistry } func Binding(module string) ModuleBinding { @@ -251,6 +252,10 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo if len(steps) == 0 { return ResolvedPipeline{}, fmt.Errorf("pipeline %q must declare at least one artifact lane", pipelineID) } + configuredLaneIDs, err := configuredProfileLaneIDs(pipelineID, steps) + if err != nil { + return ResolvedPipeline{}, err + } input := resolveBinding(profile.Input, "") if input.Module == "" { @@ -381,7 +386,7 @@ 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 { + if err := validateResolvedOptions(resolved, catalog, configuredLaneIDs); err != nil { return ResolvedPipeline{}, err } @@ -393,6 +398,29 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo return resolved, nil } +// configuredProfileLaneIDs validates the complete configured lane identity +// set before invocation-level selection removes legacy-profile lanes. +func configuredProfileLaneIDs(pipelineID string, steps []PipelineStepProfile) ([]string, error) { + seen := make(map[string]string) + var laneIDs []string + for _, step := range steps { + _, ids, err := selectedArtifactLanes(pipelineID, step.Artifacts, ResolveOptions{}) + if err != nil { + return nil, err + } + stepID := strings.TrimSpace(step.ID) + for _, laneID := range ids { + if previous, ok := seen[laneID]; ok { + return nil, fmt.Errorf("pipeline %q artifact lane id %q is duplicated across steps %q and %q", pipelineID, laneID, previous, stepID) + } + seen[laneID] = stepID + laneIDs = append(laneIDs, laneID) + } + } + sort.Strings(laneIDs) + return laneIDs, nil +} + func resolveArtifactLane( pipelineID string, stepID string, @@ -1190,11 +1218,32 @@ func cloneOptions(options map[string]any) map[string]any { copied := make(map[string]any, len(options)) for key, value := range options { - copied[key] = value + copied[key] = cloneOptionValue(value) } return copied } +func cloneOptionValue(value any) any { + switch typed := value.(type) { + case map[string]any: + return cloneOptions(typed) + case []any: + out := make([]any, len(typed)) + for i := range typed { + out[i] = cloneOptionValue(typed[i]) + } + return out + case []string: + return append([]string(nil), typed...) + case []byte: + return append([]byte(nil), typed...) + case json.RawMessage: + return append(json.RawMessage(nil), typed...) + default: + return value + } +} + func normalizeReferenceMap(values map[string]ReferenceSource) map[string]ReferenceSource { if len(values) == 0 { return nil diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 87e11ab..82cd730 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -21,15 +21,16 @@ import ( ) type Registries struct { - Inputs *InputAdapterRegistry - Chunkers *ChunkerRegistry - ArtifactCodecs *ArtifactCodecRegistry - Extractors *ExtractorRegistry - Mergers *MergerRegistry - Normalizers *NormalizerRegistry - Validators *ValidatorRegistry - ValidatorChains *ValidatorChainRegistry - Outputs *OutputEncoderRegistry + Inputs *InputAdapterRegistry + Chunkers *ChunkerRegistry + ArtifactCodecs *ArtifactCodecRegistry + ArtifactEvidence *ArtifactEvidenceRegistry + Extractors *ExtractorRegistry + Mergers *MergerRegistry + Normalizers *NormalizerRegistry + Validators *ValidatorRegistry + ValidatorChains *ValidatorChainRegistry + Outputs *OutputEncoderRegistry } type Runner struct{} diff --git a/internal/modules/dnd/register/evidence.go b/internal/modules/dnd/register/evidence.go new file mode 100644 index 0000000..6867438 --- /dev/null +++ b/internal/modules/dnd/register/evidence.go @@ -0,0 +1,74 @@ +package register + +import ( + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" + "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd" +) + +func registerEvidence(registry *pipeline.ArtifactEvidenceRegistry) error { + return runRegistrations([]registration{ + {name: "spells evidence", register: func() error { return pipeline.RegisterArtifactEvidence(registry, dnd.SpellListKind, spellEvidence) }}, + {name: "npcs evidence", register: func() error { return pipeline.RegisterArtifactEvidence(registry, dnd.NPCListKind, npcEvidence) }}, + {name: "combat turns evidence", register: func() error { + return pipeline.RegisterArtifactEvidence(registry, dnd.CombatTurnListKind, combatTurnEvidence) + }}, + {name: "item events evidence", register: func() error { + return pipeline.RegisterArtifactEvidence(registry, dnd.ItemEventListKind, itemEventEvidence) + }}, + {name: "npc interactions evidence", register: func() error { + return pipeline.RegisterArtifactEvidence(registry, dnd.NPCInteractionListKind, npcInteractionEvidence) + }}, + {name: "scene descriptions evidence", register: func() error { + return pipeline.RegisterArtifactEvidence(registry, dnd.SceneDescriptionListKind, sceneDescriptionEvidence) + }}, + }) +} + +func spellEvidence(value dnd.SpellList) []source.SourceRef { + var refs []source.SourceRef + for _, record := range value.SpellCasts { + refs = append(refs, record.SourceRefs...) + } + return append([]source.SourceRef(nil), refs...) +} + +func npcEvidence(value dnd.NPCList) []source.SourceRef { + var refs []source.SourceRef + for _, record := range value.NPCs { + refs = append(refs, record.SourceRefs...) + } + return append([]source.SourceRef(nil), refs...) +} + +func combatTurnEvidence(value dnd.CombatTurnList) []source.SourceRef { + var refs []source.SourceRef + for _, record := range value.CombatTurns { + refs = append(refs, record.SourceRefs...) + } + return append([]source.SourceRef(nil), refs...) +} + +func itemEventEvidence(value dnd.ItemEventList) []source.SourceRef { + var refs []source.SourceRef + for _, record := range value.Events { + refs = append(refs, record.SourceRefs...) + } + return append([]source.SourceRef(nil), refs...) +} + +func npcInteractionEvidence(value dnd.NPCInteractionList) []source.SourceRef { + var refs []source.SourceRef + for _, record := range value.Interactions { + refs = append(refs, record.SourceRefs...) + } + return append([]source.SourceRef(nil), refs...) +} + +func sceneDescriptionEvidence(value dnd.SceneDescriptionList) []source.SourceRef { + refs := make([]source.SourceRef, 0, len(value.Scenes)) + for _, record := range value.Scenes { + refs = append(refs, record.SourceRef) + } + return refs +} diff --git a/internal/modules/dnd/register/register.go b/internal/modules/dnd/register/register.go index dbb2f80..ce11618 100644 --- a/internal/modules/dnd/register/register.go +++ b/internal/modules/dnd/register/register.go @@ -21,6 +21,9 @@ func Register(registries pipeline.Registries, assets *llm.AssetRegistry) error { if err := registerModules(registries); err != nil { return err } + if err := registerEvidence(registries.ArtifactEvidence); err != nil { + return err + } if err := registerValidators(registries); err != nil { return err } @@ -45,6 +48,8 @@ func validateRegistries(registries pipeline.Registries, assets *llm.AssetRegistr return fmt.Errorf("dnd registrar: chunker registry must not be nil") case registries.ArtifactCodecs == nil: return fmt.Errorf("dnd registrar: artifact codec registry must not be nil") + case registries.ArtifactEvidence == nil: + return fmt.Errorf("dnd registrar: artifact evidence registry must not be nil") case registries.Extractors == nil: return fmt.Errorf("dnd registrar: extractor registry must not be nil") case registries.Mergers == nil: diff --git a/internal/modules/dnd/register/register_test.go b/internal/modules/dnd/register/register_test.go index 6004a16..1ac3fc3 100644 --- a/internal/modules/dnd/register/register_test.go +++ b/internal/modules/dnd/register/register_test.go @@ -50,6 +50,7 @@ func TestRegisterAddsDNDFamily(t *testing.T) { assertContainsKeys(t, "extractors", registries.Extractors.RegisteredKeys(), []string{"dnd/spells", npcextract.Key, combatextract.Key, itemeventextract.Key, interactionextract.Key, scenedescriptionextract.Key}) assertContainsKeys(t, "normalizers", registries.Normalizers.RegisteredKeys(), []string{spellnormalize.Key, npcnormalize.Key, combatnormalize.Key, itemeventnormalize.Key, interactionnormalize.Key, scenedescriptionnormalize.Key, pipeline.DefaultNormalizeModule}) assertContainsArtifactKinds(t, registries.ArtifactCodecs.RegisteredKinds(), []contracts.ArtifactKind{dnd.SpellListKind, dnd.NPCListKind, dnd.CombatTurnListKind, dnd.ItemEventListKind, dnd.NPCInteractionListKind, dnd.SceneDescriptionListKind}) + assertContainsArtifactKinds(t, registries.ArtifactEvidence.RegisteredKinds(), []contracts.ArtifactKind{dnd.SpellListKind, dnd.NPCListKind, dnd.CombatTurnListKind, dnd.ItemEventListKind, dnd.NPCInteractionListKind, dnd.SceneDescriptionListKind}) assertContainsArtifactKinds(t, registries.Mergers.RegisteredArtifactKinds(pipeline.DefaultMergeModule), []contracts.ArtifactKind{dnd.SpellListKind, dnd.NPCListKind, dnd.CombatTurnListKind, dnd.ItemEventListKind, dnd.NPCInteractionListKind, dnd.SceneDescriptionListKind}) assertContainsArtifactKinds(t, registries.Normalizers.RegisteredArtifactKinds(pipeline.DefaultNormalizeModule), []contracts.ArtifactKind{dnd.SpellListKind, dnd.NPCListKind, dnd.CombatTurnListKind, dnd.ItemEventListKind, dnd.NPCInteractionListKind, dnd.SceneDescriptionListKind}) assertContainsArtifactKinds(t, registries.Normalizers.RegisteredArtifactKinds(npcnormalize.Key), []contracts.ArtifactKind{dnd.NPCListKind}) @@ -318,6 +319,46 @@ func TestRegisterAddsDNDFamily(t *testing.T) { } } +func TestEvidenceProjectorsPreserveDirectReferencesWithIndependentStorage(t *testing.T) { + first := source.SourceRef{SourceID: "session", StartUnitID: 1, EndUnitID: 1} + second := source.SourceRef{SourceID: "session", StartUnitID: 2, EndUnitID: 2} + for _, test := range []struct { + name string + project func() []source.SourceRef + want []source.SourceRef + }{ + {name: "spells", project: func() []source.SourceRef { + return spellEvidence(dnd.SpellList{SpellCasts: []dnd.SpellCast{{SourceRefs: []source.SourceRef{first, second}}}}) + }, want: []source.SourceRef{first, second}}, + {name: "npcs", project: func() []source.SourceRef { + return npcEvidence(dnd.NPCList{NPCs: []dnd.NPC{{SourceRefs: []source.SourceRef{first, second}}}}) + }, want: []source.SourceRef{first, second}}, + {name: "combat turns", project: func() []source.SourceRef { + return combatTurnEvidence(dnd.CombatTurnList{CombatTurns: []dnd.CombatTurn{{SourceRefs: []source.SourceRef{first, second}}}}) + }, want: []source.SourceRef{first, second}}, + {name: "item events", project: func() []source.SourceRef { + return itemEventEvidence(dnd.ItemEventList{Events: []dnd.ItemEvent{{SourceRefs: []source.SourceRef{first, second}}}}) + }, want: []source.SourceRef{first, second}}, + {name: "npc interactions", project: func() []source.SourceRef { + return npcInteractionEvidence(dnd.NPCInteractionList{Interactions: []dnd.NPCInteraction{{SourceRefs: []source.SourceRef{first, second}}}}) + }, want: []source.SourceRef{first, second}}, + {name: "scene descriptions", project: func() []source.SourceRef { + return sceneDescriptionEvidence(dnd.SceneDescriptionList{Scenes: []dnd.SceneDescription{{SourceRef: first}, {SourceRef: second}}}) + }, want: []source.SourceRef{first, second}}, + } { + t.Run(test.name, func(t *testing.T) { + got := test.project() + if !reflect.DeepEqual(got, test.want) { + t.Fatalf("projected references = %#v, want %#v", got, test.want) + } + got[0].StartUnitID = 99 + if first.StartUnitID != 1 { + t.Fatal("projector returned aliased reference storage") + } + }) + } +} + func referenceSlot(slots []contracts.ReferenceSlot, name string) contracts.ReferenceSlot { for _, slot := range slots { if slot.Name == name { @@ -531,6 +572,7 @@ func TestRegisterRejectsMissingDNDDependenciesBeforeMutation(t *testing.T) { }{ {name: "chunkers", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.Chunkers = nil }, wantErr: "chunker registry"}, {name: "artifact codecs", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.ArtifactCodecs = nil }, wantErr: "artifact codec registry"}, + {name: "artifact evidence", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.ArtifactEvidence = nil }, wantErr: "artifact evidence registry"}, {name: "extractors", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.Extractors = nil }, wantErr: "extractor registry"}, {name: "mergers", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.Mergers = nil }, wantErr: "merger registry"}, {name: "normalizers", remove: func(r *pipeline.Registries, _ **llm.AssetRegistry) { r.Normalizers = nil }, wantErr: "normalizer registry"}, @@ -568,15 +610,16 @@ func TestRegisterReportsDuplicateDNDRegistration(t *testing.T) { func completeRegistries() pipeline.Registries { return pipeline.Registries{ - Inputs: pipeline.NewInputAdapterRegistry(), - Chunkers: pipeline.NewChunkerRegistry(), - ArtifactCodecs: pipeline.NewArtifactCodecRegistry(), - Extractors: pipeline.NewExtractorRegistry(), - Mergers: pipeline.NewMergerRegistry(), - Normalizers: pipeline.NewNormalizerRegistry(), - Validators: pipeline.NewValidatorRegistry(), - ValidatorChains: pipeline.NewValidatorChainRegistry(), - Outputs: pipeline.NewOutputEncoderRegistry(), + Inputs: pipeline.NewInputAdapterRegistry(), + Chunkers: pipeline.NewChunkerRegistry(), + ArtifactCodecs: pipeline.NewArtifactCodecRegistry(), + ArtifactEvidence: pipeline.NewArtifactEvidenceRegistry(), + Extractors: pipeline.NewExtractorRegistry(), + Mergers: pipeline.NewMergerRegistry(), + Normalizers: pipeline.NewNormalizerRegistry(), + Validators: pipeline.NewValidatorRegistry(), + ValidatorChains: pipeline.NewValidatorChainRegistry(), + Outputs: pipeline.NewOutputEncoderRegistry(), } } diff --git a/internal/modules/generic/output/json/encoder.go b/internal/modules/generic/output/json/encoder.go index bfef442..ca5c0cd 100644 --- a/internal/modules/generic/output/json/encoder.go +++ b/internal/modules/generic/output/json/encoder.go @@ -28,6 +28,7 @@ var _ contracts.OutputEncoder = (*Encoder)(nil) type Options struct { IncludeChunkMap bool + EvidenceContext pipeline.EvidenceContextPolicy } type Encoder struct { @@ -39,6 +40,7 @@ func New() *Encoder { } func NewWithOptions(options Options) *Encoder { + options.EvidenceContext.LaneIDs = append([]string(nil), options.EvidenceContext.LaneIDs...) return &Encoder{options: options} } @@ -46,6 +48,15 @@ func (e *Encoder) Key() string { return Key } +func (e *Encoder) EvidenceContextPolicy() pipeline.EvidenceContextPolicy { + if e == nil { + return pipeline.EvidenceContextPolicy{} + } + policy := e.options.EvidenceContext + policy.LaneIDs = append([]string(nil), policy.LaneIDs...) + return policy +} + func (e *Encoder) Encode(ctx context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) { if e == nil { return contracts.OutputResult{}, encoderErrorf("encoder must not be nil") @@ -74,7 +85,7 @@ func ModuleSpec() pipeline.ModuleSpec { } func Register(registry *pipeline.OutputEncoderRegistry) error { - return registry.RegisterBuilderWithSpec(ModuleSpec(), validateOptions, func(request pipeline.BuildRequest) (contracts.OutputEncoder, error) { + return registry.RegisterBuilderWithProfileValidation(ModuleSpec(), validateOptions, validateProfileOptions, func(request pipeline.BuildRequest) (contracts.OutputEncoder, error) { options, err := DecodeOptions(request.Options) if err != nil { return nil, err @@ -89,17 +100,126 @@ func validateOptions(options map[string]any) error { } func DecodeOptions(options map[string]any) (Options, error) { - if err := pipeline.RejectUnknownOptions(options, "include_chunk_map"); err != nil { + if err := pipeline.RejectUnknownOptions(options, "include_chunk_map", "evidence_context"); err != nil { return Options{}, encoderErrorf("%w", err) } + decoded := Options{} if value, ok := options["include_chunk_map"]; ok { enabled, ok := value.(bool) if !ok { return Options{}, encoderErrorf("option %q must be a boolean", "include_chunk_map") } - return Options{IncludeChunkMap: enabled}, nil + decoded.IncludeChunkMap = enabled } - return Options{}, nil + if value, ok := options["evidence_context"]; ok { + policy, err := decodeEvidenceContextPolicy(value) + if err != nil { + return Options{}, err + } + decoded.EvidenceContext = policy + } + return decoded, nil +} + +func validateProfileOptions(context pipeline.OutputProfileOptionContext, options map[string]any) error { + decoded, err := DecodeOptions(options) + if err != nil || !decoded.EvidenceContext.Enabled { + return err + } + configured := make(map[string]struct{}, len(context.LaneIDs)) + for _, laneID := range context.LaneIDs { + configured[laneID] = struct{}{} + } + for _, laneID := range decoded.EvidenceContext.LaneIDs { + if _, ok := configured[laneID]; !ok { + return encoderErrorf("evidence_context lane %q is not configured", laneID) + } + } + return nil +} + +func decodeEvidenceContextPolicy(value any) (pipeline.EvidenceContextPolicy, error) { + object, ok := value.(map[string]any) + if !ok { + return pipeline.EvidenceContextPolicy{}, encoderErrorf("option %q must be an object", "evidence_context") + } + if err := pipeline.RejectUnknownOptions(object, "enabled", "lanes", "window_units"); err != nil { + return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context: %w", err) + } + enabledValue, ok := object["enabled"] + if !ok { + return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q is required", "enabled") + } + enabled, ok := enabledValue.(bool) + if !ok { + return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q must be a boolean", "enabled") + } + if !enabled { + if _, ok := object["lanes"]; ok { + return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q is not allowed when disabled", "lanes") + } + if _, ok := object["window_units"]; ok { + return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q is not allowed when disabled", "window_units") + } + return pipeline.EvidenceContextPolicy{}, nil + } + rawLanes, ok := object["lanes"] + if !ok { + return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q is required when enabled", "lanes") + } + lanes, err := decodeEvidenceLaneIDs(rawLanes) + if err != nil { + return pipeline.EvidenceContextPolicy{}, err + } + windowUnits := 3 + if rawWindow, ok := object["window_units"]; ok { + value, ok := rawWindow.(int) + if !ok { + return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q must be an integer", "window_units") + } + if value < 0 { + return pipeline.EvidenceContextPolicy{}, encoderErrorf("evidence_context option %q must not be negative", "window_units") + } + windowUnits = value + } + return pipeline.EvidenceContextPolicy{Enabled: true, WindowUnits: windowUnits, LaneIDs: lanes}, nil +} + +func decodeEvidenceLaneIDs(value any) ([]string, error) { + var raw []any + switch typed := value.(type) { + case []any: + raw = typed + case []string: + raw = make([]any, len(typed)) + for i := range typed { + raw[i] = typed[i] + } + default: + return nil, encoderErrorf("evidence_context option %q must be an array", "lanes") + } + if len(raw) == 0 { + return nil, encoderErrorf("evidence_context option %q must not be empty", "lanes") + } + seen := make(map[string]struct{}, len(raw)) + lanes := make([]string, 0, len(raw)) + for _, value := range raw { + lane, ok := value.(string) + if !ok { + return nil, encoderErrorf("evidence_context lane values must be strings") + } + lane = strings.TrimSpace(lane) + if lane == "" { + return nil, encoderErrorf("evidence_context lane values must not be empty") + } + if _, ok := seen[lane]; ok { + return nil, encoderErrorf("evidence_context lane %q is duplicated", lane) + } + seen[lane] = struct{}{} + lanes = append(lanes, lane) + } + sort.Strings(lanes) + return lanes, nil } type indexFile struct { diff --git a/internal/modules/generic/output/json/encoder_test.go b/internal/modules/generic/output/json/encoder_test.go index 622b759..1f5c152 100644 --- a/internal/modules/generic/output/json/encoder_test.go +++ b/internal/modules/generic/output/json/encoder_test.go @@ -68,13 +68,62 @@ func TestDecodeOptions(t *testing.T) { if err != nil { t.Fatalf("DecodeOptions() error = %v, want nil", err) } - if got != test.want { + if !reflect.DeepEqual(got, test.want) { t.Fatalf("DecodeOptions() = %#v, want %#v", got, test.want) } }) } } +func TestDecodeEvidenceContextOptions(t *testing.T) { + for _, test := range []struct { + name string + options map[string]any + want pipeline.EvidenceContextPolicy + wantErr string + }{ + {name: "disabled", options: map[string]any{"evidence_context": map[string]any{"enabled": false}}}, + {name: "enabled default window", options: map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": []any{"npcs"}}}, want: pipeline.EvidenceContextPolicy{Enabled: true, WindowUnits: 3, LaneIDs: []string{"npcs"}}}, + {name: "explicit zero window and normalized lanes", options: map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": []any{" spells ", "npcs"}, "window_units": 0}}, want: pipeline.EvidenceContextPolicy{Enabled: true, WindowUnits: 0, LaneIDs: []string{"npcs", "spells"}}}, + {name: "duplicate lanes", options: map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": []any{"npcs", " npcs "}}}, wantErr: "duplicated"}, + {name: "unknown nested option", options: map[string]any{"evidence_context": map[string]any{"enabled": false, "extra": true}}, wantErr: "unknown option"}, + {name: "disabled nested fields", options: map[string]any{"evidence_context": map[string]any{"enabled": false, "lanes": []any{"npcs"}}}, wantErr: "not allowed"}, + {name: "invalid object", options: map[string]any{"evidence_context": true}, wantErr: "must be an object"}, + {name: "invalid lane type", options: map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": "npcs"}}, wantErr: "must be an array"}, + {name: "invalid window type", options: map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": []any{"npcs"}, "window_units": "3"}}, wantErr: "must be an integer"}, + } { + t.Run(test.name, func(t *testing.T) { + got, err := DecodeOptions(test.options) + if test.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), test.wantErr) { + t.Fatalf("DecodeOptions() error = %v, want %q", err, test.wantErr) + } + return + } + if err != nil { + t.Fatalf("DecodeOptions() error = %v, want nil", err) + } + if !reflect.DeepEqual(got.EvidenceContext, test.want) { + t.Fatalf("EvidenceContext = %#v, want %#v", got.EvidenceContext, test.want) + } + }) + } +} + +func TestProfileValidationRejectsUnknownEvidenceLaneAndCopiesInputs(t *testing.T) { + registry := pipeline.NewOutputEncoderRegistry() + if err := Register(registry); err != nil { + t.Fatal(err) + } + options := map[string]any{"evidence_context": map[string]any{"enabled": true, "lanes": []any{"npcs"}}} + if err := registry.ValidateProfileOptions(Key, pipeline.OutputProfileOptionContext{LaneIDs: []string{"npcs", "spells"}}, options); err != nil { + t.Fatalf("ValidateProfileOptions() error = %v, want nil", err) + } + if err := registry.ValidateProfileOptions(Key, pipeline.OutputProfileOptionContext{LaneIDs: []string{"spells"}}, options); err == nil || !strings.Contains(err.Error(), "not configured") { + t.Fatalf("ValidateProfileOptions() error = %v, want unknown lane failure", err) + } +} + func TestEncodeReturnsLogicalFilesForNormalizedOutputs(t *testing.T) { req := contracts.OutputRequest{ Manifest: artifacts.RunManifest{RunID: "run-1", PipelineID: "pipeline-1"}, diff --git a/internal/modules/integration/concurrent_runner_test.go b/internal/modules/integration/concurrent_runner_test.go index ba644f6..594ecee 100644 --- a/internal/modules/integration/concurrent_runner_test.go +++ b/internal/modules/integration/concurrent_runner_test.go @@ -153,7 +153,7 @@ func TestRunnerIndependentlyBoundsWorkersAndProviderCallsAcrossRegisteredModules if err != nil { t.Fatalf("Resolve() error = %v", err) } - registries := pipeline.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} + registries := pipeline.Registries{Inputs: catalog.Inputs, Chunkers: catalog.Chunkers, ArtifactCodecs: catalog.ArtifactCodecs, ArtifactEvidence: catalog.ArtifactEvidence, Extractors: catalog.Extractors, Mergers: catalog.Mergers, Normalizers: catalog.Normalizers, Validators: catalog.Validators, ValidatorChains: catalog.ValidatorChains, Outputs: catalog.Outputs} output, err := runPreparedPipeline(t, registries, resolved.ResolvedPipeline, client, pipeline.RunInput{RawInput: readDNDSpellsFixture(t), ExtractWorkers: 3}) if err != nil { t.Fatalf("Run() error = %v", err) diff --git a/internal/modules/integration/dnd_npcs_runner_test.go b/internal/modules/integration/dnd_npcs_runner_test.go index a4d207c..781fca1 100644 --- a/internal/modules/integration/dnd_npcs_runner_test.go +++ b/internal/modules/integration/dnd_npcs_runner_test.go @@ -287,15 +287,16 @@ func (client *fakeNPCProductionLLMClient) requestCount(promptID string) int { func productionNPCRegistries(t *testing.T) pipeline.Registries { t.Helper() registries := pipeline.Registries{ - Inputs: pipeline.NewInputAdapterRegistry(), - Chunkers: pipeline.NewChunkerRegistry(), - ArtifactCodecs: pipeline.NewArtifactCodecRegistry(), - Extractors: pipeline.NewExtractorRegistry(), - Mergers: pipeline.NewMergerRegistry(), - Normalizers: pipeline.NewNormalizerRegistry(), - Validators: pipeline.NewValidatorRegistry(), - ValidatorChains: pipeline.NewValidatorChainRegistry(), - Outputs: pipeline.NewOutputEncoderRegistry(), + Inputs: pipeline.NewInputAdapterRegistry(), + Chunkers: pipeline.NewChunkerRegistry(), + ArtifactCodecs: pipeline.NewArtifactCodecRegistry(), + ArtifactEvidence: pipeline.NewArtifactEvidenceRegistry(), + Extractors: pipeline.NewExtractorRegistry(), + Mergers: pipeline.NewMergerRegistry(), + Normalizers: pipeline.NewNormalizerRegistry(), + Validators: pipeline.NewValidatorRegistry(), + ValidatorChains: pipeline.NewValidatorChainRegistry(), + Outputs: pipeline.NewOutputEncoderRegistry(), } assets := llm.NewAssetRegistry() for _, registration := range []struct { @@ -315,7 +316,7 @@ func productionNPCRegistries(t *testing.T) pipeline.Registries { func moduleCatalog(registries pipeline.Registries) pipeline.ModuleCatalog { return pipeline.ModuleCatalog{ - Inputs: registries.Inputs, Chunkers: registries.Chunkers, ArtifactCodecs: registries.ArtifactCodecs, + Inputs: registries.Inputs, Chunkers: registries.Chunkers, ArtifactCodecs: registries.ArtifactCodecs, ArtifactEvidence: registries.ArtifactEvidence, Extractors: registries.Extractors, Mergers: registries.Mergers, Normalizers: registries.Normalizers, Validators: registries.Validators, ValidatorChains: registries.ValidatorChains, Outputs: registries.Outputs, } diff --git a/internal/modules/integration/dnd_spells_config_test.go b/internal/modules/integration/dnd_spells_config_test.go index 4de698e..c915183 100644 --- a/internal/modules/integration/dnd_spells_config_test.go +++ b/internal/modules/integration/dnd_spells_config_test.go @@ -88,6 +88,7 @@ func dndCapabilityCatalog(t *testing.T, inputSpec, extractorSpec pipeline.Module } codecs := pipeline.NewArtifactCodecRegistry() + evidence := pipeline.NewArtifactEvidenceRegistry() if err := pipeline.RegisterArtifactCodec(codecs, spellcodec.New()); err != nil { t.Fatalf("register capability codec: %v", err) } @@ -112,7 +113,7 @@ func dndCapabilityCatalog(t *testing.T, inputSpec, extractorSpec pipeline.Module } return pipeline.ModuleCatalog{ - Inputs: inputs, Chunkers: chunkers, ArtifactCodecs: codecs, Extractors: extractors, + Inputs: inputs, Chunkers: chunkers, ArtifactCodecs: codecs, ArtifactEvidence: evidence, Extractors: extractors, Mergers: mergers, Normalizers: normalizers, ValidatorChains: pipeline.NewValidatorChainRegistry(), Outputs: outputs, } } @@ -148,6 +149,7 @@ func dndSpellsTestCatalog(t *testing.T, specs dndSpellsCatalogSpecs) pipeline.Mo chunkers := pipeline.NewChunkerRegistry() extractors := pipeline.NewExtractorRegistry() codecs := pipeline.NewArtifactCodecRegistry() + evidence := pipeline.NewArtifactEvidenceRegistry() mergers := pipeline.NewMergerRegistry() normalizers := pipeline.NewNormalizerRegistry() outputs := pipeline.NewOutputEncoderRegistry() @@ -218,14 +220,15 @@ func dndSpellsTestCatalog(t *testing.T, specs dndSpellsCatalogSpecs) pipeline.Mo } return pipeline.ModuleCatalog{ - Inputs: inputs, - Chunkers: chunkers, - ArtifactCodecs: codecs, - Extractors: extractors, - Mergers: mergers, - Normalizers: normalizers, - ValidatorChains: pipeline.NewValidatorChainRegistry(), - Outputs: outputs, + Inputs: inputs, + Chunkers: chunkers, + ArtifactCodecs: codecs, + ArtifactEvidence: evidence, + Extractors: extractors, + Mergers: mergers, + Normalizers: normalizers, + ValidatorChains: pipeline.NewValidatorChainRegistry(), + Outputs: outputs, } } diff --git a/internal/modules/integration/dnd_spells_runner_test.go b/internal/modules/integration/dnd_spells_runner_test.go index 38b294e..df36d6d 100644 --- a/internal/modules/integration/dnd_spells_runner_test.go +++ b/internal/modules/integration/dnd_spells_runner_test.go @@ -284,13 +284,14 @@ func dndSpellsRunnerRegistries(t *testing.T) pipeline.Registries { catalog := dndSpellsTestCatalog(t, dndSpellsCatalogSpecs{}) return pipeline.Registries{ - Inputs: catalog.Inputs, - Chunkers: catalog.Chunkers, - ArtifactCodecs: catalog.ArtifactCodecs, - Extractors: catalog.Extractors, - Mergers: catalog.Mergers, - Normalizers: catalog.Normalizers, - Outputs: catalog.Outputs, + Inputs: catalog.Inputs, + Chunkers: catalog.Chunkers, + ArtifactCodecs: catalog.ArtifactCodecs, + ArtifactEvidence: catalog.ArtifactEvidence, + Extractors: catalog.Extractors, + Mergers: catalog.Mergers, + Normalizers: catalog.Normalizers, + Outputs: catalog.Outputs, } }