diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index e2b322f..ee92b8a 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -92,6 +92,12 @@ runner marks it stale and executes it. Session manifest is the authoritative stage-progress ledger across invocations. Run manifest is invocation-scoped audit state. +Each invocation derives campaign, session, run, local-path, and remote-prefix +metadata from the validated resolved configuration as one projection. A persisted +session manifest must agree on campaign and session identity before execution; +the current projection is refreshed for every invocation while stage progress, +inputs, and durable artifacts remain session history. + ## Invariants - stage resume/skip decisions are session-manifest driven. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index aa3512e..4409ef7 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -26,7 +26,7 @@ All stages are pending when this plan is created. | 8 | Terminate owned subprocess trees | RSK-011 | Completed | | 9 | Redact and cap subprocess diagnostics | RSK-012 | Completed | | 10 | Confine publish archive reads | COR-005 | Completed | -| 11 | Make manifest and run identity singular | COR-001, TST-006 | Pending | +| 11 | Make manifest and run identity singular | COR-001, TST-006 | Completed | | 12 | Centralize handled terminal-failure persistence | RSK-001, TST-002, SIM-001, COM-001 | Pending | | 13 | Introduce the immutable remote-commit model and legacy boundary | ARC-003 | Pending | | 14 | Publish through immutable commits and canonical mappings | COR-004, COR-011, DUP-002, TST-004 | Pending | diff --git a/internal/app/runner.go b/internal/app/runner.go index 49a5f18..c0631ae 100644 --- 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, diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index eebeccd..52bfdd0 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -71,6 +71,10 @@ type captureSelectedArtifactsStage struct { captured *[]string } +type captureConfigStage struct { + captured **config.Config +} + type captureNotariusStage struct { captured *bool } @@ -108,6 +112,13 @@ func (s captureSelectedArtifactsStage) Run(_ context.Context, env *stage.Env, _ return &stage.StageResult{Metadata: map[string]any{"captured": true}}, nil } +func (captureConfigStage) Name() string { return "prepare" } +func (captureConfigStage) Declares() stage.IODecl { return stage.IODecl{} } +func (s captureConfigStage) Run(_ context.Context, env *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { + *s.captured = env.Config + return &stage.StageResult{}, nil +} + type analyzeOutputStage struct { output artifacts.Ref } @@ -183,6 +194,24 @@ func TestExecuteStagesPropagatesSelectedArtifactsToEnv(t *testing.T) { } } +func TestExecuteStagesBindsResolvedConfigToEnvironment(t *testing.T) { + resolved := testConfig(t) + stale := testConfig(t) + var captured *config.Config + env := &Env{Config: stale} + + _, err := executeStages(context.Background(), resolved, []stage.Stage{captureConfigStage{captured: &captured}}, RunOptions{Env: env}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if env.Config != resolved { + t.Fatalf("environment config = %p, want resolved config %p", env.Config, resolved) + } + if captured != resolved { + t.Fatalf("stage config = %p, want resolved config %p", captured, resolved) + } +} + func TestExecuteStagesComposesNotariusOnlyForEnabledExtraction(t *testing.T) { cfg := testConfig(t) cfg.Pipeline.Notarius = &config.NotariusConfig{Enabled: true} @@ -916,10 +945,24 @@ func TestExecuteStagesLoadsExistingManifest(t *testing.T) { func TestExecuteStagesCreatesRunManifestPerInvocation(t *testing.T) { cfg := testConfig(t) + cfg.Pipeline.Spool.Root = t.TempDir() + cfg.Pipeline.Storage.S3 = &config.StorageS3Config{ + Bucket: "archive-bucket", + RootPrefix: "narratio", + } run1, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[0]}, RunOptions{}) if err != nil { t.Fatalf("first executeStages() error = %v", err) } + store := &manifest.LocalStore{} + sessionManifest, err := store.Load(context.Background(), run1.ManifestPath) + if err != nil { + t.Fatalf("Load first session manifest error = %v", err) + } + sessionManifest.Inputs = append(sessionManifest.Inputs, manifest.InputRecord{Kind: "audio", Path: "audio/alice.flac"}) + if err := store.Save(context.Background(), run1.ManifestPath, sessionManifest); err != nil { + t.Fatalf("Save session history error = %v", err) + } run2, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[0]}, RunOptions{Force: true}) if err != nil { t.Fatalf("second executeStages() error = %v", err) @@ -943,14 +986,111 @@ func TestExecuteStagesCreatesRunManifestPerInvocation(t *testing.T) { } } - store := &manifest.LocalStore{} - sessionManifest, err := store.Load(context.Background(), run2.ManifestPath) + sessionManifest, err = store.Load(context.Background(), run2.ManifestPath) if err != nil { t.Fatalf("Load session manifest error = %v", err) } if sessionManifest.RunID != run2.RunID { t.Fatalf("session manifest run_id = %q, want latest run id %q", sessionManifest.RunID, run2.RunID) } + if len(sessionManifest.Inputs) != 1 || sessionManifest.Inputs[0].Path != "audio/alice.flac" { + t.Fatalf("session history inputs = %#v, want preserved input", sessionManifest.Inputs) + } + + for _, run := range []*RunSummary{run1, run2} { + runManifest, err := store.LoadRun(context.Background(), run.RunManifestPath) + if err != nil { + t.Fatalf("LoadRun(%q) error = %v", run.RunManifestPath, err) + } + wantWorkDir := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, run.RunID) + wantSpoolDir := artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, cfg.Session.Campaign, cfg.Session.SessionID, run.RunID) + wantSessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID) + wantRunPrefix := artifacts.S3RunPrefix(wantSessionPrefix, run.RunID) + if runManifest.SessionID != cfg.Session.SessionID || runManifest.Campaign != cfg.Session.Campaign || runManifest.RunID != run.RunID { + t.Fatalf("run identity = (%q, %q, %q), want (%q, %q, %q)", runManifest.SessionID, runManifest.Campaign, runManifest.RunID, cfg.Session.SessionID, cfg.Session.Campaign, run.RunID) + } + if runManifest.SessionManifestPath != run.ManifestPath || runManifest.LocalWorkDir != wantWorkDir || runManifest.LocalSpoolDir != wantSpoolDir { + t.Fatalf("run paths = (%q, %q, %q), want (%q, %q, %q)", runManifest.SessionManifestPath, runManifest.LocalWorkDir, runManifest.LocalSpoolDir, run.ManifestPath, wantWorkDir, wantSpoolDir) + } + if runManifest.S3Bucket != cfg.Pipeline.Storage.S3.Bucket || runManifest.S3SessionPrefix != wantSessionPrefix || runManifest.S3RunPrefix != wantRunPrefix { + t.Fatalf("run remote identity = (%q, %q, %q), want (%q, %q, %q)", runManifest.S3Bucket, runManifest.S3SessionPrefix, runManifest.S3RunPrefix, cfg.Pipeline.Storage.S3.Bucket, wantSessionPrefix, wantRunPrefix) + } + } + if sessionManifest.LocalWorkDir == artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, run1.RunID) { + t.Fatal("session manifest retained the prior invocation work directory") + } +} + +func TestExecuteStagesRejectsPersistedManifestIdentityMismatch(t *testing.T) { + for _, tc := range []struct { + name string + configure func(*manifest.Manifest) + wantDetail string + }{ + { + name: "session", + configure: func(m *manifest.Manifest) { + m.SessionID = "2026-05-04" + }, + wantDetail: "persisted manifest session_id", + }, + { + name: "campaign", + configure: func(m *manifest.Manifest) { + m.Campaign = "other-campaign" + }, + wantDetail: "persisted manifest campaign", + }, + } { + t.Run(tc.name, func(t *testing.T) { + cfg := testConfig(t) + manifestPath := manifestPathFor(cfg) + store := &manifest.LocalStore{} + persisted := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + persisted.Campaign = cfg.Session.Campaign + tc.configure(persisted) + if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { + t.Fatalf("MkdirAll() error = %v", err) + } + if err := store.Save(context.Background(), manifestPath, persisted); err != nil { + t.Fatalf("Save manifest error = %v", err) + } + before, err := os.ReadFile(manifestPath) + if err != nil { + t.Fatalf("ReadFile() error = %v", err) + } + + runs := 0 + _, err = executeStages(context.Background(), cfg, []stage.Stage{countingStage{name: "prepare", runs: &runs}}, RunOptions{}) + if err == nil || !strings.Contains(err.Error(), tc.wantDetail) { + t.Fatalf("executeStages() error = %v, want %q", err, tc.wantDetail) + } + if runs != 0 { + t.Fatalf("stage runs = %d, want 0", runs) + } + after, err := os.ReadFile(manifestPath) + if err != nil { + t.Fatalf("ReadFile() after rejection error = %v", err) + } + if string(after) != string(before) { + t.Fatal("persisted manifest changed after identity rejection") + } + paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) + if _, err := os.Stat(paths.LockPath); !os.IsNotExist(err) { + t.Fatalf("lock path stat error = %v, want no lock side effect", err) + } + }) + } +} + +func TestExecuteStagesRejectsConfiguredCampaignDisagreement(t *testing.T) { + cfg := testConfig(t) + cfg.Session.Campaign = "other-campaign" + + _, err := executeStages(context.Background(), cfg, []stage.Stage{countingStage{name: "prepare", runs: new(int)}}, RunOptions{}) + if err == nil || !strings.Contains(err.Error(), "configured session campaign") { + t.Fatalf("executeStages() error = %v, want configured campaign disagreement", err) + } } func TestExecuteStagesRunManifestRecordsSkippedStage(t *testing.T) {