agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix CheckpointInfo.Parent always null in InProcessRunner (#3796) (#3812)

Browse Source 
Track the last CheckpointInfo in InProcessRunner so that newly created
checkpoints reference their parent. When resuming from a checkpoint,
the resumed-from checkpoint becomes the parent of the next checkpoint.

Adds tests verifying:
- First checkpoint has null parent
- Subsequent checkpoints chain parents correctly
- Checkpoint after resume references the resumed-from checkpoint
This commit is contained in:
Jacob Alber
2026-02-19 09:34:58 -05:00
committed by GitHub
Unverified
parent 6a3d22598f
commit 3507b2c532
2 changed files with 183 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs
+6 -4
View File
@@ -274,6 +274,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
}
private WorkflowInfo? _workflowInfoCache;
private CheckpointInfo? _lastCheckpointInfo;
private readonly List<CheckpointInfo> _checkpoints = [];
internal async ValueTask CheckpointAsync(CancellationToken cancellationToken = default)
{
@@ -299,10 +300,10 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
RunnerStateData runnerData = await this.RunContext.ExportStateAsync().ConfigureAwait(false);
Dictionary<ScopeKey, PortableValue> stateData = await this.RunContext.StateManager.ExportStateAsync().ConfigureAwait(false);
Checkpoint checkpoint = new(this.StepTracer.StepNumber, this._workflowInfoCache, runnerData, stateData, edgeData);
CheckpointInfo checkpointInfo = await this.CheckpointManager.CommitCheckpointAsync(this.RunId, checkpoint).ConfigureAwait(false);
this.StepTracer.TraceCheckpointCreated(checkpointInfo);
this._checkpoints.Add(checkpointInfo);
Checkpoint checkpoint = new(this.StepTracer.StepNumber, this._workflowInfoCache, runnerData, stateData, edgeData, this._lastCheckpointInfo);
this._lastCheckpointInfo = await this.CheckpointManager.CommitCheckpointAsync(this.RunId, checkpoint).ConfigureAwait(false);
this.StepTracer.TraceCheckpointCreated(this._lastCheckpointInfo);
this._checkpoints.Add(this._lastCheckpointInfo);
}
public async ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellationToken = default)
@@ -333,6 +334,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
await this.EdgeMap.ImportStateAsync(checkpoint).ConfigureAwait(false);
await Task.WhenAll(executorNotifyTask, republishRequestsTask.AsTask()).ConfigureAwait(false);
this._lastCheckpointInfo = checkpointInfo;
this.StepTracer.Reload(this.StepTracer.StepNumber);
}
dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointParentTests.cs
+177
View File
@@ -0,0 +1,177 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using FluentAssertions;
using Microsoft.Agents.AI.Workflows.Checkpointing;
namespace Microsoft.Agents.AI.Workflows.UnitTests;
/// <summary>
/// Tests for verifying that CheckpointInfo.Parent is properly populated
/// when checkpoints are created during workflow execution (GH #3796).
/// </summary>
public class CheckpointParentTests
{
[Theory]
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
internal async Task Checkpoint_FirstCheckpoint_ShouldHaveNullParentAsync(ExecutionEnvironment environment)
{
// Arrange: A simple two-step workflow that will produce at least one checkpoint.
ForwardMessageExecutor<string> executorA = new("A");
ForwardMessageExecutor<string> executorB = new("B");
Workflow workflow = new WorkflowBuilder(executorA)
.AddEdge(executorA, executorB)
.Build();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
IWorkflowExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// Act
Checkpointed<StreamingRun> checkpointed =
await env.StreamAsync(workflow, "Hello", checkpointManager);
List<CheckpointInfo> checkpoints = [];
await foreach (WorkflowEvent evt in checkpointed.Run.WatchStreamAsync())
{
if (evt is SuperStepCompletedEvent stepEvt && stepEvt.CompletionInfo?.Checkpoint is { } cp)
{
checkpoints.Add(cp);
}
}
// Assert: The first checkpoint should have been created and stored with a null parent.
checkpoints.Should().NotBeEmpty("at least one checkpoint should have been created");
CheckpointInfo firstCheckpoint = checkpoints[0];
Checkpoint storedFirst = await ((ICheckpointManager)checkpointManager)
.LookupCheckpointAsync(firstCheckpoint.RunId, firstCheckpoint);
storedFirst.Parent.Should().BeNull("the first checkpoint should have no parent");
}
[Theory]
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
internal async Task Checkpoint_SubsequentCheckpoints_ShouldChainParentsAsync(ExecutionEnvironment environment)
{
// Arrange: A workflow with a loop that will produce multiple checkpoints.
ForwardMessageExecutor<string> executorA = new("A");
ForwardMessageExecutor<string> executorB = new("B");
// A -> B -> A (loop) to generate multiple supersteps/checkpoints.
Workflow workflow = new WorkflowBuilder(executorA)
.AddEdge(executorA, executorB)
.AddEdge(executorB, executorA)
.Build();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
IWorkflowExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// Act
await using Checkpointed<StreamingRun> checkpointed =
await env.StreamAsync(workflow, "Hello", checkpointManager);
List<CheckpointInfo> checkpoints = [];
using CancellationTokenSource cts = new();
await foreach (WorkflowEvent evt in checkpointed.Run.WatchStreamAsync(cts.Token))
{
if (evt is SuperStepCompletedEvent stepEvt && stepEvt.CompletionInfo?.Checkpoint is { } cp)
{
checkpoints.Add(cp);
if (checkpoints.Count >= 3)
{
cts.Cancel();
}
}
}
// Assert: We should have at least 3 checkpoints
checkpoints.Should().HaveCountGreaterThanOrEqualTo(3);
// Verify the parent chain
Checkpoint stored0 = await ((ICheckpointManager)checkpointManager)
.LookupCheckpointAsync(checkpoints[0].RunId, checkpoints[0]);
stored0.Parent.Should().BeNull("the first checkpoint should have no parent");
Checkpoint stored1 = await ((ICheckpointManager)checkpointManager)
.LookupCheckpointAsync(checkpoints[1].RunId, checkpoints[1]);
stored1.Parent.Should().NotBeNull("the second checkpoint should have a parent");
stored1.Parent.Should().Be(checkpoints[0], "the second checkpoint's parent should be the first checkpoint");
Checkpoint stored2 = await ((ICheckpointManager)checkpointManager)
.LookupCheckpointAsync(checkpoints[2].RunId, checkpoints[2]);
stored2.Parent.Should().NotBeNull("the third checkpoint should have a parent");
stored2.Parent.Should().Be(checkpoints[1], "the third checkpoint's parent should be the second checkpoint");
}
[Theory]
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
internal async Task Checkpoint_AfterResume_ShouldHaveResumedCheckpointAsParentAsync(ExecutionEnvironment environment)
{
// Arrange: A looping workflow that produces checkpoints.
ForwardMessageExecutor<string> executorA = new("A");
ForwardMessageExecutor<string> executorB = new("B");
Workflow workflow = new WorkflowBuilder(executorA)
.AddEdge(executorA, executorB)
.AddEdge(executorB, executorA)
.Build();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
IWorkflowExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// First run: collect a checkpoint to resume from
await using Checkpointed<StreamingRun> checkpointed =
await env.StreamAsync(workflow, "Hello", checkpointManager);
List<CheckpointInfo> firstRunCheckpoints = [];
using CancellationTokenSource cts = new();
await foreach (WorkflowEvent evt in checkpointed.Run.WatchStreamAsync(cts.Token))
{
if (evt is SuperStepCompletedEvent stepEvt && stepEvt.CompletionInfo?.Checkpoint is { } cp)
{
firstRunCheckpoints.Add(cp);
if (firstRunCheckpoints.Count >= 2)
{
cts.Cancel();
}
}
}
firstRunCheckpoints.Should().HaveCountGreaterThanOrEqualTo(2);
CheckpointInfo resumePoint = firstRunCheckpoints[0];
// Dispose the first run to release workflow ownership before resuming.
await checkpointed.DisposeAsync();
// Act: Resume from the first checkpoint
Checkpointed<StreamingRun> resumed =
await env.ResumeStreamAsync(workflow, resumePoint, checkpointManager);
List<CheckpointInfo> resumedCheckpoints = [];
using CancellationTokenSource cts2 = new();
await foreach (WorkflowEvent evt in resumed.Run.WatchStreamAsync(cts2.Token))
{
if (evt is SuperStepCompletedEvent stepEvt && stepEvt.CompletionInfo?.Checkpoint is { } cp)
{
resumedCheckpoints.Add(cp);
if (resumedCheckpoints.Count >= 1)
{
cts2.Cancel();
}
}
}
// Assert: The first checkpoint after resume should have the resume point as its parent.
resumedCheckpoints.Should().NotBeEmpty();
Checkpoint storedResumed = await ((ICheckpointManager)checkpointManager)
.LookupCheckpointAsync(resumedCheckpoints[0].RunId, resumedCheckpoints[0]);
storedResumed.Parent.Should().NotBeNull("checkpoint created after resume should have a parent");
storedResumed.Parent.Should().Be(resumePoint, "checkpoint after resume should reference the checkpoint we resumed from");
}
}