package app import ( "context" "fmt" "log/slog" "os" "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" "gitea.maximumdirect.net/eric/narratio/internal/adapters/notify" "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" "gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/logging" "gitea.maximumdirect.net/eric/narratio/internal/manifest" "gitea.maximumdirect.net/eric/narratio/internal/stage" ) type RunOptions struct { Force bool Env *Env } type RunSummary struct { SessionID string RunID string ManifestPath string RunManifestPath string StageNames []string Executed []string Skipped []string } func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage, opts RunOptions) (*RunSummary, error) { env := opts.Env if env == nil { env = &Env{} } if env.Config == nil { env.Config = cfg } if env.ArtifactStore == nil { env.ArtifactStore = artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) } if env.ManifestStore == nil { env.ManifestStore = &manifest.LocalStore{} } if env.Logger == nil { env.Logger = logging.NewLogger(os.Stderr, slog.LevelInfo) } if _, err := loadSecretsFromConfig(env.Config, env.Logger); err != nil { return nil, fmt.Errorf("load secrets from files: %w", err) } if env.WhisperX == nil { client, err := buildDefaultWhisperXClient(env.Config) if err != nil { return nil, fmt.Errorf("initialize whisperx client: %w", err) } env.WhisperX = client } if env.Seriatim == nil { runner, err := buildDefaultSeriatimRunner(env.Config) if err != nil { return nil, fmt.Errorf("initialize seriatim runner: %w", err) } env.Seriatim = runner } if env.Audita == nil { runner, err := buildDefaultAuditaRunner(env.Config) if err != nil { return nil, fmt.Errorf("initialize audita runner: %w", err) } env.Audita = runner } if env.Scriptorium == nil { env.Scriptorium = scriptorium.NewSubprocessRunner() } if env.Storage == nil { env.Storage = &storage.NoopBackend{} } if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages) { objectStore, err := storage.NewObjectStoreFromConfig(ctx, env.Config) if err != nil { return nil, fmt.Errorf("initialize object store backend: %w", err) } env.ObjectStore = objectStore } if env.Notifier == nil { env.Notifier = ¬ify.NoopSender{} } artifactStore := env.ArtifactStore paths, err := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) if err != nil { return nil, fmt.Errorf("prepare workdir: %w", err) } lock, err := artifactStore.AcquireSessionLockFor(cfg.Session.Campaign, cfg.Session.SessionID) if err != nil { return nil, fmt.Errorf("acquire session lock: %w", err) } defer func() { _ = artifactStore.ReleaseSessionLock(lock) }() manifestPath := paths.ManifestPath m, err := loadOrCreateManifest(ctx, env.ManifestStore, manifestPath, cfg.Session.SessionID) if err != nil { return nil, err } runID, err := artifacts.NewRunID() if err != nil { return nil, fmt.Errorf("generate run id: %w", err) } identityChanged, err := ensureManifestIdentity(cfg, m, runID) if err != nil { return nil, fmt.Errorf("initialize manifest identity: %w", err) } if identityChanged { if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { return nil, fmt.Errorf("save manifest identity %q: %w", manifestPath, err) } } runManifestPath := artifacts.SessionRunManifestPathForCampaign( cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, runID, ) runManifestStore := &manifest.LocalStore{} runManifest, err := runManifestStore.CreateRun( ctx, cfg.Session.SessionID, cfg.Session.Campaign, runID, opts.Force, requestedStageNames(stages), ) if err != nil { return nil, fmt.Errorf("create run manifest: %w", err) } runManifest.SessionManifestPath = manifestPath syncRunManifestIdentityFromSession(m, runManifest) if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { return nil, fmt.Errorf("save initial run manifest %q: %w", runManifestPath, err) } stageEnv := env decisions := decideStageActions(stages, m, opts.Force) runNames := make([]string, 0, len(decisions)) executed := make([]string, 0, len(decisions)) skipped := make([]string, 0, len(decisions)) for _, d := range decisions { s := d.Stage runNames = append(runNames, s.Name()) if d.Action == stageActionSkip { skipped = append(skipped, s.Name()) skipAt := nowUTC() runManifest.SetStageAction(s.Name(), manifest.RunStageActionSkip, skipAt) runManifest.MarkStageSkipped(s.Name(), skipAt, "already_succeeded") if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { return nil, fmt.Errorf("save run manifest after skip %q: %w", s.Name(), err) } env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force) continue } executed = append(executed, s.Name()) now := nowUTC() runManifest.SetStageAction(s.Name(), manifest.RunStageActionRun, now) runManifest.MarkStageRunning(s.Name(), now) if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { return nil, fmt.Errorf("save run manifest before stage %q: %w", s.Name(), err) } m.MarkStageRunning(s.Name(), now) env.Logger.Info("starting stage", "stage", s.Name()) if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { return nil, fmt.Errorf("save manifest before stage %q: %w", s.Name(), err) } env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath) result, err := s.Run(ctx, stageEnv, m) if err != nil { failedAt := nowUTC() m.MarkStageFailed(s.Name(), failedAt, err.Error()) if saveErr := env.ManifestStore.Save(ctx, manifestPath, m); saveErr != nil { return nil, fmt.Errorf("stage %q failed (%v) and manifest save failed (%v)", s.Name(), err, saveErr) } runManifest.MarkStageFailed(s.Name(), failedAt, err.Error()) syncRunManifestIdentityFromSession(m, runManifest) if saveErr := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); saveErr != nil { return nil, fmt.Errorf("stage %q failed (%v) and run-manifest save failed (%v)", s.Name(), err, saveErr) } env.Logger.Info("stage failed", "stage", s.Name(), "error", err) return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err) } outputs := mapResultOutputs(result, runID) succeededAt := nowUTC() m.MarkStageSucceeded(s.Name(), succeededAt, outputs) applyStageResultToManifest(m, s.Name(), result) if opts.Force { invalidateDownstreamSucceededStages(m, s.Name(), succeededAt) } if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { return nil, fmt.Errorf("save manifest after stage %q: %w", s.Name(), err) } runManifest.MarkStageSucceeded(s.Name(), succeededAt, outputs) applyStageResultToRunManifest(runManifest, s.Name(), result) syncRunManifestIdentityFromSession(m, runManifest) if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { return nil, fmt.Errorf("save run manifest after stage %q: %w", s.Name(), err) } env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "succeeded", "path", manifestPath) env.Logger.Info("stage succeeded", "stage", s.Name()) } if err := runPostArchiveCleanup(ctx, env, manifestPath, m, executed); err != nil { failedAt := nowUTC() runManifest.MarkFailed(failedAt, err.Error()) syncRunManifestIdentityFromSession(m, runManifest) if saveErr := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); saveErr != nil { return nil, fmt.Errorf("post-archive cleanup failed (%v) and run-manifest save failed (%v)", err, saveErr) } return nil, fmt.Errorf("post-archive cleanup: %w", err) } completedAt := nowUTC() runManifest.MarkSucceeded(completedAt) syncRunManifestIdentityFromSession(m, runManifest) if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil { return nil, fmt.Errorf("save final run manifest %q: %w", runManifestPath, err) } return &RunSummary{ SessionID: cfg.Session.SessionID, RunID: runID, ManifestPath: manifestPath, RunManifestPath: runManifestPath, StageNames: runNames, Executed: executed, Skipped: skipped, }, nil } func buildDefaultWhisperXClient(cfg *config.Config) (whisperx.Client, error) { if cfg == nil || cfg.Pipeline == nil { return &whisperx.NoopClient{}, nil } wx := cfg.Pipeline.WhisperX if strings.TrimSpace(wx.TranscribeURL) == "" { // Compatibility fallback for tests or internal call paths that bypass config validation. return &whisperx.NoopClient{}, nil } retries := 0 if wx.Retries != nil { retries = *wx.Retries } client, err := whisperx.NewHTTPClientFromConfigValues( wx.TranscribeURL, wx.Language, wx.Timeout, wx.RetryDelay, retries, ) if err != nil { return nil, fmt.Errorf("from pipeline.whisperx: %w", err) } return client, nil } func buildDefaultSeriatimRunner(cfg *config.Config) (seriatim.Runner, error) { if cfg == nil || cfg.Pipeline == nil { return &seriatim.NoopRunner{}, nil } s := cfg.Pipeline.Seriatim if strings.TrimSpace(s.Binary) == "" || strings.TrimSpace(s.Timeout) == "" || strings.TrimSpace(s.OutputSchema) == "" { // Compatibility fallback for tests or internal call paths that bypass config validation/defaults. return &seriatim.NoopRunner{}, nil } report := false if s.Report != nil { report = *s.Report } runner, err := seriatim.NewSubprocessRunnerFromConfigValues( s.Binary, s.Timeout, s.OutputSchema, s.CoalesceGap, report, seriatim.EnvConfig{ OverlapWordRunGap: s.Env.OverlapWordRunGap, OverlapWordRunReorderWindow: s.Env.OverlapWordRunReorderWindow, BackchannelMaxDuration: s.Env.BackchannelMaxDuration, FillerMaxDuration: s.Env.FillerMaxDuration, }, ) if err != nil { return nil, fmt.Errorf("from pipeline.seriatim: %w", err) } return runner, nil } func buildDefaultAuditaRunner(cfg *config.Config) (audita.Runner, error) { if cfg == nil || cfg.Pipeline == nil { return &audita.NoopRunner{}, nil } a := cfg.Pipeline.Audita if strings.TrimSpace(a.Binary) == "" || strings.TrimSpace(a.Timeout) == "" { // Compatibility fallback for tests or internal call paths that bypass config validation/defaults. return &audita.NoopRunner{}, nil } report := false if a.Report != nil { report = *a.Report } runner, err := audita.NewSubprocessRunnerFromConfigValues( a.Binary, a.Timeout, a.LLMAPIKeyEnv, append([]string(nil), a.Modules...), a.BaseURL, a.Model, a.TranscriptDescription, a.ConfigPath, a.OutputSchema, a.WorkDirRetention, a.TotalLLMConcurrency, a.ProposalLLMConcurrency, a.ValidationModel, a.ValidationLLMConcurrency, report, ) if err != nil { return nil, fmt.Errorf("from pipeline.audita: %w", err) } return runner, nil } func loadOrCreateManifest(ctx context.Context, store manifest.Store, path, sessionID string) (*manifest.Manifest, error) { exists, err := fileExists(path) if err != nil { return nil, fmt.Errorf("check manifest existence %q: %w", path, err) } if exists { m, err := store.Load(ctx, path) if err != nil { return nil, fmt.Errorf("load manifest %q: %w", path, err) } return m, nil } m, err := store.Create(ctx, sessionID) if err != nil { return nil, fmt.Errorf("create manifest: %w", err) } if err := store.Save(ctx, path, m); err != nil { return nil, fmt.Errorf("save new manifest %q: %w", path, err) } return m, nil } func fileExists(path string) (bool, error) { _, err := os.Stat(path) if err == nil { return true, nil } if os.IsNotExist(err) { return false, nil } return false, err } func mapResultOutputs(result *stage.StageResult, runID string) []manifest.ArtifactRecord { if result == nil || len(result.Outputs) == 0 { return nil } runID = strings.TrimSpace(runID) out := make([]manifest.ArtifactRecord, 0, len(result.Outputs)) for _, ref := range result.Outputs { localPath := ref.AbsolutePath if localPath == "" { localPath = ref.RelativePath } out = append(out, manifest.ArtifactRecord{ Kind: ref.Kind, LocalPath: localPath, ProducerRunID: runID, RemoteKey: ref.RemoteKey, Checksum: ref.Checksum, }) } return out } func applyStageResultToManifest(m *manifest.Manifest, stageName string, result *stage.StageResult) { if m == nil || result == nil { return } sr := m.Stages[stageName] if sr == nil { return } if len(result.Logs) > 0 { sr.Logs = append([]string(nil), result.Logs...) } if len(result.GeneratedConfigs) > 0 { sr.GeneratedConfigs = append([]string(nil), result.GeneratedConfigs...) } if len(result.Metadata) > 0 { sr.Metadata = result.Metadata } } func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest, runID string) (bool, error) { if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil || m == nil { return false, nil } changed := false campaign := strings.TrimSpace(cfg.Session.Campaign) sessionID := strings.TrimSpace(cfg.Session.SessionID) if sessionID == "" { sessionID = strings.TrimSpace(m.SessionID) } if m.Campaign == "" && campaign != "" { m.Campaign = campaign changed = true } runID = strings.TrimSpace(runID) if runID != "" && m.RunID != runID { m.RunID = runID changed = true } if m.LocalWorkDir == "" && campaign != "" && sessionID != "" && m.RunID != "" { m.LocalWorkDir = artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, campaign, sessionID, m.RunID) changed = true } if m.LocalSpoolDir == "" && campaign != "" && sessionID != "" && m.RunID != "" && strings.TrimSpace(cfg.Pipeline.Spool.Root) != "" { m.LocalSpoolDir = artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, campaign, sessionID, m.RunID) changed = true } if cfg.Pipeline.Storage.S3 != nil { if m.S3Bucket == "" && strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket) != "" { m.S3Bucket = strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket) changed = true } sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, campaign, sessionID) if m.S3SessionPrefix == "" && sessionPrefix != "" { m.S3SessionPrefix = sessionPrefix changed = true } runPrefix := artifacts.S3RunPrefix(sessionPrefix, m.RunID) if m.S3RunPrefix == "" && runPrefix != "" { m.S3RunPrefix = runPrefix changed = true } } return changed, nil } func requestedStageNames(stages []stage.Stage) []string { out := make([]string, 0, len(stages)) for _, s := range stages { if s == nil { continue } out = append(out, s.Name()) } return out } func applyStageResultToRunManifest(m *manifest.RunManifest, stageName string, result *stage.StageResult) { if m == nil || result == nil { return } sr := m.Stages[stageName] if sr == nil { return } if len(result.Logs) > 0 { sr.Logs = append([]string(nil), result.Logs...) } if len(result.GeneratedConfigs) > 0 { sr.GeneratedConfigs = append([]string(nil), result.GeneratedConfigs...) } if len(result.Metadata) > 0 { sr.Metadata = result.Metadata } } func syncRunManifestIdentityFromSession(session *manifest.Manifest, run *manifest.RunManifest) { if session == nil || run == nil { return } run.Campaign = session.Campaign run.LocalWorkDir = session.LocalWorkDir run.LocalSpoolDir = session.LocalSpoolDir run.S3Bucket = session.S3Bucket run.S3SessionPrefix = session.S3SessionPrefix run.S3RunPrefix = session.S3RunPrefix } func manifestPathFor(cfg *config.Config) string { return artifacts.SessionManifestPathForCampaign( cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, ) } func needsObjectStoreForRun(cfg *config.Config, stages []stage.Stage) bool { if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { return false } stageRequested := func(name string) bool { for _, s := range stages { if s != nil && s.Name() == name { return true } } return false } if cfg.Session.Inputs.AudioS3 != nil && stageRequested("prepare") { return true } if !stageRequested("archive") { return false } if cfg.Pipeline.Archive == nil { return false } if cfg.Pipeline.Archive.Enabled != nil && !*cfg.Pipeline.Archive.Enabled { return false } if cfg.Pipeline.Archive.UploadRun != nil && !*cfg.Pipeline.Archive.UploadRun { return false } return true }