Forward passthrough headers on the Serve path

pkg/vmcp/client/client.go

@@ -433,11 +433,19 @@ func (h *httpBackendClient) defaultClientFactory(ctx context.Context, target *vm
 		target:       target,
 	}
 
-	// Inject per-backend HTTP headers from MCPServerEntry.spec.headerForward.
-	// Resolves plaintext + secret-backed headers once here; auth (inner) always
-	// wins over user-supplied headers because it runs after this tripper.
+	// Merge the live per-request forwarded headers (captured by headerforward.CaptureMiddleware
+	// into the request context) with the static per-backend header-forward config. This factory
+	// runs on every backend call, so forwarding is per-request: each call reflects the current
+	// incoming header value. Restricted names and static-config collisions are rejected by
+	// MergeForwardedHeaders.
+	mergedHeaderForward, mergeErr := headerforward.MergeForwardedHeaders(
+		target.HeaderForward, headerforward.ForwardedHeadersFromContext(ctx),
+	)
+	if mergeErr != nil {
+		return nil, fmt.Errorf("failed to merge forwarded headers for backend %s: %w", target.WorkloadID, mergeErr)
+	}
 	baseTransport, err = headerforward.BuildHeaderForwardTripper(
-		ctx, baseTransport, target.HeaderForward, h.secretsProvider, target.WorkloadID,
+		ctx, baseTransport, mergedHeaderForward, h.secretsProvider, target.WorkloadID,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to build header-forward transport: %w", err)

pkg/vmcp/core/core_telemetry.go

@@ -0,0 +1,134 @@
+// SPDX-FileCopyrightText: Copyright 2025 Stacklok, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package core
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/trace"
+
+	"github.com/stacklok/toolhive/pkg/telemetry"
+	"github.com/stacklok/toolhive/pkg/vmcp/composer"
+)
+
+// instrumentationName is the OTEL scope used for all metrics and traces emitted
+// by the core. Must match the scope used by pkg/vmcp/internal/backendtelemetry
+// and pkg/vmcp/server/sessionmanager so they share the same Prometheus namespace.
+const instrumentationName = "github.com/stacklok/toolhive/pkg/vmcp"
+
+// workflowInstruments holds pre-built OTEL instruments for workflow execution
+// telemetry. Created once in core.New when TelemetryProvider is set and reused
+// across all per-call composer factories.
+type workflowInstruments struct {
+	tracer            trace.Tracer
+	executionsTotal   metric.Int64Counter
+	errorsTotal       metric.Int64Counter
+	executionDuration metric.Float64Histogram
+}
+
+// newWorkflowInstruments creates the OTEL instruments for workflow execution
+// telemetry. Returns nil if provider is nil (telemetry disabled).
+// The metric names match those emitted by the session-factory composite-tool
+// decorator in pkg/vmcp/server/sessionmanager so all telemetry appears under
+// the same Prometheus metrics regardless of which path executed the workflow.
+func newWorkflowInstruments(provider *telemetry.Provider) (*workflowInstruments, error) {
+	if provider == nil {
+		return nil, nil
+	}
+
+	meter := provider.MeterProvider().Meter(instrumentationName)
+
+	executions, err := meter.Int64Counter(
+		"toolhive_vmcp_workflow_executions",
+		metric.WithDescription("Total number of workflow executions"),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create workflow executions counter: %w", err)
+	}
+
+	errors, err := meter.Int64Counter(
+		"toolhive_vmcp_workflow_errors",
+		metric.WithDescription("Total number of workflow execution errors"),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create workflow errors counter: %w", err)
+	}
+
+	duration, err := meter.Float64Histogram(
+		"toolhive_vmcp_workflow_duration",
+		metric.WithDescription("Duration of workflow executions in seconds"),
+		metric.WithUnit("s"),
+		metric.WithExplicitBucketBoundaries(telemetry.MCPHistogramBuckets...),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create workflow duration histogram: %w", err)
+	}
+
+	return &workflowInstruments{
+		tracer:            provider.TracerProvider().Tracer(instrumentationName),
+		executionsTotal:   executions,
+		errorsTotal:       errors,
+		executionDuration: duration,
+	}, nil
+}
+
+// telemetryComposer wraps composer.Composer.ExecuteWorkflow with OTEL metrics
+// and tracing. ValidateWorkflow is delegated to the base without instrumentation
+// (validation is called once at startup, not on the hot path).
+type telemetryComposer struct {
+	base        composer.Composer
+	instruments *workflowInstruments
+}
+
+var _ composer.Composer = (*telemetryComposer)(nil)
+
+// ExecuteWorkflow records execution count, duration, and errors before delegating
+// to the wrapped composer. Uses def.Name as the workflow_name metric attribute to
+// match the label emitted by the session-factory path in sessionmanager.
+func (c *telemetryComposer) ExecuteWorkflow(
+	ctx context.Context, def *composer.WorkflowDefinition, params map[string]any,
+) (*composer.WorkflowResult, error) {
+	commonAttrs := []attribute.KeyValue{attribute.String("workflow.name", def.Name)}
+
+	ctx, span := c.instruments.tracer.Start(ctx, "core.ExecuteWorkflow",
+		trace.WithAttributes(commonAttrs...),
+	)
+	defer span.End()
+
+	metricAttrs := metric.WithAttributes(commonAttrs...)
+	start := time.Now()
+	c.instruments.executionsTotal.Add(ctx, 1, metricAttrs)
+
+	result, err := c.base.ExecuteWorkflow(ctx, def, params)
+
+	c.instruments.executionDuration.Record(ctx, time.Since(start).Seconds(), metricAttrs)
+
+	if err != nil {
+		c.instruments.errorsTotal.Add(ctx, 1, metricAttrs)
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+	}
+
+	return result, err
+}
+
+// ValidateWorkflow delegates to the base composer without instrumentation.
+func (c *telemetryComposer) ValidateWorkflow(ctx context.Context, def *composer.WorkflowDefinition) error {
+	return c.base.ValidateWorkflow(ctx, def)
+}
+
+// GetWorkflowStatus delegates to the base composer without instrumentation.
+func (c *telemetryComposer) GetWorkflowStatus(ctx context.Context, workflowID string) (*composer.WorkflowStatus, error) {
+	return c.base.GetWorkflowStatus(ctx, workflowID)
+}
+
+// CancelWorkflow delegates to the base composer without instrumentation.
+func (c *telemetryComposer) CancelWorkflow(ctx context.Context, workflowID string) error {
+	return c.base.CancelWorkflow(ctx, workflowID)
+}

pkg/vmcp/core/core_telemetry_test.go

@@ -0,0 +1,171 @@
+// SPDX-FileCopyrightText: Copyright 2025 Stacklok, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package core
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	tracesdk "go.opentelemetry.io/otel/sdk/trace"
+	tracenoop "go.opentelemetry.io/otel/trace/noop"
+
+	"github.com/stacklok/toolhive/pkg/vmcp/composer"
+)
+
+// stubComposer is already declared in core_calls_test.go (same package).
+// Its fields are: result *composer.WorkflowResult, err error.
+
+// newTestInstruments creates a workflowInstruments backed by an in-memory OTEL SDK
+// and returns the instruments plus a ManualReader for metric assertions.
+// The metric names match production names exactly so the assertions mirror what
+// Prometheus would expose.
+func newTestInstruments(t *testing.T) (*workflowInstruments, *sdkmetric.ManualReader) {
+	t.Helper()
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	tp := tracesdk.NewTracerProvider()
+	meter := mp.Meter(instrumentationName)
+
+	executions, err := meter.Int64Counter("toolhive_vmcp_workflow_executions")
+	require.NoError(t, err)
+	errs, err := meter.Int64Counter("toolhive_vmcp_workflow_errors")
+	require.NoError(t, err)
+	duration, err := meter.Float64Histogram("toolhive_vmcp_workflow_duration")
+	require.NoError(t, err)
+
+	return &workflowInstruments{
+		tracer:            tp.Tracer(instrumentationName),
+		executionsTotal:   executions,
+		errorsTotal:       errs,
+		executionDuration: duration,
+	}, reader
+}
+
+// collectMetrics gathers all metrics from the reader into a snapshot.
+func collectMetrics(t *testing.T, reader *sdkmetric.ManualReader) metricdata.ResourceMetrics {
+	t.Helper()
+	var rm metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(context.Background(), &rm))
+	return rm
+}
+
+// findMetricByName returns the first metric with the given name, or nil.
+func findMetricByName(rm metricdata.ResourceMetrics, name string) *metricdata.Metrics {
+	for _, sm := range rm.ScopeMetrics {
+		for i := range sm.Metrics {
+			if sm.Metrics[i].Name == name {
+				return &sm.Metrics[i]
+			}
+		}
+	}
+	return nil
+}
+
+func int64CounterValue(m *metricdata.Metrics) int64 {
+	if m == nil {
+		return 0
+	}
+	s, ok := m.Data.(metricdata.Sum[int64])
+	if !ok {
+		return 0
+	}
+	var total int64
+	for _, dp := range s.DataPoints {
+		total += dp.Value
+	}
+	return total
+}
+
+func float64HistogramCount(m *metricdata.Metrics) uint64 {
+	if m == nil {
+		return 0
+	}
+	h, ok := m.Data.(metricdata.Histogram[float64])
+	if !ok {
+		return 0
+	}
+	var total uint64
+	for _, dp := range h.DataPoints {
+		total += dp.Count
+	}
+	return total
+}
+
+// TestTelemetryComposer_Success verifies that on a successful ExecuteWorkflow call:
+// - the executions counter increments by 1
+// - the error counter stays at 0
+// - the duration histogram records 1 observation
+// - the result from the inner composer is returned unchanged
+func TestTelemetryComposer_Success(t *testing.T) {
+	t.Parallel()
+
+	instruments, reader := newTestInstruments(t)
+	want := &composer.WorkflowResult{Output: map[string]any{"key": "val"}}
+	tc := &telemetryComposer{
+		base:        stubComposer{result: want},
+		instruments: instruments,
+	}
+
+	def := &composer.WorkflowDefinition{Name: "test-workflow"}
+	got, err := tc.ExecuteWorkflow(context.Background(), def, nil)
+	require.NoError(t, err)
+	assert.Equal(t, want, got)
+
+	rm := collectMetrics(t, reader)
+	assert.Equal(t, int64(1), int64CounterValue(findMetricByName(rm, "toolhive_vmcp_workflow_executions")),
+		"executions counter must increment on success")
+	assert.Equal(t, int64(0), int64CounterValue(findMetricByName(rm, "toolhive_vmcp_workflow_errors")),
+		"errors counter must remain zero on success")
+	assert.Equal(t, uint64(1), float64HistogramCount(findMetricByName(rm, "toolhive_vmcp_workflow_duration")),
+		"duration histogram must record exactly one observation")
+}
+
+// TestTelemetryComposer_Error verifies that on a failed ExecuteWorkflow call:
+// - the executions counter still increments by 1
+// - the error counter also increments by 1
+// - the duration histogram records 1 observation
+// - the error from the inner composer is propagated
+func TestTelemetryComposer_Error(t *testing.T) {
+	t.Parallel()
+
+	instruments, reader := newTestInstruments(t)
+	boom := errors.New("backend exploded")
+	tc := &telemetryComposer{
+		base:        stubComposer{err: boom},
+		instruments: instruments,
+	}
+
+	def := &composer.WorkflowDefinition{Name: "failing-workflow"}
+	_, err := tc.ExecuteWorkflow(context.Background(), def, nil)
+	require.ErrorIs(t, err, boom)
+
+	rm := collectMetrics(t, reader)
+	assert.Equal(t, int64(1), int64CounterValue(findMetricByName(rm, "toolhive_vmcp_workflow_executions")),
+		"executions counter must increment even on failure")
+	assert.Equal(t, int64(1), int64CounterValue(findMetricByName(rm, "toolhive_vmcp_workflow_errors")),
+		"errors counter must increment on failure")
+	assert.Equal(t, uint64(1), float64HistogramCount(findMetricByName(rm, "toolhive_vmcp_workflow_duration")),
+		"duration histogram must record one observation even on failure")
+}
+
+// TestTelemetryComposer_DelegatesNonExecuteMethods verifies that ValidateWorkflow,
+// GetWorkflowStatus, and CancelWorkflow delegate to the base without instrumentation.
+func TestTelemetryComposer_DelegatesNonExecuteMethods(t *testing.T) {
+	t.Parallel()
+
+	tc := &telemetryComposer{
+		base:        stubComposer{},
+		instruments: &workflowInstruments{tracer: tracenoop.Tracer{}},
+	}
+
+	require.NoError(t, tc.ValidateWorkflow(context.Background(), &composer.WorkflowDefinition{}))
+	_, err := tc.GetWorkflowStatus(context.Background(), "any-id")
+	require.NoError(t, err)
+	require.NoError(t, tc.CancelWorkflow(context.Background(), "any-id"))
+}

pkg/vmcp/core/core_vmcp.go

@@ -150,17 +150,30 @@ func New(cfg *Config) (VMCP, error) {
 			"type", fmt.Sprintf("%T", stateStore))
 	}
 
+	// Build workflow telemetry instruments once here so they are reused across all
+	// per-call composer factories. Nil when TelemetryProvider is absent (disabled).
+	// Uses the same metric names as the session-factory composite-tool decorator in
+	// pkg/vmcp/server/sessionmanager so both execution paths emit into the same
+	// Prometheus counters and histograms.
+	instruments, err := newWorkflowInstruments(cfg.TelemetryProvider)
+	if err != nil {
+		stopStore()
+		return nil, fmt.Errorf("failed to create workflow telemetry instruments: %w", err)
+	}
+
 	// composerFactory builds a composite-tool engine bound to a specific routing
-	// table, generalizing server.New's sessionComposerFactory (server.go:393-398).
-	// Workflow-level telemetry (toolhive_vmcp_workflow_*) is intentionally NOT wired
-	// here: that instrumentation lives in the session/Serve layer
-	// (sessionmanager/factory.go:174-176) and is added when the core is wired into
-	// Serve (#5439). This mirrors server.go:393-398, which is also uninstrumented.
+	// table. When telemetry is configured, it wraps the engine with OTEL metrics so
+	// workflow executions are instrumented the same way as the session-factory path
+	// (sessionmanager/factory.go). The telemetryComposer is in core_telemetry.go.
 	composerFactory := func(sessionRT *vmcp.RoutingTable, sessionTools []vmcp.Tool) composer.Composer {
-		return composer.NewWorkflowEngine(
+		engine := composer.NewWorkflowEngine(
 			router.NewSessionRouter(sessionRT), backendClient, elicitationHandler,
 			stateStore, workflowAuditor, sessionTools,
 		)
+		if instruments != nil {
+			return &telemetryComposer{base: engine, instruments: instruments}
+		}
+		return engine
 	}
 
 	// Validate workflows fail-fast (server.go:400-405). The validation engine uses