feat: unify pipeline execution paths — all triggers preserve StepOutputs via PipelineContextHolder

Adds PipelineContextHolder (module.PipelineContextHolder / module.PipelineContextKey)
alongside the existing PipelineResultHolder. Pipeline.Run() now populates the holder
when it is present in the context, so every caller of TriggerWorkflow — HTTP triggers,
EventBus triggers, scheduler triggers, messaging triggers — gets full StepOutputs
visibility without a separate code path.

ExecutePipeline and ExecutePipelineContext are unified through TriggerWorkflow;
both inject a PipelineContextHolder before calling TriggerWorkflow and read the
result back, eliminating the bypass that previously discarded step output data.

FireEvent in wftest now also injects a PipelineContextHolder so event-driven test
assertions get StepResults populated, matching the behaviour of FireSchedule and
ExecutePipeline. Three TDD tests assert the new behaviour across all paths.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: intel352

engine.go

@@ -694,15 +694,20 @@ func (e *StdEngine) TriggerWorkflow(ctx context.Context, workflowType string, ac
 //
 // If the pipeline uses step.pipeline_output, the explicitly marked output
 // is returned. Otherwise, the pipeline's merged Current state is returned.
+//
+// This goes through TriggerWorkflow so all trigger paths share the same
+// execution code, including StepOutputs preservation via PipelineContextHolder.
 func (e *StdEngine) ExecutePipeline(ctx context.Context, name string, data map[string]any) (map[string]any, error) {
-	pipeline, ok := e.pipelineRegistry[name]
-	if !ok {
-		return nil, fmt.Errorf("pipeline %q not found", name)
+	holder := &module.PipelineContextHolder{}
+	ctx = context.WithValue(ctx, module.PipelineContextKey, holder)
+
+	if err := e.TriggerWorkflow(ctx, "pipeline:"+name, "", data); err != nil {
+		return nil, err
 	}
 
-	pc, err := pipeline.Execute(ctx, data)
-	if err != nil {
-		return nil, fmt.Errorf("pipeline %q: %w", name, err)
+	pc := holder.Get()
+	if pc == nil {
+		return nil, fmt.Errorf("pipeline %q: no context returned", name)
 	}
 
 	// Prefer explicit pipeline output if step.pipeline_output was used.
@@ -720,15 +725,20 @@ func (e *StdEngine) ExecutePipeline(ctx context.Context, name string, data map[s
 // ExecutePipelineContext runs a named pipeline and returns the full PipelineContext,
 // including StepOutputs for each step. This is intended for test harnesses that need
 // per-step output inspection. For production callers, use ExecutePipeline instead.
+//
+// This goes through TriggerWorkflow so all trigger paths share the same
+// execution code. The PipelineContextHolder captures the full context.
 func (e *StdEngine) ExecutePipelineContext(ctx context.Context, name string, data map[string]any) (*interfaces.PipelineContext, error) {
-	pipeline, ok := e.pipelineRegistry[name]
-	if !ok {
-		return nil, fmt.Errorf("pipeline %q not found", name)
+	holder := &module.PipelineContextHolder{}
+	ctx = context.WithValue(ctx, module.PipelineContextKey, holder)
+
+	if err := e.TriggerWorkflow(ctx, "pipeline:"+name, "", data); err != nil {
+		return nil, err
 	}
 
-	pc, err := pipeline.Execute(ctx, data)
-	if err != nil {
-		return nil, fmt.Errorf("pipeline %q: %w", name, err)
+	pc := holder.Get()
+	if pc == nil {
+		return nil, fmt.Errorf("pipeline %q: no context returned", name)
 	}
 
 	return pc, nil

engine_pipeline_test.go

@@ -736,3 +736,156 @@ func TestExecutePipeline_UnknownPipeline(t *testing.T) {
 		t.Errorf("expected error to mention pipeline name, got: %v", err)
 	}
 }
+
+// TestTriggerWorkflow_PreservesStepOutputs verifies that injecting a
+// PipelineContextHolder into the context before calling TriggerWorkflow causes
+// StepOutputs to be populated after the pipeline runs.
+func TestTriggerWorkflow_PreservesStepOutputs(t *testing.T) {
+	engine, _ := setupPipelineEngine(t)
+
+	pipelineCfg := map[string]any{
+		"step_output_pipeline": map[string]any{
+			"steps": []any{
+				map[string]any{
+					"name": "set_value",
+					"type": "step.set",
+					"config": map[string]any{
+						"values": map[string]any{
+							"color": "blue",
+						},
+					},
+				},
+			},
+		},
+	}
+	if err := engine.configurePipelines(pipelineCfg); err != nil {
+		t.Fatalf("configurePipelines failed: %v", err)
+	}
+
+	holder := &module.PipelineContextHolder{}
+	ctx := context.WithValue(context.Background(), module.PipelineContextKey, holder)
+
+	if err := engine.TriggerWorkflow(ctx, "pipeline:step_output_pipeline", "", map[string]any{}); err != nil {
+		t.Fatalf("TriggerWorkflow failed: %v", err)
+	}
+
+	pc := holder.Get()
+	if pc == nil {
+		t.Fatal("expected PipelineContextHolder to be populated after TriggerWorkflow")
+	}
+	if pc.StepOutputs == nil {
+		t.Fatal("expected StepOutputs to be non-nil")
+	}
+	stepOut, ok := pc.StepOutputs["set_value"]
+	if !ok {
+		t.Fatalf("expected StepOutputs to contain 'set_value', got keys: %v", stepKeys(pc.StepOutputs))
+	}
+	if stepOut["color"] != "blue" {
+		t.Errorf("expected step output color=blue, got %v", stepOut["color"])
+	}
+}
+
+// TestExecutePipeline_UsesUnifiedPath verifies that ExecutePipeline and
+// TriggerWorkflow with a PipelineContextHolder produce identical results.
+func TestExecutePipeline_UsesUnifiedPath(t *testing.T) {
+	engine, _ := setupPipelineEngine(t)
+
+	pipelineCfg := map[string]any{
+		"unified_path_pipeline": map[string]any{
+			"steps": []any{
+				map[string]any{
+					"name": "set_msg",
+					"type": "step.set",
+					"config": map[string]any{
+						"values": map[string]any{
+							"msg": "hello",
+						},
+					},
+				},
+			},
+		},
+	}
+	if err := engine.configurePipelines(pipelineCfg); err != nil {
+		t.Fatalf("configurePipelines failed: %v", err)
+	}
+
+	// Path 1: ExecutePipeline
+	result1, err := engine.ExecutePipeline(context.Background(), "unified_path_pipeline", map[string]any{})
+	if err != nil {
+		t.Fatalf("ExecutePipeline failed: %v", err)
+	}
+
+	// Path 2: TriggerWorkflow with PipelineContextHolder
+	holder := &module.PipelineContextHolder{}
+	ctx := context.WithValue(context.Background(), module.PipelineContextKey, holder)
+	if err := engine.TriggerWorkflow(ctx, "pipeline:unified_path_pipeline", "", map[string]any{}); err != nil {
+		t.Fatalf("TriggerWorkflow failed: %v", err)
+	}
+	pc := holder.Get()
+	if pc == nil {
+		t.Fatal("holder should be populated")
+	}
+
+	if result1["msg"] != pc.Current["msg"] {
+		t.Errorf("ExecutePipeline and TriggerWorkflow produced different results: %v vs %v", result1["msg"], pc.Current["msg"])
+	}
+	if pc.StepOutputs["set_msg"]["msg"] != "hello" {
+		t.Errorf("expected StepOutputs[set_msg][msg]=hello, got %v", pc.StepOutputs["set_msg"]["msg"])
+	}
+}
+
+// TestExecutePipelineContext_PreservesStepOutputs verifies that ExecutePipelineContext
+// returns a PipelineContext with StepOutputs populated for all executed steps.
+func TestExecutePipelineContext_PreservesStepOutputs(t *testing.T) {
+	engine, _ := setupPipelineEngine(t)
+
+	pipelineCfg := map[string]any{
+		"ctx_step_output_pipeline": map[string]any{
+			"steps": []any{
+				map[string]any{
+					"name": "step_a",
+					"type": "step.set",
+					"config": map[string]any{
+						"values": map[string]any{"a": "1"},
+					},
+				},
+				map[string]any{
+					"name": "step_b",
+					"type": "step.set",
+					"config": map[string]any{
+						"values": map[string]any{"b": "2"},
+					},
+				},
+			},
+		},
+	}
+	if err := engine.configurePipelines(pipelineCfg); err != nil {
+		t.Fatalf("configurePipelines failed: %v", err)
+	}
+
+	pc, err := engine.ExecutePipelineContext(context.Background(), "ctx_step_output_pipeline", map[string]any{})
+	if err != nil {
+		t.Fatalf("ExecutePipelineContext failed: %v", err)
+	}
+	if pc == nil {
+		t.Fatal("expected non-nil PipelineContext")
+	}
+	if len(pc.StepOutputs) != 2 {
+		t.Errorf("expected 2 step outputs, got %d: %v", len(pc.StepOutputs), stepKeys(pc.StepOutputs))
+	}
+	if pc.StepOutputs["step_a"]["a"] != "1" {
+		t.Errorf("expected step_a output a=1, got %v", pc.StepOutputs["step_a"]["a"])
+	}
+	if pc.StepOutputs["step_b"]["b"] != "2" {
+		t.Errorf("expected step_b output b=2, got %v", pc.StepOutputs["step_b"]["b"])
+	}
+}
+
+// stepKeys returns the keys of a step output map for test failure messages.
+func stepKeys(m map[string]map[string]any) []string {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	return keys
+}

module/http_trigger.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/GoCodeAlone/modular"
 	"github.com/GoCodeAlone/workflow/interfaces"
@@ -61,6 +62,37 @@ func (h *PipelineResultHolder) Get() map[string]any {
 	return h.result
 }
 
+// pipelineContextKey is the unexported type for the PipelineContextHolder context key.
+type pipelineContextKey struct{}
+
+// PipelineContextKey is the context key used to capture the full PipelineContext
+// (including StepOutputs) from TriggerWorkflow. Callers that need per-step output
+// visibility inject a *PipelineContextHolder into the context before calling
+// TriggerWorkflow; the pipeline handler populates it after execution.
+var PipelineContextKey = pipelineContextKey{}
+
+// PipelineContextHolder is a thread-safe container used to pass the full
+// PipelineContext (including StepOutputs) back through the context from the
+// pipeline handler to any caller of TriggerWorkflow that needs per-step visibility.
+type PipelineContextHolder struct {
+	mu  sync.Mutex
+	ctx *PipelineContext
+}
+
+// Set stores the PipelineContext in the holder.
+func (h *PipelineContextHolder) Set(pc *PipelineContext) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.ctx = pc
+}
+
+// Get returns the stored PipelineContext, or nil if not set.
+func (h *PipelineContextHolder) Get() *PipelineContext {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.ctx
+}
+
 // coercePipelineStatus coerces common numeric/string types into an HTTP status
 // code. Pipeline steps may emit response_status as int, int64, float64 (common
 // after generic JSON decoding), json.Number, or a numeric string.

module/pipeline_executor.go

@@ -359,10 +359,18 @@ func (p *Pipeline) SetEventRecorder(recorder interfaces.EventRecorder) {
 // Run executes the pipeline and returns the merged result data map.
 // It implements interfaces.PipelineRunner by wrapping Execute and
 // returning PipelineContext.Current so callers need not import PipelineContext.
+// If a PipelineContextHolder is stored in ctx under PipelineContextKey, it is
+// populated with the full PipelineContext (including StepOutputs) so that callers
+// can inspect per-step results without a separate code path.
 func (p *Pipeline) Run(ctx context.Context, data map[string]any) (map[string]any, error) {
 	pc, err := p.Execute(ctx, data)
 	if err != nil {
 		return nil, err
 	}
+	// Populate the PipelineContextHolder if present in context, so all callers
+	// of TriggerWorkflow can receive full step output visibility.
+	if holder, ok := ctx.Value(PipelineContextKey).(*PipelineContextHolder); ok && holder != nil {
+		holder.Set(pc)
+	}
 	return pc.Current, nil
 }

wftest/events.go

@@ -25,8 +25,10 @@ func (h *Harness) FireEvent(topic string, data map[string]any) *Result {
 		data = map[string]any{}
 	}
 
-	holder := &module.PipelineResultHolder{}
-	ctx := context.WithValue(h.t.Context(), module.PipelineResultContextKey, holder)
+	resultHolder := &module.PipelineResultHolder{}
+	ctxHolder := &module.PipelineContextHolder{}
+	ctx := context.WithValue(h.t.Context(), module.PipelineResultContextKey, resultHolder)
+	ctx = context.WithValue(ctx, module.PipelineContextKey, ctxHolder)
 
 	start := time.Now()
 	for _, trigger := range h.engine.Triggers() {
@@ -39,8 +41,21 @@ func (h *Harness) FireEvent(topic string, data map[string]any) *Result {
 		}
 	}
 
+	// Prefer the full PipelineContext (with StepOutputs) when available.
+	if pc := ctxHolder.Get(); pc != nil {
+		output := pc.Current
+		if pipeOut, ok := pc.Metadata["_pipeline_output"].(map[string]any); ok {
+			output = pipeOut
+		}
+		return &Result{
+			Output:      output,
+			StepResults: pc.StepOutputs,
+			Duration:    time.Since(start),
+		}
+	}
+
 	return &Result{
-		Output:   holder.Get(),
+		Output:   resultHolder.Get(),
 		Duration: time.Since(start),
 	}
 }