Unify invocation manifest identity

2026-08-10 19:25:03 +00:00
parent 9900211fa4
commit a1ceb457e9
4 changed files with 296 additions and 100 deletions
--- a/internal/app/runner.go
+++ b/internal/app/runner.go
@@ -41,20 +41,41 @@ type RunSummary struct {
 var executeStagesFn = executeStages

 func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage, opts RunOptions) (summary *RunSummary, resultErr error) {
+	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{}
 	}
-	if env.Config == nil {
-		env.Config = cfg
-	}
+	// 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...)
-	if env.ArtifactStore == nil {
-		env.ArtifactStore = artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root)
-	}
 	if env.ManifestStore == nil {
 		env.ManifestStore = &manifest.LocalStore{}
 	}
+	manifestPath := artifacts.SessionManifestPathForCampaign(
+		cfg.Pipeline.Workspace.Root,
+		identity.Campaign,
+		identity.SessionID,
+	)
+	if existing, present, err := loadManifestAtPathIfPresent(ctx, env.ManifestStore, manifestPath); err != nil {
+		return nil, err
+	} else if present {
+		if err := identity.validateSessionManifest(existing); err != nil {
+			return nil, err
+		}
+	}
+	if env.ArtifactStore == nil {
+		env.ArtifactStore = artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root)
+	}
 	if env.Logger == nil {
 		env.Logger = logging.NewLogger(os.Stderr, slog.LevelInfo)
 	}
@@ -107,12 +128,12 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
 	}

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

-	lock, err := artifactStore.AcquireSessionLockForContext(ctx, cfg.Session.Campaign, cfg.Session.SessionID)
+	lock, err := artifactStore.AcquireSessionLockForContext(ctx, identity.Campaign, identity.SessionID)
 	if err != nil {
 		return nil, fmt.Errorf("acquire session lock: %w", err)
 	}
@@ -126,44 +147,46 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
 		}
 	}()

-	manifestPath := paths.ManifestPath
-	m, err := loadOrCreateManifest(ctx, env.ManifestStore, manifestPath, cfg.Session.SessionID)
+	if paths.ManifestPath != manifestPath {
+		return nil, fmt.Errorf("prepared manifest path %q does not match resolved manifest path %q", paths.ManifestPath, manifestPath)
+	}
+	m, present, err := loadManifestAtPathIfPresent(ctx, env.ManifestStore, manifestPath)
 	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)
+	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,
-		cfg.Session.Campaign,
-		cfg.Session.SessionID,
-		runID,
+		identity.Campaign,
+		identity.SessionID,
+		identity.RunID,
 	)
 	runManifestStore := &manifest.LocalStore{}
 	runManifest, err := runManifestStore.CreateRun(
 		ctx,
-		cfg.Session.SessionID,
-		cfg.Session.Campaign,
-		runID,
+		identity.SessionID,
+		identity.Campaign,
+		identity.RunID,
 		opts.Force,
 		requestedStageNames(stages),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create run manifest: %w", err)
 	}
-	runManifest.SessionManifestPath = manifestPath
-	syncRunManifestIdentityFromSession(m, runManifest)
+	identity.applyToRunManifest(runManifest, manifestPath)
 	if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
 		return nil, fmt.Errorf("save initial run manifest %q: %w", runManifestPath, err)
 	}
@@ -244,7 +267,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
 				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)
+			identity.applyToRunManifest(runManifest, manifestPath)
 			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)
 			}
@@ -264,7 +287,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
 			}
 			runManifest.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason)
 			applyStageResultToRunManifest(runManifest, s.Name(), result)
-			syncRunManifestIdentityFromSession(m, runManifest)
+			identity.applyToRunManifest(runManifest, manifestPath)
 			if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
 				return nil, fmt.Errorf("save run manifest after self-skip %q: %w", s.Name(), err)
 			}
@@ -286,7 +309,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
 		}
 		runManifest.MarkStageSucceeded(s.Name(), succeededAt, outputs)
 		applyStageResultToRunManifest(runManifest, s.Name(), result)
-		syncRunManifestIdentityFromSession(m, runManifest)
+		identity.applyToRunManifest(runManifest, manifestPath)
 		if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
 			return nil, fmt.Errorf("save run manifest after stage %q: %w", s.Name(), err)
 		}
@@ -297,7 +320,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
 	if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil {
 		failedAt := nowUTC()
 		runManifest.MarkFailed(failedAt, err.Error())
-		syncRunManifestIdentityFromSession(m, runManifest)
+		identity.applyToRunManifest(runManifest, manifestPath)
 		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)
 		}
@@ -305,7 +328,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
 	}
 	completedAt := nowUTC()
 	runManifest.MarkSucceeded(completedAt)
-	syncRunManifestIdentityFromSession(m, runManifest)
+	identity.applyToRunManifest(runManifest, manifestPath)
 	if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
 		return nil, fmt.Errorf("save final run manifest %q: %w", runManifestPath, err)
 	}
@@ -423,27 +446,19 @@ func buildDefaultAuditaRunner(cfg *config.Config) (audita.Runner, error) {
 	return runner, nil
 }

-func loadOrCreateManifest(ctx context.Context, store manifest.Store, path, sessionID string) (*manifest.Manifest, error) {
+func loadManifestAtPathIfPresent(ctx context.Context, store manifest.Store, path string) (*manifest.Manifest, bool, error) {
 	exists, err := fileExists(path)
 	if err != nil {
-		return nil, fmt.Errorf("check manifest existence %q: %w", path, err)
+		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, fmt.Errorf("load manifest %q: %w", path, err)
+			return nil, false, fmt.Errorf("load manifest %q: %w", path, err)
 		}
-		return m, nil
+		return m, true, 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
+	return nil, false, nil
 }

 func fileExists(path string) (bool, error) {
@@ -569,53 +584,100 @@ func applyStageResultToManifest(m *manifest.Manifest, stageName string, result *
 	}
 }

-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
+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")
 	}

-	changed := false
 	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 == "" {
-		sessionID = strings.TrimSpace(m.SessionID)
-	}
-
-	if m.Campaign == "" && campaign != "" {
-		m.Campaign = campaign
-		changed = true
+		return invocationIdentity{}, fmt.Errorf("configured session_id is required")
 	}
 	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
-		}
+	if runID == "" {
+		return invocationIdentity{}, fmt.Errorf("run id is required")
 	}

-	return changed, nil
+	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 {
@@ -648,18 +710,6 @@ func applyStageResultToRunManifest(m *manifest.RunManifest, stageName string, re
 	}
 }

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