Add body_from config to step.http_call for raw body forwarding (#227)

* Initial plan

* feat: add body_from config to step.http_call for raw body forwarding

Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

* feat: add body_from config to step.http_call for raw body forwarding

Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

* fix: address review feedback - fix data race in tests, update comments

Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

Author: Copilot

module/pipeline_step_http_call.go

@@ -100,6 +100,7 @@ type HTTPCallStep struct {
 	method     string
 	headers    map[string]string
 	body       map[string]any
+	bodyFrom   string // dot-path into pc.Current or a prior step result via "steps.<name>..."; if set, the resolved value is used as the request body (strings/[]byte sent as-is, other types JSON-marshaled)
 	timeout    time.Duration
 	tmpl       *TemplateEngine
 	auth       *oauthConfig
@@ -142,6 +143,10 @@ func NewHTTPCallStepFactory() StepFactory {
 			step.body = body
 		}
 
+		if bodyFrom, ok := config["body_from"].(string); ok {
+			step.bodyFrom = bodyFrom
+		}
+
 		if timeout, ok := config["timeout"].(string); ok && timeout != "" {
 			if d, err := time.ParseDuration(timeout); err == nil {
 				step.timeout = d
@@ -284,36 +289,56 @@ func (s *HTTPCallStep) getToken(ctx context.Context) (string, error) {
 }
 
 // buildBodyReader constructs the request body reader from the step configuration.
-func (s *HTTPCallStep) buildBodyReader(pc *PipelineContext) (io.Reader, error) {
+func (s *HTTPCallStep) buildBodyReader(pc *PipelineContext) (io.Reader, bool, error) {
+	if s.bodyFrom != "" {
+		val := resolveBodyFrom(s.bodyFrom, pc)
+		switch v := val.(type) {
+		case []byte:
+			return bytes.NewReader(v), true, nil
+		case string:
+			return strings.NewReader(v), true, nil
+		case nil:
+			return nil, true, nil
+		default:
+			data, marshalErr := json.Marshal(v)
+			if marshalErr != nil {
+				return nil, false, fmt.Errorf("http_call step %q: failed to marshal body_from value: %w", s.name, marshalErr)
+			}
+			return bytes.NewReader(data), false, nil
+		}
+	}
 	if s.body != nil {
 		resolvedBody, resolveErr := s.tmpl.ResolveMap(s.body, pc)
 		if resolveErr != nil {
-			return nil, fmt.Errorf("http_call step %q: failed to resolve body: %w", s.name, resolveErr)
+			return nil, false, fmt.Errorf("http_call step %q: failed to resolve body: %w", s.name, resolveErr)
 		}
 		data, marshalErr := json.Marshal(resolvedBody)
 		if marshalErr != nil {
-			return nil, fmt.Errorf("http_call step %q: failed to marshal body: %w", s.name, marshalErr)
+			return nil, false, fmt.Errorf("http_call step %q: failed to marshal body: %w", s.name, marshalErr)
 		}
-		return bytes.NewReader(data), nil
+		return bytes.NewReader(data), false, nil
 	}
 	if s.method != "GET" && s.method != "HEAD" {
 		data, marshalErr := json.Marshal(pc.Current)
 		if marshalErr != nil {
-			return nil, fmt.Errorf("http_call step %q: failed to marshal current data: %w", s.name, marshalErr)
+			return nil, false, fmt.Errorf("http_call step %q: failed to marshal current data: %w", s.name, marshalErr)
 		}
-		return bytes.NewReader(data), nil
+		return bytes.NewReader(data), false, nil
 	}
-	return nil, nil
+	return nil, false, nil
 }
 
 // buildRequest constructs the HTTP request with resolved headers and optional bearer token.
-func (s *HTTPCallStep) buildRequest(ctx context.Context, resolvedURL string, bodyReader io.Reader, pc *PipelineContext, bearerToken string) (*http.Request, error) {
+// rawBody, when true, indicates that the request body is a raw value (string/[]byte/nil,
+// typically provided via body_from) and should not have its Content-Type automatically
+// overridden with application/json.
+func (s *HTTPCallStep) buildRequest(ctx context.Context, resolvedURL string, bodyReader io.Reader, rawBody bool, pc *PipelineContext, bearerToken string) (*http.Request, error) {
 	req, err := http.NewRequestWithContext(ctx, s.method, resolvedURL, bodyReader)
 	if err != nil {
 		return nil, fmt.Errorf("http_call step %q: failed to create request: %w", s.name, err)
 	}
 
-	if bodyReader != nil {
+	if bodyReader != nil && !rawBody {
 		req.Header.Set("Content-Type", "application/json")
 	}
 	for k, v := range s.headers {
@@ -373,7 +398,7 @@ func (s *HTTPCallStep) Execute(ctx context.Context, pc *PipelineContext) (*StepR
 		return nil, fmt.Errorf("http_call step %q: failed to resolve url: %w", s.name, err)
 	}
 
-	bodyReader, err := s.buildBodyReader(pc)
+	bodyReader, rawBody, err := s.buildBodyReader(pc)
 	if err != nil {
 		return nil, err
 	}
@@ -387,7 +412,7 @@ func (s *HTTPCallStep) Execute(ctx context.Context, pc *PipelineContext) (*StepR
 		}
 	}
 
-	req, err := s.buildRequest(ctx, resolvedURL, bodyReader, pc, bearerToken)
+	req, err := s.buildRequest(ctx, resolvedURL, bodyReader, rawBody, pc, bearerToken)
 	if err != nil {
 		return nil, err
 	}
@@ -413,11 +438,11 @@ func (s *HTTPCallStep) Execute(ctx context.Context, pc *PipelineContext) (*StepR
 			return nil, tokenErr
 		}
 
-		retryBody, buildErr := s.buildBodyReader(pc)
+		retryBody, rawBody2, buildErr := s.buildBodyReader(pc)
 		if buildErr != nil {
 			return nil, buildErr
 		}
-		retryReq, buildErr := s.buildRequest(ctx, resolvedURL, retryBody, pc, newToken)
+		retryReq, buildErr := s.buildRequest(ctx, resolvedURL, retryBody, rawBody2, pc, newToken)
 		if buildErr != nil {
 			return nil, buildErr
 		}

module/pipeline_step_http_call_test.go

@@ -1,9 +1,11 @@
 package module
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -501,3 +503,208 @@ func TestHTTPCallStep_OAuth2_ConcurrentFetch(t *testing.T) {
 		t.Errorf("expected exactly 1 token request via singleflight, got %d", n)
 	}
 }
+
+// TestHTTPCallStep_BodyFrom_String verifies that body_from with a string value sends raw bytes
+// without JSON-encoding and without auto-setting Content-Type: application/json.
+func TestHTTPCallStep_BodyFrom_String(t *testing.T) {
+	type captured struct {
+		body        []byte
+		contentType string
+	}
+	ch := make(chan captured, 1)
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Errorf("failed to read request body: %v", err)
+		}
+		ch <- captured{body: b, contentType: r.Header.Get("Content-Type")}
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(`{"ok":true}`))
+	}))
+	defer srv.Close()
+
+	factory := NewHTTPCallStepFactory()
+	step, err := factory("body-from-string", map[string]any{
+		"url":       srv.URL,
+		"method":    "POST",
+		"body_from": "raw_payload",
+	}, nil)
+	if err != nil {
+		t.Fatalf("factory error: %v", err)
+	}
+	step.(*HTTPCallStep).httpClient = srv.Client()
+
+	pc := NewPipelineContext(nil, nil)
+	pc.Current["raw_payload"] = `{"hello":"world"}`
+
+	if _, err := step.Execute(context.Background(), pc); err != nil {
+		t.Fatalf("execute error: %v", err)
+	}
+
+	got := <-ch
+	if string(got.body) != `{"hello":"world"}` {
+		t.Errorf("expected raw body %q, got %q", `{"hello":"world"}`, string(got.body))
+	}
+	// Content-Type should NOT be auto-set to application/json for raw bodies
+	if got.contentType == "application/json" {
+		t.Errorf("expected Content-Type not to be application/json for body_from, got %q", got.contentType)
+	}
+}
+
+// TestHTTPCallStep_BodyFrom_Bytes verifies that body_from with a []byte value sends raw bytes.
+func TestHTTPCallStep_BodyFrom_Bytes(t *testing.T) {
+	ch := make(chan []byte, 1)
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Errorf("failed to read request body: %v", err)
+		}
+		ch <- b
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(`{}`))
+	}))
+	defer srv.Close()
+
+	factory := NewHTTPCallStepFactory()
+	step, err := factory("body-from-bytes", map[string]any{
+		"url":       srv.URL,
+		"method":    "POST",
+		"body_from": "raw_data",
+	}, nil)
+	if err != nil {
+		t.Fatalf("factory error: %v", err)
+	}
+	step.(*HTTPCallStep).httpClient = srv.Client()
+
+	pc := NewPipelineContext(nil, nil)
+	pc.Current["raw_data"] = []byte("binary\x00data")
+
+	if _, err := step.Execute(context.Background(), pc); err != nil {
+		t.Fatalf("execute error: %v", err)
+	}
+
+	gotBody := <-ch
+	if !bytes.Equal(gotBody, []byte("binary\x00data")) {
+		t.Errorf("expected raw bytes, got %q", string(gotBody))
+	}
+}
+
+// TestHTTPCallStep_BodyFrom_StepOutput verifies that body_from can resolve from step outputs.
+func TestHTTPCallStep_BodyFrom_StepOutput(t *testing.T) {
+	ch := make(chan []byte, 1)
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Errorf("failed to read request body: %v", err)
+		}
+		ch <- b
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(`{}`))
+	}))
+	defer srv.Close()
+
+	factory := NewHTTPCallStepFactory()
+	step, err := factory("body-from-step", map[string]any{
+		"url":       srv.URL,
+		"method":    "POST",
+		"body_from": "steps.parse.raw_body",
+	}, nil)
+	if err != nil {
+		t.Fatalf("factory error: %v", err)
+	}
+	step.(*HTTPCallStep).httpClient = srv.Client()
+
+	pc := NewPipelineContext(nil, nil)
+	pc.StepOutputs["parse"] = map[string]any{
+		"raw_body": `{"event":"push"}`,
+	}
+
+	if _, err := step.Execute(context.Background(), pc); err != nil {
+		t.Fatalf("execute error: %v", err)
+	}
+
+	gotBody := <-ch
+	if string(gotBody) != `{"event":"push"}` {
+		t.Errorf("expected raw body from step output, got %q", string(gotBody))
+	}
+}
+
+// TestHTTPCallStep_BodyFrom_ContentTypeOverride verifies that Content-Type set in headers
+// takes effect even with body_from.
+func TestHTTPCallStep_BodyFrom_ContentTypeOverride(t *testing.T) {
+	ch := make(chan string, 1)
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ch <- r.Header.Get("Content-Type")
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(`{}`))
+	}))
+	defer srv.Close()
+
+	factory := NewHTTPCallStepFactory()
+	step, err := factory("body-from-ct", map[string]any{
+		"url":       srv.URL,
+		"method":    "POST",
+		"body_from": "payload",
+		"headers": map[string]any{
+			"Content-Type": "application/xml",
+		},
+	}, nil)
+	if err != nil {
+		t.Fatalf("factory error: %v", err)
+	}
+	step.(*HTTPCallStep).httpClient = srv.Client()
+
+	pc := NewPipelineContext(nil, nil)
+	pc.Current["payload"] = `<root><item>1</item></root>`
+
+	if _, err := step.Execute(context.Background(), pc); err != nil {
+		t.Fatalf("execute error: %v", err)
+	}
+
+	gotCT := <-ch
+	if gotCT != "application/xml" {
+		t.Errorf("expected Content-Type application/xml, got %q", gotCT)
+	}
+}
+
+// TestHTTPCallStep_BodyFrom_NilValue verifies that body_from with a missing path sends no body.
+func TestHTTPCallStep_BodyFrom_NilValue(t *testing.T) {
+	ch := make(chan []byte, 1)
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Errorf("failed to read request body: %v", err)
+		}
+		ch <- b
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(`{}`))
+	}))
+	defer srv.Close()
+
+	factory := NewHTTPCallStepFactory()
+	step, err := factory("body-from-nil", map[string]any{
+		"url":       srv.URL,
+		"method":    "POST",
+		"body_from": "nonexistent.path",
+	}, nil)
+	if err != nil {
+		t.Fatalf("factory error: %v", err)
+	}
+	step.(*HTTPCallStep).httpClient = srv.Client()
+
+	pc := NewPipelineContext(nil, nil)
+
+	if _, err := step.Execute(context.Background(), pc); err != nil {
+		t.Fatalf("execute error: %v", err)
+	}
+
+	gotBody := <-ch
+	if len(gotBody) != 0 {
+		t.Errorf("expected empty body for nil body_from, got %q", string(gotBody))
+	}
+}