Add typed workflow edge callback options

docs/dotnet-go-sdk-feature-comparison.md

@@ -73,7 +73,7 @@ Within overlapping features, the main misalignments are API shape and ecosystem
 | Purview | `Microsoft.Agents.AI.Purview` models and end-to-end sample. | No equivalent package. | .NET only | No Go governance/Purview integration. |
 | Cosmos DB storage | Cosmos chat history provider and workflow checkpoint store. | No built-in Cosmos package. | .NET only | Go has public in-memory and JSON/file workflow checkpoint stores plus a custom store interface, but no Cosmos DB provider. |
 | Agent workflow builders | Sequential, concurrent, handoff, group chat builders. | `agentworkflow.NewSequentialWorkflowBuilder`, `agentworkflow.NewConcurrentWorkflowBuilder`, `agentworkflow.NewGroupChatWorkflowBuilder`; manual builder plus `AddChain`, `AddSwitch`, direct/fan-out/fan-in edges; workflow-as-agent, group chat, and agents-in-workflows examples. | Partial | Go now has first-class sequential, concurrent, and group chat builders with explicit output designation support. Handoff and Magentic builders are not yet implemented. |
-| Workflow graph builder | `WorkflowBuilder`, direct edges, fan-out, fan-in barrier, labels, conditions, switch/case samples. | `workflow.Builder`, `AddEdge`, `AddDirectEdge`, `AddFanOutEdge`, `AddFanInBarrierEdge`, `WithEdgeLabel`, `WithEdgeAssigner`, `AddSwitch`. | Aligned | .NET has more overloads/extension methods; Go uses simpler methods and option functions. |
+| Workflow graph builder | `WorkflowBuilder`, direct edges, fan-out, fan-in barrier, labels, conditions, switch/case samples. | `workflow.Builder`, `AddEdge`, `AddFanOutEdge`, `AddFanInBarrierEdge`, `WithEdgeLabel`, `WithEdgeCondition`, `WithEdgeCondition0`, `WithEdgeAssigner`, `IdempotentEdge`, `AddSwitch`. | Aligned | .NET has more overloads/extension methods; Go uses typed option functions. |
 | Workflow executor model | Generic `Executor<TInput>` and `Executor<TInput,TOutput>`, function executors, aggregating executor, protocol builder. | `Executor`, `NewExecutor`, `Executor.Bind`, `Executor.Extend`, `RouteBuilder`, `StatefulExecutorCache`. | Partial | .NET has more overloads and an explicit `AggregatingExecutor`; Go mirrors .NET's executor-level cross-run declaration and binding-level concurrent-run gate, while route configuration and lifecycle hooks live on `Executor`. |
 | Workflow protocol description | Accepts/yields/sends/catch-all protocol descriptor and chat protocol helpers. | `ProtocolDescriptor` exposes accepted, yielded, and sent types plus catch-all acceptance; `messageworkflow.Configure` contributes chat-message protocol metadata. | Aligned | Go now exposes the same protocol shape while keeping chat helpers in the Go-specific message workflow adapter. |
 | Workflow execution modes | In-process OffThread, Concurrent, Lockstep; durable execution in separate package. | In-process OffThread, Concurrent, Lockstep; subworkflow execution mode used by `workflow/inproc`. | Partial | Durable execution is .NET only. |

examples/03-workflows/conditional-edges/01_edge_condition/main.go

@@ -42,9 +42,9 @@ func main() {
 	}).Bind()
 
 	wf, err := workflow.NewBuilder(detect).
-		AddDirectEdge(detect, assistant, false, func(msg any) bool { return !msg.(DetectionResult).IsSpam }).
+		AddEdge(detect, assistant, workflow.WithEdgeCondition(func(result DetectionResult) bool { return !result.IsSpam })).
 		AddEdge(assistant, send).
-		AddDirectEdge(detect, spam, false, func(msg any) bool { return msg.(DetectionResult).IsSpam }).
+		AddEdge(detect, spam, workflow.WithEdgeCondition(func(result DetectionResult) bool { return result.IsSpam })).
 		WithOutputFrom(send, spam).
 		Build()
 	if err != nil {

examples/03-workflows/conditional-edges/03_multi_selection/main.go

@@ -69,7 +69,9 @@ func main() {
 	b := workflow.NewBuilder(analyze)
 	b.AddFanOutEdge(analyze, []workflow.ExecutorBinding{spam, assistant, summary, uncertain}, workflow.WithEdgeAssigner(routeAnalysis)).
 		AddEdge(assistant, send).
-		AddDirectEdge(analyze, log, false, func(msg any) bool { return msg.(AnalysisResult).EmailLength <= longEmailThreshold }).
+		AddEdge(analyze, log, workflow.WithEdgeCondition(func(result AnalysisResult) bool {
+			return result.EmailLength <= longEmailThreshold
+		})).
 		AddEdge(summary, log).
 		WithOutputFrom(spam, send, uncertain, log)
 
@@ -104,9 +106,8 @@ func analyzeEmail(email string) AnalysisResult {
 	return AnalysisResult{Email: email, Decision: decision, Reason: reason, EmailLength: len(email)}
 }
 
-func routeAnalysis(_ int, msg any) iter.Seq[int] {
+func routeAnalysis(_ int, result AnalysisResult) iter.Seq[int] {
 	return func(yield func(int) bool) {
-		result := msg.(AnalysisResult)
 		switch result.Decision {
 		case Spam:
 			yield(0)

examples/03-workflows/subworkflows/request_interception/main.go

@@ -113,10 +113,10 @@ func buildWorkflow() *workflow.Workflow {
 	//   interceptor -> escalation  (large requests are forwarded to a top-level port)
 	//   escalation -> host         (the user's answer flows back to the child)
 	parent, err := workflow.NewBuilder(host).
-		AddDirectEdge(host, interceptor, false, externalRequestOnly).
-		AddDirectEdge(interceptor, host, false, externalResponseOnly).
-		AddDirectEdge(interceptor, escalation, false, externalRequestOnly).
-		AddDirectEdge(escalation, host, false, externalResponseOnly).
+		AddEdge(host, interceptor, workflow.WithEdgeCondition(externalRequestOnly)).
+		AddEdge(interceptor, host, workflow.WithEdgeCondition(externalResponseOnly)).
+		AddEdge(interceptor, escalation, workflow.WithEdgeCondition(externalRequestOnly)).
+		AddEdge(escalation, host, workflow.WithEdgeCondition(externalResponseOnly)).
 		WithOutputFrom(host).
 		Build()
 	if err != nil {
@@ -211,12 +211,10 @@ func approvalInterceptor(id, hostID, escalationPortID string) workflow.ExecutorB
 	})
 }
 
-func externalRequestOnly(msg any) bool {
-	_, ok := msg.(*workflow.ExternalRequest)
-	return ok
+func externalRequestOnly(request *workflow.ExternalRequest) bool {
+	return request != nil
 }
 
-func externalResponseOnly(msg any) bool {
-	_, ok := msg.(*workflow.ExternalResponse)
-	return ok
+func externalResponseOnly(response *workflow.ExternalResponse) bool {
+	return response != nil
 }

workflow/agentworkflow/workflow_test.go

@@ -2441,14 +2441,14 @@ func emitTextUpdate(ctx *workflow.Context, executorID, text string) error {
 
 func addCrossExecutorEdges(builder *workflow.Builder, startBinding, downstreamBinding workflow.ExecutorBinding) *workflow.Builder {
 	return builder.
-		AddDirectEdge(startBinding, downstreamBinding, false, func(value any) bool {
+		AddEdge(startBinding, downstreamBinding, workflow.WithEdgeCondition(func(value any) bool {
 			_, ok := value.([]*message.Message)
 			return ok
-		}).
-		AddDirectEdge(startBinding, downstreamBinding, false, func(value any) bool {
+		})).
+		AddEdge(startBinding, downstreamBinding, workflow.WithEdgeCondition(func(value any) bool {
 			_, ok := value.(workflow.TurnToken)
 			return ok
-		})
+		}))
 }
 
 func requireWorkflowFunctionCallID(t *testing.T, response *agent.Response) string {

workflow/builder.go

@@ -127,27 +127,20 @@ func (wb *Builder) BindExecutor(binding ExecutorBinding) *Builder {
 	return wb
 }
 
-// AddEdge adds a single unconditional edge from source to target. It is a
-// convenience wrapper over [Builder.AddDirectEdge] with idempotent=false and a
-// nil condition.
+// AddEdge adds a direct edge from source to target. Pass [WithEdgeCondition] to
+// make the edge conditional. Adding a duplicate conditionless edge records an
+// error unless [IdempotentEdge] is supplied. Only conditionless edges
+// participate in the duplicate-edge check.
 func (wb *Builder) AddEdge(source ExecutorBinding, target ExecutorBinding, opts ...EdgeOption) *Builder {
-	return wb.AddDirectEdge(source, target, false, nil, opts...)
-}
-
-// AddDirectEdge adds an edge from source to target. A non-nil condition makes
-// the edge fire only when condition returns true; a nil condition makes it
-// unconditional. When a matching conditionless edge already exists, idempotent=true
-// silently skips the duplicate while idempotent=false records an error. Only
-// conditionless edges participate in the duplicate-edge check.
-func (wb *Builder) AddDirectEdge(source ExecutorBinding, target ExecutorBinding, idempotent bool, condition func(any) bool, opts ...EdgeOption) *Builder {
 	if wb.err != nil {
 		return wb
 	}
 	conn := newDirectEdgeConnection(source.ID, target.ID)
-	if condition == nil && slices.ContainsFunc(wb.conditionlessConnections, func(c EdgeConnection) bool {
+	config := normalizeEdgeOptions(opts)
+	if config.condition == nil && slices.ContainsFunc(wb.conditionlessConnections, func(c EdgeConnection) bool {
 		return conn.Equal(c)
 	}) {
-		if idempotent {
+		if config.idempotent {
 			return wb
 		}
 		wb.err = fmt.Errorf("an edge from '%s' to '%s' already exists without a condition", source.ID, target.ID)
@@ -158,15 +151,14 @@ func (wb *Builder) AddDirectEdge(source ExecutorBinding, target ExecutorBinding,
 	}
 	edge := Edge{
 		Connection: conn,
-		Condition:  condition,
 		Index:      wb.edgeIdx(),
 	}
-	applyEdgeOptions(&edge, opts)
+	config.apply(&edge)
 	wb.addEdgeForSource(source.ID, edge)
 	// Only conditionless edges participate in the conditionless-edge dedup set;
 	// appending conditional edges here would wrongly make a later conditionless
 	// edge on the same source→target pair look like a duplicate.
-	if condition == nil {
+	if edge.Condition == nil {
 		wb.conditionlessConnections = append(wb.conditionlessConnections, conn)
 	}
 	return wb
@@ -199,7 +191,7 @@ func (wb *Builder) AddFanOutEdge(source ExecutorBinding, targets []ExecutorBindi
 		Connection: conn,
 		Index:      wb.edgeIdx(),
 	}
-	applyEdgeOptions(&edge, opts)
+	normalizeEdgeOptions(opts).apply(&edge)
 	wb.addEdgeForSource(source.ID, edge)
 	return wb
 }
@@ -228,7 +220,7 @@ func (wb *Builder) AddFanInBarrierEdge(sources []ExecutorBinding, target Executo
 		Connection: newEdgeConnection(sourceIDs, []string{target.ID}),
 		Index:      wb.edgeIdx(),
 	}
-	applyEdgeOptions(&edge, opts)
+	normalizeEdgeOptions(opts).apply(&edge)
 	for _, id := range sourceIDs {
 		wb.addEdgeForSource(id, edge)
 	}
@@ -579,7 +571,7 @@ func (wb *Builder) AddChain(source ExecutorBinding, executors []ExecutorBinding,
 			}
 			seen[exec.ID] = struct{}{}
 		}
-		wb.AddDirectEdge(current, exec, true /*idempotent*/, nil)
+		wb.AddEdge(current, exec, IdempotentEdge())
 		if wb.err != nil {
 			return wb
 		}

workflow/builder_test.go

@@ -734,7 +734,7 @@ func TestBuilder_Validation_SelfLoopWarning(t *testing.T) {
 	start := newNoOpExecutor("start")
 
 	wf, err := workflow.NewBuilder(start).
-		AddDirectEdge(start, start, true, func(any) bool { return false }).
+		AddEdge(start, start, workflow.WithEdgeCondition(func(any) bool { return false }), workflow.IdempotentEdge()).
 		Build()
 	if err != nil {
 		t.Fatalf("expected no error for self-loop, got %v", err)
@@ -948,6 +948,40 @@ func TestBuilder_Validation_TypeCompatibility_CatchAllSourceSkipped(t *testing.T
 	}
 }
 
+func TestBuilder_DuplicateConditionlessEdgeRejected(t *testing.T) {
+	start := newNoOpExecutor("start")
+	target := newNoOpExecutor("target")
+
+	_, err := workflow.NewBuilder(start).
+		AddEdge(start, target).
+		AddEdge(start, target).
+		Build()
+	if err == nil || !strings.Contains(err.Error(), "already exists without a condition") {
+		t.Fatalf("Build error = %v, want duplicate conditionless edge error", err)
+	}
+}
+
+func TestBuilder_EdgeIdempotencySkipsDuplicateConditionlessEdge(t *testing.T) {
+	start := newNoOpExecutor("start")
+	middle := newNoOpExecutor("middle")
+	target := newNoOpExecutor("target")
+
+	wf, err := workflow.NewBuilder(start).
+		AddEdge(start, middle).
+		AddEdge(start, middle, workflow.IdempotentEdge()).
+		AddEdge(middle, target).
+		Build()
+	if err != nil {
+		t.Fatalf("Build: %v", err)
+	}
+	if got := len(wf.Edges()[start.ID]); got != 1 {
+		t.Fatalf("start edge count = %d, want 1", got)
+	}
+	if got := wf.Edges()[middle.ID][0].Index; got != 2 {
+		t.Fatalf("edge index after skipped duplicate = %d, want 2", got)
+	}
+}
+
 // A conditional edge on a source→target pair must not populate the
 // conditionless-edge dedup set: adding a legitimate conditionless edge on the
 // same pair afterwards should succeed, not be rejected as a duplicate.
@@ -956,23 +990,23 @@ func TestBuilder_ConditionalEdgeDoesNotBlockConditionlessEdge(t *testing.T) {
 	target := newNoOpExecutor("target")
 
 	_, err := workflow.NewBuilder(start).
-		AddDirectEdge(start, target, false, func(any) bool { return true }).
+		AddEdge(start, target, workflow.WithEdgeCondition(func(any) bool { return true })).
 		AddEdge(start, target).
 		Build()
 	if err != nil {
 		t.Fatalf("conditionless edge after a conditional edge on the same pair should be allowed, got error: %v", err)
 	}
 }
 
-// The idempotent path (AddChain / idempotent=true) must likewise not silently
+// The idempotent path (AddChain / IdempotentEdge) must likewise not silently
 // drop a conditionless edge just because a conditional edge preceded it.
 func TestBuilder_ConditionalEdgeDoesNotDropIdempotentConditionlessEdge(t *testing.T) {
 	start := newNoOpExecutor("start")
 	target := newNoOpExecutor("target")
 
 	wf, err := workflow.NewBuilder(start).
-		AddDirectEdge(start, target, false, func(any) bool { return true }).
-		AddDirectEdge(start, target, true, nil). // idempotent conditionless edge
+		AddEdge(start, target, workflow.WithEdgeCondition(func(any) bool { return true })).
+		AddEdge(start, target, workflow.IdempotentEdge()).
 		Build()
 	if err != nil {
 		t.Fatalf("unexpected build error: %v", err)

workflow/edge.go

@@ -4,32 +4,109 @@ package workflow
 
 import (
 	"iter"
+	"reflect"
 	"slices"
 )
 
+type edgeOptions struct {
+	label      string
+	condition  func(any) bool
+	assigner   func(int, any) iter.Seq[int]
+	idempotent bool
+}
+
 // EdgeOption configures an [Edge] when it is added to a workflow via [Builder].
-type EdgeOption func(*Edge)
+type EdgeOption func(*edgeOptions)
 
 // WithEdgeLabel sets an optional label on the edge. Labels can be used by
 // visualizers to annotate edges.
 func WithEdgeLabel(label string) EdgeOption {
-	return func(e *Edge) { e.Label = label }
+	return func(options *edgeOptions) { options.label = label }
+}
+
+// IdempotentEdge makes adding a duplicate conditionless direct edge a
+// no-op instead of a build error. It has no effect on conditional, fan-out, or
+// fan-in edges.
+func IdempotentEdge() EdgeOption {
+	return func(options *edgeOptions) { options.idempotent = true }
+}
+
+// WithEdgeCondition attaches a condition that receives messages as T to a
+// direct edge. PortableValue messages are decoded as T before the condition is
+// invoked. Messages that cannot be assigned to T are passed as the zero value
+// of T. When T is any, the original message is passed unchanged.
+func WithEdgeCondition[T any](condition func(T) bool) EdgeOption {
+	return func(options *edgeOptions) {
+		if condition == nil {
+			options.condition = nil
+			return
+		}
+		options.condition = func(message any) bool {
+			return condition(edgeMessageFor[T](message))
+		}
+	}
+}
+
+// WithEdgeCondition0 attaches a message-independent condition to a direct
+// edge. The condition is invoked once for each incoming message.
+func WithEdgeCondition0(condition func() bool) EdgeOption {
+	return func(options *edgeOptions) {
+		if condition == nil {
+			options.condition = nil
+			return
+		}
+		options.condition = func(any) bool { return condition() }
+	}
 }
 
 // WithEdgeAssigner attaches an [Edge.Assigner] callback to a fan-out edge.
 // The assigner is invoked for each message and must return the indexes of
-// the targets that should receive it. Has no effect on direct or fan-in
-// edges.
-func WithEdgeAssigner(assigner func(targetCount int, message any) iter.Seq[int]) EdgeOption {
-	return func(e *Edge) { e.Assigner = assigner }
+// the targets that should receive it. PortableValue messages are decoded as T
+// before the assigner is invoked. Messages that cannot be assigned to T are
+// passed as the zero value of T. When T is any, the original message is passed
+// unchanged. Has no effect on direct or fan-in edges.
+func WithEdgeAssigner[T any](assigner func(targetCount int, message T) iter.Seq[int]) EdgeOption {
+	return func(options *edgeOptions) {
+		if assigner == nil {
+			options.assigner = nil
+			return
+		}
+		options.assigner = func(targetCount int, message any) iter.Seq[int] {
+			return assigner(targetCount, edgeMessageFor[T](message))
+		}
+	}
 }
 
-func applyEdgeOptions(e *Edge, opts []EdgeOption) {
+func edgeMessageFor[T any](message any) T {
+	var zero T
+	if message == nil {
+		return zero
+	}
+	if reflect.TypeFor[T]() == reflect.TypeFor[any]() {
+		return message.(T)
+	}
+	if portable, ok := message.(PortableValue); ok {
+		value, _ := PortableValueAs[T](portable)
+		return value
+	}
+	value, _ := message.(T)
+	return value
+}
+
+func normalizeEdgeOptions(opts []EdgeOption) edgeOptions {
+	var config edgeOptions
 	for _, opt := range opts {
 		if opt != nil {
-			opt(e)
+			opt(&config)
 		}
 	}
+	return config
+}
+
+func (config edgeOptions) apply(edge *Edge) {
+	edge.Label = config.label
+	edge.Condition = config.condition
+	edge.Assigner = config.assigner
 }
 
 // Edge represents a connection or relationship between nodes.