Agentic Workflow Deep Dive — From Concepts to C# Implementation with Microsoft Agent Framework

// Sequential: Building a translation chain
// AgentWorkflowBuilder.BuildSequential() internally generates
// AddEdge(agent[0], agent[1]), AddEdge(agent[1], agent[2]), ...
Workflow workflow = AgentWorkflowBuilder.BuildSequential(
    from lang in new[] { "French", "Spanish", "English" }
    select new ChatClientAgent(
        chatClient,
        instructions: $"You are a translation assistant. Translate the input to {lang}.",
        name: $"{lang}Translator"
    )
);
 
// Execute the workflow
await using StreamingRun run =
    await InProcessExecution.RunStreamingAsync(workflow, "Hello, world!");
 
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is AgentResponseUpdateEvent e)
        Console.Write(e.Update.Text);
    else if (evt is WorkflowOutputEvent output)
        Console.WriteLine($"\nFinal: {output.Data}");
}

// Concurrent: Parallel translation
// BuildConcurrent() internally generates Fan-out and Fan-in Barrier Edges
// Fan-out: Delivers messages from input node to all agents simultaneously
// Fan-in Barrier: Waits for all agents to complete before aggregating results
Workflow workflow = AgentWorkflowBuilder.BuildConcurrent(
    from lang in new[] { "French", "Spanish", "English" }
    select new ChatClientAgent(
        chatClient,
        instructions: $"Translate the input to {lang}. Output only the translation.",
        name: $"{lang}Translator"
    )
);

// Conditional Routing: Spam classification example
// AddSwitch() branches based on the ClassificationResult's Category property
// Type-safe condition evaluation via lambda expressions
WorkflowBuilder builder = new WorkflowBuilder(classifier)
    .AddSwitch(classifier, sw => sw
        .AddCase<ClassificationResult>(
            result => result?.Category == "spam",
            spamHandler)
        .AddCase<ClassificationResult>(
            result => result?.Category == "normal",
            replyGenerator)
        .WithDefault(manualReviewHandler)
    )
    .WithOutputFrom(spamHandler, replyGenerator, manualReviewHandler);

// Handoff: Triage → Specialist agent delegation
// CreateHandoffBuilderWith() builds starting from the triage agent
// WithHandoffs() defines bidirectional handoff routes
ChatClientAgent triageAgent = new(chatClient,
    instructions: "Determine which specialist to use. ALWAYS hand off.",
    name: "triage_agent",
    description: "Routes to the right specialist.");
 
ChatClientAgent mathTutor = new(chatClient,
    instructions: "You help with math. Explain step by step.",
    name: "math_tutor",
    description: "Math specialist.");
 
ChatClientAgent historyTutor = new(chatClient,
    instructions: "You help with history. Explain events and context.",
    name: "history_tutor",
    description: "History specialist.");
 
// Build the Handoff workflow
// WithHandoffs(source, targets) injects targets' descriptions
// as tool definitions into source, so the LLM calls a handoff tool
// and the workflow engine automatically routes to the correct Executor
Workflow workflow = AgentWorkflowBuilder
    .CreateHandoffBuilderWith(triageAgent)
    .WithHandoffs(triageAgent, [mathTutor, historyTutor])
    .WithHandoffs([mathTutor, historyTutor], triageAgent)
    .Build();

// Writer-Critic Loop: Iterative refinement workflow
// WorkflowBuilder declaratively defines the graph structure
WorkflowBuilder builder = new WorkflowBuilder(writer)
    .AddEdge(writer, critic)                    // Writer → Critic
    .AddSwitch(critic, sw => sw
        .AddCase<CriticDecision>(                // Approved → Summary
            cd => cd?.Approved == true, summary)
        .AddCase<CriticDecision>(                // Rejected → back to Writer
            cd => cd?.Approved == false, writer))
    .WithOutputFrom(summary);                    // Summary output = workflow output

// Group Chat: Manager controls turn-taking among multiple agents
// CreateGroupChatBuilderWith() takes a GroupChatManager factory
// RoundRobinGroupChatManager: simplest manager — agents speak in order
Workflow workflow = AgentWorkflowBuilder
    .CreateGroupChatBuilderWith(
        agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = 3 })
    .AddParticipants(researchAgent, criticAgent, synthesizerAgent)
    .Build();

// Sub-Workflow: Embedding a workflow as an Executor
// 1. Build the child workflow normally
Workflow textProcessingWorkflow = new WorkflowBuilder(uppercase)
    .AddEdge(uppercase, reverse)
    .AddEdge(reverse, append)
    .WithOutputFrom(append)
    .Build();
 
// 2. Convert workflow to Executor via BindAsExecutor()
ExecutorBinding subWorkflowExecutor =
    textProcessingWorkflow.BindAsExecutor("TextProcessingSubWorkflow");
 
// 3. Connect it in the parent workflow like any other Executor
Workflow mainWorkflow = new WorkflowBuilder(prefix)
    .AddEdge(prefix, subWorkflowExecutor)
    .AddEdge(subWorkflowExecutor, postProcess)
    .WithOutputFrom(postProcess)
    .Build();

// Workflow-as-Agent: Expose a workflow as an AIAgent
// AsAIAgent() wraps the workflow in the AIAgent interface
Workflow workflow = WorkflowFactory.BuildWorkflow(chatClient);
AIAgent agent = workflow.AsAIAgent("workflow-agent", "Workflow Agent");
 
// Interact just like with any other agent
AgentSession session = await agent.CreateSessionAsync();
await foreach (AgentResponseUpdate update in
    agent.RunStreamingAsync("Hello!", session))
{
    Console.Write(update.Text);
}

<!-- Add to your .csproj -->
<PackageReference Include="Microsoft.Agents.AI" Version="1.1.0" />
<PackageReference Include="Microsoft.Agents.AI.Workflows" Version="1.1.0" />
<PackageReference Include="Microsoft.Agents.AI.Workflows.Generators" Version="1.1.0"
                OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
<PackageReference Include="Microsoft.Agents.AI.OpenAI" Version="1.1.0" />
<!-- or Microsoft.Agents.AI.Foundry for Azure AI Foundry -->

// Basic Executor form
// The partial keyword and source generators auto-generate
// message dispatch code
internal sealed partial class UpperCaseExecutor() : Executor("UpperCase")
{
    // [MessageHandler] attribute defines a message handler
    // The argument type (string) determines message routing
    [MessageHandler]
    public ValueTask<string> HandleAsync(
        string message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        return ValueTask.FromResult(message.ToUpperInvariant());
    }
}

// Executor that calls an LLM
// ChatClientAgent wraps IChatClient and provides the AIAgent interface
internal sealed partial class WriterExecutor : Executor
{
    private readonly AIAgent _agent;
 
    public WriterExecutor(IChatClient chatClient) : base("Writer")
    {
        // ChatClientAgent: IChatClient → AIAgent adapter
        // instructions are sent to the LLM as the system prompt
        this._agent = new ChatClientAgent(
            chatClient,
            name: "Writer",
            instructions: """
                You are a skilled writer. Create clear, engaging content.
                If you receive feedback, revise to address all concerns.
                """
        );
    }
 
    // Handler for initial requests (string type)
    [MessageHandler]
    public async ValueTask<ChatMessage> HandleInitialRequestAsync(
        string message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        // RunStreamingAsync sends a streaming request to the LLM
        StringBuilder sb = new();
        await foreach (AgentResponseUpdate update in
            this._agent.RunStreamingAsync(
                new ChatMessage(ChatRole.User, message),
                cancellationToken: cancellationToken))
        {
            if (!string.IsNullOrEmpty(update.Text))
            {
                sb.Append(update.Text);
                Console.Write(update.Text); // Real-time output
            }
        }
 
        return new ChatMessage(ChatRole.User, sb.ToString());
    }
 
    // Handler for revision requests (CriticDecision type)
    // The same Executor handles different message types
    [MessageHandler]
    public async ValueTask<ChatMessage> HandleRevisionAsync(
        CriticDecision decision,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        string prompt =
            $"Revise based on this feedback:\n{decision.Feedback}\n\n" +
            $"Original:\n{decision.Content}";
 
        StringBuilder sb = new();
        await foreach (AgentResponseUpdate update in
            this._agent.RunStreamingAsync(
                new ChatMessage(ChatRole.User, prompt),
                cancellationToken: cancellationToken))
        {
            if (!string.IsNullOrEmpty(update.Text))
                sb.Append(update.Text);
        }
 
        return new ChatMessage(ChatRole.User, sb.ToString());
    }
}

// Direct Edge: Unconditionally routes messages from A → B
builder.AddEdge(executorA, executorB);
 
// Conditional Edge: Routes only when the condition is met
builder.AddEdge<ClassificationResult>(executorA, executorB,
    condition: r => r?.Category == "normal");
 
// Switch-Case Edge: Multiple conditional branches
builder.AddSwitch(executorA, sw => sw
    .AddCase<ResultType>(r => r?.IsSuccess == true, successHandler)
    .AddCase<ResultType>(r => r?.IsSuccess == false, failureHandler)
    .WithDefault(fallbackHandler));
 
// Fan-out Edge: 1 → N parallel distribution
builder.AddFanOutEdge<RoutingResult>(
    sourceExecutor,
    targets: [handlerA, handlerB, handlerC],
    targetSelector: (result, count) =>
    {
        // Select targets based on result content
        // Return value is a list of target indices
        if (result?.NeedsAll == true) return Enumerable.Range(0, count);
        return [0]; // handlerA only
    });
 
// Fan-in Barrier Edge: N → 1 aggregation (waits for all sources)
builder.AddFanInBarrierEdge(
    sources: [workerA, workerB, workerC],
    target: aggregator);

// Full WorkflowBuilder usage example
Workflow workflow = new WorkflowBuilder(entryExecutor)
    // Direct Edge: entry → processor
    .AddEdge(entryExecutor, processorExecutor)
    // Switch-Case: Branch based on processor output
    .AddSwitch(processorExecutor, sw => sw
        .AddCase<ProcessResult>(r => r?.Status == "approved", approvedHandler)
        .AddCase<ProcessResult>(r => r?.Status == "rejected", rejectedHandler)
        .WithDefault(manualReviewHandler))
    // Output sources: These executors' outputs become the workflow's output
    .WithOutputFrom(approvedHandler, rejectedHandler, manualReviewHandler)
    // Build: Generate the Workflow instance
    .Build();

// Streaming execution
// InProcessExecution.RunStreamingAsync() returns a StreamingRun
// StreamingRun is IAsyncDisposable, so using ensures resource cleanup
await using StreamingRun run =
    await InProcessExecution.RunStreamingAsync(workflow, "Input message");
 
// WatchStreamAsync() subscribes to events in real-time
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case AgentResponseUpdateEvent agentUpdate:
            // Agent streaming output (per-token)
            Console.Write(agentUpdate.Update.Text);
            break;
 
        case WorkflowOutputEvent output:
            // Workflow's final output
            Console.WriteLine($"\n=== Final Output ===");
            Console.WriteLine(output.Data);
            break;
    }
}

// InProcessRunner.RunSuperStepAsync() internals (simplified)
// https://github.com/microsoft/agent-framework/blob/main/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs
async ValueTask<bool> RunSuperStepAsync(CancellationToken cancellationToken)
{
    // 1. AdvanceAsync(): Move messages queued in the previous superstep
    //    to the current superstep's StepContext
    StepContext currentStep = await RunContext.AdvanceAsync(cancellationToken);
 
    // No messages means no superstep needed
    if (!currentStep.HasMessages && !RunContext.HasQueuedExternalDeliveries)
        return false;
 
    // 2. Raise SuperStepStartedEvent
    await RaiseWorkflowEventAsync(StepTracer.Advance(currentStep));
 
    // 3. Deliver messages to all target Executors in parallel
    //    Task.WhenAll runs all Executors within the same superstep concurrently
    List<Task> receiverTasks = currentStep.QueuedMessages.Keys
        .Select(receiverId =>
            DeliverMessagesAsync(receiverId,
                currentStep.MessagesFor(receiverId),
                cancellationToken).AsTask())
        .ToList();
    await Task.WhenAll(receiverTasks);
 
    // 4. Drive sub-workflows in parallel if any
    List<Task> subworkflowTasks = [];
    foreach (var subRunner in RunContext.JoinedSubworkflowRunners)
        subworkflowTasks.Add(subRunner.RunSuperStepAsync(cancellationToken).AsTask());
    await Task.WhenAll(subworkflowTasks);
 
    // 5. Barrier synchronization: Batch-apply state updates
    //    StateManager.PublishUpdatesAsync() applies all updates
    //    queued via QueueStateUpdateAsync() in one batch
    await StateManager.PublishUpdatesAsync(StepTracer);
 
    // 6. Create checkpoint (if enabled)
    await CheckpointAsync(cancellationToken);
 
    // 7. Raise SuperStepCompletedEvent
    await RaiseWorkflowEventAsync(
        StepTracer.Complete(
            RunContext.NextStepHasActions,
            RunContext.HasUnservicedRequests));
}

// StepContext: Superstep message queue management
// https://github.com/microsoft/agent-framework/blob/main/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StepContext.cs
internal sealed class StepContext
{
    // Target Executor ID → Message queue
    // ConcurrentDictionary + ConcurrentQueue for thread safety
    public ConcurrentDictionary<string, ConcurrentQueue<MessageEnvelope>>
        QueuedMessages { get; } = [];
 
    public bool HasMessages =>
        !QueuedMessages.IsEmpty &&
        QueuedMessages.Values.Any(q => !q.IsEmpty);
}

// Execution mode selection
// Default (OffThread): Real-time streaming
await using var run = await InProcessExecution.RunStreamingAsync(workflow, input);
 
// Lockstep: For testing/debugging (events delivered in batches)
await using var run = await InProcessExecution.Lockstep
    .RunStreamingAsync(workflow, input);
 
// Concurrent: Parallel runs of the same workflow
// All Executors in the workflow must implement IResettableExecutor
await using var run = await InProcessExecution.Concurrent
    .RunStreamingAsync(workflow, input);

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case SuperStepStartedEvent stepStarted:
            // SendingExecutors: IDs of Executors that sent messages in the previous step
            Console.WriteLine(
                $"[Step Start] Senders: {string.Join(", ", stepStarted.StartInfo.SendingExecutors)}");
            break;
 
        case SuperStepCompletedEvent stepCompleted:
            // ActivatedExecutors: IDs of Executors that ran in this step
            // HasPendingMessages: Whether messages are queued for the next step
            // StateUpdated: Whether state was updated
            Console.WriteLine(
                $"[Step Done] Activated: {string.Join(", ", stepCompleted.CompletionInfo.ActivatedExecutors)}, " +
                $"Pending: {stepCompleted.CompletionInfo.HasPendingMessages}, " +
                $"StateUpdated: {stepCompleted.CompletionInfo.StateUpdated}");
            break;
 
        case AgentResponseUpdateEvent agentUpdate:
            Console.Write(agentUpdate.Update.Text);
            break;
 
        case WorkflowOutputEvent output:
            Console.WriteLine($"\nFinal: {output.Data}");
            break;
    }
}

Executor A calls context.SendMessageAsync(output)
  → InProcessRunnerContext.SendMessageAsync(sourceId, output) is called
    → All Edges attached to sourceId are traversed
      → EdgeMap.PrepareDeliveryForEdgeAsync() evaluates conditions
        → DeliveryMapping.MapInto(_nextStep) enqueues into
          the StepContext for the next superstep
            → _nextStep.MessagesFor(targetId).Enqueue(envelope)

// InProcessRunnerContext.AdvanceAsync() — Superstep transition
// https://github.com/microsoft/agent-framework/blob/main/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs
public async ValueTask<StepContext> AdvanceAsync(CancellationToken cancellationToken = default)
{
    // First, route external messages into _nextStep via Edges
    while (_queuedExternalDeliveries.TryDequeue(out var deliveryPrep))
        await deliveryPrep();
 
    // Atomically swap _nextStep with a fresh empty StepContext
    return Interlocked.Exchange(ref _nextStep, new StepContext());
}

// InProcessRunner.DeliverMessagesAsync() — Process one Executor's queue in order (excerpt)
// https://github.com/microsoft/agent-framework/blob/main/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs
private async ValueTask DeliverMessagesAsync(string receiverId, ConcurrentQueue<MessageEnvelope> envelopes, CancellationToken cancellationToken)
{
    Executor executor = await this.RunContext.EnsureExecutorAsync(receiverId, this.StepTracer, cancellationToken).ConfigureAwait(false);
 
    this.StepTracer.TraceActivated(receiverId);
 
    // ...
    IWorkflowContext tracelessContext = this.RunContext.BindWorkflowContext(receiverId);
 
    try
    {
        await executor.OnMessageDeliveryStartingAsync(tracelessContext, cancellationToken)
                      .ConfigureAwait(false);
 
        while (envelopes.TryDequeue(out var envelope))
        {
            (object message, TypeId messageType) = await TranslateMessageAsync(envelope).ConfigureAwait(false);
 
            await executor.ExecuteCoreAsync(
                message,
                messageType,
                this.RunContext.BindWorkflowContext(receiverId, envelope.TraceContext),
                this.TelemetryContext,
                cancellationToken
            ).ConfigureAwait(false);
        }
    }
    finally
    {
        await executor.OnMessageDeliveryFinishedAsync(tracelessContext, cancellationToken)
                      .ConfigureAwait(false);
    }
 
    // (The local function TranslateMessageAsync, which restores each message's type, is omitted)
}

// ConcurrentEventSink — Event notification mechanism
// https://github.com/microsoft/agent-framework/blob/main/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/ConcurrentEventSink.cs
internal class ConcurrentEventSink : IEventSink
{
    public ValueTask EnqueueAsync(WorkflowEvent workflowEvent)
        => this.EventRaised?.Invoke(this, workflowEvent) ?? default;
 
    public event Func<object?, WorkflowEvent, ValueTask>? EventRaised;
}

// CriticDecision: DTO for Structured Output
// Description attributes map to JSON Schema description fields,
// giving the LLM hints about each field's meaning
[Description("Critic's review decision")]
internal sealed class CriticDecision
{
    [JsonPropertyName("approved")]
    [Description("Whether the content is approved (true) or needs revision (false)")]
    public bool Approved { get; set; }
 
    [JsonPropertyName("feedback")]
    [Description("Specific feedback for improvements if not approved")]
    public string Feedback { get; set; } = "";
 
    // Fields for workflow-internal use (not included in JSON)
    [JsonIgnore]
    public string Content { get; set; } = "";
 
    [JsonIgnore]
    public int Iteration { get; set; }
}
 
// Critic Executor: Review using Structured Output
internal sealed class CriticExecutor : Executor<ChatMessage, CriticDecision>
{
    private readonly AIAgent _agent;
 
    public CriticExecutor(IChatClient chatClient) : base("Critic")
    {
        this._agent = new ChatClientAgent(chatClient, new ChatClientAgentOptions
        {
            Name = "Critic",
            ChatOptions = new()
            {
                Instructions = """
                    You are a constructive critic. Review content and provide feedback.
                    Provide your decision as structured output.
                    """,
                // ChatResponseFormat.ForJsonSchema<T>() sends
                // CriticDecision's JSON Schema to the LLM
                // The LLM is guaranteed to return JSON conforming to this schema
                ResponseFormat = ChatResponseFormat.ForJsonSchema<CriticDecision>()
            }
        });
    }
 
    public override async ValueTask<CriticDecision> HandleAsync(
        ChatMessage message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        // Get streaming response
        IAsyncEnumerable<AgentResponseUpdate> updates =
            this._agent.RunStreamingAsync(message, cancellationToken: cancellationToken);
 
        // Display streaming output in real-time
        await foreach (AgentResponseUpdate update in updates)
        {
            if (!string.IsNullOrEmpty(update.Text))
                Console.Write(update.Text);
        }
 
        // Convert stream to AgentResponse and deserialize JSON
        AgentResponse response = await updates.ToAgentResponseAsync(cancellationToken);
        CriticDecision decision = JsonSerializer.Deserialize<CriticDecision>(
            response.Text, JsonSerializerOptions.Web)
            ?? throw new JsonException("Failed to deserialize CriticDecision");
 
        return decision;
    }
}

// Reading and writing shared state
// IWorkflowContext is injected into each Executor
// scopeName isolates scopes, key identifies individual state entries
 
// Read state
FlowState? state = await context.ReadStateAsync<FlowState>(
    key: "singleton",
    scopeName: "FlowStateScope");
 
// Write state (queued, applied at superstep end)
await context.QueueStateUpdateAsync(
    key: "singleton",
    value: updatedState,
    scopeName: "FlowStateScope");

// IResettableExecutor: Reusing stateful shared Executors
// Framework-provided Executors (ChatForwardingExecutor, etc.) already implement this
// Custom stateful Executors must implement it themselves
internal sealed class AggregationExecutor()
    : Executor<List<ChatMessage>>("Aggregator"), IResettableExecutor
{
    private readonly List<ChatMessage> _messages = [];
 
    public override async ValueTask HandleAsync(
        List<ChatMessage> message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        this._messages.AddRange(message);
        if (this._messages.Count >= 2)
            await context.YieldOutputAsync(
                string.Join("\n", this._messages.Select(m => m.Text)),
                cancellationToken);
    }
 
    // Framework calls this automatically between workflow runs
    public ValueTask ResetAsync()
    {
        this._messages.Clear();
        return default;
    }
}

// ⚠️ If both conditions are true, BOTH processorA and processorB execute!
builder
    .AddEdge<MyResult>(source, processorA, r => r?.Score > 50)
    .AddEdge<MyResult>(source, processorB, r => r?.Score > 30);
// Score = 70 → Messages delivered to BOTH processorA and processorB

// InProcessRunnerContext.SendMessageAsync() — All Edges evaluated independently
foreach (Edge edge in edges)  // No break, no return
{
    DeliveryMapping? maybeMapping =
        await _edgeMap.PrepareDeliveryForEdgeAsync(edge, envelope, cancellationToken);
    maybeMapping?.MapInto(this._nextStep);  // Delivers to ALL matches
}

// SwitchBuilder's internal EdgeSelector (simplified)
IEnumerable<int> EdgeSelector(object? input, int targetCount)
{
    for (int i = 0; i < caseMap.Count; i++)
    {
        if (caseMap[i].Predicate(input))
            return caseMap[i].OutgoingIndicies;  // Returns on first match
    }
    return defaultIndicies;  // Falls through to default
}

// DirectEdgeRunner.ChaseEdgeAsync() — Type mismatch = silent drop
if (CanHandle(target, messageType))
{
    return new DeliveryMapping(envelope, target);  // Delivered
}
// ↓ Falls through here: DroppedTypeMismatch — no exception
return null;

// ⚠️ No targetSelector → delivers to ALL targets
builder.AddFanOutEdge(source, [handlerA, handlerB, handlerC]);
 
// ✅ targetSelector controls delivery
builder.AddFanOutEdge<MyResult>(source, [handlerA, handlerB, handlerC],
    targetSelector: (result, count) =>
    {
        if (result?.Category == "urgent") return [0];  // handlerA only
        return Enumerable.Range(0, count);              // All targets
    });

// ⚠️ If workerC never sends a message, aggregator never executes
builder.AddFanInBarrierEdge(
    sources: [workerA, workerB, workerC],
    target: aggregator);

// ✅ Within the same Executor, your own writes are immediately readable
await context.QueueStateUpdateAsync("key", newValue, scopeName: "MyScope");
var value = await context.ReadStateAsync<MyState>("key", scopeName: "MyScope");
// value IS newValue (StateManager checks _queuedUpdates first)
 
// ⚠️ But writes from OTHER Executors in the same superstep are NOT visible
// Executor A's queued update cannot be read by Executor B in the same superstep
// It only becomes visible after PublishUpdatesAsync() at the superstep boundary

// ⚠️ Throws exception
builder.AddEdge(source, target);
builder.AddEdge(source, target);  // InvalidOperationException!
 
// ✅ Safe with idempotent
builder.AddEdge(source, target, idempotent: true);
builder.AddEdge(source, target, idempotent: true);  // OK (2nd is NoOp)

// Program.cs — Writer-Critic workflow entry point
// https://github.com/microsoft/agent-framework/blob/main/dotnet/samples/03-workflows/_StartHere/07_WriterCriticWorkflow/Program.cs
using System.ComponentModel;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
 
public static class Program
{
    public const int MaxIterations = 3; // Safety valve: max 3 iterations
 
    private static async Task Main()
    {
        // ── 1. Initialize an IChatClient ──
        string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT")
            ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
        string deploymentName =
            Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME")
            ?? "gpt-4o-mini";
 
        // IChatClient: unified interface from Microsoft.Extensions.AI.
        // The current official sample initializes via AIProjectClient.AsAIAgent();
        // this variant uses ChatClientAgent over IChatClient to keep providers swappable.
        IChatClient chatClient = new AzureOpenAIClient(
            new Uri(endpoint),
            new AzureCliCredential()
        ).GetChatClient(deploymentName).AsIChatClient();
 
        // ── 2. Create Executors ──
        WriterExecutor writer = new(chatClient);
        CriticExecutor critic = new(chatClient);
        SummaryExecutor summary = new(chatClient);
 
        // ── 3. Build workflow graph ──
        // WorkflowBuilder constructor arg = entry point
        WorkflowBuilder workflowBuilder = new WorkflowBuilder(writer)
            .AddEdge(writer, critic)           // Writer → Critic
            .AddSwitch(critic, sw => sw        // Conditional branch at Critic
                .AddCase<CriticDecision>(
                    cd => cd?.Approved == true,
                    summary)                   //   Approved → Summary
                .AddCase<CriticDecision>(
                    cd => cd?.Approved == false,
                    writer))                   //   Rejected → Writer (loop)
            .WithOutputFrom(summary);          // Summary output = final output
 
        // ── 4. Execute workflow ──
        Workflow workflow = workflowBuilder.Build();
        await using StreamingRun run =
            await InProcessExecution.RunStreamingAsync(
                workflow,
                "Write a 200-word blog post about AI ethics.");
 
        // ── 5. Subscribe to events ──
        await foreach (WorkflowEvent evt in run.WatchStreamAsync())
        {
            switch (evt)
            {
                case AgentResponseUpdateEvent agentUpdate:
                    if (!string.IsNullOrEmpty(agentUpdate.Update.Text))
                        Console.Write(agentUpdate.Update.Text);
                    break;
 
                case WorkflowOutputEvent output:
                    Console.WriteLine($"\n\n=== FINAL APPROVED CONTENT ===");
                    Console.WriteLine(output.Data);
                    break;
            }
        }
    }
}

// FlowState: Shared state for iteration tracking across Executors
// Read/written via IWorkflowContext's ReadStateAsync / QueueStateUpdateAsync
internal sealed class FlowState
{
    public int Iteration { get; set; } = 1;
    public List<ChatMessage> History { get; } = [];
}
 
// Constants for state scope and key
internal static class FlowStateShared
{
    public const string Scope = "FlowStateScope";
    public const string Key = "singleton";
}
 
// Helper methods
internal static class FlowStateHelpers
{
    public static async Task<FlowState> ReadFlowStateAsync(IWorkflowContext context)
    {
        FlowState? state = await context.ReadStateAsync<FlowState>(
            FlowStateShared.Key,
            scopeName: FlowStateShared.Scope);
        return state ?? new FlowState();
    }
 
    public static ValueTask SaveFlowStateAsync(
        IWorkflowContext context, FlowState state)
        => context.QueueStateUpdateAsync(
            FlowStateShared.Key,
            state,
            scopeName: FlowStateShared.Scope);
}

// Safety valve logic inside CriticExecutor
if (!decision.Approved && state.Iteration >= Program.MaxIterations)
{
    Console.WriteLine(
        $"⚠️ Max iterations ({Program.MaxIterations}) reached - auto-approving");
    decision.Approved = true;
    decision.Feedback = "";
}

// Retry and error handling inside an Executor
[MessageHandler]
public async ValueTask<ChatMessage> HandleAsync(
    string message,
    IWorkflowContext context,
    CancellationToken cancellationToken = default)
{
    const int maxRetries = 3;
    for (int attempt = 1; attempt <= maxRetries; attempt++)
    {
        try
        {
            StringBuilder sb = new();
            await foreach (AgentResponseUpdate update in
                this._agent.RunStreamingAsync(
                    new ChatMessage(ChatRole.User, message),
                    cancellationToken: cancellationToken))
            {
                if (!string.IsNullOrEmpty(update.Text))
                    sb.Append(update.Text);
            }
            return new ChatMessage(ChatRole.User, sb.ToString());
        }
        catch (HttpRequestException ex) when (attempt < maxRetries)
        {
            // Exponential backoff for rate limits and timeouts
            int delay = (int)Math.Pow(2, attempt) * 1000;
            Console.WriteLine($"⚠️ Attempt {attempt} failed: {ex.Message}. Retrying in {delay}ms...");
            await Task.Delay(delay, cancellationToken);
        }
    }
    throw new InvalidOperationException("All retry attempts exhausted");
}

// Checkpoint storage configuration
// FileSystemJsonCheckpointStore persists state to the file system
// Production typically uses Cosmos DB or Azure Table Storage-based ICheckpointStore
var store = new FileSystemJsonCheckpointStore("./checkpoints");
var checkpointManager = new CheckpointManager(store);
 
// Workflow execution with checkpointing
await using StreamingRun run =
    await InProcessExecution.RunStreamingAsync(
        workflow,
        input,
        checkpointManager);

// OpenTelemetry setup example
using var tracerProvider = Sdk.CreateTracerProviderBuilder()
    .SetResourceBuilder(ResourceBuilder.CreateDefault()
        .AddService("WriterCriticWorkflow"))
    .AddSource("Microsoft.Agents.AI")     // Agent Framework traces
    .AddOtlpExporter()                     // Send to Jaeger / Azure Monitor etc.
    .Build();