narratio/internal/app/runner.go

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/whisperx"
	"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
	"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
	SelectedArtifacts []string
	Env               *Env
}

type RunSummary struct {
	SessionID       string
	RunID           string
	ManifestPath    string
	RunManifestPath string
	StageNames      []string
	Executed        []string
	Skipped         []string
}

var executeStagesFn = executeStages

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
	}
	env.SelectedArtifactKeys = append([]string(nil), opts.SelectedArtifacts...)
	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.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages) {
		objectStore, err := newCommandObjectStore(ctx, env.Config, nil)
		if err != nil {
			return nil, err
		}
		env.ObjectStore = objectStore
	}
	if needsRemoteLocksForRun(env.Config, stages) {
		locks, err := loadEffectiveLocks(ctx, env.Config, env.ObjectStore)
		if err != nil {
			return nil, fmt.Errorf("load remote publish locks: %w", err)
		}
		applyEffectiveLocks(env.Config, locks.All)
	}
	if env.Notifier == nil {
		env.Notifier = &notify.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(s.Name(), 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 := runPostPublishCleanup(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-publish cleanup failed (%v) and run-manifest save failed (%v)", err, saveErr)
		}
		return nil, fmt.Errorf("post-publish 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(stageName string, 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
		}
		kind := ref.Kind
		sourceID := ""
		if stageName == "analyze" {
			sourceID = artifacts.ConfiguredArtifactSourceID(ref.Kind)
			kind = "scriptorium_artifact"
		} else {
			sourceID = sourceIDForOutputKind(kind)
		}
		out = append(out, manifest.ArtifactRecord{
			Kind:          kind,
			SourceID:      sourceID,
			LocalPath:     localPath,
			ProducerRunID: runID,
			RemoteKey:     ref.RemoteKey,
			Checksum:      ref.Checksum,
		})
	}

	return out
}

func sourceIDForOutputKind(kind string) string {
	trimmed := strings.TrimSpace(kind)
	if trimmed == "" {
		return ""
	}
	if trimmed == "session_bounds" {
		return artifacts.ArtifactBoundsSession
	}
	for _, spec := range artifactmodel.RuntimeTranscriptArtifacts() {
		if spec.OutputKind == trimmed {
			return spec.SourceID
		}
	}
	return ""
}

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("prepare") {
		requirements := artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg))
		if len(requirements) > 0 && strings.TrimSpace(cfg.Session.PreviousSessionID) != "" {
			return true
		}
	}
	if !stageRequested("publish") {
		return false
	}
	if cfg.Pipeline.Publish == nil {
		return false
	}
	if cfg.Pipeline.Publish.Enabled != nil && !*cfg.Pipeline.Publish.Enabled {
		return false
	}
	if cfg.Pipeline.Publish.UploadRun != nil && !*cfg.Pipeline.Publish.UploadRun {
		return false
	}
	return true
}

func needsRemoteLocksForRun(cfg *config.Config, stages []stage.Stage) bool {
	if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil {
		return false
	}
	publishRequested := false
	for _, s := range stages {
		if s != nil && s.Name() == "publish" {
			publishRequested = true
			break
		}
	}
	if !publishRequested {
		return false
	}
	if cfg.Pipeline.Publish == nil {
		return false
	}
	if cfg.Pipeline.Publish.Enabled != nil && !*cfg.Pipeline.Publish.Enabled {
		return false
	}
	if cfg.Pipeline.Publish.UploadRun != nil && !*cfg.Pipeline.Publish.UploadRun {
		return false
	}
	return cfg.Pipeline.Storage.S3 != nil
}

func configuredScriptoriumArtifacts(cfg *config.Config) map[string]config.ScriptoriumArtifactConfig {
	if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Scriptorium == nil {
		return nil
	}
	return cfg.Pipeline.Scriptorium.Artifacts
}