fix: harden aggregate Helm result reporting

Author: shirkevich

pkg/ci/plugins/helm/aggregate.go

@@ -277,13 +277,9 @@ func writeHelmAggregateDetails(builder *strings.Builder, command string, compone
 		detail.WriteString(helmMarkdownCell(component.Result.Stack + "/" + component.Result.Component + ": " + component.Status))
 		detail.WriteString("</summary>\n\n")
 		if component.Status == "failed" {
-			detail.WriteString("```text\n")
-			detail.WriteString(plugin.TruncateDetail(component.Result.Error))
-			detail.WriteString("\n```\n")
+			detail.WriteString(helmMarkdownCodeBlock("text", plugin.TruncateDetail(component.Result.Error)))
 		} else {
-			detail.WriteString("```diff\n")
-			detail.WriteString(plugin.TruncateDetail(component.Summary.Diff))
-			detail.WriteString("\n```\n")
+			detail.WriteString(helmMarkdownCodeBlock("diff", plugin.TruncateDetail(component.Summary.Diff)))
 		}
 		detail.WriteString("\n</details>\n\n")
 		if builder.Len()+detail.Len() > helmAggregateMarkdownMaxBytes {
@@ -294,6 +290,14 @@ func writeHelmAggregateDetails(builder *strings.Builder, command string, compone
 	}
 }
 
+func helmMarkdownCodeBlock(language, value string) string {
+	fence := "```"
+	for strings.Contains(value, fence) {
+		fence += "`"
+	}
+	return fence + language + "\n" + value + "\n" + fence + "\n"
+}
+
 func formatHelmAggregateDuration(milliseconds int64) string {
 	if milliseconds <= 0 {
 		return "-"

pkg/ci/plugins/helm/aggregate_test.go

@@ -72,6 +72,29 @@ func TestOnAfterAggregateRendersApplySummary(t *testing.T) {
 	assert.Contains(t, writer.summary, "| dev | api | succeeded | api | api | apps | kubernetes | - |")
 }
 
+func TestOnAfterAggregateUsesSafeDetailFences(t *testing.T) {
+	writer := &fakeWriter{}
+	err := (&Plugin{}).onAfterAggregate(&plugin.HookContext{
+		Provider: fakeProvider{writer: writer},
+		Aggregate: schema.HelmCIResultSet{
+			Command: "plan",
+			Results: []schema.HelmCIResult{
+				{
+					Stack: "dev", Component: "api", Processed: true,
+					Summary: map[string]any{"diff": "+ change\n```\nnot summary Markdown"},
+				},
+				{
+					Stack: "dev", Component: "worker", Status: "failed",
+					Error: "render failed\n```\nnot summary Markdown",
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+	assert.Contains(t, writer.summary, "````diff\n+ change\n```\nnot summary Markdown\n````")
+	assert.Contains(t, writer.summary, "````text\nrender failed\n```\nnot summary Markdown\n````")
+}
+
 func TestOnAfterAggregateSkipsInvalidOrDisabledAndReturnsWriterError(t *testing.T) {
 	pluginUnderTest := &Plugin{}
 	require.NoError(t, pluginUnderTest.onAfterAggregate(&plugin.HookContext{Provider: fakeProvider{}, Aggregate: "invalid"}))

pkg/component/graph.go

@@ -39,6 +39,12 @@ type GraphExecutionOptions struct {
 	Selection     *GraphSelection
 }
 
+// GraphNodeSkipObserver is implemented by providers that need to record graph
+// nodes skipped after execution stops before reaching them.
+type GraphNodeSkipObserver interface {
+	OnGraphNodeSkipped(node *dependency.Node)
+}
+
 // ExecuteGraph runs selected components in dependency order and stops before
 // starting another component when the caller context is canceled.
 func ExecuteGraph(ctx context.Context, opts *GraphExecutionOptions) error {
@@ -63,24 +69,37 @@ func ExecuteGraph(ctx context.Context, opts *GraphExecutionOptions) error {
 	for i := range order {
 		select {
 		case <-ctx.Done():
+			notifyGraphNodeSkips(opts.Provider, order[i:])
 			return fmt.Errorf(errUtils.ErrWrapFormat, errUtils.ErrGraphExecutionCanceled, ctx.Err())
 		default:
 		}
 
 		if err := executeGraphNode(ctx, opts, &order[i]); err != nil {
+			notifyGraphNodeSkips(opts.Provider, order[i+1:])
 			if ctxErr := ctx.Err(); ctxErr != nil {
 				return fmt.Errorf(errUtils.ErrWrapFormat, errUtils.ErrGraphExecutionCanceled, errors.Join(ctxErr, err))
 			}
 			return err
 		}
 		if ctxErr := ctx.Err(); ctxErr != nil {
+			notifyGraphNodeSkips(opts.Provider, order[i+1:])
 			return fmt.Errorf(errUtils.ErrWrapFormat, errUtils.ErrGraphExecutionCanceled, ctxErr)
 		}
 	}
 
 	return nil
 }
 
+func notifyGraphNodeSkips(provider ComponentProvider, nodes dependency.ExecutionOrder) {
+	observer, ok := provider.(GraphNodeSkipObserver)
+	if !ok {
+		return
+	}
+	for i := range nodes {
+		observer.OnGraphNodeSkipped(&nodes[i])
+	}
+}
+
 // prepareExecutionOrder validates options, builds and filters the graph, and returns
 // the topologically sorted execution order. An empty order indicates no matching components.
 func prepareExecutionOrder(opts *GraphExecutionOptions) (dependency.ExecutionOrder, error) {

pkg/component/helm/aggregate_ci.go

@@ -7,6 +7,7 @@ import (
 
 	errUtils "github.com/cloudposse/atmos/errors"
 	"github.com/cloudposse/atmos/pkg/component"
+	"github.com/cloudposse/atmos/pkg/dependency"
 	"github.com/cloudposse/atmos/pkg/hooks"
 	log "github.com/cloudposse/atmos/pkg/logger"
 	"github.com/cloudposse/atmos/pkg/schema"
@@ -32,7 +33,7 @@ func newHelmBulkCICollector(command string) *helmBulkCICollector {
 // bulkCollectingProvider wraps the native Helm provider and records failures
 // that happen before an operation produces its structured summary.
 type bulkCollectingProvider struct {
-	*ComponentProvider
+	component.ComponentProvider
 	collector *helmBulkCICollector
 }
 
@@ -43,6 +44,13 @@ func (p *bulkCollectingProvider) Execute(ctx *component.ExecutionContext) error
 	return err
 }
 
+func (p *bulkCollectingProvider) OnGraphNodeSkipped(node *dependency.Node) {
+	if p == nil || p.collector == nil || node == nil {
+		return
+	}
+	p.collector.markSkipped(node.Stack, node.Component)
+}
+
 func helmBulkCollector(ctx *component.ExecutionContext) *helmBulkCICollector {
 	if ctx == nil {
 		return nil
@@ -82,6 +90,17 @@ func (c *helmBulkCICollector) finish(ctx *component.ExecutionContext, startedAt,
 	applyHelmResultError(result, execErr)
 }
 
+func (c *helmBulkCICollector) markSkipped(stack, componentName string) {
+	if c == nil {
+		return
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	result := c.ensure(stack, componentName)
+	result.Status = "skipped"
+}
+
 func (c *helmBulkCICollector) ensure(stack, componentName string) *schema.HelmCIResult {
 	nodeID := component.GraphNodeID(componentName, stack)
 	if result, ok := c.results[nodeID]; ok {
@@ -137,11 +156,28 @@ func cloneHelmSummary(summary map[string]any) map[string]any {
 	}
 	cloned := make(map[string]any, len(summary))
 	for key, value := range summary {
-		cloned[key] = value
+		cloned[key] = cloneHelmSummaryValue(value)
 	}
 	return cloned
 }
 
+func cloneHelmSummaryValue(value any) any {
+	switch typed := value.(type) {
+	case map[string]any:
+		return cloneHelmSummary(typed)
+	case []any:
+		cloned := make([]any, len(typed))
+		for i := range typed {
+			cloned[i] = cloneHelmSummaryValue(typed[i])
+		}
+		return cloned
+	case []string:
+		return append([]string(nil), typed...)
+	default:
+		return value
+	}
+}
+
 func supportsHelmAggregateCI(command string) bool {
 	switch command {
 	case "plan", "diff", "apply", "deploy":

pkg/component/helm/aggregate_ci_test.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/cloudposse/atmos/pkg/auth"
 	"github.com/cloudposse/atmos/pkg/component"
+	cfg "github.com/cloudposse/atmos/pkg/config"
 	"github.com/cloudposse/atmos/pkg/hooks"
 	"github.com/cloudposse/atmos/pkg/schema"
 )
@@ -39,6 +40,87 @@ func TestHelmBulkCICollectorRecordsAndSortsResults(t *testing.T) {
 	assert.Equal(t, "+ change", collector.resultSet().Results[1].Summary["diff"])
 }
 
+func TestHelmBulkCICollectorDeepCopiesNestedSummary(t *testing.T) {
+	collector := newHelmBulkCICollector("apply")
+	wait := map[string]any{"strategy": "watcher"}
+	kinds := []string{"Deployment", "Service"}
+	summary := map[string]any{
+		"release":      map[string]any{"wait": wait},
+		"object_kinds": kinds,
+	}
+	collector.setSummary(&schema.ConfigAndStacksInfo{Stack: "dev", ComponentFromArg: "api"}, summary, nil)
+
+	wait["strategy"] = "legacy"
+	kinds[0] = "Job"
+	result := collector.resultSet().Results[0]
+	release := result.Summary["release"].(map[string]any)
+	assert.Equal(t, "watcher", release["wait"].(map[string]any)["strategy"])
+	assert.Equal(t, []string{"Deployment", "Service"}, result.Summary["object_kinds"])
+
+	release["wait"].(map[string]any)["strategy"] = "mutated"
+	result.Summary["object_kinds"].([]string)[1] = "ConfigMap"
+	retained := collector.resultSet().Results[0].Summary
+	assert.Equal(t, "watcher", retained["release"].(map[string]any)["wait"].(map[string]any)["strategy"])
+	assert.Equal(t, []string{"Deployment", "Service"}, retained["object_kinds"])
+}
+
+type helmBulkGraphTestProvider struct {
+	failComponent string
+}
+
+func (p *helmBulkGraphTestProvider) GetType() string                               { return cfg.HelmComponentType }
+func (p *helmBulkGraphTestProvider) GetGroup() string                              { return "test" }
+func (p *helmBulkGraphTestProvider) GetBasePath(*schema.AtmosConfiguration) string { return "" }
+func (p *helmBulkGraphTestProvider) ListComponents(context.Context, string, map[string]any) ([]string, error) {
+	return nil, nil
+}
+func (p *helmBulkGraphTestProvider) ValidateComponent(map[string]any) error { return nil }
+func (p *helmBulkGraphTestProvider) Execute(ctx *component.ExecutionContext) error {
+	if ctx.Component == p.failComponent {
+		return errors.New("operation failed")
+	}
+	return nil
+}
+func (p *helmBulkGraphTestProvider) GenerateArtifacts(*component.ExecutionContext) error { return nil }
+func (p *helmBulkGraphTestProvider) GetAvailableCommands() []string                      { return nil }
+
+func TestHelmBulkCICollectorRecordsDependencyBlockedComponents(t *testing.T) {
+	collector := newHelmBulkCICollector("apply")
+	provider := &bulkCollectingProvider{
+		ComponentProvider: &helmBulkGraphTestProvider{failComponent: "base"},
+		collector:         collector,
+	}
+	stacks := map[string]any{
+		"dev": map[string]any{
+			cfg.ComponentsSectionName: map[string]any{
+				cfg.HelmComponentType: map[string]any{
+					"base": map[string]any{},
+					"api": map[string]any{
+						cfg.SettingsSectionName: map[string]any{"depends_on": []any{"base"}},
+					},
+				},
+			},
+		},
+	}
+
+	err := component.ExecuteGraph(context.Background(), &component.GraphExecutionOptions{
+		Provider:      provider,
+		Info:          &schema.ConfigAndStacksInfo{},
+		Stacks:        stacks,
+		ComponentType: cfg.HelmComponentType,
+		SubCommand:    "apply",
+	})
+	require.Error(t, err)
+
+	results := collector.resultSet().Results
+	require.Len(t, results, 2)
+	assert.Equal(t, "api", results[0].Component)
+	assert.Equal(t, "skipped", results[0].Status)
+	assert.False(t, results[0].Processed)
+	assert.Equal(t, "base", results[1].Component)
+	assert.Equal(t, "failed", results[1].Status)
+}
+
 func TestHelmBulkCICollectorUsesProcessedComponentIdentity(t *testing.T) {
 	collector := newHelmBulkCICollector("plan")
 	info := schema.ConfigAndStacksInfo{Stack: "dev", ComponentFromArg: "apps/app"}