diff --git a/docs/config.md b/docs/config.md index 328df40..06c3f49 100644 --- a/docs/config.md +++ b/docs/config.md @@ -55,6 +55,10 @@ remote state with an unsafe legacy identity must be migrated before use. - YAML decode is strict (`KnownFields(true)`) and accepts exactly one document: unknown fields or trailing documents fail load. +- A pipeline file may explicitly import additive YAML fragments through the + root-only `composition.imports` list. Imported files contribute fields to one + logical pipeline document; they do not override fields supplied by the root + or another import. - Configured timeout and retry-delay durations must be positive. An omitted artifact timeout continues to inherit its configured Scriptorium timeout. - Session files must be concrete; unresolved `{{ ... }}` placeholders fail load. @@ -68,6 +72,38 @@ remote state with an unsafe legacy identity must be migrated before use. - local (`audio_dir` or `audio_files`), or - S3 (`audio_s3.prefix`). +### Pipeline imports + +Large pipeline configurations may be split into explicitly named fragments: + +```yaml +composition: + imports: + - config/storage.yml + - config/integrations.yaml + +campaigns: + root: /usr/local/share/narratio/campaigns +``` + +Imports are resolved relative to the directory containing the root pipeline +file and are loaded in declaration order. Narratio does not scan directories or +infer fragments. Each import must be a confined regular `.yml` or `.yaml` file: +absolute paths, traversal, symlinks, directories, duplicate files, and an +import of the root pipeline itself are rejected. Only the root pipeline may +contain `composition`; nested composition is rejected. + +Composition is additive. A map may be extended by multiple files when every +leaf is distinct, but a scalar, list, or map/list/scalar kind cannot be claimed +more than once, even when the repeated values are identical. Conflict errors +name the full field path and every source that claimed it. The assembled YAML is +then decoded against the normal strict pipeline schema and defaults are applied +once. + +An imported field has the same meaning it would have in a monolithic root +pipeline. In particular, ordinary relative pipeline paths continue to resolve +from the root pipeline directory, not from the importing fragment's directory. + ## Minimal Working Configuration `pipeline.yml` @@ -155,6 +191,7 @@ Rules: | Field | Type | Required | Default / Rule | | --- | --- | --- | --- | +| `composition.imports[]` | list of strings | No | explicit additive pipeline fragments relative to the root pipeline directory; `.yml` or `.yaml` regular files only | | `pipeline.workspace.root` | string | No | `/var/lib/narratio` | | `pipeline.workspace.cleanup_after_publish` | bool | No | `false` | | `pipeline.campaigns.root` | string | No | `/usr/local/share/narratio/campaigns` | diff --git a/docs/internal/configuration.md b/docs/internal/configuration.md new file mode 100644 index 0000000..bffda3e --- /dev/null +++ b/docs/internal/configuration.md @@ -0,0 +1,48 @@ +# Configuration Internals + +User-visible fields, defaults, and selection behavior belong in the +[Configuration Reference](../config.md). This document describes the internal +pipeline-loading boundary implemented by `internal/config`. + +## Pipeline Loading + +`LoadPipeline` assembles and validates a pipeline in this order: + +1. Parse the root YAML into a presence-aware composition tree. The tree retains + source names, full field paths, node kinds, declaration order, and explicit + zero, false, empty-map, and empty-list values. +2. Remove the root-only `composition` envelope and validate its explicit + `imports` list. +3. Open each import relative to the root pipeline directory through the + confined regular-file boundary. Imports must use a `.yml` or `.yaml` + extension and cannot traverse, use symlinks, repeat a file, import the root, + or contain another composition envelope. +4. Additively merge the root body and imports. Distinct map leaves compose; + repeated scalar or list paths and node-kind disagreements are conflicts. +5. Emit deterministic canonical YAML and strictly decode it into + `PipelineConfig`. +6. Apply pipeline defaults once, then resolve ordinary relative pipeline paths + from the root pipeline file. + +This ordering preserves monolithic configuration behavior. Moving a field to +an imported fragment changes its source ownership, not its path base, default, +or schema semantics. + +## Diagnostics And Runtime Metadata + +Syntax, duplicate-key, composition, conflict, and schema failures include the +relevant source name and full field path. Additive conflicts report every +claiming source so operators can repair the split without repeatedly +rediscovering additional conflicts. + +The loaded pipeline retains private runtime metadata for the absolute root +path, ordered imports, contributing sources, and field ownership. This metadata +does not participate in YAML decoding or alter the public configuration model. + +## Test Surfaces + +`composition_test.go` protects the presence and merge algebra independently of +the public schema. `pipeline_composition_test.go` exercises explicit imports, +confinement, conflicts, strict decoding, metadata, and root-relative path +behavior through `LoadPipeline`. Other configuration tests continue to protect +defaults and validation after assembly. diff --git a/docs/internal/overview.md b/docs/internal/overview.md index 04f9881..1dabec1 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -28,7 +28,7 @@ progress and artifact services resolve durable inputs and outputs. | --- | --- | --- | | Executable | `cmd/narratio` | Process entry, standard stream wiring, argument handoff, and exit status. | | Application orchestration | `internal/app` | Command dispatch, configuration selection, secret-file environment loading, production composition, session locking, planning, execution, restore, cleanup gates, and user-facing reporting. | -| Configuration | `internal/config` | Strict YAML loading, discovery, defaults, normalization, session templating, and validation. | +| Configuration | [`internal/config`](configuration.md) | Presence-aware pipeline composition, strict YAML loading, discovery, defaults, normalization, session templating, and validation. | | Pipeline stages | `internal/stage` | Canonical stage registry, shared stage contract, execution dependencies, and implemented stage behavior. | | External boundaries | `internal/adapters`, `internal/audio` | WhisperX HTTP, downstream subprocesses, notification, object storage, and S3 audio materialization behind Narratio contracts. | | Manifests | `internal/manifest` | Durable session progress, invocation audit state, stage transitions, validation, and atomic persistence. | @@ -82,6 +82,8 @@ trimmed transcript state: neither invalidates the other, while either can stale ## Focused Documentation +- [Configuration Internals](configuration.md): pipeline composition, import + confinement, field ownership, decoding, and root-relative path semantics. - [Adapter Internals](adapters.md): external adapter boundaries, composition, failure behavior, and test surfaces. - [Artifact Internals](artifacts.md): source identities, runtime catalog, diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index c4069fe..89b0671 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -206,7 +206,7 @@ needed by imports and profiles without changing the public pipeline schema. ## Stage 2 — Explicit Additive Imports -**Status: Pending** +**Status: Completed** ### Goal diff --git a/internal/config/composition.go b/internal/config/composition.go index 2a8ed80..d133f55 100644 --- a/internal/config/composition.go +++ b/internal/config/composition.go @@ -166,22 +166,92 @@ func buildCompositionNode(node *yaml.Node, source, path string) (*compositionNod // list, or final keyed value may have only one base owner, regardless of // whether duplicate values happen to be equal. func mergeAdditiveComposition(base, incoming *compositionDocument) (*compositionDocument, error) { - if err := validateCompositionDocument(base, "base"); err != nil { + return mergeAdditiveCompositions(base, incoming) +} + +// mergeAdditiveCompositions validates the complete base source set before +// merging so a conflict names every source that claims the same final path. +func mergeAdditiveCompositions(documents ...*compositionDocument) (*compositionDocument, error) { + if len(documents) == 0 { + return nil, fmt.Errorf("configuration additive base merge requires at least one document") + } + for index, document := range documents { + if err := validateCompositionDocument(document, fmt.Sprintf("base[%d]", index)); err != nil { + return nil, err + } + } + if err := validateAdditiveClaims(documents); err != nil { return nil, err } - if err := validateCompositionDocument(incoming, "incoming"); err != nil { - return nil, err + + result := &compositionDocument{ + root: cloneCompositionNode(documents[0].root), + sources: append([]string(nil), documents[0].sources...), } - merged, err := mergeAdditiveNodes(cloneCompositionNode(base.root), incoming.root) - if err != nil { - return nil, err + for _, incoming := range documents[1:] { + merged, err := mergeAdditiveNodes(result.root, incoming.root) + if err != nil { + return nil, err + } + result.root = merged + result.sources = appendUniqueStrings(result.sources, incoming.sources...) } - return &compositionDocument{ - root: merged, - sources: appendUniqueStrings( - append([]string(nil), base.sources...), incoming.sources..., - ), - }, nil + return result, nil +} + +func validateAdditiveClaims(documents []*compositionDocument) error { + claims := make(map[string][]*compositionNode) + for _, document := range documents { + appendCompositionClaims(document.root, claims) + } + paths := make([]string, 0, len(claims)) + for path := range claims { + paths = append(paths, path) + } + sort.Slice(paths, func(i, j int) bool { + leftDepth := compositionPathDepth(paths[i]) + rightDepth := compositionPathDepth(paths[j]) + if leftDepth != rightDepth { + return leftDepth < rightDepth + } + return paths[i] < paths[j] + }) + for _, path := range paths { + values := claims[path] + if len(values) < 2 { + continue + } + allPopulatedMappings := true + for _, value := range values { + if value.kind != yaml.MappingNode || len(value.fields) == 0 { + allPopulatedMappings = false + break + } + } + if !allPopulatedMappings { + return newCompositionConflict("additive base merge", path, values...) + } + } + return nil +} + +func appendCompositionClaims(node *compositionNode, claims map[string][]*compositionNode) { + if node == nil { + return + } + if node.path != "" { + claims[node.path] = append(claims[node.path], node) + } + for _, field := range node.fields { + appendCompositionClaims(field.value, claims) + } +} + +func compositionPathDepth(path string) int { + if path == "" { + return 0 + } + return strings.Count(path, ".") + strings.Count(path, "[") + 1 } func mergeAdditiveNodes(base, incoming *compositionNode) (*compositionNode, error) { diff --git a/internal/config/composition_test.go b/internal/config/composition_test.go index 9c89ff2..9128e8c 100644 --- a/internal/config/composition_test.go +++ b/internal/config/composition_test.go @@ -1,8 +1,6 @@ package config import ( - "os" - "path/filepath" "reflect" "strings" "testing" @@ -165,6 +163,20 @@ func TestMergeAdditiveCompositionReportsAllClaimingSources(t *testing.T) { t.Fatalf("error = %q, want %q", err, want) } } + + _, err = mergeAdditiveCompositions( + mustParseComposition(t, "first.yml", "value: 1\n"), + mustParseComposition(t, "second.yml", "value: 2\n"), + mustParseComposition(t, "third.yml", "value: 3\n"), + ) + if err == nil { + t.Fatal("mergeAdditiveCompositions() error = nil") + } + for _, want := range []string{"value", "first.yml", "second.yml", "third.yml"} { + if !strings.Contains(err.Error(), want) { + t.Fatalf("error = %q, want %q", err, want) + } + } } func TestMergeOverlayCompositionRecursesMapsAndReplacesAtomicValues(t *testing.T) { @@ -307,21 +319,6 @@ zeta: 1 } } -func TestPipelineCompositionEnvelopeIsNotPublicYet(t *testing.T) { - path := filepath.Join(t.TempDir(), "pipeline.yml") - if err := os.WriteFile(path, []byte(`composition: - imports: [] -whisperx: - transcribe_url: https://transcription.example.com/transcribe -`), 0o644); err != nil { - t.Fatal(err) - } - _, err := LoadPipeline(path) - if err == nil || !strings.Contains(err.Error(), "field composition not found") { - t.Fatalf("LoadPipeline() error = %v, want strict public-schema rejection", err) - } -} - func mustParseComposition(t *testing.T, source, input string) *compositionDocument { t.Helper() document, err := parseCompositionBytes(source, []byte(input)) diff --git a/internal/config/config.go b/internal/config/config.go index 3e3b4ca..7e0cb1e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -32,6 +32,8 @@ type PipelineConfig struct { Scriptorium *ScriptoriumConfig `yaml:"scriptorium"` Notarius *NotariusConfig `yaml:"notarius"` Notification NotificationConfig `yaml:"notification"` + + resolution *pipelineResolutionMetadata `yaml:"-"` } // CampaignsConfig configures the local campaign registry. diff --git a/internal/config/load.go b/internal/config/load.go index 8402b63..0f46237 100644 --- a/internal/config/load.go +++ b/internal/config/load.go @@ -14,15 +14,15 @@ import ( // LoadPipeline loads pipeline configuration from a YAML file with strict field checking. func LoadPipeline(path string) (*PipelineConfig, error) { - var cfg PipelineConfig - if err := decodeStrictYAML("pipeline", path, &cfg); err != nil { + cfg, err := loadComposedPipeline(path) + if err != nil { return nil, fmt.Errorf("load pipeline config: %w", err) } - applyPipelineDefaults(&cfg) - if err := resolveNotariusPaths(&cfg, path); err != nil { + applyPipelineDefaults(cfg) + if err := resolveNotariusPaths(cfg, path); err != nil { return nil, fmt.Errorf("load pipeline config: %w", err) } - return &cfg, nil + return cfg, nil } // LoadCampaign loads campaign configuration from a YAML file with strict field checking. diff --git a/internal/config/notarius_test.go b/internal/config/notarius_test.go index e57bdc3..2ebffa3 100644 --- a/internal/config/notarius_test.go +++ b/internal/config/notarius_test.go @@ -117,8 +117,13 @@ func TestNotariusStrictYAML(t *testing.T) { if err := os.WriteFile(path, []byte(testPipelineBaseYAML+"\n"+tt.yaml), 0o644); err != nil { t.Fatalf("write pipeline: %v", err) } - if _, err := LoadPipeline(path); err == nil || !strings.Contains(err.Error(), "strict decode failed") { - t.Fatalf("LoadPipeline() error = %v, want strict decode failure", err) + _, err := LoadPipeline(path) + want := "strict decode failed" + if tt.name == "duplicate reference selector" { + want = "duplicate YAML key" + } + if err == nil || !strings.Contains(err.Error(), want) { + t.Fatalf("LoadPipeline() error = %v, want containing %q", err, want) } }) } diff --git a/internal/config/pipeline_composition.go b/internal/config/pipeline_composition.go new file mode 100644 index 0000000..fc95102 --- /dev/null +++ b/internal/config/pipeline_composition.go @@ -0,0 +1,224 @@ +package config + +import ( + "fmt" + "os" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/fileops" + "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" + "gopkg.in/yaml.v3" +) + +type pipelineResolutionMetadata struct { + rootPath string + imports []string + sources []string + ownership []pipelineFieldOwnership +} + +type pipelineFieldOwnership struct { + path string + sources []string +} + +type pipelineCompositionEnvelope struct { + imports []string +} + +func loadComposedPipeline(path string) (*PipelineConfig, error) { + rootPath, err := filepath.Abs(path) + if err != nil { + return nil, fmt.Errorf("resolve root pipeline path %q: %w", path, err) + } + rootFile, err := os.Open(rootPath) + if err != nil { + return nil, fmt.Errorf("pipeline file %q: open: %w", path, err) + } + rootDocument, parseErr := parseCompositionDocument(rootPath, rootFile) + closeErr := rootFile.Close() + if parseErr != nil { + return nil, parseErr + } + if closeErr != nil { + return nil, fmt.Errorf("pipeline file %q: close: %w", rootPath, closeErr) + } + + baseRoot, envelope, err := splitPipelineCompositionEnvelope(rootDocument) + if err != nil { + return nil, err + } + imports, err := loadPipelineImports(rootPath, envelope.imports) + if err != nil { + return nil, err + } + documents := make([]*compositionDocument, 0, len(imports)+1) + documents = append(documents, baseRoot) + for _, imported := range imports { + documents = append(documents, imported.document) + } + merged, err := mergeAdditiveCompositions(documents...) + if err != nil { + return nil, err + } + + rendered, err := merged.canonicalYAML() + if err != nil { + return nil, err + } + var cfg PipelineConfig + if err := decodeStrictYAMLFromReader("pipeline", rootPath, strings.NewReader(string(rendered)), &cfg); err != nil { + return nil, fmt.Errorf("assembled pipeline sources %s: %w", formatCompositionSources(merged.sources), err) + } + records, err := merged.semanticRecords() + if err != nil { + return nil, err + } + metadata := &pipelineResolutionMetadata{ + rootPath: rootPath, + sources: append([]string(nil), merged.sources...), + } + for _, imported := range imports { + metadata.imports = append(metadata.imports, imported.path) + } + for _, record := range records { + metadata.ownership = append(metadata.ownership, pipelineFieldOwnership{ + path: record.Path, sources: append([]string(nil), record.Sources...), + }) + } + cfg.resolution = metadata + return &cfg, nil +} + +func splitPipelineCompositionEnvelope(document *compositionDocument) (*compositionDocument, pipelineCompositionEnvelope, error) { + if err := validateCompositionDocument(document, "root pipeline"); err != nil { + return nil, pipelineCompositionEnvelope{}, err + } + root := cloneCompositionNode(document.root) + index := compositionFieldIndex(root.fields, "composition") + if index < 0 { + return &compositionDocument{root: root, sources: append([]string(nil), document.sources...)}, pipelineCompositionEnvelope{}, nil + } + envelopeNode := root.fields[index].value + if envelopeNode.kind != yaml.MappingNode { + return nil, pipelineCompositionEnvelope{}, fmt.Errorf( + "configuration source %s at composition: expected a mapping, got %s", + formatCompositionSources(envelopeNode.sources), yamlKindName(envelopeNode.kind), + ) + } + + var envelope pipelineCompositionEnvelope + for _, field := range envelopeNode.fields { + switch field.key { + case "imports": + if field.value.kind != yaml.SequenceNode { + return nil, pipelineCompositionEnvelope{}, fmt.Errorf( + "configuration source %s at composition.imports: expected a list, got %s", + formatCompositionSources(field.value.sources), yamlKindName(field.value.kind), + ) + } + for itemIndex, item := range field.value.items { + if item.kind != yaml.ScalarNode || item.tag != "!!str" { + return nil, pipelineCompositionEnvelope{}, fmt.Errorf( + "configuration source %s at composition.imports[%d]: expected a string path", + formatCompositionSources(item.sources), itemIndex, + ) + } + envelope.imports = append(envelope.imports, item.value) + } + default: + return nil, pipelineCompositionEnvelope{}, fmt.Errorf( + "configuration source %s at composition.%s: unknown composition field %q", + formatCompositionSources(field.value.sources), field.key, field.key, + ) + } + } + root.fields = append(root.fields[:index], root.fields[index+1:]...) + for fieldIndex := range root.fields { + root.fields[fieldIndex].order = fieldIndex + } + return &compositionDocument{root: root, sources: append([]string(nil), document.sources...)}, envelope, nil +} + +type loadedPipelineImport struct { + path string + document *compositionDocument + info os.FileInfo +} + +func loadPipelineImports(rootPath string, declared []string) ([]loadedPipelineImport, error) { + if len(declared) == 0 { + return nil, nil + } + rootDir := filepath.Dir(rootPath) + rootInfo, err := os.Stat(rootPath) + if err != nil { + return nil, fmt.Errorf("inspect root pipeline file %q: %w", rootPath, err) + } + seenPaths := make(map[string]int, len(declared)) + loaded := make([]loadedPipelineImport, 0, len(declared)) + for index, raw := range declared { + if strings.TrimSpace(raw) != raw || raw == "" { + return nil, fmt.Errorf("composition.imports[%d] must be a non-empty path without surrounding whitespace", index) + } + normalized, err := pathsafe.NormalizeRelativeDestination(raw) + if err != nil { + return nil, fmt.Errorf("composition.imports[%d] path %q is invalid: %w", index, raw, err) + } + extension := filepath.Ext(filepath.FromSlash(normalized)) + if extension != ".yml" && extension != ".yaml" { + return nil, fmt.Errorf("composition.imports[%d] path %q must use .yml or .yaml", index, raw) + } + if prior, duplicate := seenPaths[normalized]; duplicate { + return nil, fmt.Errorf( + "composition.imports[%d] path %q duplicates composition.imports[%d] after normalization", + index, raw, prior, + ) + } + seenPaths[normalized] = index + resolved := filepath.Join(rootDir, filepath.FromSlash(normalized)) + if resolved == rootPath { + return nil, fmt.Errorf("composition.imports[%d] path %q imports the root pipeline itself", index, raw) + } + + file, err := fileops.OpenConfinedRegularFile(rootDir, normalized) + if err != nil { + return nil, fmt.Errorf("open composition.imports[%d] path %q beneath root pipeline directory: %w", index, raw, err) + } + info, statErr := file.Stat() + if statErr != nil { + _ = file.Close() + return nil, fmt.Errorf("inspect composition.imports[%d] path %q: %w", index, raw, statErr) + } + if os.SameFile(rootInfo, info) { + _ = file.Close() + return nil, fmt.Errorf("composition.imports[%d] path %q imports the root pipeline itself", index, raw) + } + for priorIndex, prior := range loaded { + if os.SameFile(prior.info, info) { + _ = file.Close() + return nil, fmt.Errorf( + "composition.imports[%d] path %q references the same file as composition.imports[%d] %q", + index, raw, priorIndex, declared[priorIndex], + ) + } + } + document, parseErr := parseCompositionDocument(resolved, file) + closeErr := file.Close() + if parseErr != nil { + return nil, parseErr + } + if closeErr != nil { + return nil, fmt.Errorf("close composition.imports[%d] path %q: %w", index, raw, closeErr) + } + if compositionFieldIndex(document.root.fields, "composition") >= 0 { + return nil, fmt.Errorf( + "imported configuration source %q declares composition; only the root pipeline may declare composition", + resolved, + ) + } + loaded = append(loaded, loadedPipelineImport{path: resolved, document: document, info: info}) + } + return loaded, nil +} diff --git a/internal/config/pipeline_composition_test.go b/internal/config/pipeline_composition_test.go new file mode 100644 index 0000000..b819284 --- /dev/null +++ b/internal/config/pipeline_composition_test.go @@ -0,0 +1,370 @@ +package config + +import ( + "os" + "path/filepath" + "reflect" + "runtime" + "strings" + "testing" +) + +func TestLoadPipelineCompositionImportsDisjointFieldsAndTracksOwnership(t *testing.T) { + dir := t.TempDir() + rootPath := writePipelineSource(t, dir, "pipeline.yml", `composition: + imports: + - conf.d/platform.yml + - conf.d/artifacts.yml +whisperx: + transcribe_url: https://transcription.example.com/transcribe +`) + platformPath := writePipelineSource(t, dir, "conf.d/platform.yml", `workspace: + root: /srv/narratio/work +storage: + backend: local +`) + artifactsPath := writePipelineSource(t, dir, "conf.d/artifacts.yml", `scriptorium: + artifacts: + player_handout: + enabled: false + session_recap: + enabled: true + prompt_id: dnd.session_recap + output_path: artifacts/session_recap.md +`) + + cfg, err := LoadPipeline(rootPath) + if err != nil { + t.Fatalf("LoadPipeline() error = %v", err) + } + if cfg.Workspace.Root != "/srv/narratio/work" || cfg.Storage.Backend != StorageBackendLocal { + t.Fatalf("imported platform config = workspace=%q storage=%#v", cfg.Workspace.Root, cfg.Storage) + } + if cfg.Scriptorium == nil || len(cfg.Scriptorium.Artifacts) != 2 || cfg.Scriptorium.Artifacts["player_handout"].Enabled { + t.Fatalf("imported artifacts = %#v", cfg.Scriptorium) + } + if cfg.WhisperX.TranscribeURL != "https://transcription.example.com/transcribe" { + t.Fatalf("root field = %q", cfg.WhisperX.TranscribeURL) + } + if cfg.resolution == nil { + t.Fatal("pipeline resolution metadata = nil") + } + wantSources := []string{absolutePath(t, rootPath), absolutePath(t, platformPath), absolutePath(t, artifactsPath)} + if !reflect.DeepEqual(cfg.resolution.sources, wantSources) || !reflect.DeepEqual(cfg.resolution.imports, wantSources[1:]) { + t.Fatalf("resolution sources=%#v imports=%#v, want %#v / %#v", cfg.resolution.sources, cfg.resolution.imports, wantSources, wantSources[1:]) + } + assertPipelineFieldOwner(t, cfg, "whisperx.transcribe_url", absolutePath(t, rootPath)) + assertPipelineFieldOwner(t, cfg, "workspace.root", absolutePath(t, platformPath)) + assertPipelineFieldOwner(t, cfg, "scriptorium.artifacts.session_recap.prompt_id", absolutePath(t, artifactsPath)) +} + +func TestLoadPipelineCompositionMergesDisjointKeyedEntries(t *testing.T) { + dir := t.TempDir() + rootPath := writePipelineSource(t, dir, "pipeline.yml", `composition: + imports: [first.yml, second.yml] +whisperx: + transcribe_url: https://transcription.example.com/transcribe +scriptorium: + artifacts: + root_artifact: + enabled: false +`) + writePipelineSource(t, dir, "first.yml", `scriptorium: + artifacts: + first_artifact: + enabled: false +`) + writePipelineSource(t, dir, "second.yml", `scriptorium: + artifacts: + second_artifact: + enabled: false +`) + + cfg, err := LoadPipeline(rootPath) + if err != nil { + t.Fatal(err) + } + if got := len(cfg.Scriptorium.Artifacts); got != 3 { + t.Fatalf("artifact count = %d, want 3: %#v", got, cfg.Scriptorium.Artifacts) + } +} + +func TestLoadPipelineCompositionRejectsBaseConflictsWithAllSources(t *testing.T) { + tests := []struct { + name string + root string + imports map[string]string + path string + sources []string + }{ + { + name: "root and import identical scalar", + root: "whisperx:\n language: en\n", + imports: map[string]string{"one.yml": "whisperx:\n language: en\n"}, + path: "whisperx.language", sources: []string{"pipeline.yml", "one.yml"}, + }, + { + name: "all import claimants", + imports: map[string]string{ + "one.yml": "workspace:\n root: /one\n", + "two.yml": "workspace:\n root: /two\n", + "three.yml": "workspace:\n root: /three\n", + }, + path: "workspace.root", sources: []string{"one.yml", "two.yml", "three.yml"}, + }, + { + name: "atomic list", + root: "audita:\n modules: [one]\n", + imports: map[string]string{"one.yml": "audita:\n modules: [two]\n"}, + path: "audita.modules", sources: []string{"pipeline.yml", "one.yml"}, + }, + { + name: "kind conflict", + root: "workspace:\n root: /work\n", + imports: map[string]string{"one.yml": "workspace: invalid\n"}, + path: "workspace", sources: []string{"pipeline.yml", "one.yml"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dir := t.TempDir() + order := make([]string, 0, len(tt.imports)) + for _, name := range []string{"one.yml", "two.yml", "three.yml"} { + if _, ok := tt.imports[name]; ok { + order = append(order, name) + } + } + root := "composition:\n imports:\n" + for _, name := range order { + root += " - " + name + "\n" + } + root += tt.root + rootPath := writePipelineSource(t, dir, "pipeline.yml", root) + for name, content := range tt.imports { + writePipelineSource(t, dir, name, content) + } + + _, err := LoadPipeline(rootPath) + if err == nil { + t.Fatal("LoadPipeline() error = nil") + } + if !strings.Contains(err.Error(), tt.path) { + t.Fatalf("error = %q, want path %q", err, tt.path) + } + for _, source := range tt.sources { + if !strings.Contains(err.Error(), source) { + t.Fatalf("error = %q, want source %q", err, source) + } + } + }) + } +} + +func TestLoadPipelineCompositionRejectsUnsafeOrInvalidImports(t *testing.T) { + tests := []struct { + name string + imports []string + setup func(*testing.T, string) + want string + }{ + {name: "empty", imports: []string{""}, want: "non-empty path"}, + {name: "surrounding whitespace", imports: []string{" one.yml "}, want: "surrounding whitespace"}, + {name: "absolute", imports: []string{"/tmp/one.yml"}, want: "invalid"}, + {name: "traversal", imports: []string{"../one.yml"}, want: "invalid"}, + {name: "unsupported extension", imports: []string{"one.json"}, want: ".yml or .yaml"}, + {name: "missing", imports: []string{"missing.yml"}, want: "open composition.imports"}, + {name: "duplicate normalized", imports: []string{"one.yml", "./one.yml"}, setup: func(t *testing.T, dir string) { + writePipelineSource(t, dir, "one.yml", "workspace:\n root: /work\n") + }, want: "duplicates composition.imports"}, + {name: "root self import", imports: []string{"pipeline.yml"}, want: "root pipeline itself"}, + {name: "directory", imports: []string{"directory.yml"}, setup: func(t *testing.T, dir string) { + if err := os.Mkdir(filepath.Join(dir, "directory.yml"), 0o755); err != nil { + t.Fatal(err) + } + }, want: "regular file"}, + {name: "symlink file", imports: []string{"link.yml"}, setup: func(t *testing.T, dir string) { + writePipelineSource(t, dir, "target.yml", "workspace:\n root: /work\n") + if err := os.Symlink("target.yml", filepath.Join(dir, "link.yml")); err != nil { + t.Skipf("symlink unavailable: %v", err) + } + }, want: "not a regular file"}, + {name: "symlink directory", imports: []string{"linked/one.yml"}, setup: func(t *testing.T, dir string) { + writePipelineSource(t, dir, "actual/one.yml", "workspace:\n root: /work\n") + if err := os.Symlink("actual", filepath.Join(dir, "linked")); err != nil { + t.Skipf("symlink unavailable: %v", err) + } + }, want: "not a regular directory"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dir := t.TempDir() + if tt.setup != nil { + tt.setup(t, dir) + } + rootPath := writeImportRoot(t, dir, tt.imports) + _, err := LoadPipeline(rootPath) + if err == nil || !strings.Contains(strings.ToLower(err.Error()), strings.ToLower(tt.want)) { + t.Fatalf("LoadPipeline() error = %v, want containing %q", err, tt.want) + } + }) + } +} + +func TestLoadPipelineCompositionRejectsSameFileAliasesAndImportedComposition(t *testing.T) { + t.Run("same file through hard link", func(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("hard-link identity behavior is platform-specific") + } + dir := t.TempDir() + writePipelineSource(t, dir, "one.yml", "workspace:\n root: /work\n") + if err := os.Link(filepath.Join(dir, "one.yml"), filepath.Join(dir, "two.yml")); err != nil { + t.Skipf("hard links unavailable: %v", err) + } + rootPath := writeImportRoot(t, dir, []string{"one.yml", "two.yml"}) + _, err := LoadPipeline(rootPath) + if err == nil || !strings.Contains(err.Error(), "same file") { + t.Fatalf("LoadPipeline() error = %v, want same-file rejection", err) + } + }) + + t.Run("imported composition", func(t *testing.T) { + dir := t.TempDir() + rootPath := writeImportRoot(t, dir, []string{"nested.yml"}) + writePipelineSource(t, dir, "nested.yml", "composition:\n imports: []\n") + _, err := LoadPipeline(rootPath) + if err == nil || !strings.Contains(err.Error(), "only the root pipeline") || !strings.Contains(err.Error(), "nested.yml") { + t.Fatalf("LoadPipeline() error = %v, want imported composition rejection", err) + } + }) + + t.Run("future profile field", func(t *testing.T) { + dir := t.TempDir() + rootPath := writePipelineSource(t, dir, "pipeline.yml", "composition:\n profiles: {}\n") + _, err := LoadPipeline(rootPath) + if err == nil || !strings.Contains(err.Error(), "unknown composition field") || !strings.Contains(err.Error(), "profiles") { + t.Fatalf("LoadPipeline() error = %v, want profile field rejected before its implementation", err) + } + }) +} + +func TestLoadPipelineCompositionReportsImportedParseAndSchemaSources(t *testing.T) { + t.Run("malformed imported YAML", func(t *testing.T) { + dir := t.TempDir() + rootPath := writeImportRoot(t, dir, []string{"broken.yml"}) + brokenPath := writePipelineSource(t, dir, "broken.yml", "workspace: [\n") + _, err := LoadPipeline(rootPath) + if err == nil || !strings.Contains(err.Error(), absolutePath(t, brokenPath)) || !strings.Contains(err.Error(), "decode YAML") { + t.Fatalf("LoadPipeline() error = %v, want imported parse source", err) + } + }) + + t.Run("unknown imported field", func(t *testing.T) { + dir := t.TempDir() + rootPath := writeImportRoot(t, dir, []string{"unknown.yml"}) + unknownPath := writePipelineSource(t, dir, "unknown.yml", "unknown_field: true\n") + _, err := LoadPipeline(rootPath) + if err == nil || !strings.Contains(err.Error(), absolutePath(t, unknownPath)) || !strings.Contains(err.Error(), "strict decode failed") { + t.Fatalf("LoadPipeline() error = %v, want assembled source-aware strict error", err) + } + }) +} + +func TestLoadPipelineCompositionKeepsRelativePathsRootBased(t *testing.T) { + rootDir := t.TempDir() + monolithicPath := writePipelineSource(t, rootDir, "monolithic.yml", testPipelineBaseYAML+` +notarius: + enabled: true + config_path: tool/notarius.yml + pipeline_id: dnd-session + outputs: + npc_registry: + lane_id: npc-registry + media_type: application/json + schema_id: notarius.dnd.npc_registry + schema_version: v1 +`) + composedPath := writePipelineSource(t, rootDir, "pipeline.yml", `composition: + imports: [conf.d/extraction.yml] +`+testPipelineBaseYAML) + writePipelineSource(t, rootDir, "conf.d/extraction.yml", `notarius: + enabled: true + config_path: tool/notarius.yml + pipeline_id: dnd-session + outputs: + npc_registry: + lane_id: npc-registry + media_type: application/json + schema_id: notarius.dnd.npc_registry + schema_version: v1 +`) + + monolithic, err := LoadPipeline(monolithicPath) + if err != nil { + t.Fatal(err) + } + composed, err := LoadPipeline(composedPath) + if err != nil { + t.Fatal(err) + } + want := filepath.Join(rootDir, "tool", "notarius.yml") + if monolithic.Notarius.ConfigPath != want || composed.Notarius.ConfigPath != want { + t.Fatalf("config paths = monolithic %q composed %q, want %q", monolithic.Notarius.ConfigPath, composed.Notarius.ConfigPath, want) + } + if monolithic.Notarius.WorkingDirectory != filepath.Dir(want) || composed.Notarius.WorkingDirectory != filepath.Dir(want) { + t.Fatalf("working directories = %q / %q", monolithic.Notarius.WorkingDirectory, composed.Notarius.WorkingDirectory) + } +} + +func writeImportRoot(t *testing.T, dir string, imports []string) string { + t.Helper() + var builder strings.Builder + builder.WriteString("composition:\n imports:\n") + for _, imported := range imports { + builder.WriteString(" - ") + if imported == "" { + builder.WriteString(`""`) + } else { + builder.WriteString(`"` + imported + `"`) + } + builder.WriteByte('\n') + } + builder.WriteString("whisperx:\n transcribe_url: https://transcription.example.com/transcribe\n") + return writePipelineSource(t, dir, "pipeline.yml", builder.String()) +} + +func writePipelineSource(t *testing.T, root, relative, content string) string { + t.Helper() + path := filepath.Join(root, filepath.FromSlash(relative)) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte(content), 0o644); err != nil { + t.Fatal(err) + } + return path +} + +func absolutePath(t *testing.T, path string) string { + t.Helper() + absolute, err := filepath.Abs(path) + if err != nil { + t.Fatal(err) + } + return absolute +} + +func assertPipelineFieldOwner(t *testing.T, cfg *PipelineConfig, path, source string) { + t.Helper() + if cfg == nil || cfg.resolution == nil { + t.Fatal("pipeline resolution metadata is absent") + } + for _, ownership := range cfg.resolution.ownership { + if ownership.path == path { + if !reflect.DeepEqual(ownership.sources, []string{source}) { + t.Fatalf("owner of %s = %#v, want %q", path, ownership.sources, source) + } + return + } + } + t.Fatalf("ownership path %q not found: %#v", path, cfg.resolution.ownership) +}