From 06ead347e4f1ca7a5983b0e82ea2d7fe02bd06db Mon Sep 17 00:00:00 2001 From: PratikDhanave Date: Fri, 24 Jul 2026 08:08:46 +0530 Subject: [PATCH 1/2] Unwrap PortableValue before invoking edge Condition and fan-out Assigner PrepareDeliveryForEdge passed the raw envelope.Message to the caller-supplied Condition (func(any) bool) and fan-out Assigner (func(int, any)). When the message is a workflow.PortableValue, a user delegate that type-asserts the concrete type (e.g. m.(Order).Total > 100) sees the wrapper and fails: the Condition drops or misroutes the message, and an Assigner assertion can panic the superstep since PrepareDeliveryForEdge has no recover(). Resolve the PortableValue to its declared runtime type once (via messageRuntimeType) and use the unwrapped value for both delegates, falling back to the raw message when the type cannot be resolved. This mirrors messageRouter.routeMessage and the .NET runtime, which unwrap PortableValue before invoking the delegate. --- workflow/internal/execution/edgerunner.go | 31 ++++- .../internal/execution/edgerunner_test.go | 118 ++++++++++++++++++ 2 files changed, 145 insertions(+), 4 deletions(-) diff --git a/workflow/internal/execution/edgerunner.go b/workflow/internal/execution/edgerunner.go index 1b3debaf..0f65a9c8 100644 --- a/workflow/internal/execution/edgerunner.go +++ b/workflow/internal/execution/edgerunner.go @@ -198,12 +198,16 @@ 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. + 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 +283,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]) diff --git a/workflow/internal/execution/edgerunner_test.go b/workflow/internal/execution/edgerunner_test.go index 63a516a5..127db3a4 100644 --- a/workflow/internal/execution/edgerunner_test.go +++ b/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. From a54d3c05fd36ea613a94590339eb21184eec37c2 Mon Sep 17 00:00:00 2001 From: PratikDhanave Date: Fri, 24 Jul 2026 10:36:00 +0530 Subject: [PATCH 2/2] workflow: unwrap edge message only when a delegate consumes it --- workflow/internal/execution/edgerunner.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/workflow/internal/execution/edgerunner.go b/workflow/internal/execution/edgerunner.go index 0f65a9c8..0cd4e93a 100644 --- a/workflow/internal/execution/edgerunner.go +++ b/workflow/internal/execution/edgerunner.go @@ -201,7 +201,12 @@ func (em *EdgeRunner) PrepareDeliveryForEdge(ctx context.Context, edge workflow. // 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. - message := em.unwrapMessage(ctx, envelope) + // 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)