Unwrap PortableValue before invoking edge Condition and fan-out Assigner

workflow/internal/execution/edgerunner.go

@@ -198,12 +198,21 @@ func (em *EdgeRunner) PrepareDeliveryForEdge(ctx context.Context, edge workflow.
 		span.End()
 	}()
 
-	if edge.Condition != nil && !edge.Condition(envelope.Message) {
+	// Unwrap a PortableValue to its declared runtime type before invoking the
+	// caller-supplied Condition and Assigner delegates, so a user predicate like
+	// m.(Order).Total > 100 sees the concrete type rather than the wrapper.
+	// Only unwrap when a delegate will actually consume the message; otherwise
+	// avoid the cost (and potential side-effects) of resolving the runtime type.
+	message := envelope.Message
+	if edge.Condition != nil || edge.Assigner != nil {
+		message = em.unwrapMessage(ctx, envelope)
+	}
+	if edge.Condition != nil && !edge.Condition(message) {
 		// Condition not met; do not route message.
 		span.SetDeliveryStatus(observability.DeliveryStatusDroppedConditionFalse)
 		return nil, nil
 	}
-	targetIDs := selectedTargetIDs(edge, envelope)
+	targetIDs := selectedTargetIDs(edge, envelope, message)
 	if len(targetIDs) == 0 {
 		span.SetDeliveryStatus(observability.DeliveryStatusDroppedTargetMismatch)
 		return nil, nil
@@ -279,14 +288,33 @@ func (em *EdgeRunner) filterEnvelopesForTarget(ctx context.Context, envelopes []
 	return filtered, nil
 }
 
-func selectedTargetIDs(edge workflow.Edge, envelope *MessageEnvelope) []string {
+// unwrapMessage resolves a workflow.PortableValue envelope to its declared
+// runtime type so the caller-supplied Condition and Assigner delegates receive
+// the concrete message value, mirroring messageRouter.routeMessage. It falls
+// back to the raw message whenever the runtime type cannot be resolved.
+func (em *EdgeRunner) unwrapMessage(ctx context.Context, envelope *MessageEnvelope) any {
+	portable, ok := envelope.Message.(workflow.PortableValue)
+	if !ok {
+		return envelope.Message
+	}
+	runtimeType, err := em.messageRuntimeType(ctx, envelope)
+	if err != nil || runtimeType == nil {
+		return envelope.Message
+	}
+	if value, ok := portable.As(runtimeType); ok {
+		return value
+	}
+	return envelope.Message
+}
+
+func selectedTargetIDs(edge workflow.Edge, envelope *MessageEnvelope, message any) []string {
 	targetIDs := edge.Connection.SinkIDs
 	if edge.Assigner != nil {
 		targetIDs = make([]string, 0, len(edge.Connection.SinkIDs))
 		// Assigner is caller-supplied (WithEdgeAssigner). Guard against both a
 		// nil sequence (ranging over a nil iter.Seq panics) and out-of-range
 		// indices, so a misbehaving assigner cannot crash the workflow runtime.
-		if seq := edge.Assigner(len(edge.Connection.SinkIDs), envelope.Message); seq != nil {
+		if seq := edge.Assigner(len(edge.Connection.SinkIDs), message); seq != nil {
 			for id := range seq {
 				if id >= 0 && id < len(edge.Connection.SinkIDs) {
 					targetIDs = append(targetIDs, edge.Connection.SinkIDs[id])

workflow/internal/execution/edgerunner_test.go

@@ -5,11 +5,129 @@ package execution_test
 import (
 	"context"
 	"iter"
+	"reflect"
 	"testing"
 
 	"github.com/microsoft/agent-framework-go/workflow"
+	"github.com/microsoft/agent-framework-go/workflow/internal/execution"
 )
 
+// edgeUnwrapOrder is a concrete message type used to verify that caller-supplied
+// edge delegates (Condition, Assigner) receive the unwrapped value rather than a
+// workflow.PortableValue wrapper.
+type edgeUnwrapOrder struct{ Total int }
+
+func edgeUnwrapExecutor(id string) *workflow.Executor {
+	orderType := reflect.TypeFor[edgeUnwrapOrder]()
+	return &workflow.Executor{
+		ID: id,
+		ConfigureProtocol: func(b *workflow.ProtocolBuilder) (*workflow.ProtocolBuilder, error) {
+			b.RouteBuilder.AddHandlerRaw(orderType, nil, func(*workflow.Context, any) (any, error) {
+				return nil, nil
+			})
+			b.SendsMessageType(orderType)
+			return b, nil
+		},
+	}
+}
+
+func edgeUnwrapBinding(id string) workflow.ExecutorBinding {
+	return workflow.ExecutorBinding{
+		ID:               id,
+		ImplementationID: "workflow_test.edgeUnwrapExecutor",
+		NewExecutorFunc:  func(string) (*workflow.Executor, error) { return edgeUnwrapExecutor(id), nil },
+	}
+}
+
+func newEdgeUnwrapRunner(t *testing.T, wf *workflow.Workflow) *execution.EdgeRunner {
+	t.Helper()
+	return execution.NewEdgeRunner(wf, nil, func(_ context.Context, id string, _ execution.StepTracer) (*workflow.Executor, error) {
+		return edgeUnwrapExecutor(id), nil
+	})
+}
+
+// A PortableValue message whose runtime type resolves must be unwrapped to its
+// concrete type before the edge Condition is invoked, so a user predicate that
+// type-asserts the concrete type routes the message instead of seeing (and
+// rejecting) the PortableValue wrapper. Mirrors messageRouter.routeMessage and
+// the .NET runtime, which unwraps PortableValue before invoking the delegate.
+func TestPrepareDeliveryForEdge_UnwrapsPortableValueForCondition(t *testing.T) {
+	source := edgeUnwrapBinding("source")
+	sink := edgeUnwrapBinding("sink")
+
+	builder := workflow.NewBuilder(source)
+	var sawConcrete bool
+	builder.AddDirectEdge(source, sink, false, func(m any) bool {
+		_, ok := m.(edgeUnwrapOrder)
+		sawConcrete = ok
+		return ok
+	})
+	wf, err := builder.Build()
+	if err != nil {
+		t.Fatalf("Build: %v", err)
+	}
+
+	runner := newEdgeUnwrapRunner(t, wf)
+	envelope := &execution.MessageEnvelope{
+		Message:  workflow.AnyPortableValue(edgeUnwrapOrder{Total: 150}),
+		SourceID: "source",
+	}
+
+	mapping, err := runner.PrepareDeliveryForEdge(context.Background(), wf.Edges()["source"][0], envelope)
+	if err != nil {
+		t.Fatalf("PrepareDeliveryForEdge: %v", err)
+	}
+	if !sawConcrete {
+		t.Fatal("Condition saw a PortableValue, want the concrete edgeUnwrapOrder")
+	}
+	if mapping == nil || len(mapping.Targets) != 1 || mapping.Targets[0].ID != "sink" {
+		t.Fatalf("mapping = %+v, want delivery to sink", mapping)
+	}
+}
+
+// The fan-out Assigner is also caller-supplied and must receive the unwrapped
+// concrete message. Before unwrapping, an Assigner that type-asserts the
+// concrete type yields no indices, dropping the delivery.
+func TestPrepareDeliveryForEdge_UnwrapsPortableValueForAssigner(t *testing.T) {
+	source := edgeUnwrapBinding("source")
+	sinkA := edgeUnwrapBinding("sinkA")
+	sinkB := edgeUnwrapBinding("sinkB")
+
+	builder := workflow.NewBuilder(source)
+	var sawConcrete bool
+	assigner := func(_ int, m any) iter.Seq[int] {
+		_, ok := m.(edgeUnwrapOrder)
+		sawConcrete = ok
+		return func(yield func(int) bool) {
+			if ok {
+				yield(0) // route to sinkA only
+			}
+		}
+	}
+	builder.AddFanOutEdge(source, []workflow.ExecutorBinding{sinkA, sinkB}, workflow.WithEdgeAssigner(assigner))
+	wf, err := builder.Build()
+	if err != nil {
+		t.Fatalf("Build: %v", err)
+	}
+
+	runner := newEdgeUnwrapRunner(t, wf)
+	envelope := &execution.MessageEnvelope{
+		Message:  workflow.AnyPortableValue(edgeUnwrapOrder{Total: 150}),
+		SourceID: "source",
+	}
+
+	mapping, err := runner.PrepareDeliveryForEdge(context.Background(), wf.Edges()["source"][0], envelope)
+	if err != nil {
+		t.Fatalf("PrepareDeliveryForEdge: %v", err)
+	}
+	if !sawConcrete {
+		t.Fatal("Assigner saw a PortableValue, want the concrete edgeUnwrapOrder")
+	}
+	if mapping == nil || len(mapping.Targets) != 1 || mapping.Targets[0].ID != "sinkA" {
+		t.Fatalf("mapping = %+v, want delivery to sinkA", mapping)
+	}
+}
+
 // A caller-supplied fan-out Assigner (WithEdgeAssigner) may yield an index that
 // is out of range for the edge's SinkIDs. Target selection must ignore such
 // indices rather than panic the workflow runtime with an out-of-range access.