narratio/internal/app/runner.go

package app

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"strings"

	"gitea.maximumdirect.net/eric/narratio/internal/adapters/audita"
	"gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius"
	"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
	EffectiveArtifacts artifacts.EffectiveArtifactSet
	Env                *Env
	RunManifestStore   manifest.RunStore
}

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) (summary *RunSummary, resultErr error) {
	effectiveArtifacts := opts.EffectiveArtifacts
	if !effectiveArtifacts.Resolved() && cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Scriptorium != nil {
		var err error
		effectiveArtifacts, err = resolveEffectiveArtifacts(cfg, nil)
		if err != nil {
			return nil, fmt.Errorf("resolve effective artifacts: %w", err)
		}
	}
	if !effectiveArtifacts.Resolved() {
		var err error
		effectiveArtifacts, err = artifacts.ResolveEffectiveArtifactSet(nil, nil)
		if err != nil {
			return nil, fmt.Errorf("resolve default effective artifacts: %w", err)
		}
	}
	runID, err := artifacts.NewRunID()
	if err != nil {
		return nil, fmt.Errorf("generate run id: %w", err)
	}
	identity, err := resolveInvocationIdentity(cfg, runID)
	if err != nil {
		return nil, fmt.Errorf("resolve invocation identity: %w", err)
	}

	env := opts.Env
	if env == nil {
		env = &Env{}
	}
	// The resolved configuration is the invocation's single source of truth.
	// Injected environments supply collaborators, never an alternate config.
	env.Config = cfg
	env.SelectedArtifactKeys = append([]string(nil), opts.SelectedArtifacts...)
	env.EffectiveArtifacts = effectiveArtifacts
	if env.ManifestStore == nil {
		env.ManifestStore = &manifest.LocalStore{}
	}
	manifestPath := artifacts.SessionManifestPathForCampaign(
		cfg.Pipeline.Workspace.Root,
		identity.Campaign,
		identity.SessionID,
	)
	if env.ArtifactStore == nil {
		env.ArtifactStore = artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root)
	}
	if env.Logger == nil {
		env.Logger = logging.NewLogger(os.Stderr, slog.LevelInfo)
	}

	artifactStore := env.ArtifactStore
	paths, err := artifactStore.EnsureLayoutFor(identity.Campaign, identity.SessionID)
	if err != nil {
		return nil, fmt.Errorf("prepare workdir: %w", err)
	}

	lock, err := artifactStore.AcquireSessionLockForContext(ctx, identity.Campaign, identity.SessionID)
	if err != nil {
		return nil, fmt.Errorf("acquire session lock: %w", err)
	}
	defer func() {
		if releaseErr := artifactStore.ReleaseSessionLock(lock); releaseErr != nil {
			if resultErr == nil {
				resultErr = fmt.Errorf("release session lock: %w", releaseErr)
			} else {
				resultErr = errors.Join(resultErr, fmt.Errorf("release session lock: %w", releaseErr))
			}
		}
	}()

	if paths.ManifestPath != manifestPath {
		return nil, fmt.Errorf("prepared manifest path %q does not match resolved manifest path %q", paths.ManifestPath, manifestPath)
	}
	if err := requireCompleteRestore(paths); err != nil {
		return nil, err
	}
	m, present, err := loadManifestAtPathIfPresent(ctx, env.ManifestStore, manifestPath)
	if err != nil {
		return nil, err
	}
	if present {
		if err := identity.validateSessionManifest(m); err != nil {
			return nil, err
		}
	} else {
		m, err = env.ManifestStore.Create(ctx, identity.SessionID)
		if err != nil {
			return nil, fmt.Errorf("create manifest: %w", err)
		}
	}
	identity.applyToSessionManifest(m)
	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,
		identity.Campaign,
		identity.SessionID,
		identity.RunID,
	)
	runManifestStore := opts.RunManifestStore
	if runManifestStore == nil {
		runManifestStore = &manifest.LocalStore{}
	}
	runManifest, err := runManifestStore.CreateRun(
		ctx,
		identity.SessionID,
		identity.Campaign,
		identity.RunID,
		opts.Force,
		requestedStageNames(stages),
	)
	if err != nil {
		return nil, fmt.Errorf("create run manifest: %w", err)
	}
	identity.applyToRunManifest(runManifest, manifestPath)
	if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
		return nil, persistTerminalFailure(
			ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
			fmt.Errorf("save initial run manifest %q: %w", runManifestPath, err),
		)
	}
	if _, err := loadSecretsFromConfig(env.Config, env.Logger); err != nil {
		return nil, persistTerminalFailure(
			ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
			fmt.Errorf("load secrets from files: %w", err),
		)
	}
	if env.WhisperX == nil {
		client, err := buildDefaultWhisperXClient(env.Config)
		if err != nil {
			return nil, persistTerminalFailure(
				ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
				fmt.Errorf("initialize whisperx client: %w", err),
			)
		}
		env.WhisperX = client
	}
	if env.Seriatim == nil {
		runner, err := buildDefaultSeriatimRunner(env.Config)
		if err != nil {
			return nil, persistTerminalFailure(
				ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
				fmt.Errorf("initialize seriatim runner: %w", err),
			)
		}
		env.Seriatim = runner
	}
	if env.Audita == nil {
		runner, err := buildDefaultAuditaRunner(env.Config)
		if err != nil {
			return nil, persistTerminalFailure(
				ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
				fmt.Errorf("initialize audita runner: %w", err),
			)
		}
		env.Audita = runner
	}
	if env.Notarius == nil && needsNotariusForRun(env.Config, stages) {
		env.Notarius = notarius.NewSubprocessRunner()
	}
	if env.Scriptorium == nil {
		env.Scriptorium = scriptorium.NewSubprocessRunner()
	}
	if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages, effectiveArtifacts) {
		objectStore, err := newCommandObjectStore(ctx, env.Config, nil)
		if err != nil {
			return nil, persistTerminalFailure(
				ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, err,
			)
		}
		env.ObjectStore = objectStore
	}
	if needsRemoteLocksForRun(env.Config, stages) {
		locks, err := loadEffectiveLocks(ctx, env.Config, env.ObjectStore)
		if err != nil {
			return nil, persistTerminalFailure(
				ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
				fmt.Errorf("load remote publish locks: %w", err),
			)
		}
		applyEffectiveLocks(env.Config, locks.All)
		staticLocks := append([]config.PublishLockRule(nil), locks.Static...)
		env.RevalidatePublishLocks = func(recheckCtx context.Context) ([]config.PublishLockRule, error) {
			remote, _, _, err := loadRemoteLockStore(recheckCtx, env.Config, env.ObjectStore)
			if err != nil {
				return nil, err
			}
			return config.MergePublishLockRules(staticLocks, remote.Locks), nil
		}
	}
	if env.Notifier == nil {
		env.Notifier = &notify.NoopSender{}
	}

	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())
		d.Action = decideStageAction(s, m, opts.Force)

		if d.Action == stageActionSkip {
			if validator, ok := s.(stage.ResumeValidator); ok {
				validation, err := validator.ValidateResume(ctx, stageEnv, m)
				if err != nil {
					return nil, persistTerminalFailure(
						ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
						fmt.Errorf("validate resume for stage %q: %w", s.Name(), err),
					)
				}
				validation = validation.Normalized()
				if !validation.Resumable {
					staleAt := nowUTC()
					m.MarkStageStale(s.Name(), staleAt, validation.Reason)
					invalidateDownstreamSucceededStagesWithReason(
						m, s.Name(), staleAt, staleReasonNotResumable,
					)
					if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
						return nil, persistTerminalFailure(
							ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
							fmt.Errorf("save manifest after resume validation for stage %q: %w", s.Name(), err),
						)
					}
					env.Logger.Info("stage result is not resumable", "stage", s.Name(), "reason", validation.Reason)
					d.Action = stageActionRun
				}
			}
		}

		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, persistTerminalFailure(
					ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
					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())
		priorOutcome := capturePriorStageOutcome(m, 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 {
			operationErr := fmt.Errorf("save run manifest before stage %q: %w", s.Name(), err)
			runManifest.MarkStageFailed(s.Name(), nowUTC(), operationErr.Error())
			return nil, persistTerminalFailure(
				ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, operationErr,
			)
		}
		m.MarkStageRunning(s.Name(), now)
		if opts.Force {
			invalidateDownstreamSucceededStagesWithReason(m, s.Name(), now, staleReasonForcedReplacement)
		}
		env.Logger.Info("starting stage", "stage", s.Name())
		if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
			operationErr := fmt.Errorf("save manifest before stage %q: %w", s.Name(), err)
			m.MarkStageFailed(s.Name(), nowUTC(), operationErr.Error())
			invalidateDownstreamSucceededStagesWithReason(m, s.Name(), nowUTC(), staleReasonFailure)
			return nil, persistTerminalFailure(
				ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, operationErr,
			)
		}
		env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath)

		result, err := s.Run(ctx, stageEnv, m)
		if err == nil {
			err = validateStageResult(result)
		}
		if err != nil {
			failedAt := nowUTC()
			m.MarkStageFailed(s.Name(), failedAt, err.Error())
			invalidateDownstreamSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure)
			runManifest.MarkStageFailed(s.Name(), failedAt, err.Error())
			identity.applyToRunManifest(runManifest, manifestPath)
			env.Logger.Info("stage failed", "stage", s.Name(), "error", err)
			return nil, persistTerminalFailure(
				ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
				fmt.Errorf("stage %q failed: %w", s.Name(), err),
			)
		}
		if result != nil && result.Disposition == stage.StageDispositionSkipped {
			skipped = append(skipped, s.Name())
			skippedAt := nowUTC()
			m.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason)
			applyStageResultToManifest(m, s.Name(), result)
			if !priorOutcome.isSameSelfSkip(result.SkipReason) {
				invalidateDownstreamSucceededStagesWithReason(m, s.Name(), skippedAt, staleReasonSelfSkip)
			}
			if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
				return nil, persistTerminalFailure(
					ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
					fmt.Errorf("save manifest after self-skip %q: %w", s.Name(), err),
				)
			}
			runManifest.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason)
			applyStageResultToRunManifest(runManifest, s.Name(), result)
			identity.applyToRunManifest(runManifest, manifestPath)
			if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
				return nil, persistTerminalFailure(
					ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
					fmt.Errorf("save run manifest after self-skip %q: %w", s.Name(), err),
				)
			}
			env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "skipped", "path", manifestPath)
			env.Logger.Info("stage skipped", "stage", s.Name(), "reason", result.SkipReason)
			continue
		}

		outputs := mapResultOutputs(s.Name(), result, runID)
		succeededAt := nowUTC()
		m.MarkStageSucceeded(s.Name(), succeededAt, outputs)
		applyStageResultToManifest(m, s.Name(), result)
		if !priorOutcome.exists || priorOutcome.status != manifest.StatusSucceeded {
			invalidateDownstreamSucceededStagesWithReason(m, s.Name(), succeededAt, staleReasonChangedResult)
		}

		if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
			return nil, persistTerminalFailure(
				ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
				fmt.Errorf("save manifest after stage %q: %w", s.Name(), err),
			)
		}
		runManifest.MarkStageSucceeded(s.Name(), succeededAt, outputs)
		applyStageResultToRunManifest(runManifest, s.Name(), result)
		identity.applyToRunManifest(runManifest, manifestPath)
		if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
			return nil, persistTerminalFailure(
				ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
				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())
	}

	completedAt := nowUTC()
	runManifest.MarkSucceeded(completedAt)
	identity.applyToRunManifest(runManifest, manifestPath)
	if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
		return nil, persistTerminalFailure(
			ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
			fmt.Errorf("save final run manifest %q: %w", runManifestPath, err),
		)
	}
	// The run record lives inside the run work directory, which cleanup may
	// remove. Persist its completed publishing result before cleanup starts so a
	// successful deletion cannot be undone by a later diagnostic write.
	if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil {
		return nil, persistPostPublishCleanupFailure(
			ctx, env.ManifestStore, manifestPath, m,
			fmt.Errorf("post-publish cleanup incomplete: %w", err),
		)
	}

	return &RunSummary{
		SessionID:       cfg.Session.SessionID,
		RunID:           runID,
		ManifestPath:    manifestPath,
		RunManifestPath: runManifestPath,
		StageNames:      runNames,
		Executed:        executed,
		Skipped:         skipped,
	}, nil
}

func persistPostPublishCleanupFailure(
	ctx context.Context,
	sessionStore manifest.Store,
	sessionManifestPath string,
	sessionManifest *manifest.Manifest,
	operationErr error,
) error {
	if operationErr == nil {
		return nil
	}
	if sessionManifest != nil {
		sessionManifest.RecordFailure(nowUTC(), operationErr.Error())
	}
	// Do not rewrite the run manifest here: cleanup may have deliberately
	// removed the directory that contains it. The completed run record remains
	// the authoritative evidence of the remote publish; the session manifest
	// records the outstanding local cleanup for the next invocation.
	sessionErr := sessionStore.Save(ctx, sessionManifestPath, sessionManifest)
	return errors.Join(operationErr, wrapTerminalPersistenceError("save terminal session manifest", sessionErr))
}

func persistTerminalFailure(
	ctx context.Context,
	sessionStore manifest.Store,
	sessionManifestPath string,
	sessionManifest *manifest.Manifest,
	runStore manifest.RunStore,
	runManifestPath string,
	runManifest *manifest.RunManifest,
	operationErr error,
) error {
	if operationErr == nil {
		return nil
	}

	at := nowUTC()
	if sessionManifest != nil {
		sessionManifest.RecordFailure(at, operationErr.Error())
	}

	// A running run record is durable before execution. At termination, session
	// state controls resume while the run record only diagnoses this invocation,
	// so session persistence comes first and any disagreement remains visible.
	sessionErr := sessionStore.Save(ctx, sessionManifestPath, sessionManifest)
	if runManifest != nil {
		runManifest.MarkFailed(at, operationErr.Error())
	}
	runErr := runStore.SaveRun(ctx, runManifestPath, runManifest)
	return errors.Join(
		operationErr,
		wrapTerminalPersistenceError("save terminal session manifest", sessionErr),
		wrapTerminalPersistenceError("save terminal run manifest", runErr),
	)
}

func wrapTerminalPersistenceError(operation string, err error) error {
	if err == nil {
		return nil
	}
	return fmt.Errorf("%s: %w", operation, err)
}

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 loadManifestAtPathIfPresent(ctx context.Context, store manifest.Store, path string) (*manifest.Manifest, bool, error) {
	exists, err := fileExists(path)
	if err != nil {
		return nil, false, fmt.Errorf("check manifest existence %q: %w", path, err)
	}
	if exists {
		m, err := store.Load(ctx, path)
		if err != nil {
			return nil, false, fmt.Errorf("load manifest %q: %w", path, err)
		}
		return m, true, nil
	}
	return nil, false, 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 := strings.TrimSpace(ref.SourceID)
		if 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,
			ArchivePath:        ref.ArchivePath,
			Contract:           cloneContractMetadata(ref.Contract),
			ExternalProvenance: cloneExternalProvenance(ref.ExternalProvenance),
			ProducerRunID:      runID,
			RemoteKey:          ref.RemoteKey,
			Checksum:           ref.Checksum,
		})
	}

	return out
}

func validateStageResult(result *stage.StageResult) error {
	if result == nil {
		return nil
	}
	switch result.Disposition {
	case stage.StageDispositionSucceeded:
		if strings.TrimSpace(result.SkipReason) != "" {
			return fmt.Errorf("successful result contains a skip reason")
		}
		return nil
	case stage.StageDispositionSkipped:
		if strings.TrimSpace(result.SkipReason) == "" {
			return fmt.Errorf("skipped result requires a skip reason")
		}
		if len(result.Outputs) != 0 {
			return fmt.Errorf("skipped result contains %d output(s)", len(result.Outputs))
		}
		return nil
	default:
		return fmt.Errorf("unsupported stage result disposition %q", result.Disposition)
	}
}

func cloneContractMetadata(value *artifactmodel.ContractMetadata) *artifactmodel.ContractMetadata {
	if value == nil {
		return nil
	}
	cloned := *value
	return &cloned
}

func cloneExternalProvenance(value *artifactmodel.ExternalProvenance) *artifactmodel.ExternalProvenance {
	if value == nil {
		return nil
	}
	cloned := *value
	return &cloned
}

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
	}
}

type invocationIdentity struct {
	Campaign        string
	SessionID       string
	RunID           string
	LocalWorkDir    string
	LocalSpoolDir   string
	S3Bucket        string
	S3SessionPrefix string
	S3RunPrefix     string
}

func resolveInvocationIdentity(cfg *config.Config, runID string) (invocationIdentity, error) {
	if cfg == nil || cfg.Pipeline == nil || cfg.Campaign == nil || cfg.Session == nil {
		return invocationIdentity{}, fmt.Errorf("resolved configuration requires pipeline, campaign, and session")
	}

	campaign := strings.TrimSpace(cfg.Session.Campaign)
	configuredCampaign := strings.TrimSpace(config.CampaignID(cfg.Campaign))
	if campaign == "" {
		return invocationIdentity{}, fmt.Errorf("configured session campaign is required")
	}
	if configuredCampaign == "" {
		return invocationIdentity{}, fmt.Errorf("configured campaign is required")
	}
	if campaign != configuredCampaign {
		return invocationIdentity{}, fmt.Errorf("configured session campaign %q does not match configured campaign %q", campaign, configuredCampaign)
	}

	sessionID := strings.TrimSpace(cfg.Session.SessionID)
	if sessionID == "" {
		return invocationIdentity{}, fmt.Errorf("configured session_id is required")
	}
	runID = strings.TrimSpace(runID)
	if runID == "" {
		return invocationIdentity{}, fmt.Errorf("run id is required")
	}

	identity := invocationIdentity{
		Campaign:     campaign,
		SessionID:    sessionID,
		RunID:        runID,
		LocalWorkDir: artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, campaign, sessionID, runID),
	}
	if spoolRoot := strings.TrimSpace(cfg.Pipeline.Spool.Root); spoolRoot != "" {
		identity.LocalSpoolDir = artifacts.SessionSpoolAudioDir(spoolRoot, campaign, sessionID, runID)
	}
	if s3 := cfg.Pipeline.Storage.S3; s3 != nil {
		identity.S3Bucket = strings.TrimSpace(s3.Bucket)
		identity.S3SessionPrefix = artifacts.S3SessionPrefix(s3.RootPrefix, campaign, sessionID)
		identity.S3RunPrefix = artifacts.S3RunPrefix(identity.S3SessionPrefix, runID)
	}
	return identity, nil
}

func (identity invocationIdentity) validateSessionManifest(m *manifest.Manifest) error {
	if m == nil {
		return fmt.Errorf("persisted manifest is nil")
	}
	if sessionID := strings.TrimSpace(m.SessionID); sessionID != "" && sessionID != identity.SessionID {
		return fmt.Errorf("persisted manifest session_id %q does not match configured session_id %q", sessionID, identity.SessionID)
	}
	if campaign := strings.TrimSpace(m.Campaign); campaign != "" && campaign != identity.Campaign {
		return fmt.Errorf("persisted manifest campaign %q does not match configured campaign %q", campaign, identity.Campaign)
	}
	return nil
}

func (identity invocationIdentity) applyToSessionManifest(m *manifest.Manifest) {
	if m == nil {
		return
	}
	m.SessionID = identity.SessionID
	m.Campaign = identity.Campaign
	m.RunID = identity.RunID
	m.LocalWorkDir = identity.LocalWorkDir
	m.LocalSpoolDir = identity.LocalSpoolDir
	m.S3Bucket = identity.S3Bucket
	m.S3SessionPrefix = identity.S3SessionPrefix
	m.S3RunPrefix = identity.S3RunPrefix
}

func (identity invocationIdentity) applyToRunManifest(m *manifest.RunManifest, sessionManifestPath string) {
	if m == nil {
		return
	}
	m.SessionID = identity.SessionID
	m.Campaign = identity.Campaign
	m.RunID = identity.RunID
	m.SessionManifestPath = sessionManifestPath
	m.LocalWorkDir = identity.LocalWorkDir
	m.LocalSpoolDir = identity.LocalSpoolDir
	m.S3Bucket = identity.S3Bucket
	m.S3SessionPrefix = identity.S3SessionPrefix
	m.S3RunPrefix = identity.S3RunPrefix
}

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 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, effectiveSets ...artifacts.EffectiveArtifactSet) bool {
	if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil {
		return false
	}
	effective := artifacts.EffectiveArtifactSet{}
	if len(effectiveSets) > 0 {
		effective = effectiveSets[0]
	}
	if !effective.Resolved() && cfg.Pipeline.Scriptorium != nil {
		var err error
		effective, err = artifacts.ResolveEffectiveArtifactSet(
			artifacts.ConfiguredArtifactDefinitions(cfg.Pipeline.Scriptorium.Artifacts),
			nil,
		)
		if err != 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), effective)
		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 needsNotariusForRun(cfg *config.Config, stages []stage.Stage) bool {
	if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Notarius == nil || !cfg.Pipeline.Notarius.Enabled {
		return false
	}
	for _, candidate := range stages {
		if candidate != nil && candidate.Name() == "extract" {
			return true
		}
	}
	return false
}

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
}