diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 1e8115d5..ec5c02e7 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -217,6 +217,8 @@ cache paths, and local backend presence. No live profile resolution. ## Stage 3 — Implement Pipeline Document Rules +**Status:** Completed 2026-09-13. + **Goal:** Validate and normalize the complete pipeline file shape before catalog resolution. diff --git a/internal/core/config/load.go b/internal/core/config/load.go index 1bd8d9e0..4189bc7b 100644 --- a/internal/core/config/load.go +++ b/internal/core/config/load.go @@ -20,17 +20,20 @@ func loadDocument(path string, options LoadOptions) (*configkit.Result[Document] } func loadDocumentSource(source configkit.Source, options LoadOptions) (*configkit.Result[Document], error) { + state := newDocumentLoadState() return configkit.Load(source, configkit.Options[Document]{ - Definition: documentDefinition(), + Definition: documentDefinition(state), Profile: options.Profile, Overrides: options.Overrides, }) } -func documentDefinition() configkit.Definition[Document] { +func documentDefinition(state *documentLoadState) configkit.Definition[Document] { return configkit.Definition[Document]{ Defaults: documentDefaults, - Resolve: resolveDocumentOperations, + Resolve: func(document *Document, resolution *configkit.Resolution) error { + return resolveDocumentOperations(document, resolution, state) + }, Validate: validateDocumentOperations, Version: &configkit.VersionRule{ Path: configkit.Keys("version"), diff --git a/internal/core/config/operational_document.go b/internal/core/config/operational_document.go index c11002ea..01067a6a 100644 --- a/internal/core/config/operational_document.go +++ b/internal/core/config/operational_document.go @@ -10,12 +10,13 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) -func resolveDocumentOperations(document *Document, resolution *configkit.Resolution) error { +func resolveDocumentOperations(document *Document, resolution *configkit.Resolution, state *documentLoadState) error { if document == nil { return nil } normalizeDocumentPromptKit(document.PromptKit, resolution) normalizeDocumentState(document, resolution) + normalizeDocumentPipelines(document, resolution, state) deriveDocumentExtractWorkers(document, resolution) return nil } @@ -94,6 +95,7 @@ func validateDocumentOperations(document *Document, validation *configkit.Valida validateDocumentConcurrency(document.Concurrency, validation) validateDocumentPromptKit(document.PromptKit, validation) validateDocumentState(document, validation) + validateDocumentPipelines(document, validation) } func validateDocumentConcurrency(concurrency *ConcurrencyDocument, validation *configkit.Validation) { diff --git a/internal/core/config/pipeline_document.go b/internal/core/config/pipeline_document.go new file mode 100644 index 00000000..8c283ec4 --- /dev/null +++ b/internal/core/config/pipeline_document.go @@ -0,0 +1,504 @@ +package config + +import ( + "sort" + "strings" + + configkit "gitea.maximumdirect.net/eric/ConfigKit" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +// documentLoadState keeps the spelling supplied by the user while resolver +// operations move map entries to their canonical keys. It is intentionally +// load-local: source-location handling consumes it during a later resolver +// pass and it never becomes runtime configuration state. +type documentLoadState struct { + originalPaths map[string]configkit.Path +} + +func newDocumentLoadState() *documentLoadState { + return &documentLoadState{originalPaths: make(map[string]configkit.Path)} +} + +func (state *documentLoadState) rememberPath(normalized, original configkit.Path) { + if state != nil { + state.originalPaths[normalized.String()] = original + } +} + +// originalPath follows recorded map-key moves back to the spelling present in +// the source. Resolver code that needs a source coordinate must use this +// rather than reconstructing a path from a normalized identifier. +func (state *documentLoadState) originalPath(path configkit.Path) configkit.Path { + if state == nil || len(state.originalPaths) == 0 { + return path + } + value := path.String() + for { + changed := false + segments := strings.Split(strings.TrimPrefix(value, "/"), "/") + for length := 1; length <= len(segments); length++ { + prefix := "/" + strings.Join(segments[:length], "/") + original, ok := state.originalPaths[prefix] + if !ok { + continue + } + value = original.String() + strings.TrimPrefix(value, prefix) + changed = true + break + } + if !changed { + break + } + } + if original, err := configkit.ParsePath(value); err == nil { + return original + } + return path +} + +func normalizeDocumentPipelines(document *Document, resolution *configkit.Resolution, state *documentLoadState) { + if document == nil { + return + } + validateDocumentPipelinePresence(document, resolution) + + pipelinesPath := configkit.Keys("pipelines") + for _, id := range sortedPipelineDocumentKeys(document.Pipelines) { + profile := document.Pipelines[id] + normalizePipelineDocument(&profile, pipelinesPath.Key(id), resolution, state) + document.Pipelines[id] = profile + } + document.Pipelines = normalizeDocumentMap(document.Pipelines, pipelinesPath, "pipeline identifier", resolution, state) +} + +func validateDocumentPipelinePresence(document *Document, resolution *configkit.Resolution) { + pipelinesPath := configkit.Keys("pipelines") + for _, id := range sortedPipelineDocumentKeys(document.Pipelines) { + profile := document.Pipelines[id] + path := pipelinesPath.Key(id) + validatePipelineDocumentPresence(profile, path, resolution) + } +} + +func validatePipelineDocumentPresence(profile PipelineDocument, path configkit.Path, resolution *configkit.Resolution) { + rejectDocumentNull(resolution, path.Key("llm_profile"), profile.LLMProfile == nil) + rejectDocumentNull(resolution, path.Key("structured_output_repair_attempts"), profile.StructuredOutputRepairAttempts == nil) + rejectDocumentNull(resolution, path.Key("validation_policy"), profile.ValidationPolicy == nil) + validateDocumentPolicyPresence(profile.ValidationPolicy, path.Key("validation_policy"), resolution) + + stepsPath := path.Key("steps") + if resolution.IsExplicit(stepsPath) && profile.Steps == nil { + resolution.Error(stepsPath, "must not be null") + } + if resolution.IsExplicit(stepsPath) && len(profile.Steps) == 0 { + resolution.Error(stepsPath, "must contain at least one ordered step") + } + if resolution.IsExplicit(stepsPath) && resolution.IsExplicit(path.Key("artifacts")) { + resolution.Error(stepsPath, "must not be declared with artifacts", path.Key("artifacts")) + } + + validateBindingDocumentPresence(profile.Input, path.Key("input"), resolution) + validateOptionalBindingDocumentPresence(profile.Chunk, path.Key("chunk"), resolution) + validateOptionalBindingDocumentPresence(profile.Output, path.Key("output"), resolution) + validateReferenceDocumentMapPresence(profile.References, path.Key("references"), resolution) + for _, laneID := range sortedLaneDocumentKeys(profile.Artifacts) { + validateArtifactLaneDocumentPresence(profile.Artifacts[laneID], path.Key("artifacts").Key(laneID), resolution) + } + for index, step := range profile.Steps { + stepPath := stepsPath.Index(index) + validateReferenceDocumentMapPresence(step.References, stepPath.Key("references"), resolution) + for _, laneID := range sortedLaneDocumentKeys(step.Artifacts) { + validateArtifactLaneDocumentPresence(step.Artifacts[laneID], stepPath.Key("artifacts").Key(laneID), resolution) + } + } +} + +func validateArtifactLaneDocumentPresence(lane ArtifactLaneDocument, path configkit.Path, resolution *configkit.Resolution) { + if resolution.IsExplicit(path.Key("references")) { + resolution.Error(path.Key("references"), "artifact-lane references are not supported; use binding references") + } + validateBindingDocumentPresence(lane.Extract, path.Key("extract"), resolution) + validateOptionalBindingDocumentPresence(lane.Merge, path.Key("merge"), resolution) + validateOptionalBindingDocumentPresence(lane.Normalize, path.Key("normalize"), resolution) +} + +func validateOptionalBindingDocumentPresence(binding *BindingDocument, path configkit.Path, resolution *configkit.Resolution) { + if binding == nil { + if resolution.IsExplicit(path) { + resolution.Error(path, "must not be null when configured") + } + return + } + validateBindingDocumentPresence(*binding, path, resolution) +} + +func validateBindingDocumentPresence(binding BindingDocument, path configkit.Path, resolution *configkit.Resolution) { + rejectDocumentNull(resolution, path.Key("llm_profile"), binding.LLMProfile == nil) + rejectDocumentNull(resolution, path.Key("structured_output_repair_attempts"), binding.StructuredOutputRepairAttempts == nil) + rejectDocumentNull(resolution, path.Key("validation_policy"), binding.ValidationPolicy == nil) + validateDocumentPolicyPresence(binding.ValidationPolicy, path.Key("validation_policy"), resolution) + if binding.Validators == nil && resolution.IsExplicit(path.Key("validators")) { + resolution.Error(path.Key("validators"), "must not be null") + } + validateReferenceDocumentMapPresence(binding.References, path.Key("references"), resolution) + for index, validator := range binding.Validators { + validatorPath := path.Key("validators").Index(index) + rejectDocumentNull(resolution, validatorPath.Key("llm_profile"), validator.LLMProfile == nil) + rejectDocumentNull(resolution, validatorPath.Key("structured_output_repair_attempts"), validator.StructuredOutputRepairAttempts == nil) + } +} + +func validateReferenceDocumentMapPresence(references map[string]ReferenceDocument, path configkit.Path, resolution *configkit.Resolution) { + for _, name := range sortedReferenceDocumentKeys(references) { + source := references[name] + sourcePath := path.Key(name) + rejectDocumentNull(resolution, sourcePath.Key("path"), source.Path == nil) + rejectDocumentNull(resolution, sourcePath.Key("artifact"), source.Artifact == nil) + } +} + +func rejectDocumentNull(resolution *configkit.Resolution, path configkit.Path, isNil bool) { + if isNil && resolution.IsExplicit(path) { + resolution.Error(path, "must not be null") + } +} + +func validateDocumentPolicyPresence(policy *ValidationPolicyDocument, path configkit.Path, resolution *configkit.Resolution) { + if policy == nil { + return + } + rejectDocumentNull(resolution, path.Key("producer_structural_failure"), policy.ProducerStructuralFailure == nil) + rejectDocumentNull(resolution, path.Key("semantic_rejection"), policy.SemanticRejection == nil) + rejectDocumentNull(resolution, path.Key("validator_failure"), policy.ValidatorFailure == nil) +} + +func normalizePipelineDocument(profile *PipelineDocument, path configkit.Path, resolution *configkit.Resolution, state *documentLoadState) { + normalizePipelineDocumentString(resolution, path.Key("llm_profile"), &profile.LLMProfile, "trimmed pipeline LLM profile") + normalizeDocumentBinding(&profile.Input, path.Key("input"), resolution, state) + if profile.Chunk != nil { + normalizeDocumentBinding(profile.Chunk, path.Key("chunk"), resolution, state) + } + if profile.Output != nil { + normalizeDocumentBinding(profile.Output, path.Key("output"), resolution, state) + } + profile.References = normalizeDocumentReferences(profile.References, path.Key("references"), resolution, state) + for _, laneID := range sortedLaneDocumentKeys(profile.Artifacts) { + lane := profile.Artifacts[laneID] + normalizeArtifactLaneDocument(&lane, path.Key("artifacts").Key(laneID), resolution, state) + profile.Artifacts[laneID] = lane + } + profile.Artifacts = normalizeDocumentMap(profile.Artifacts, path.Key("artifacts"), "artifact lane identifier", resolution, state) + for index := range profile.Steps { + step := &profile.Steps[index] + stepPath := path.Key("steps").Index(index) + normalizeDocumentValue(resolution, stepPath.Key("id"), &step.ID, strings.TrimSpace(step.ID), "trimmed step identifier") + step.References = normalizeDocumentReferences(step.References, stepPath.Key("references"), resolution, state) + for _, laneID := range sortedLaneDocumentKeys(step.Artifacts) { + lane := step.Artifacts[laneID] + normalizeArtifactLaneDocument(&lane, stepPath.Key("artifacts").Key(laneID), resolution, state) + step.Artifacts[laneID] = lane + } + step.Artifacts = normalizeDocumentMap(step.Artifacts, stepPath.Key("artifacts"), "artifact lane identifier", resolution, state) + } +} + +func normalizeArtifactLaneDocument(lane *ArtifactLaneDocument, path configkit.Path, resolution *configkit.Resolution, state *documentLoadState) { + normalizeDocumentBinding(&lane.Extract, path.Key("extract"), resolution, state) + if lane.Merge != nil { + normalizeDocumentBinding(lane.Merge, path.Key("merge"), resolution, state) + } + if lane.Normalize != nil { + normalizeDocumentBinding(lane.Normalize, path.Key("normalize"), resolution, state) + } + for index := range lane.Validators { + normalizeValidatorDocument(&lane.Validators[index], path.Key("validators").Index(index), resolution) + } +} + +func normalizeDocumentBinding(binding *BindingDocument, path configkit.Path, resolution *configkit.Resolution, state *documentLoadState) { + normalizeDocumentValue(resolution, path.Key("module"), &binding.Module, strings.TrimSpace(binding.Module), "trimmed module identifier") + normalizePipelineDocumentString(resolution, path.Key("llm_profile"), &binding.LLMProfile, "trimmed binding LLM profile") + binding.References = normalizeDocumentReferences(binding.References, path.Key("references"), resolution, state) + for index := range binding.Validators { + normalizeValidatorDocument(&binding.Validators[index], path.Key("validators").Index(index), resolution) + } +} + +func normalizeValidatorDocument(validator *ValidatorDocument, path configkit.Path, resolution *configkit.Resolution) { + normalizeDocumentValue(resolution, path.Key("module"), &validator.Module, strings.TrimSpace(validator.Module), "trimmed validator module identifier") + normalizePipelineDocumentString(resolution, path.Key("llm_profile"), &validator.LLMProfile, "trimmed validator LLM profile") +} + +func normalizeDocumentReferences(references map[string]ReferenceDocument, path configkit.Path, resolution *configkit.Resolution, state *documentLoadState) map[string]ReferenceDocument { + for _, name := range sortedReferenceDocumentKeys(references) { + source := references[name] + sourcePath := path.Key(name) + normalizePipelineDocumentString(resolution, sourcePath.Key("path"), &source.Path, "trimmed reference path") + if source.Artifact != nil { + normalizeDocumentValue(resolution, sourcePath.Key("artifact").Key("step"), &source.Artifact.Step, strings.TrimSpace(source.Artifact.Step), "trimmed artifact step selector") + normalizeDocumentValue(resolution, sourcePath.Key("artifact").Key("lane"), &source.Artifact.Lane, strings.TrimSpace(source.Artifact.Lane), "trimmed artifact lane selector") + } + references[name] = source + } + return normalizeDocumentMap(references, path, "reference name", resolution, state) +} + +func normalizeDocumentValue[T comparable](resolution *configkit.Resolution, path configkit.Path, target *T, value T, reason string) { + if target == nil || !resolution.IsExplicit(path) { + return + } + if *target != value { + resolution.Record(path, configkit.ChangeNormalization, reason) + *target = value + } +} + +func normalizePipelineDocumentString(resolution *configkit.Resolution, path configkit.Path, target **string, reason string) { + if target == nil || *target == nil || !resolution.IsExplicit(path) { + return + } + value := strings.TrimSpace(**target) + if **target != value { + resolution.Record(path, configkit.ChangeNormalization, reason) + **target = value + } +} + +func normalizeDocumentMap[T any](values map[string]T, path configkit.Path, kind string, resolution *configkit.Resolution, state *documentLoadState) map[string]T { + if len(values) == 0 { + return values + } + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + canonical := make(map[string]string, len(values)) + valid := true + for _, key := range keys { + trimmed := strings.TrimSpace(key) + if trimmed == "" { + resolution.Error(path.Key(key), kind+" must not be empty") + valid = false + continue + } + if first, exists := canonical[trimmed]; exists { + resolution.Error(path.Key(key), kind+" collides after whitespace normalization", path.Key(first)) + valid = false + continue + } + canonical[trimmed] = key + } + if !valid { + return values + } + result := make(map[string]T, len(values)) + changed := false + dependencies := make([]configkit.Path, 0, len(keys)) + for _, key := range keys { + trimmed := strings.TrimSpace(key) + result[trimmed] = values[key] + dependencies = append(dependencies, path.Key(key)) + if trimmed != key { + changed = true + state.rememberPath(path.Key(trimmed), path.Key(key)) + } + } + if changed { + resolution.Record(path, configkit.ChangeNormalization, "normalized "+kind+" keys", dependencies...) + values = result + } + return values +} + +func validateDocumentPipelines(document *Document, validation *configkit.Validation) { + if document == nil { + return + } + pipelinesPath := configkit.Keys("pipelines") + for _, id := range sortedPipelineDocumentKeys(document.Pipelines) { + profile := document.Pipelines[id] + path := pipelinesPath.Key(id) + validatePipelineDocument(profile, path, validation) + } +} + +func validatePipelineDocument(profile PipelineDocument, path configkit.Path, validation *configkit.Validation) { + validateSelectedDocumentProfile(profile.LLMProfile, path.Key("llm_profile"), validation) + validateDocumentRepairAttempts(profile.StructuredOutputRepairAttempts, path.Key("structured_output_repair_attempts"), validation) + validateDocumentPolicy(profile.ValidationPolicy, path.Key("validation_policy"), validation) + validateDocumentBinding(profile.Input, path.Key("input"), "input", false, false, validation) + if profile.Chunk != nil { + validateDocumentBinding(*profile.Chunk, path.Key("chunk"), "chunk", true, true, validation) + } + if profile.Output != nil { + validateDocumentBinding(*profile.Output, path.Key("output"), "output", false, false, validation) + } + validateDocumentReferences(profile.References, path.Key("references"), false, validation) + + seenLanes := make(map[string]configkit.Path) + if profile.Steps != nil { + seenSteps := make(map[string]configkit.Path) + for index, step := range profile.Steps { + stepPath := path.Key("steps").Index(index) + if strings.TrimSpace(step.ID) == "" { + validation.Error(stepPath.Key("id"), "must not be empty") + } else if first, exists := seenSteps[step.ID]; exists { + validation.Error(stepPath.Key("id"), "must be unique", first) + } else { + seenSteps[step.ID] = stepPath.Key("id") + } + validateDocumentReferences(step.References, stepPath.Key("references"), true, validation) + validateDocumentLanes(step.Artifacts, stepPath.Key("artifacts"), seenLanes, validation) + } + return + } + validateDocumentLanes(profile.Artifacts, path.Key("artifacts"), seenLanes, validation) +} + +func validateDocumentLanes(lanes map[string]ArtifactLaneDocument, path configkit.Path, seen map[string]configkit.Path, validation *configkit.Validation) { + for _, id := range sortedLaneDocumentKeys(lanes) { + lanePath := path.Key(id) + if first, exists := seen[id]; exists { + validation.Error(lanePath, "artifact lane identifier must be unique across steps", first) + } else { + seen[id] = lanePath + } + lane := lanes[id] + validateDocumentBinding(lane.Extract, lanePath.Key("extract"), "extract", true, true, validation) + if lane.Merge != nil { + validateDocumentBinding(*lane.Merge, lanePath.Key("merge"), "merge", true, true, validation) + } + if lane.Normalize != nil { + validateDocumentBinding(*lane.Normalize, lanePath.Key("normalize"), "normalize", true, true, validation) + } + if len(lane.Validators) > 0 { + validation.Error(lanePath.Key("validators"), "artifact-lane validators are not supported; attach validators to a binding") + } + } +} + +func validateDocumentBinding(binding BindingDocument, path configkit.Path, slot string, referencesAllowed, validatorsAllowed bool, validation *configkit.Validation) { + if strings.TrimSpace(binding.Module) == "" { + validation.Error(path.Key("module"), "must not be empty") + } + validateSelectedDocumentProfile(binding.LLMProfile, path.Key("llm_profile"), validation) + validateDocumentRepairAttempts(binding.StructuredOutputRepairAttempts, path.Key("structured_output_repair_attempts"), validation) + if binding.Retries < 0 { + validation.Error(path.Key("retries"), "must be greater than or equal to zero") + } + if binding.ValidationPolicy != nil { + if slot != "chunk" && slot != "extract" && slot != "merge" && slot != "normalize" { + validation.Error(path.Key("validation_policy"), "is not supported for this binding") + } + validateDocumentPolicy(binding.ValidationPolicy, path.Key("validation_policy"), validation) + } + if len(binding.References) > 0 && !referencesAllowed { + validation.Error(path.Key("references"), "are not supported for this binding") + } + if referencesAllowed { + validateDocumentReferences(binding.References, path.Key("references"), true, validation) + } + if len(binding.Validators) > 0 && !validatorsAllowed { + validation.Error(path.Key("validators"), "are not supported for this binding") + } + for index, validator := range binding.Validators { + validateValidatorDocument(validator, path.Key("validators").Index(index), validation) + } +} + +func validateValidatorDocument(validator ValidatorDocument, path configkit.Path, validation *configkit.Validation) { + if strings.TrimSpace(validator.Module) == "" { + validation.Error(path.Key("module"), "must not be empty") + } + validateSelectedDocumentProfile(validator.LLMProfile, path.Key("llm_profile"), validation) + validateDocumentRepairAttempts(validator.StructuredOutputRepairAttempts, path.Key("structured_output_repair_attempts"), validation) + if validator.Retries < 0 { + validation.Error(path.Key("retries"), "must be greater than or equal to zero") + } +} + +func validateDocumentReferences(references map[string]ReferenceDocument, path configkit.Path, generatedAllowed bool, validation *configkit.Validation) { + for _, name := range sortedReferenceDocumentKeys(references) { + source := references[name] + sourcePath := path.Key(name) + hasPath := source.Path != nil && strings.TrimSpace(*source.Path) != "" + hasArtifact := source.Artifact != nil + if hasPath == hasArtifact { + validation.Error(sourcePath, "must select exactly one reference source") + continue + } + if hasPath { + continue + } + if !generatedAllowed { + validation.Error(sourcePath.Key("artifact"), "is not supported here; use an external path") + continue + } + if strings.TrimSpace(source.Artifact.Step) == "" { + validation.Error(sourcePath.Key("artifact").Key("step"), "must not be empty") + } + if strings.TrimSpace(source.Artifact.Lane) == "" { + validation.Error(sourcePath.Key("artifact").Key("lane"), "must not be empty") + } + } +} + +func validateSelectedDocumentProfile(value *string, path configkit.Path, validation *configkit.Validation) { + if value != nil && strings.TrimSpace(*value) == "" { + validation.Error(path, "must not be empty when set") + } +} + +func validateDocumentRepairAttempts(value *int, path configkit.Path, validation *configkit.Validation) { + if value != nil && (*value < 0 || *value > 3) { + validation.Error(path, "must be between zero and three") + } +} + +func validateDocumentPolicy(policy *ValidationPolicyDocument, path configkit.Path, validation *configkit.Validation) { + if policy == nil { + return + } + if policy.ProducerStructuralFailure != nil { + switch *policy.ProducerStructuralFailure { + case pipeline.ProducerStructuralFailureFailRun, pipeline.ProducerStructuralFailureRejectOutput: + default: + validation.Error(path.Key("producer_structural_failure"), "must be fail_run or reject_output") + } + } + if policy.SemanticRejection != nil { + switch *policy.SemanticRejection { + case pipeline.SemanticRejectionFailRun, pipeline.SemanticRejectionRejectOutput: + default: + validation.Error(path.Key("semantic_rejection"), "must be fail_run or reject_output") + } + } + if policy.ValidatorFailure != nil { + switch *policy.ValidatorFailure { + case pipeline.ValidatorFailureWarnContinue, pipeline.ValidatorFailureFailRun: + default: + validation.Error(path.Key("validator_failure"), "must be warn_continue or fail_run") + } + } +} + +func sortedPipelineDocumentKeys(values map[string]PipelineDocument) []string { + return sortedDocumentKeys(values) +} +func sortedLaneDocumentKeys(values map[string]ArtifactLaneDocument) []string { + return sortedDocumentKeys(values) +} +func sortedReferenceDocumentKeys(values map[string]ReferenceDocument) []string { + return sortedDocumentKeys(values) +} + +func sortedDocumentKeys[T any](values map[string]T) []string { + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} diff --git a/internal/core/config/pipeline_document_test.go b/internal/core/config/pipeline_document_test.go new file mode 100644 index 00000000..b69828fe --- /dev/null +++ b/internal/core/config/pipeline_document_test.go @@ -0,0 +1,197 @@ +package config + +import ( + "errors" + "testing" + + configkit "gitea.maximumdirect.net/eric/ConfigKit" +) + +func TestDocumentPipelineNormalizesCanonicalIdentifiers(t *testing.T) { + t.Parallel() + + loaded, err := loadDocumentSource(configkit.Bytes("document.yml", []byte(`version: 5 +pipelines: + " session ": + llm_profile: " local " + input: + module: " seriatim " + artifacts: + " spells ": + extract: + module: " dnd/spells " + structured_output_repair_attempts: 0 + validators: [] + merge: + module: dnd/merge + references: + previous: + artifact: + step: " default " + lane: " spells " +`)), LoadOptions{}) + if err != nil { + var loadError *configkit.LoadError + if errors.As(err, &loadError) { + t.Fatalf("loadDocumentSource() diagnostics = %#v", loadError.Diagnostics()) + } + t.Fatalf("loadDocumentSource() error = %v", err) + } + profile, ok := loaded.Config.Pipelines["session"] + if !ok || profile.Input.Module != "seriatim" || profile.LLMProfile == nil || *profile.LLMProfile != "local" { + t.Fatalf("normalized pipeline = %#v", loaded.Config.Pipelines) + } + lane, ok := profile.Artifacts["spells"] + if !ok || lane.Extract.StructuredOutputRepairAttempts == nil || *lane.Extract.StructuredOutputRepairAttempts != 0 || lane.Extract.Validators == nil || len(lane.Extract.Validators) != 0 { + t.Fatalf("normalized lane = %#v", profile.Artifacts) + } + selector := lane.Merge.References["previous"].Artifact + if selector == nil || selector.Step != "default" || selector.Lane != "spells" { + t.Fatalf("artifact selector = %#v", selector) + } +} + +func TestDocumentPipelineRejectsInvalidShapes(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + source string + path string + }{ + { + name: "both pipeline forms", + source: `version: 5 +pipelines: + session: + input: {module: seriatim} + artifacts: {} + steps: + - id: first + artifacts: {} +`, + path: "/pipelines/session/steps", + }, + { + name: "empty ordered steps", + source: `version: 5 +pipelines: + session: + input: {module: seriatim} + steps: [] +`, + path: "/pipelines/session/steps", + }, + { + name: "normalized pipeline collision", + source: `version: 5 +pipelines: + session: {input: {module: seriatim}} + " session ": {input: {module: seriatim}} +`, + path: "/pipelines/session", + }, + { + name: "duplicate lane across ordered steps", + source: `version: 5 +pipelines: + session: + input: {module: seriatim} + steps: + - id: first + artifacts: {summary: {extract: {module: dnd/summary}}} + - id: second + artifacts: {summary: {extract: {module: dnd/summary}}} +`, + path: "/pipelines/session/steps/1/artifacts/summary", + }, + { + name: "invalid reference union", + source: `version: 5 +pipelines: + session: + input: {module: seriatim} + references: + party: {path: ./party.txt, artifact: {step: one, lane: summary}} +`, + path: "/pipelines/session/references/party", + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, err := loadDocumentSource(configkit.Bytes("document.yml", []byte(test.source)), LoadOptions{}) + assertDocumentDiagnosticAt(t, err, test.path) + }) + } +} + +func TestDocumentPipelineRejectsInvalidBindingPlacementAndPresence(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + source string + path string + }{ + { + name: "input validators", + source: `version: 5 +pipelines: + session: + input: + module: seriatim + validators: [{module: generic/valid_json}] +`, + path: "/pipelines/session/input/validators", + }, + { + name: "blank selected profile", + source: `version: 5 +pipelines: + session: + llm_profile: " " + input: {module: seriatim} +`, + path: "/pipelines/session/llm_profile", + }, + { + name: "null validators", + source: `version: 5 +pipelines: + session: + input: {module: seriatim} + chunk: {module: generic/chunk, validators: null} +`, + path: "/pipelines/session/chunk/validators", + }, + { + name: "negative retry budget", + source: `version: 5 +pipelines: + session: + input: {module: seriatim, retries: -1} +`, + path: "/pipelines/session/input/retries", + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, err := loadDocumentSource(configkit.Bytes("document.yml", []byte(test.source)), LoadOptions{}) + assertDocumentDiagnosticAt(t, err, test.path) + }) + } +} + +func assertDocumentDiagnosticAt(t *testing.T, err error, path string) { + t.Helper() + var loadError *configkit.LoadError + if !errors.As(err, &loadError) { + t.Fatalf("error = %v, want ConfigKit load error", err) + } + for _, diagnostic := range loadError.Diagnostics() { + if diagnostic.Path.String() == path { + return + } + } + t.Fatalf("diagnostics = %#v, want a diagnostic at %s", loadError.Diagnostics(), path) +}