eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 27 Wiki Activity

Separate config normalization from validation

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-07 03:32:50 +00:00
parent f8e3e651e6
commit 10a2e89965
6 changed files with 202 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
docs/internal/configuration.md
View File

@@ -30,8 +30,9 @@ pipeline-loading boundary implemented by `internal/config`.
lists are atomic replacements, and null or kind changes fail.
7. Emit deterministic canonical YAML and strictly decode it into
`PipelineConfig`.
8. Apply pipeline defaults once, resolve ordinary relative pipeline paths from
the root pipeline file, and digest the normalized effective mapping.
8. Apply pipeline defaults once, explicitly normalize reusable domain values,
resolve ordinary relative pipeline paths from the root pipeline file, and
digest the normalized effective mapping.
This ordering preserves monolithic configuration behavior. Moving a field to
an imported fragment changes its source ownership, not its path base, default,
@@ -117,12 +118,14 @@ for core pipeline/session rules, external adapters, publish/storage, Notarius,
Scriptorium, and shared scalar/path checks. These are package-local ownership
boundaries rather than extension points or a general validation framework.
Validation fails fast and does not define which of several independent invalid
fields is reported first. Map-backed declarations are nevertheless traversed
deterministically. Notarius validates scalar policy, builds normalized reference
bindings and output lanes independently, and installs all candidate values only
after every check succeeds. Scriptorium validates artifact declarations and
outputs, dependencies and inputs, variables, referenced outputs, and the final
Validation is read-only, fails fast, and does not define which of several
independent invalid fields is reported first. Map-backed declarations are
nevertheless traversed deterministically. Reusable domain normalization is an
explicit preceding operation for both loaded and programmatically constructed
pipelines. Notarius normalization commits canonical scalar, reference, and output
values only after the complete normalization succeeds; its validator then checks
that candidate without changing it. Scriptorium validates artifact declarations
and outputs, dependencies and inputs, variables, referenced outputs, and the final
dependency graph in deterministic artifact order.
Diagnostics identify the relevant semantic field. Composition conflicts also

4
docs/roadmap/implementation.md
View File

@@ -120,7 +120,9 @@ or asset list retains Windows or macOS amd64.
## Stage 2 — Separate configuration normalization from validation
Status: Pending
Status: Complete — Added explicit deterministic pipeline normalization before
validation, made Notarius validation read-only on accepted and rejected inputs,
documented the boundary, and verified config race tests plus app/stage consumers.
Read: `internal/config/load.go`, `defaults.go`, `normalize.go`, `validate*.go`,
`artifact_families.go`, and ConfigKit's “Recorded application resolution”,

3
internal/config/load.go
View File

@@ -95,6 +95,9 @@ func LoadPipelineProfilePair(path, leftProfile, rightProfile string) (*PipelineC
func finalizeLoadedPipeline(path string, cfg *PipelineConfig) (*PipelineConfig, error) {
cfg.resolution.publishDeclared = cfg.Publish != nil
applyPipelineDefaults(cfg)
if err := NormalizePipelineConfig(cfg); err != nil {
return nil, fmt.Errorf("load pipeline config: %w", err)
}
captureLogicalNotariusPaths(cfg)
if err := resolveNotariusPaths(cfg, path); err != nil {
return nil, fmt.Errorf("load pipeline config: %w", err)

48
internal/config/notarius_test.go
View File

@@ -169,8 +169,11 @@ func TestNotariusReferenceValidationAndNormalization(t *testing.T) {
"spells": "narratio.input.spell_catalog",
}
if err := NormalizePipelineConfig(&PipelineConfig{Notarius: cfg}); err != nil {
t.Fatalf("NormalizePipelineConfig() error = %v", err)
}
if err := validateNotarius(cfg, nil); err != nil {
t.Fatalf("validateNotarius() error = %v", err)
t.Fatalf("validateNotarius() after normalization error = %v", err)
}
want := map[string]string{
"party": "narratio.input.party",
@@ -207,7 +210,7 @@ func TestNotariusReferenceValidationRejectsInvalidBindings(t *testing.T) {
" chunk . party ": "narratio.input.players",
},
wantField: "pipeline.notarius.references",
wantReason: "normalize to",
wantReason: "must be normalized",
},
}
@@ -264,19 +267,17 @@ func TestNotariusValidationFailureDoesNotMutateConfig(t *testing.T) {
{
name: "reference",
mutate: func(cfg *NotariusConfig) {
cfg.Binary = " notarius "
cfg.References = map[string]string{" party ": "unsupported"}
},
want: "pipeline.notarius.references.party",
want: "pipeline.notarius.references selector",
},
{
name: "output",
mutate: func(cfg *NotariusConfig) {
cfg.Binary = " notarius "
cfg.References = map[string]string{" party ": " narratio.input.party "}
cfg.References = map[string]string{"party": "narratio.input.party"}
cfg.Outputs["npc_registry"] = NotariusOutputConfig{LaneID: " missing-schema "}
},
want: "pipeline.notarius.outputs.npc_registry.media_type",
want: "pipeline.notarius.outputs.npc_registry.lane_id",
},
}
@@ -314,8 +315,11 @@ func TestNotariusNormalizationProducesCanonicalValues(t *testing.T) {
},
}
if err := NormalizePipelineConfig(&PipelineConfig{Notarius: cfg}); err != nil {
t.Fatalf("NormalizePipelineConfig() error = %v", err)
}
if err := validateNotarius(cfg, nil); err != nil {
t.Fatalf("validateNotarius() error = %v", err)
t.Fatalf("validateNotarius() after normalization error = %v", err)
}
wantReferences := map[string]string{
"party": "narratio.input.party",
@@ -330,6 +334,24 @@ func TestNotariusNormalizationProducesCanonicalValues(t *testing.T) {
if got := cfg.Outputs["npc_registry"]; got.LaneID != "npc-registry" || got.MediaType != "application/json" || got.SchemaID != "notarius.dnd.npc_registry" || got.SchemaVersion != "v1" {
t.Fatalf("normalized npc_registry output = %#v", got)
}
beforeSecondPass := cloneNotariusConfigForTest(cfg)
if err := NormalizePipelineConfig(&PipelineConfig{Notarius: cfg}); err != nil {
t.Fatalf("second NormalizePipelineConfig() error = %v", err)
}
if !reflect.DeepEqual(cfg, beforeSecondPass) {
t.Fatalf("normalization is not deterministic:\n got: %#v\nwant: %#v", cfg, beforeSecondPass)
}
}
func TestNotariusValidationDoesNotMutateAcceptedConfig(t *testing.T) {
cfg := validNotariusConfig()
before := cloneNotariusConfigForTest(cfg)
if err := validateNotarius(cfg, nil); err != nil {
t.Fatalf("validateNotarius() error = %v", err)
}
if !reflect.DeepEqual(cfg, before) {
t.Fatalf("accepted config mutated during validation:\n got: %#v\nwant: %#v", cfg, before)
}
}
func TestNotariusDiagnosticsUseSortedMapKeys(t *testing.T) {
@@ -421,7 +443,7 @@ func validNotariusConfig() *NotariusConfig {
return &NotariusConfig{
Enabled: true,
Binary: "notarius",
ConfigPath: "./notarius.yml",
ConfigPath: "notarius.yml",
PipelineID: "dnd-session",
Timeout: "45m",
WorkingDirectory: ".",
@@ -497,6 +519,14 @@ func TestNotariusEnabledValidation(t *testing.T) {
pipelinePath, sessionPath := writeConfigFiles(t, testPipelineBaseYAML+"\n"+tt.section, testSessionBaseYAML)
cfg, err := Load(pipelinePath, sessionPath)
if err != nil {
if len(tt.wantErr) > 0 {
for _, fragment := range tt.wantErr {
if !strings.Contains(err.Error(), fragment) {
t.Fatalf("Load() error = %v, want containing %q", err, fragment)
}
}
return
}
t.Fatalf("Load() error = %v", err)
}
err = Validate(cfg)

42
internal/config/pipeline_normalize.go Normal file
View File

@@ -0,0 +1,42 @@
package config
import (
"path/filepath"
"strings"
)
// NormalizePipelineConfig applies reusable domain canonicalization before
// validation. It is deterministic and commits no partial Notarius changes when
// canonicalization fails.
func NormalizePipelineConfig(cfg *PipelineConfig) error {
if cfg == nil || cfg.Notarius == nil || !cfg.Notarius.Enabled {
return nil
}
notarius := *cfg.Notarius
notarius.Binary = strings.TrimSpace(notarius.Binary)
notarius.ConfigPath = cleanNotariusPath(notarius.ConfigPath)
notarius.PipelineID = strings.TrimSpace(notarius.PipelineID)
notarius.Timeout = strings.TrimSpace(notarius.Timeout)
notarius.WorkingDirectory = cleanNotariusPath(notarius.WorkingDirectory)
references, err := normalizeNotariusReferences(notarius.References)
if err != nil {
return err
}
outputs, err := normalizeNotariusOutputs(notarius.Outputs)
if err != nil {
return err
}
notarius.References = references
notarius.Outputs = outputs
*cfg.Notarius = notarius
return nil
}
func cleanNotariusPath(value string) string {
value = strings.TrimSpace(value)
if value == "" {
return ""
}
return filepath.Clean(value)
}

157
internal/config/validate_notarius.go
View File

@@ -2,7 +2,6 @@ package config
import (
"fmt"
"path/filepath"
"sort"
"strings"
@@ -16,24 +15,19 @@ func validateNotarius(cfg *NotariusConfig, scriptorium *ScriptoriumConfig) error
return nil
}
binary := strings.TrimSpace(cfg.Binary)
configPath := filepath.Clean(cfg.ConfigPath)
pipelineID := strings.TrimSpace(cfg.PipelineID)
timeout := strings.TrimSpace(cfg.Timeout)
workingDirectory := filepath.Clean(cfg.WorkingDirectory)
if binary == "" {
if strings.TrimSpace(cfg.Binary) == "" {
return fmt.Errorf("pipeline.notarius.binary is required when pipeline.notarius.enabled is true")
}
if strings.TrimSpace(cfg.ConfigPath) == "" {
return fmt.Errorf("pipeline.notarius.config_path is required when pipeline.notarius.enabled is true")
}
if pipelineID == "" {
if strings.TrimSpace(cfg.PipelineID) == "" {
return fmt.Errorf("pipeline.notarius.pipeline_id is required when pipeline.notarius.enabled is true")
}
if timeout == "" {
if strings.TrimSpace(cfg.Timeout) == "" {
return fmt.Errorf("pipeline.notarius.timeout must be a valid duration")
}
if err := validateDuration("pipeline.notarius.timeout", timeout); err != nil {
if err := validateDuration("pipeline.notarius.timeout", cfg.Timeout); err != nil {
return err
}
if strings.TrimSpace(cfg.WorkingDirectory) == "" {
@@ -46,22 +40,32 @@ func validateNotarius(cfg *NotariusConfig, scriptorium *ScriptoriumConfig) error
return fmt.Errorf("pipeline.notarius.references must contain at most %d bindings", MaxNotariusReferenceBindings)
}
references, err := normalizeNotariusReferences(cfg.References)
if err != nil {
if err := validateNotariusCanonicalScalars(cfg); err != nil {
return err
}
outputs, err := normalizeNotariusOutputs(cfg.Outputs, scriptorium)
if err != nil {
if err := validateNotariusReferences(cfg.References); err != nil {
return err
}
return validateNotariusOutputs(cfg.Outputs, scriptorium)
}
cfg.Binary = binary
cfg.ConfigPath = configPath
cfg.PipelineID = pipelineID
cfg.Timeout = timeout
cfg.WorkingDirectory = workingDirectory
cfg.References = references
cfg.Outputs = outputs
func validateNotariusCanonicalScalars(cfg *NotariusConfig) error {
fields := []struct {
name string
got string
want string
}{
{name: "binary", got: cfg.Binary, want: strings.TrimSpace(cfg.Binary)},
{name: "config_path", got: cfg.ConfigPath, want: cleanNotariusPath(cfg.ConfigPath)},
{name: "pipeline_id", got: cfg.PipelineID, want: strings.TrimSpace(cfg.PipelineID)},
{name: "timeout", got: cfg.Timeout, want: strings.TrimSpace(cfg.Timeout)},
{name: "working_directory", got: cfg.WorkingDirectory, want: cleanNotariusPath(cfg.WorkingDirectory)},
}
for _, field := range fields {
if field.got != field.want {
return fmt.Errorf("pipeline.notarius.%s must be normalized", field.name)
}
}
return nil
}
@@ -88,19 +92,12 @@ func normalizeNotariusReferences(references map[string]string) (map[string]strin
owners[selector] = rawSelector
source := strings.TrimSpace(references[rawSelector])
if source == "" {
return nil, fmt.Errorf("pipeline.notarius.references.%s source is required", selector)
}
if _, ok := artifactpolicy.DescribePreparedInputSource(source); !ok {
return nil, fmt.Errorf("pipeline.notarius.references.%s source %q is not a supported prepared input source", selector, source)
}
normalized[selector] = source
}
return normalized, nil
}
func normalizeNotariusOutputs(outputs map[string]NotariusOutputConfig, scriptorium *ScriptoriumConfig) (map[string]NotariusOutputConfig, error) {
reservedSources := notariusReservedSources(scriptorium)
func normalizeNotariusOutputs(outputs map[string]NotariusOutputConfig) (map[string]NotariusOutputConfig, error) {
keys := make([]string, 0, len(outputs))
for key := range outputs {
keys = append(keys, key)
@@ -109,20 +106,13 @@ func normalizeNotariusOutputs(outputs map[string]NotariusOutputConfig, scriptori
normalized := make(map[string]NotariusOutputConfig, len(outputs))
sourceOwners := make(map[string]string, len(outputs))
laneOwners := make(map[string]string, len(outputs))
for _, rawKey := range keys {
output := outputs[rawKey]
key := strings.TrimSpace(rawKey)
if !artifactpolicy.IsConfiguredKey(key) {
return nil, fmt.Errorf("pipeline.notarius.outputs keys must match ^[a-z][a-z0-9_]*$")
}
sourceID := artifactpolicy.ExtractionSourceID(key)
if previousKey, ok := sourceOwners[sourceID]; ok {
return nil, fmt.Errorf("pipeline.notarius.outputs keys %q and %q normalize to duplicate source %q", previousKey, rawKey, sourceID)
}
if owner, ok := reservedSources[sourceID]; ok {
return nil, fmt.Errorf("pipeline.notarius.outputs.%s source %q collides with %s", key, sourceID, owner)
}
sourceOwners[sourceID] = rawKey
output.LaneID = strings.TrimSpace(output.LaneID)
@@ -130,28 +120,89 @@ func normalizeNotariusOutputs(outputs map[string]NotariusOutputConfig, scriptori
output.SchemaID = strings.TrimSpace(output.SchemaID)
output.SchemaVersion = strings.TrimSpace(output.SchemaVersion)
output.ModuleKey = strings.TrimSpace(output.ModuleKey)
prefix := "pipeline.notarius.outputs." + key
if output.LaneID == "" {
return nil, fmt.Errorf("%s.lane_id is required", prefix)
}
if previousKey, ok := laneOwners[output.LaneID]; ok {
return nil, fmt.Errorf("%s.lane_id %q duplicates pipeline.notarius.outputs.%s.lane_id", prefix, output.LaneID, previousKey)
}
laneOwners[output.LaneID] = key
if output.MediaType == "" {
return nil, fmt.Errorf("%s.media_type is required", prefix)
}
if output.SchemaID == "" {
return nil, fmt.Errorf("%s.schema_id is required", prefix)
}
if output.SchemaVersion == "" {
return nil, fmt.Errorf("%s.schema_version is required", prefix)
}
normalized[key] = output
}
return normalized, nil
}
func validateNotariusReferences(references map[string]string) error {
selectors := make([]string, 0, len(references))
for selector := range references {
selectors = append(selectors, selector)
}
sort.Strings(selectors)
for _, selector := range selectors {
canonical, err := notariusref.NormalizeSelector(selector)
if err != nil {
return fmt.Errorf("pipeline.notarius.references selector %q is invalid: %w", selector, err)
}
if canonical != selector {
return fmt.Errorf("pipeline.notarius.references selector %q must be normalized", selector)
}
source := references[selector]
if strings.TrimSpace(source) == "" {
return fmt.Errorf("pipeline.notarius.references.%s source is required", selector)
}
if source != strings.TrimSpace(source) {
return fmt.Errorf("pipeline.notarius.references.%s source must be normalized", selector)
}
if _, ok := artifactpolicy.DescribePreparedInputSource(source); !ok {
return fmt.Errorf("pipeline.notarius.references.%s source %q is not a supported prepared input source", selector, source)
}
}
return nil
}
func validateNotariusOutputs(outputs map[string]NotariusOutputConfig, scriptorium *ScriptoriumConfig) error {
reservedSources := notariusReservedSources(scriptorium)
keys := make([]string, 0, len(outputs))
for key := range outputs {
keys = append(keys, key)
}
sort.Strings(keys)
laneOwners := make(map[string]string, len(outputs))
for _, key := range keys {
output := outputs[key]
if strings.TrimSpace(key) != key || !artifactpolicy.IsConfiguredKey(key) {
return fmt.Errorf("pipeline.notarius.outputs keys must match ^[a-z][a-z0-9_]*$")
}
sourceID := artifactpolicy.ExtractionSourceID(key)
if owner, ok := reservedSources[sourceID]; ok {
return fmt.Errorf("pipeline.notarius.outputs.%s source %q collides with %s", key, sourceID, owner)
}
prefix := "pipeline.notarius.outputs." + key
fields := []struct{ name, value string }{
{name: "lane_id", value: output.LaneID},
{name: "media_type", value: output.MediaType},
{name: "schema_id", value: output.SchemaID},
{name: "schema_version", value: output.SchemaVersion},
{name: "module_key", value: output.ModuleKey},
}
for _, field := range fields {
if field.value != strings.TrimSpace(field.value) {
return fmt.Errorf("%s.%s must be normalized", prefix, field.name)
}
}
if output.LaneID == "" {
return fmt.Errorf("%s.lane_id is required", prefix)
}
if previousKey, ok := laneOwners[output.LaneID]; ok {
return fmt.Errorf("%s.lane_id %q duplicates pipeline.notarius.outputs.%s.lane_id", prefix, output.LaneID, previousKey)
}
laneOwners[output.LaneID] = key
if output.MediaType == "" {
return fmt.Errorf("%s.media_type is required", prefix)
}
if output.SchemaID == "" {
return fmt.Errorf("%s.schema_id is required", prefix)
}
if output.SchemaVersion == "" {
return fmt.Errorf("%s.schema_version is required", prefix)
}
}
return nil
}
func notariusReservedSources(scriptorium *ScriptoriumConfig) map[string]string {
reserved := make(map[string]string)
for _, spec := range artifactmodel.RuntimeTranscriptArtifacts() {