narratio/internal/config/pipeline_configkit.go

package config

import (
	"bytes"
	"errors"
	"fmt"
	"path/filepath"
	"reflect"
	"sort"
	"strings"

	configkit "gitea.maximumdirect.net/eric/ConfigKit"
)

type pipelineResolutionMetadata struct {
	rootPath                 string
	imports                  []string
	sources                  []string
	selectedOverlay          *pipelineOverlaySelection
	effectiveDigest          string
	logicalNotariusConfig    string
	logicalNotariusWorking   string
	logicalNotariusCaptured  bool
	ownership                []pipelineFieldOwnership
	artifactFamilies         map[string]ScriptoriumArtifactFamilyConfig
	artifactFamiliesExpanded bool
	publishDeclared          bool
	configkitMetadata        configkit.Metadata
	configkitSources         []configkit.SourceInfo
}

type pipelineOverlaySelection struct {
	name        string
	source      string
	overlayPath string
}

type pipelineFieldOwnership struct {
	path    string
	sources []string
}

func loadPipelineWithConfigKit(path string, opts PipelineLoadOptions) (*PipelineConfig, error) {
	overlay := selectedPipelineOverlayName(opts)
	result, err := loadPipelineConfigKitResult(path, overlay)
	if err != nil {
		return nil, describePipelineLoadError(err)
	}
	cfg := result.Config
	metadata, err := pipelineMetadataFromConfigKit(&cfg, result, opts)
	if err != nil {
		return nil, fmt.Errorf("load pipeline config: %w", err)
	}
	cfg.resolution = metadata
	return &cfg, nil
}

func loadPipelineConfigKitResult(path, overlay string) (*configkit.Result[PipelineConfig], error) {
	return configkit.Load(configkit.File(path), configkit.Options[PipelineConfig]{
		Imports: true,
		Overlay: overlay,
		Definition: configkit.Definition[PipelineConfig]{
			Defaults: pipelineConfigDefaults,
			Resolve:  resolvePipelineConfig,
			Validate: validatePipelineConfigKit,
		},
	})
}

func describePipelineLoadError(err error) error {
	return describeConfigKitLoadError("pipeline", err)
}

func describeConfigKitLoadError(kind string, err error) error {
	var loadError *configkit.LoadError
	if !errors.As(err, &loadError) {
		return fmt.Errorf("load %s config: %w", kind, err)
	}
	var diagnostics bytes.Buffer
	resolved := loadError.Diagnostics()
	sourcePaths := make(map[string]string, len(loadError.Sources()))
	for _, source := range loadError.Sources() {
		if source.Path != "" {
			sourcePaths[source.ID] = source.Path
		} else if source.Label != "" {
			sourcePaths[source.ID] = source.Label
		}
	}
	for index := range resolved {
		if resolved[index].Primary != nil {
			if name := sourcePaths[resolved[index].Primary.SourceID]; name != "" {
				resolved[index].Primary.SourceID = name
			}
		}
		for relatedIndex := range resolved[index].Related {
			if name := sourcePaths[resolved[index].Related[relatedIndex].SourceID]; name != "" {
				resolved[index].Related[relatedIndex].SourceID = name
			}
		}
	}
	if writeErr := configkit.WriteDiagnostics(&diagnostics, resolved); writeErr != nil {
		return fmt.Errorf("load %s config: %w", kind, err)
	}
	return fmt.Errorf("load %s config: strict decode failed or validation failed: %w: %s", kind, err, strings.TrimSpace(diagnostics.String()))
}

func selectedPipelineOverlayName(opts PipelineLoadOptions) string {
	if opts.Overlay == nil || *opts.Overlay == "" || *opts.Overlay == "default" {
		return ""
	}
	return *opts.Overlay
}

func pipelineConfigDefaults() PipelineConfig {
	return PipelineConfig{
		Workspace: WorkspaceConfig{Root: DefaultWorkspaceRoot},
		Campaigns: CampaignsConfig{Root: DefaultCampaignsRoot},
		Storage:   StorageConfig{Backend: StorageBackendLocal},
		Spool:     SpoolConfig{Root: DefaultSpoolRoot},
		Cache:     CacheConfig{Root: DefaultCacheRoot, S3Audio: boolPtr(DefaultCacheS3Audio)},
		Publish: &PublishConfig{
			Enabled: boolPtr(DefaultArchiveEnabled), UploadRun: boolPtr(DefaultArchiveUploadRun),
			Outputs: defaultPublishOutputs(),
		},
		WhisperX: WhisperXConfig{
			Language: DefaultWhisperXLanguage, Timeout: DefaultWhisperXTimeout,
			RetryDelay: DefaultWhisperXRetryDelay, Concurrency: intPtr(DefaultWhisperXConcurrency),
			Retries: intPtr(DefaultWhisperXRetries),
		},
		Seriatim: SeriatimConfig{
			Binary: DefaultSeriatimBinary, Timeout: DefaultSeriatimTimeout,
			OutputSchema: DefaultSeriatimOutputSchema, CoalesceGap: float64Ptr(DefaultSeriatimCoalesceGap),
			Report: boolPtr(DefaultSeriatimReport),
		},
		Audita:    AuditaConfig{Binary: DefaultAuditaBinary, Timeout: DefaultAuditaTimeout},
		Normalize: &NormalizeConfig{OutputPath: DefaultNormalizeOutputPath, OutputSchema: DefaultNormalizeOutputSchema, Report: boolPtr(DefaultNormalizeReport)},
		Trim: &TrimConfig{
			Enabled: boolPtr(DefaultTrimEnabled), OutputPath: DefaultTrimOutputPath,
			Bounds: TrimBoundsConfig{
				PromptID: DefaultTrimBoundsPromptID, TranscriptInputName: DefaultTrimBoundsTranscriptInputName,
				OutputPath: DefaultTrimBoundsOutputPath, Timeout: DefaultTrimBoundsTimeout,
			},
			Seriatim: TrimSeriatimConfig{Report: boolPtr(DefaultTrimSeriatimReport)},
		},
		Render: &RenderConfig{
			Enabled: boolPtr(DefaultRenderEnabled), Format: DefaultRenderFormat, Title: DefaultRenderTitle,
			IncludeTimestamps: boolPtr(DefaultRenderTimestamps), IncludeSegmentIDs: boolPtr(DefaultRenderSegmentIDs),
		},
		Notification: NotificationConfig{Mode: DefaultNotificationMode},
	}
}

func defaultPublishOutputs() []PublishOutputRule {
	outputs := append([]PublishOutputRule(nil), DefaultPublishOutputs...)
	for index := range outputs {
		outputs[index].Required = boolPtr(true)
	}
	return outputs
}

func resolvePipelineConfig(cfg *PipelineConfig, resolution *configkit.Resolution) error {
	backend := strings.ToLower(strings.TrimSpace(cfg.Storage.Backend))
	if backend != cfg.Storage.Backend {
		configkit.Normalize(resolution, configkit.Keys("storage", "backend"), &cfg.Storage.Backend, backend, "canonical storage backend")
	}
	if cfg.Storage.S3 != nil {
		path := configkit.Keys("storage", "s3")
		configkit.Default(resolution, path.Key("root_prefix"), &cfg.Storage.S3.RootPrefix, DefaultStorageS3RootPrefix, "default S3 root prefix")
	}
	if cfg.Trim != nil && cfg.Trim.Enabled != nil && *cfg.Trim.Enabled && cfg.Scriptorium == nil {
		value := &ScriptoriumConfig{Binary: DefaultScriptoriumBinary, Timeout: DefaultScriptoriumTimeout}
		configkit.Default(resolution, configkit.Keys("scriptorium"), &cfg.Scriptorium, value, "trim requires Scriptorium")
	}
	if cfg.Scriptorium != nil {
		path := configkit.Keys("scriptorium")
		configkit.Default(resolution, path.Key("binary"), &cfg.Scriptorium.Binary, DefaultScriptoriumBinary, "default Scriptorium binary")
		configkit.Default(resolution, path.Key("timeout"), &cfg.Scriptorium.Timeout, DefaultScriptoriumTimeout, "default Scriptorium timeout")
	}
	if cfg.Notarius != nil {
		path := configkit.Keys("notarius")
		configkit.Default(resolution, path.Key("binary"), &cfg.Notarius.Binary, DefaultNotariusBinary, "default Notarius binary")
		configkit.Default(resolution, path.Key("timeout"), &cfg.Notarius.Timeout, DefaultNotariusTimeout, "default Notarius timeout")
		if cfg.Notarius.Enabled && cfg.Notarius.ConfigPath != "" {
			configkit.Default(resolution, path.Key("working_directory"), &cfg.Notarius.WorkingDirectory,
				filepath.Dir(cfg.Notarius.ConfigPath), "derived from Notarius config path", path.Key("config_path"))
		}
	}
	if cfg.Publish != nil {
		defaultedOutput := false
		for index := range cfg.Publish.Outputs {
			output := cfg.Publish.Outputs[index]
			path := configkit.Keys("publish", "outputs").Index(index).Key("required")
			if !resolution.IsExplicit(path) {
				defaultedOutput = true
			}
			configkit.Default(resolution, path, &output.Required, boolPtr(true), "publish outputs are required by default")
			cfg.Publish.Outputs[index] = output
		}
		if defaultedOutput {
			resolution.Record(configkit.Keys("publish", "outputs"), configkit.ChangeDerivedDefault, "publish output entry defaults")
		}
	}
	before := cloneNotariusForResolution(cfg.Notarius)
	if err := NormalizePipelineConfig(cfg); err != nil {
		resolution.Error(configkit.Keys("notarius"), err.Error())
		return nil
	}
	if !reflect.DeepEqual(before, cfg.Notarius) {
		resolution.Record(configkit.Keys("notarius"), configkit.ChangeNormalization, "canonical Notarius configuration")
	}
	if cfg.Publish != nil {
		destinations := deriveMissingPublishDestinations(cfg.Publish, cfg.Scriptorium, cfg.Notarius)
		for _, derived := range destinations {
			output := cfg.Publish.Outputs[derived.index]
			configkit.Normalize(resolution, configkit.Keys("publish", "outputs").Index(derived.index).Key("dest"), &output.Dest, derived.dest, "derived from publish source")
			cfg.Publish.Outputs[derived.index] = output
		}
		if len(destinations) > 0 {
			// ConfigKit compares sequences atomically; child records alone do not
			// cover the list mutation when requiredness was already explicit.
			resolution.Record(configkit.Keys("publish", "outputs"), configkit.ChangeNormalization, "derived publish destinations")
		}
	}
	return nil
}

func cloneNotariusForResolution(value *NotariusConfig) *NotariusConfig {
	if value == nil {
		return nil
	}
	clone := *value
	clone.References = cloneStringMap(value.References)
	if value.Outputs != nil {
		clone.Outputs = make(map[string]NotariusOutputConfig, len(value.Outputs))
		for key, output := range value.Outputs {
			clone.Outputs[key] = output
		}
	}
	return &clone
}

func cloneStringMap(value map[string]string) map[string]string {
	if value == nil {
		return nil
	}
	clone := make(map[string]string, len(value))
	for key, item := range value {
		clone[key] = item
	}
	return clone
}

func validatePipelineConfigKit(cfg *PipelineConfig, validation *configkit.Validation) {
	// Checks that depend on family expansion or campaign/session context remain
	// at Narratio's explicit resolution boundary. This early definition check is
	// deliberately limited to an invariant that is wholly file-local.
	if strings.TrimSpace(cfg.Workspace.Root) == "" {
		validation.Error(configkit.Keys("workspace", "root"), "workspace root is required")
	}
}

func pipelineMetadataFromConfigKit(cfg *PipelineConfig, result *configkit.Result[PipelineConfig], opts PipelineLoadOptions) (*pipelineResolutionMetadata, error) {
	sources := result.Sources()
	metadata := result.Metadata()
	if err := rejectDuplicateBaseDefinitions(metadata, sources); err != nil {
		return nil, err
	}
	resolved := &pipelineResolutionMetadata{
		configkitMetadata: metadata, configkitSources: sources,
		publishDeclared: metadata.IsExplicit(configkit.Keys("publish")),
	}
	pathsByID := make(map[string]string, len(sources))
	for index, source := range sources {
		path := source.Path
		if path == "" {
			path = source.Label
		}
		pathsByID[source.ID] = path
		resolved.sources = append(resolved.sources, path)
		if index == 0 {
			resolved.rootPath = path
		} else if source.Layer == configkit.LayerBase {
			resolved.imports = append(resolved.imports, path)
		}
		if source.Layer == configkit.LayerOverlay {
			resolved.selectedOverlay = &pipelineOverlaySelection{name: selectedPipelineOverlayName(opts), source: "explicit", overlayPath: path}
		}
	}
	detailed, err := pipelineMetadataPaths(cfg)
	if err != nil {
		return nil, err
	}
	for _, value := range detailed {
		info, ok := metadata.Lookup(configkit.Keys(value.segments...))
		sourceNames := []string{pipelineDefaultOwnershipSource}
		if ok {
			sourceNames = ownershipSources(info.Origin, pathsByID)
		}
		resolved.ownership = append(resolved.ownership, pipelineFieldOwnership{path: value.record.Path, sources: sourceNames})
	}
	return resolved, nil
}

func pipelineMetadataPaths(cfg *PipelineConfig) ([]projectedPipelineValue, error) {
	effective, err := projectionDetailedRecords(projectEffectivePipeline(cfg, false))
	if err != nil {
		return nil, err
	}
	if cfg.Scriptorium == nil || len(cfg.Scriptorium.ArtifactFamilies) == 0 {
		return effective, nil
	}
	families, err := projectionDetailedRecords(map[string]any{"scriptorium": map[string]any{"artifact_families": cfg.Scriptorium.ArtifactFamilies}})
	if err != nil {
		return nil, err
	}
	return append(effective, families...), nil
}

func ownershipSources(origin configkit.Origin, pathsByID map[string]string) []string {
	if origin.Layer == configkit.LayerDefault || (origin.Layer == configkit.LayerResolution && origin.Operation == configkit.ChangeDerivedDefault) {
		return []string{pipelineDefaultOwnershipSource}
	}
	if origin.Layer == configkit.LayerResolution {
		return []string{"resolution"}
	}
	seen := map[string]struct{}{}
	var sources []string
	for _, location := range origin.Locations {
		value := pathsByID[location.SourceID]
		if value == "" {
			value = location.SourceID
		}
		if _, ok := seen[value]; !ok {
			seen[value] = struct{}{}
			sources = append(sources, value)
		}
	}
	if len(sources) == 0 {
		return []string{pipelineDefaultOwnershipSource}
	}
	return sources
}

func rejectDuplicateBaseDefinitions(metadata configkit.Metadata, sources []configkit.SourceInfo) error {
	pathsByID := make(map[string]string, len(sources))
	for _, source := range sources {
		pathsByID[source.ID] = source.Path
	}
	// Enumerate retained declarations, not just surviving effective keys. Loading
	// the base separately would also resolve credentials removed by the overlay.
	// ConfigKit returns canonical JSON Pointers in deterministic order.
	for _, path := range metadata.HistoricalPaths() {
		parts := strings.Split(path.String(), "/")
		if len(parts) != 4 || parts[1] != "scriptorium" {
			continue
		}
		var kind string
		switch parts[2] {
		case "artifacts":
			kind = "artifact"
		case "artifact_families":
			kind = "artifact family"
		default:
			continue
		}
		seen := map[string]struct{}{}
		for _, change := range metadata.History(path) {
			if change.Origin.Layer != configkit.LayerBase {
				continue
			}
			for _, location := range change.Origin.Locations {
				seen[location.SourceID] = struct{}{}
			}
		}
		if len(seen) > 1 {
			claims := make([]string, 0, len(seen))
			for sourceID := range seen {
				claims = append(claims, pathsByID[sourceID])
			}
			sort.Strings(claims)
			key := strings.NewReplacer("~1", "/", "~0", "~").Replace(parts[3])
			return fmt.Errorf("pipeline Scriptorium %s %q is declared by multiple base files: %s", kind, key, strings.Join(claims, ", "))
		}
	}
	return nil
}

func intPtr(value int) *int             { return &value }
func float64Ptr(value float64) *float64 { return &value }
func boolPtr(value bool) *bool          { return &value }