diff --git a/workflow/internal/execution/edgerunner.go b/workflow/internal/execution/edgerunner.go index 1b3debaf..0cd4e93a 100644 --- a/workflow/internal/execution/edgerunner.go +++ b/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]) 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.