narratio/internal/config/pipeline_composition_test.go

package config

import (
	"os"
	"path/filepath"
	"reflect"
	"runtime"
	"strings"
	"testing"
)

func TestSplitPipelineCompositionEnvelopePreservesInputAndDeclarationOrder(t *testing.T) {
	document, err := parseCompositionBytes("pipeline.yml", []byte(`composition:
  imports: [first.yml, second.yml]
  default_profile: production
  profiles:
    production:
      overlay: production.yml
    testing:
      overlay: testing.yml
workspace:
  root: /srv/narratio
`))
	if err != nil {
		t.Fatal(err)
	}
	originalRoot := cloneCompositionNode(document.root)
	originalSources := append([]string(nil), document.sources...)

	body, envelope, err := splitPipelineCompositionEnvelope(document)
	if err != nil {
		t.Fatalf("splitPipelineCompositionEnvelope() error = %v", err)
	}
	if !reflect.DeepEqual(document.root, originalRoot) || !reflect.DeepEqual(document.sources, originalSources) {
		t.Fatalf("splitPipelineCompositionEnvelope() mutated its input")
	}
	if compositionFieldIndex(body.root.fields, "composition") >= 0 {
		t.Fatalf("pipeline body still contains composition envelope")
	}
	if compositionFieldIndex(body.root.fields, "workspace") < 0 {
		t.Fatalf("pipeline body lost ordinary root fields")
	}
	if want := []string{"first.yml", "second.yml"}; !reflect.DeepEqual(envelope.imports, want) {
		t.Fatalf("imports = %#v, want %#v", envelope.imports, want)
	}
	if len(envelope.profiles) != 2 || envelope.profiles[0].name != "production" || envelope.profiles[1].name != "testing" {
		t.Fatalf("profiles = %#v, want declaration order", envelope.profiles)
	}
}

func TestSplitPipelineCompositionEnvelopeRejectsMalformedShapesWithoutMutation(t *testing.T) {
	tests := []struct {
		name string
		yaml string
		want string
	}{
		{name: "envelope", yaml: "composition: []\n", want: "expected a mapping"},
		{name: "imports mapping", yaml: "composition:\n  imports: {}\n", want: "expected a list"},
		{name: "import value", yaml: "composition:\n  imports: [1]\n", want: "expected a string path"},
		{name: "default profile", yaml: "composition:\n  default_profile: {}\n", want: "expected a string profile name"},
		{name: "profiles sequence", yaml: "composition:\n  profiles: []\n", want: "expected a mapping"},
		{name: "empty profiles", yaml: "composition:\n  profiles: {}\n", want: "at least one named profile"},
		{name: "profile declaration", yaml: "composition:\n  profiles:\n    production: production.yml\n", want: "expected a mapping"},
		{name: "invalid profile name", yaml: "composition:\n  profiles:\n    ' production ':\n      overlay: production.yml\n", want: "surrounding whitespace"},
		{name: "unknown field", yaml: "composition:\n  unexpected: true\n", want: "unknown composition field"},
		{name: "unknown default", yaml: "composition:\n  default_profile: testing\n  profiles:\n    production:\n      overlay: production.yml\n", want: "does not name a declared profile"},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			document, err := parseCompositionBytes("pipeline.yml", []byte(test.yaml))
			if err != nil {
				t.Fatal(err)
			}
			originalRoot := cloneCompositionNode(document.root)

			_, _, err = splitPipelineCompositionEnvelope(document)
			if err == nil || !strings.Contains(err.Error(), test.want) {
				t.Fatalf("splitPipelineCompositionEnvelope() error = %v, want containing %q", err, test.want)
			}
			if !reflect.DeepEqual(document.root, originalRoot) {
				t.Fatalf("splitPipelineCompositionEnvelope() mutated invalid input")
			}
		})
	}
}

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: "is a 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("empty profiles", 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(), "must declare at least one named profile") {
			t.Fatalf("LoadPipeline() error = %v, want empty profiles rejection", 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)
}