narratio/internal/app/runner.go

package app

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"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
	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 _, 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.Notifier == nil {
		env.Notifier = &notify.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
	}
	identityChanged, err := ensureManifestIdentity(cfg, m)
	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)
		}
	}

	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())
			env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force)
			continue
		}
		executed = append(executed, s.Name())

		now := nowUTC()
		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 {
			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)
			}
			env.Logger.Info("stage failed", "stage", s.Name(), "error", err)
			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)
		}
		env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "succeeded", "path", manifestPath)
		env.Logger.Info("stage succeeded", "stage", s.Name())
	}

	return &RunSummary{
		SessionID:    cfg.Session.SessionID,
		ManifestPath: manifestPath,
		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) == "" || strings.TrimSpace(a.BaseURL) == "" || strings.TrimSpace(a.Model) == "" {
		// 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) []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 ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (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
	}
	if m.RunID == "" {
		runID, err := artifacts.NewRunID()
		if err != nil {
			return false, err
		}
		m.RunID = runID
		changed = true
	}
	if m.LocalWorkDir == "" && campaign != "" && sessionID != "" && m.RunID != "" {
		m.LocalWorkDir = artifacts.SessionRunWorkDir(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 manifestPathFor(cfg *config.Config) string {
	return filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.SessionID, "manifest.json")
}