package app import ( "context" "fmt" "log/slog" "os" "path/filepath" "gitea.maximumdirect.net/eric/narratio/internal/adapters/analyzer" "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" "gitea.maximumdirect.net/eric/narratio/internal/adapters/notify" "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 ManifestPath 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 env.WhisperX == nil { env.WhisperX = &whisperx.NoopClient{} } if env.Seriatim == nil { env.Seriatim = &seriatim.NoopRunner{} } if env.Audita == nil { env.Audita = &audita.NoopRunner{} } if env.Analyzer == nil { env.Analyzer = &analyzer.NoopRunner{} } if env.Storage == nil { env.Storage = &storage.NoopBackend{} } if env.Notifier == nil { env.Notifier = ¬ify.NoopSender{} } artifactStore := env.ArtifactStore paths, err := artifactStore.EnsureLayout(cfg.Session.SessionID) if err != nil { return nil, fmt.Errorf("prepare workdir: %w", err) } lock, err := artifactStore.AcquireSessionLock(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 } stageEnv := toStageEnv(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()) continue } executed = append(executed, s.Name()) now := nowUTC() m.MarkStageRunning(s.Name(), now) if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { return nil, fmt.Errorf("save manifest before stage %q: %w", s.Name(), err) } result, err := s.Run(ctx, stageEnv, m) if err != nil { m.MarkStageFailed(s.Name(), nowUTC(), 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) } return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err) } outputs := mapResultOutputs(result) m.MarkStageSucceeded(s.Name(), nowUTC(), outputs) applyStageResultToManifest(m, s.Name(), result) if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { return nil, fmt.Errorf("save manifest after stage %q: %w", s.Name(), err) } } return &RunSummary{ SessionID: cfg.Session.SessionID, ManifestPath: manifestPath, StageNames: runNames, Executed: executed, Skipped: skipped, }, 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) []manifest.ArtifactRecord { if result == nil || len(result.Outputs) == 0 { return nil } 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, 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 manifestPathFor(cfg *config.Config) string { return filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.SessionID, "manifest.json") }