Require explicit manifest schemas

internal/manifest/store.go

@@ -98,6 +98,9 @@ func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error {
 	if m == nil {
 		return fmt.Errorf("save manifest: manifest is nil")
 	}
+	if err := validateSessionManifestSchemaVersion(m.SchemaVersion); err != nil {
+		return fmt.Errorf("save manifest: %w", err)
+	}
 	if strings.TrimSpace(m.SessionID) == "" {
 		return fmt.Errorf("save manifest: session_id is required")
 	}
@@ -110,10 +113,10 @@ func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error {
 	if err := validatePostPublishCleanup(m.PostPublishCleanup); err != nil {
 		return fmt.Errorf("save manifest: %w", err)
 	}
+	if err := validateSessionStages(m.Stages); err != nil {
+		return fmt.Errorf("save manifest: %w", err)
+	}
 	for name, stage := range m.Stages {
-		if stage == nil {
-			continue
-		}
 		normalizeAnalyzeArtifactCollection(stage.AnalyzeArtifacts)
 		if err := validateAnalyzeStageState(name, stage.AnalyzeStateVersion, stage.AnalyzeArtifacts); err != nil {
 			return fmt.Errorf("save manifest: stages.%s: %w", name, err)
@@ -121,10 +124,6 @@ func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error {
 	}
 
 	m.UpdatedAt = time.Now().UTC()
-	if m.Stages == nil {
-		m.Stages = map[string]*StageRecord{}
-	}
-
 	data, err := json.MarshalIndent(m, "", "  ")
 	if err != nil {
 		return fmt.Errorf("save manifest: marshal: %w", err)
@@ -213,6 +212,9 @@ func (s *LocalStore) SaveRun(ctx context.Context, path string, m *RunManifest) e
 	if m == nil {
 		return fmt.Errorf("save run manifest: manifest is nil")
 	}
+	if err := validateRunManifestSchemaVersion(m.SchemaVersion); err != nil {
+		return fmt.Errorf("save run manifest: %w", err)
+	}
 	if strings.TrimSpace(m.SessionID) == "" {
 		return fmt.Errorf("save run manifest: session_id is required")
 	}
@@ -225,10 +227,10 @@ func (s *LocalStore) SaveRun(ctx context.Context, path string, m *RunManifest) e
 	if err := validateManifestIdentities(m.SessionID, m.Campaign, m.RunID); err != nil {
 		return fmt.Errorf("save run manifest: %w", err)
 	}
+	if err := validateRunStages(m.Stages); err != nil {
+		return fmt.Errorf("save run manifest: %w", err)
+	}
 	for name, stage := range m.Stages {
-		if stage == nil {
-			continue
-		}
 		normalizeAnalyzeArtifactCollection(stage.AnalyzeArtifacts)
 		if err := validateAnalyzeStageState(name, stage.AnalyzeStateVersion, stage.AnalyzeArtifacts); err != nil {
 			return fmt.Errorf("save run manifest: stages.%s: %w", name, err)
@@ -236,10 +238,6 @@ func (s *LocalStore) SaveRun(ctx context.Context, path string, m *RunManifest) e
 	}
 
 	m.UpdatedAt = time.Now().UTC()
-	if m.Stages == nil {
-		m.Stages = map[string]*RunStageRecord{}
-	}
-
 	data, err := json.MarshalIndent(m, "", "  ")
 	if err != nil {
 		return fmt.Errorf("save run manifest: marshal: %w", err)
@@ -253,6 +251,9 @@ func validateLoadedManifest(m *Manifest) error {
 	if m == nil {
 		return fmt.Errorf("manifest is nil")
 	}
+	if err := validateSessionManifestSchemaVersion(m.SchemaVersion); err != nil {
+		return err
+	}
 	if strings.TrimSpace(m.SessionID) == "" {
 		return fmt.Errorf("session_id is required")
 	}
@@ -268,10 +269,10 @@ func validateLoadedManifest(m *Manifest) error {
 	if err := validatePostPublishCleanup(m.PostPublishCleanup); err != nil {
 		return err
 	}
+	if err := validateSessionStages(m.Stages); err != nil {
+		return err
+	}
 	for name, stage := range m.Stages {
-		if stage == nil {
-			continue
-		}
 		if err := validateAnalyzeStageState(name, stage.AnalyzeStateVersion, stage.AnalyzeArtifacts); err != nil {
 			return fmt.Errorf("stages.%s: %w", name, err)
 		}
@@ -286,17 +287,7 @@ func validateLoadedManifest(m *Manifest) error {
 }
 
 func normalizeManifest(m *Manifest) {
-	if m.Stages == nil {
-		m.Stages = map[string]*StageRecord{}
-	}
-	for name, stage := range m.Stages {
-		if stage == nil {
-			stage = &StageRecord{Name: name, Status: StatusPending, CreatedAt: m.CreatedAt, UpdatedAt: m.UpdatedAt}
-			m.Stages[name] = stage
-		}
-		if stage.Name == "" {
-			stage.Name = name
-		}
+	for _, stage := range m.Stages {
 		normalizeAnalyzeArtifactCollection(stage.AnalyzeArtifacts)
 	}
 }
@@ -326,6 +317,9 @@ func validateLoadedRunManifest(m *RunManifest) error {
 	if m == nil {
 		return fmt.Errorf("manifest is nil")
 	}
+	if err := validateRunManifestSchemaVersion(m.SchemaVersion); err != nil {
+		return err
+	}
 	if strings.TrimSpace(m.SessionID) == "" {
 		return fmt.Errorf("session_id is required")
 	}
@@ -341,10 +335,10 @@ func validateLoadedRunManifest(m *RunManifest) error {
 	if err := validateManifestIdentities(m.SessionID, m.Campaign, m.RunID); err != nil {
 		return err
 	}
+	if err := validateRunStages(m.Stages); err != nil {
+		return err
+	}
 	for name, stage := range m.Stages {
-		if stage == nil {
-			continue
-		}
 		if err := validateAnalyzeStageState(name, stage.AnalyzeStateVersion, stage.AnalyzeArtifacts); err != nil {
 			return fmt.Errorf("stages.%s: %w", name, err)
 		}
@@ -386,27 +380,65 @@ func validateManifestIdentities(sessionID, campaign, runID string) error {
 }
 
 func normalizeRunManifest(m *RunManifest) {
-	if m.Stages == nil {
-		m.Stages = map[string]*RunStageRecord{}
-	}
-	for name, stage := range m.Stages {
-		if stage == nil {
-			stage = &RunStageRecord{
-				Name:      name,
-				Action:    RunStageActionRun,
-				Status:    StatusPending,
-				CreatedAt: m.CreatedAt,
-				UpdatedAt: m.UpdatedAt,
-			}
-			m.Stages[name] = stage
-		}
-		if stage.Name == "" {
-			stage.Name = name
-		}
+	for _, stage := range m.Stages {
 		normalizeAnalyzeArtifactCollection(stage.AnalyzeArtifacts)
 	}
 }
 
+func validateSessionManifestSchemaVersion(version string) error {
+	return validateManifestSchemaVersion("session manifest", version, SessionManifestSchemaVersion)
+}
+
+func validateRunManifestSchemaVersion(version string) error {
+	return validateManifestSchemaVersion("run manifest", version, RunManifestSchemaVersion)
+}
+
+func validateManifestSchemaVersion(kind, version, supported string) error {
+	if version == supported {
+		return nil
+	}
+	if version == "" {
+		return fmt.Errorf("%s schema_version is required; move aside and regenerate this unsupported local state (supported: %s)", kind, supported)
+	}
+	return fmt.Errorf("unsupported %s schema_version %q; move aside and regenerate this unsupported local state (supported: %s)", kind, version, supported)
+}
+
+func validateSessionStages(stages map[string]*StageRecord) error {
+	if stages == nil {
+		return fmt.Errorf("stages is required")
+	}
+	for name, stage := range stages {
+		if stage == nil {
+			return fmt.Errorf("stages.%s is nil", name)
+		}
+		if stage.Name == "" {
+			return fmt.Errorf("stages.%s.name is required", name)
+		}
+		if stage.Name != name {
+			return fmt.Errorf("stages.%s.name %q does not match map key", name, stage.Name)
+		}
+	}
+	return nil
+}
+
+func validateRunStages(stages map[string]*RunStageRecord) error {
+	if stages == nil {
+		return fmt.Errorf("stages is required")
+	}
+	for name, stage := range stages {
+		if stage == nil {
+			return fmt.Errorf("stages.%s is nil", name)
+		}
+		if stage.Name == "" {
+			return fmt.Errorf("stages.%s.name is required", name)
+		}
+		if stage.Name != name {
+			return fmt.Errorf("stages.%s.name %q does not match map key", name, stage.Name)
+		}
+	}
+	return nil
+}
+
 func writeJSONAtomically(ctx context.Context, path string, data []byte) error {
 	dir := filepath.Dir(path)
 	if err := fileops.EnsureWorkspaceDirectory(dir); err != nil {