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 selectedProfile *pipelineProfileSelection 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 pipelineProfileSelection struct { name string source string overlayPath string } type pipelineFieldOwnership struct { path string sources []string } func loadPipelineWithConfigKit(path string, opts PipelineLoadOptions) (*PipelineConfig, error) { profile := selectedPipelineProfileName(opts) result, err := loadPipelineConfigKitResult(path, profile) 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, profile string) (*configkit.Result[PipelineConfig], error) { return configkit.Load(configkit.File(path), configkit.Options[PipelineConfig]{ Imports: true, Profile: profile, 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 selectedPipelineProfileName(opts PipelineLoadOptions) string { if opts.Profile == nil || *opts.Profile == "" || *opts.Profile == "default" { return "" } return *opts.Profile } 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, Report: boolPtr(DefaultAuditaReport)}, 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.LayerProfile { resolved.selectedProfile = &pipelineProfileSelection{name: selectedPipelineProfileName(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 profile. // 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 }