From efdacb9d27980ecf1a5da05c70023901cf7da959 Mon Sep 17 00:00:00 2001 From: franklinic Date: Mon, 8 Jun 2026 22:53:45 -0400 Subject: [PATCH 1/8] =?UTF-8?q?feat(agentic):=20Phase=201=20=E2=80=94=20ty?= =?UTF-8?q?ped=20StateGraph=20runtime=20core=20(#1544)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduces the typed graph runtime (AiDotNet.Agentic.Graph), the LangGraph-surpassing core: - StateGraph builder: AddNode (sync/async), AddEdge, AddConditionalEdges, SetEntryPoint, Compile() with validation (entry exists, edges resolve, a node can't have both a fixed edge and conditional edges). - CompiledStateGraph: InvokeAsync (final state) + StreamAsync (per-node GraphStepUpdate); fully typed state threading; cycles bounded by GraphRunOptions.MaxSteps (default 25) -> GraphRecursionException. - GraphSpecialNodes.End sentinel; conditional routers return a node name or End (validated at runtime). - 7 tests (linear, conditional cycle, streaming, recursion-limit, async node, compile validation, dup/reserved names) green on net10.0 + net471. No null-forgiving. Checkpointing/HITL/fan-out/subgraphs are the next slices. Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 --- src/Agentic/Graph/CompiledStateGraph.cs | 135 ++++++++++++ src/Agentic/Graph/GraphRecursionException.cs | 30 +++ src/Agentic/Graph/GraphRunOptions.cs | 19 ++ src/Agentic/Graph/GraphSpecialNodes.cs | 18 ++ src/Agentic/Graph/GraphStepUpdate.cs | 38 ++++ src/Agentic/Graph/StateGraph.cs | 195 ++++++++++++++++++ .../Agentic/Graph/StateGraphTests.cs | 121 +++++++++++ 7 files changed, 556 insertions(+) create mode 100644 src/Agentic/Graph/CompiledStateGraph.cs create mode 100644 src/Agentic/Graph/GraphRecursionException.cs create mode 100644 src/Agentic/Graph/GraphRunOptions.cs create mode 100644 src/Agentic/Graph/GraphSpecialNodes.cs create mode 100644 src/Agentic/Graph/GraphStepUpdate.cs create mode 100644 src/Agentic/Graph/StateGraph.cs create mode 100644 tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs diff --git a/src/Agentic/Graph/CompiledStateGraph.cs b/src/Agentic/Graph/CompiledStateGraph.cs new file mode 100644 index 0000000000..6c9580cec1 --- /dev/null +++ b/src/Agentic/Graph/CompiledStateGraph.cs @@ -0,0 +1,135 @@ +using System.Runtime.CompilerServices; + +namespace AiDotNet.Agentic.Graph; + +/// +/// An executable, validated state graph produced by . Runs nodes +/// according to the graph's edges, threading the state from one node to the next until it reaches the end. +/// +/// The graph's state type (threaded through every node). +/// +/// +/// Execution starts at the entry node. After a node runs, the next node is chosen by: its conditional +/// router (if any), else its fixed edge, else the run ends. Cycles are allowed and bounded by +/// . Use for the final state or +/// to observe each step. +/// +/// For Beginners: This is the "compiled program" form of your graph. You give it a starting +/// state and it walks the nodes you wired up, handing each node the latest state, until it reaches the +/// end — then returns the final state. +/// +/// +public sealed class CompiledStateGraph +{ + private readonly IReadOnlyDictionary>> _nodes; + private readonly IReadOnlyDictionary _edges; + private readonly IReadOnlyDictionary> _conditionalEdges; + private readonly string _entryPoint; + + internal CompiledStateGraph( + IReadOnlyDictionary>> nodes, + IReadOnlyDictionary edges, + IReadOnlyDictionary> conditionalEdges, + string entryPoint) + { + _nodes = nodes; + _edges = edges; + _conditionalEdges = conditionalEdges; + _entryPoint = entryPoint; + } + + /// + /// Runs the graph from the supplied initial state and returns the final state. + /// + /// The starting state. + /// Run options (step budget). null uses defaults. + /// Token used to cancel the run. + /// The final state when flow reaches the end node. + /// Thrown when the step budget is exceeded. + public async Task InvokeAsync( + TState initialState, + GraphRunOptions? options = null, + CancellationToken cancellationToken = default) + { + var maxSteps = ResolveMaxSteps(options); + var state = initialState; + var current = _entryPoint; + var steps = 0; + + while (current != GraphSpecialNodes.End) + { + cancellationToken.ThrowIfCancellationRequested(); + if (++steps > maxSteps) + { + throw new GraphRecursionException(maxSteps); + } + + state = await _nodes[current](state, cancellationToken).ConfigureAwait(false); + current = NextNode(current, state); + } + + return state; + } + + /// + /// Runs the graph and yields an update after each node executes, ending when flow reaches the end node. + /// + /// The starting state. + /// Run options (step budget). null uses defaults. + /// Token used to cancel the run. + /// A stream of ; the final update carries the final state. + /// Thrown when the step budget is exceeded. + public async IAsyncEnumerable> StreamAsync( + TState initialState, + GraphRunOptions? options = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + var maxSteps = ResolveMaxSteps(options); + var state = initialState; + var current = _entryPoint; + var steps = 0; + + while (current != GraphSpecialNodes.End) + { + cancellationToken.ThrowIfCancellationRequested(); + if (++steps > maxSteps) + { + throw new GraphRecursionException(maxSteps); + } + + var executed = current; + state = await _nodes[current](state, cancellationToken).ConfigureAwait(false); + yield return new GraphStepUpdate(executed, state); + current = NextNode(current, state); + } + } + + private static int ResolveMaxSteps(GraphRunOptions? options) + { + var max = options?.MaxSteps ?? new GraphRunOptions().MaxSteps; + return max > 0 ? max : new GraphRunOptions().MaxSteps; + } + + private string NextNode(string current, TState state) + { + if (_conditionalEdges.TryGetValue(current, out var router)) + { + var next = router(state); + if (string.IsNullOrEmpty(next)) + { + throw new InvalidOperationException( + $"The conditional router for node '{current}' returned a null or empty target."); + } + + if (next != GraphSpecialNodes.End && !_nodes.ContainsKey(next)) + { + throw new InvalidOperationException( + $"The conditional router for node '{current}' returned unknown target '{next}'."); + } + + return next; + } + + return _edges.TryGetValue(current, out var target) ? target : GraphSpecialNodes.End; + } +} diff --git a/src/Agentic/Graph/GraphRecursionException.cs b/src/Agentic/Graph/GraphRecursionException.cs new file mode 100644 index 0000000000..3fceeb8a61 --- /dev/null +++ b/src/Agentic/Graph/GraphRecursionException.cs @@ -0,0 +1,30 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// Thrown when a graph run exceeds its configured maximum number of steps, which usually indicates a +/// cycle that never reaches the end node. +/// +/// +/// For Beginners: Graphs can loop (a node can route back to an earlier node). To stop an +/// accidental infinite loop, every run has a step budget. If the graph keeps going past that budget, +/// this exception is raised so you can fix the routing or raise the limit deliberately. +/// +/// +public sealed class GraphRecursionException : Exception +{ + /// + /// Initializes a new instance of the class. + /// + /// The step budget that was exceeded. + public GraphRecursionException(int maxSteps) + : base($"Graph exceeded its maximum of {maxSteps} steps without reaching the end node. " + + "This usually means a cycle never terminates; check your conditional edges or raise GraphRunOptions.MaxSteps.") + { + MaxSteps = maxSteps; + } + + /// + /// Gets the step budget that was exceeded. + /// + public int MaxSteps { get; } +} diff --git a/src/Agentic/Graph/GraphRunOptions.cs b/src/Agentic/Graph/GraphRunOptions.cs new file mode 100644 index 0000000000..7dea5db4e0 --- /dev/null +++ b/src/Agentic/Graph/GraphRunOptions.cs @@ -0,0 +1,19 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// Per-run settings for executing a . +/// +/// +/// For Beginners: Knobs for a single graph run. The most important one is +/// , which caps how many node executions a run may take before giving up — a +/// safety net against cycles that never end. +/// +/// +public sealed class GraphRunOptions +{ + /// + /// Gets or sets the maximum number of node executions allowed in a single run before a + /// is thrown. Default: 25. Must be positive. + /// + public int MaxSteps { get; set; } = 25; +} diff --git a/src/Agentic/Graph/GraphSpecialNodes.cs b/src/Agentic/Graph/GraphSpecialNodes.cs new file mode 100644 index 0000000000..54092d23fa --- /dev/null +++ b/src/Agentic/Graph/GraphSpecialNodes.cs @@ -0,0 +1,18 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// Reserved node names recognized by the graph runtime. +/// +/// +/// For Beginners: A graph finishes when flow reaches the special node. +/// You route to it with AddEdge("lastNode", GraphSpecialNodes.End) (or return it from a +/// conditional router) to say "we're done". +/// +/// +public static class GraphSpecialNodes +{ + /// + /// The terminal node name. When flow reaches this node, the run completes and returns the current state. + /// + public const string End = "__end__"; +} diff --git a/src/Agentic/Graph/GraphStepUpdate.cs b/src/Agentic/Graph/GraphStepUpdate.cs new file mode 100644 index 0000000000..3e094b1414 --- /dev/null +++ b/src/Agentic/Graph/GraphStepUpdate.cs @@ -0,0 +1,38 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// A streamed update emitted after a single node finishes executing during a graph run: the node's name +/// and the graph state as it stands after that node. +/// +/// The graph's state type. +/// +/// For Beginners: When you stream a graph run, you receive one of these each time a step +/// completes — letting you watch the state evolve node by node (for progress UIs, logging, or +/// debugging). The last update's is the final result. +/// +/// +public sealed class GraphStepUpdate +{ + /// + /// Initializes a new step update. + /// + /// The name of the node that just executed. + /// The graph state after the node executed. + /// Thrown when is null. + public GraphStepUpdate(string nodeName, TState state) + { + Guard.NotNull(nodeName); + NodeName = nodeName; + State = state; + } + + /// + /// Gets the name of the node that just executed. + /// + public string NodeName { get; } + + /// + /// Gets the graph state after the node executed. + /// + public TState State { get; } +} diff --git a/src/Agentic/Graph/StateGraph.cs b/src/Agentic/Graph/StateGraph.cs new file mode 100644 index 0000000000..b69bb8fef8 --- /dev/null +++ b/src/Agentic/Graph/StateGraph.cs @@ -0,0 +1,195 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// A builder for a typed state graph: register nodes (state transformers), wire them with fixed or +/// conditional edges (cycles allowed), set an entry point, then into an executable +/// . +/// +/// +/// The state type threaded through the graph. Each node receives the current state and returns the next +/// state (it may mutate and return the same instance, or return a new one). +/// +/// +/// +/// This is the AiDotNet counterpart to LangGraph's StateGraph, but fully typed: there are no +/// stringly-typed state dictionaries — is your own type, checked by the +/// compiler. Routing is explicit: a node has at most one fixed edge or one conditional router; reaching +/// finishes the run. +/// +/// For Beginners: Think of building a flowchart. adds a box that does +/// some work on your data; draws an arrow to the next box; +/// draws arrows that depend on the data; marks where to start. Arrows can +/// loop back to create cycles. turns the flowchart into something you can run. +/// +/// Example: +/// +/// var graph = new StateGraph<MyState>() +/// .AddNode("plan", (s, ct) => Task.FromResult(s.WithPlan())) +/// .AddNode("act", (s, ct) => Task.FromResult(s.Act())) +/// .AddConditionalEdges("act", s => s.IsDone ? StateGraph<MyState>.End : "act") +/// .AddEdge("plan", "act") +/// .SetEntryPoint("plan") +/// .Compile(); +/// var result = await graph.InvokeAsync(new MyState()); +/// +/// +/// +public sealed class StateGraph +{ + /// + /// The terminal node name; route here to finish the run. Alias of . + /// + public const string End = GraphSpecialNodes.End; + + private readonly Dictionary>> _nodes = new(StringComparer.Ordinal); + private readonly Dictionary _edges = new(StringComparer.Ordinal); + private readonly Dictionary> _conditionalEdges = new(StringComparer.Ordinal); + private string? _entryPoint; + + /// + /// Adds an asynchronous node that transforms the state. + /// + /// The unique node name (cannot be empty or the reserved end name). + /// The node body: receives the current state and returns the next state. + /// This builder, for chaining. + /// Thrown when or is null. + /// Thrown when the name is empty/whitespace, reserved, or already used. + public StateGraph AddNode(string name, Func> node) + { + Guard.NotNullOrWhiteSpace(name); + Guard.NotNull(node); + if (name == End) + { + throw new ArgumentException($"'{End}' is a reserved node name.", nameof(name)); + } + + if (_nodes.ContainsKey(name)) + { + throw new ArgumentException($"A node named '{name}' has already been added.", nameof(name)); + } + + _nodes[name] = node; + return this; + } + + /// + /// Adds a synchronous node that transforms the state. + /// + /// The unique node name. + /// The node body: receives the current state and returns the next state. + /// This builder, for chaining. + /// Thrown when or is null. + public StateGraph AddNode(string name, Func node) + { + Guard.NotNull(node); + return AddNode(name, (state, _) => Task.FromResult(node(state))); + } + + /// + /// Adds a fixed edge: after runs, flow always moves to . + /// + /// The source node name. + /// The destination node name, or . + /// This builder, for chaining. + /// Thrown when names are empty or already has a fixed edge. + public StateGraph AddEdge(string from, string to) + { + Guard.NotNullOrWhiteSpace(from); + Guard.NotNullOrWhiteSpace(to); + if (_edges.ContainsKey(from)) + { + throw new ArgumentException($"Node '{from}' already has a fixed edge.", nameof(from)); + } + + _edges[from] = to; + return this; + } + + /// + /// Adds conditional edges: after runs, the router inspects the state and + /// returns the next node name (or ). + /// + /// The source node name. + /// Selects the next node from the current state. + /// This builder, for chaining. + /// Thrown when is null. + /// Thrown when is empty or already has conditional edges. + public StateGraph AddConditionalEdges(string from, Func router) + { + Guard.NotNullOrWhiteSpace(from); + Guard.NotNull(router); + if (_conditionalEdges.ContainsKey(from)) + { + throw new ArgumentException($"Node '{from}' already has conditional edges.", nameof(from)); + } + + _conditionalEdges[from] = router; + return this; + } + + /// + /// Sets the node where execution begins. + /// + /// The entry node name. + /// This builder, for chaining. + /// Thrown when is empty/whitespace. + public StateGraph SetEntryPoint(string name) + { + Guard.NotNullOrWhiteSpace(name); + _entryPoint = name; + return this; + } + + /// + /// Validates the graph and produces an executable . + /// + /// The compiled graph. + /// Thrown when the graph is invalid (no/unknown entry point, edges to unknown nodes, or a node with both a fixed edge and conditional edges). + public CompiledStateGraph Compile() + { + var entry = _entryPoint; + if (entry is null || entry.Length == 0) + { + throw new InvalidOperationException("No entry point set. Call SetEntryPoint(...) before Compile()."); + } + + if (!_nodes.ContainsKey(entry)) + { + throw new InvalidOperationException($"Entry point '{entry}' is not a registered node."); + } + + foreach (var edge in _edges) + { + if (!_nodes.ContainsKey(edge.Key)) + { + throw new InvalidOperationException($"Edge source '{edge.Key}' is not a registered node."); + } + + if (edge.Value != End && !_nodes.ContainsKey(edge.Value)) + { + throw new InvalidOperationException($"Edge target '{edge.Value}' (from '{edge.Key}') is not a registered node or End."); + } + } + + foreach (var conditional in _conditionalEdges) + { + if (!_nodes.ContainsKey(conditional.Key)) + { + throw new InvalidOperationException($"Conditional-edge source '{conditional.Key}' is not a registered node."); + } + + if (_edges.ContainsKey(conditional.Key)) + { + throw new InvalidOperationException( + $"Node '{conditional.Key}' has both a fixed edge and conditional edges; use one or the other."); + } + } + + // Snapshot so post-compile builder mutations don't affect the compiled graph. + return new CompiledStateGraph( + new Dictionary>>(_nodes, StringComparer.Ordinal), + new Dictionary(_edges, StringComparer.Ordinal), + new Dictionary>(_conditionalEdges, StringComparer.Ordinal), + entry); + } +} diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs new file mode 100644 index 0000000000..704c81c40a --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs @@ -0,0 +1,121 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class StateGraphTests + { + [Fact(Timeout = 60000)] + public async Task Linear_RunsNodesInOrder_AndThreadsState() + { + var graph = new StateGraph() + .AddNode("inc", s => s + 1) + .AddNode("double", s => s * 2) + .AddEdge("inc", "double") + .AddEdge("double", StateGraph.End) + .SetEntryPoint("inc") + .Compile(); + + var result = await graph.InvokeAsync(1); + Assert.Equal(4, result); // (1 + 1) * 2 + } + + [Fact(Timeout = 60000)] + public async Task ConditionalEdges_SupportCycles() + { + var graph = new StateGraph() + .AddNode("inc", s => s + 1) + .AddConditionalEdges("inc", s => s < 3 ? "inc" : StateGraph.End) + .SetEntryPoint("inc") + .Compile(); + + var result = await graph.InvokeAsync(0); + Assert.Equal(3, result); // loops inc until value reaches 3 + } + + [Fact(Timeout = 60000)] + public async Task Stream_EmitsUpdatePerNode_WithFinalState() + { + var graph = new StateGraph() + .AddNode("a", s => s + 1) + .AddNode("b", s => s + 10) + .AddEdge("a", "b") + .AddEdge("b", StateGraph.End) + .SetEntryPoint("a") + .Compile(); + + var updates = new List>(); + await foreach (var u in graph.StreamAsync(0)) + { + updates.Add(u); + } + + Assert.Equal(new[] { "a", "b" }, updates.Select(u => u.NodeName)); + Assert.Equal(1, updates[0].State); + Assert.Equal(11, updates[1].State); + } + + [Fact(Timeout = 60000)] + public async Task RecursionLimit_Throws_OnRunawayCycle() + { + var graph = new StateGraph() + .AddNode("loop", s => s + 1) + .AddConditionalEdges("loop", _ => "loop") // never terminates + .SetEntryPoint("loop") + .Compile(); + + var ex = await Assert.ThrowsAsync( + () => graph.InvokeAsync(0, new GraphRunOptions { MaxSteps = 5 })); + Assert.Equal(5, ex.MaxSteps); + } + + [Fact(Timeout = 60000)] + public async Task AsyncNode_IsAwaited() + { + var graph = new StateGraph() + .AddNode("delay", async (s, ct) => { await Task.Yield(); return s + 41; }) + .AddEdge("delay", StateGraph.End) + .SetEntryPoint("delay") + .Compile(); + + Assert.Equal(42, await graph.InvokeAsync(1)); + } + + [Fact(Timeout = 60000)] + public async Task Compile_Validates_EntryPointEdgesAndExclusiveRouting() + { + // No entry point. + Assert.Throws(() => + new StateGraph().AddNode("a", s => s).Compile()); + + // Edge to unknown node. + Assert.Throws(() => + new StateGraph().AddNode("a", s => s).AddEdge("a", "ghost").SetEntryPoint("a").Compile()); + + // A node cannot have both a fixed edge and conditional edges. + Assert.Throws(() => + new StateGraph() + .AddNode("a", s => s) + .AddEdge("a", StateGraph.End) + .AddConditionalEdges("a", _ => StateGraph.End) + .SetEntryPoint("a") + .Compile()); + + await Task.CompletedTask; + } + + [Fact(Timeout = 60000)] + public async Task AddNode_RejectsDuplicateAndReservedNames() + { + var graph = new StateGraph().AddNode("a", s => s); + Assert.Throws(() => graph.AddNode("a", s => s)); // duplicate + Assert.Throws(() => new StateGraph().AddNode(StateGraph.End, s => s)); // reserved + await Task.CompletedTask; + } + } +} From 205d9ea22923fd8ca4ed995987763d50abadafc0 Mon Sep 17 00:00:00 2001 From: franklinic Date: Mon, 8 Jun 2026 23:10:22 -0400 Subject: [PATCH 2/8] =?UTF-8?q?feat(agentic):=20Phase=201=20=E2=80=94=20du?= =?UTF-8?q?rable=20checkpointing,=20resume=20&=20time-travel=20(#1544)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds durable execution to the graph runtime (AiDotNet.Agentic.Graph.Checkpointing): - GraphCheckpoint (threadId, checkpointId, step, nextNode, state, IsComplete); IGraphCheckpointer (Save/GetLatest/Get/GetHistory); InMemoryGraphCheckpointer (thread-safe). - CompiledStateGraph: checkpointed InvokeAsync(state, checkpointer, threadId) auto-resumes from the latest checkpoint (saves after every step; cumulative step budget); ResumeFromAsync(threadId, checkpointId) replays from any past checkpoint (time-travel). Shared RunCoreAsync powers plain + checkpointed runs. - 5 tests (fresh+history, resume-skips-earlier, re-invoke-completed, time-travel replay, unknown-checkpoint throws); 12 graph tests total green on net10.0 + net471. No null-forgiving. NEXT: HITL interrupts, channels/reducers+fan-out, DB-backed checkpointers. Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 --- .../Graph/Checkpointing/GraphCheckpoint.cs | 62 ++++++++++ .../Graph/Checkpointing/IGraphCheckpointer.cs | 46 ++++++++ .../InMemoryGraphCheckpointer.cs | 91 ++++++++++++++ src/Agentic/Graph/CompiledStateGraph.cs | 111 +++++++++++++++++- .../Agentic/Graph/GraphCheckpointingTests.cs | 86 ++++++++++++++ 5 files changed, 392 insertions(+), 4 deletions(-) create mode 100644 src/Agentic/Graph/Checkpointing/GraphCheckpoint.cs create mode 100644 src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs create mode 100644 src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs create mode 100644 tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCheckpointingTests.cs diff --git a/src/Agentic/Graph/Checkpointing/GraphCheckpoint.cs b/src/Agentic/Graph/Checkpointing/GraphCheckpoint.cs new file mode 100644 index 0000000000..1f8f504ec6 --- /dev/null +++ b/src/Agentic/Graph/Checkpointing/GraphCheckpoint.cs @@ -0,0 +1,62 @@ +namespace AiDotNet.Agentic.Graph.Checkpointing; + +/// +/// An immutable snapshot of a graph run at a point in time: which node runs next and the state as of +/// that point. Saved after each step so a run can be resumed, inspected, or replayed. +/// +/// The graph's state type. +/// +/// +/// A checkpoint records the (where execution will continue) plus the +/// at that boundary, tagged with a (the run/conversation) and a +/// monotonically increasing . The sequence of checkpoints for a thread is its history — +/// the basis for durable resume and time-travel. +/// +/// For Beginners: Like a save-game. It remembers exactly where the graph was about to go +/// next and what the data looked like, so you can stop and pick up later (or rewind to an earlier save). +/// +/// +public sealed class GraphCheckpoint +{ + /// + /// Initializes a new checkpoint. + /// + /// The run/thread this checkpoint belongs to. + /// A unique id for this checkpoint within the thread. + /// The zero-based step index (monotonically increasing within a thread). + /// The node that will execute next (or when complete). + /// The state at this boundary. + /// Thrown when a required string is null. + /// Thrown when a required string is empty/whitespace. + /// Thrown when is negative. + public GraphCheckpoint(string threadId, string checkpointId, int step, string nextNode, TState state) + { + Guard.NotNullOrWhiteSpace(threadId); + Guard.NotNullOrWhiteSpace(checkpointId); + Guard.NotNullOrWhiteSpace(nextNode); + Guard.NonNegative(step); + ThreadId = threadId; + CheckpointId = checkpointId; + Step = step; + NextNode = nextNode; + State = state; + } + + /// Gets the run/thread this checkpoint belongs to. + public string ThreadId { get; } + + /// Gets the unique id of this checkpoint within the thread. + public string CheckpointId { get; } + + /// Gets the zero-based, monotonically increasing step index. + public int Step { get; } + + /// Gets the node that will execute next (or when the run is complete). + public string NextNode { get; } + + /// Gets the state captured at this boundary. + public TState State { get; } + + /// Gets a value indicating whether this checkpoint represents a completed run. + public bool IsComplete => NextNode == GraphSpecialNodes.End; +} diff --git a/src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs b/src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs new file mode 100644 index 0000000000..c258bdd92c --- /dev/null +++ b/src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs @@ -0,0 +1,46 @@ +namespace AiDotNet.Agentic.Graph.Checkpointing; + +/// +/// Persists and retrieves s, enabling durable resume and time-travel +/// for graph runs. Implementations back onto memory, SQLite, Postgres, Redis, etc. +/// +/// The graph's state type. +/// +/// For Beginners: The save-game store. The graph asks it to save a checkpoint after each +/// step, to fetch the latest checkpoint when resuming, and to list the full history when rewinding. +/// +/// +public interface IGraphCheckpointer +{ + /// + /// Saves (appends) a checkpoint. + /// + /// The checkpoint to persist. + /// Token used to cancel the operation. + Task SaveAsync(GraphCheckpoint checkpoint, CancellationToken cancellationToken = default); + + /// + /// Gets the most recent checkpoint for a thread, or null if the thread has none. + /// + /// The run/thread id. + /// Token used to cancel the operation. + /// The latest checkpoint, or null. + Task?> GetLatestAsync(string threadId, CancellationToken cancellationToken = default); + + /// + /// Gets a specific checkpoint by id (used for time-travel / replay from a past point). + /// + /// The run/thread id. + /// The checkpoint id. + /// Token used to cancel the operation. + /// The checkpoint, or null if not found. + Task?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default); + + /// + /// Gets the full checkpoint history for a thread, ordered by step ascending. + /// + /// The run/thread id. + /// Token used to cancel the operation. + /// The ordered history (possibly empty). + Task>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default); +} diff --git a/src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs b/src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs new file mode 100644 index 0000000000..f4fc8a8c5d --- /dev/null +++ b/src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs @@ -0,0 +1,91 @@ +namespace AiDotNet.Agentic.Graph.Checkpointing; + +/// +/// An in-memory that keeps each thread's checkpoint history in a +/// dictionary. Suitable for tests, single-process runs, and as the default when no durable store is wired. +/// +/// The graph's state type. +/// +/// +/// Thread-safe via a per-instance lock. Because it stores the state object by reference, callers that use +/// a mutable reference type for should treat the state as immutable per step +/// (return a new/copied instance from nodes) to get true snapshots; durable checkpointers serialize and so +/// snapshot inherently. +/// +/// For Beginners: Keeps your save-games in memory. Great for tests and short-lived runs; for +/// durability across process restarts, use a database-backed checkpointer (coming in later slices). +/// +/// +public sealed class InMemoryGraphCheckpointer : IGraphCheckpointer +{ + private readonly object _gate = new(); + private readonly Dictionary>> _byThread = new(StringComparer.Ordinal); + + /// + public Task SaveAsync(GraphCheckpoint checkpoint, CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpoint); + lock (_gate) + { + if (!_byThread.TryGetValue(checkpoint.ThreadId, out var list)) + { + list = new List>(); + _byThread[checkpoint.ThreadId] = list; + } + + list.Add(checkpoint); + } + + return Task.CompletedTask; + } + + /// + public Task?> GetLatestAsync(string threadId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + lock (_gate) + { + if (_byThread.TryGetValue(threadId, out var list) && list.Count > 0) + { + return Task.FromResult?>(list[list.Count - 1]); + } + } + + return Task.FromResult?>(null); + } + + /// + public Task?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + Guard.NotNull(checkpointId); + lock (_gate) + { + if (_byThread.TryGetValue(threadId, out var list)) + { + foreach (var cp in list) + { + if (cp.CheckpointId == checkpointId) + { + return Task.FromResult?>(cp); + } + } + } + } + + return Task.FromResult?>(null); + } + + /// + public Task>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + lock (_gate) + { + IReadOnlyList> history = _byThread.TryGetValue(threadId, out var list) + ? list.ToList() + : new List>(); + return Task.FromResult(history); + } + } +} diff --git a/src/Agentic/Graph/CompiledStateGraph.cs b/src/Agentic/Graph/CompiledStateGraph.cs index 6c9580cec1..a9b8b8c706 100644 --- a/src/Agentic/Graph/CompiledStateGraph.cs +++ b/src/Agentic/Graph/CompiledStateGraph.cs @@ -1,4 +1,5 @@ using System.Runtime.CompilerServices; +using AiDotNet.Agentic.Graph.Checkpointing; namespace AiDotNet.Agentic.Graph; @@ -46,26 +47,128 @@ internal CompiledStateGraph( /// Token used to cancel the run. /// The final state when flow reaches the end node. /// Thrown when the step budget is exceeded. + public Task InvokeAsync( + TState initialState, + GraphRunOptions? options = null, + CancellationToken cancellationToken = default) + => RunCoreAsync(initialState, _entryPoint, startStep: 0, ResolveMaxSteps(options), checkpointer: null, threadId: null, cancellationToken); + + /// + /// Runs the graph with durable checkpointing under a thread id, resuming automatically from the latest + /// saved checkpoint if one exists for that thread (otherwise starting fresh from ). + /// + /// The starting state (used only when the thread has no prior checkpoint). + /// The checkpoint store. + /// The run/thread id under which checkpoints are saved and resumed. + /// Run options (step budget). null uses defaults. The budget is cumulative across resumes. + /// Token used to cancel the run. + /// The final state. + /// Thrown when or is null. + /// Thrown when the step budget is exceeded. public async Task InvokeAsync( TState initialState, + IGraphCheckpointer checkpointer, + string threadId, GraphRunOptions? options = null, CancellationToken cancellationToken = default) { + Guard.NotNull(checkpointer); + Guard.NotNullOrWhiteSpace(threadId); var maxSteps = ResolveMaxSteps(options); - var state = initialState; - var current = _entryPoint; - var steps = 0; + + var latest = await checkpointer.GetLatestAsync(threadId, cancellationToken).ConfigureAwait(false); + string startNode; + TState state; + int startStep; + if (latest is not null) + { + startNode = latest.NextNode; + state = latest.State; + startStep = latest.Step; + } + else + { + startNode = _entryPoint; + state = initialState; + startStep = 0; + await checkpointer.SaveAsync( + new GraphCheckpoint(threadId, $"{threadId}-0", 0, _entryPoint, initialState), cancellationToken).ConfigureAwait(false); + } + + if (startNode == GraphSpecialNodes.End) + { + return state; // thread already complete + } + + return await RunCoreAsync(state, startNode, startStep, maxSteps, checkpointer, threadId, cancellationToken).ConfigureAwait(false); + } + + /// + /// Resumes (replays) a thread from a specific past checkpoint — the basis for time-travel — continuing + /// execution from that checkpoint's next node and state, and saving new checkpoints as it goes. + /// + /// The checkpoint store. + /// The run/thread id. + /// The id of the checkpoint to resume from. + /// Run options (step budget). null uses defaults. + /// Token used to cancel the run. + /// The final state. + /// Thrown when , , or is null. + /// Thrown when the checkpoint does not exist. + public async Task ResumeFromAsync( + IGraphCheckpointer checkpointer, + string threadId, + string checkpointId, + GraphRunOptions? options = null, + CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpointer); + Guard.NotNullOrWhiteSpace(threadId); + Guard.NotNullOrWhiteSpace(checkpointId); + + var checkpoint = await checkpointer.GetAsync(threadId, checkpointId, cancellationToken).ConfigureAwait(false); + if (checkpoint is null) + { + throw new InvalidOperationException($"Checkpoint '{checkpointId}' was not found for thread '{threadId}'."); + } + + if (checkpoint.NextNode == GraphSpecialNodes.End) + { + return checkpoint.State; + } + + return await RunCoreAsync( + checkpoint.State, checkpoint.NextNode, checkpoint.Step, ResolveMaxSteps(options), checkpointer, threadId, cancellationToken).ConfigureAwait(false); + } + + private async Task RunCoreAsync( + TState state, + string startNode, + int startStep, + int maxSteps, + IGraphCheckpointer? checkpointer, + string? threadId, + CancellationToken cancellationToken) + { + var current = startNode; + var step = startStep; while (current != GraphSpecialNodes.End) { cancellationToken.ThrowIfCancellationRequested(); - if (++steps > maxSteps) + if (++step > maxSteps) { throw new GraphRecursionException(maxSteps); } state = await _nodes[current](state, cancellationToken).ConfigureAwait(false); current = NextNode(current, state); + + if (checkpointer is not null && threadId is not null) + { + await checkpointer.SaveAsync( + new GraphCheckpoint(threadId, $"{threadId}-{step}", step, current, state), cancellationToken).ConfigureAwait(false); + } } return state; diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCheckpointingTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCheckpointingTests.cs new file mode 100644 index 0000000000..2b86a1648f --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCheckpointingTests.cs @@ -0,0 +1,86 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using AiDotNet.Agentic.Graph.Checkpointing; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class GraphCheckpointingTests + { + // a: +1, b: +10, a -> b -> End + private static CompiledStateGraph BuildGraph() => + new StateGraph() + .AddNode("a", s => s + 1) + .AddNode("b", s => s + 10) + .AddEdge("a", "b") + .AddEdge("b", StateGraph.End) + .SetEntryPoint("a") + .Compile(); + + [Fact(Timeout = 60000)] + public async Task CheckpointedRun_Completes_AndRecordsHistory() + { + var graph = BuildGraph(); + var cp = new InMemoryGraphCheckpointer(); + + var result = await graph.InvokeAsync(0, cp, "t1"); + Assert.Equal(11, result); + + var history = await cp.GetHistoryAsync("t1"); + // step0(next=a,0), step1(next=b,1), step2(next=End,11) + Assert.Equal(3, history.Count); + Assert.Equal(new[] { 0, 1, 2 }, history.Select(h => h.Step)); + Assert.Equal(StateGraph.End, history[2].NextNode); + Assert.True(history[2].IsComplete); + Assert.Equal(11, history[2].State); + } + + [Fact(Timeout = 60000)] + public async Task Resume_PicksUpFromLatestCheckpoint_SkippingEarlierNodes() + { + var graph = BuildGraph(); + var cp = new InMemoryGraphCheckpointer(); + + // Simulate that node 'a' already ran (produced 100, next is 'b'). + await cp.SaveAsync(new GraphCheckpoint("t2", "t2-1", 1, "b", 100)); + + var result = await graph.InvokeAsync(initialState: 0, cp, "t2"); + // Resumes at 'b' with state 100 -> 110; 'a' is NOT re-run (would have given 11). + Assert.Equal(110, result); + } + + [Fact(Timeout = 60000)] + public async Task ReinvokingCompletedThread_ReturnsFinalState_WithoutRerunning() + { + var graph = BuildGraph(); + var cp = new InMemoryGraphCheckpointer(); + + await graph.InvokeAsync(0, cp, "t3"); // completes at 11 + var again = await graph.InvokeAsync(999, cp, "t3"); // initialState ignored; thread is complete + Assert.Equal(11, again); + } + + [Fact(Timeout = 60000)] + public async Task TimeTravel_ResumeFromEarlierCheckpoint_Replays() + { + var graph = BuildGraph(); + var cp = new InMemoryGraphCheckpointer(); + await graph.InvokeAsync(0, cp, "t4"); // history: t4-0(a,0), t4-1(b,1), t4-2(End,11) + + // Rewind to the checkpoint after 'a' (next='b', state=1) and replay forward. + var result = await graph.ResumeFromAsync(cp, "t4", "t4-1"); + Assert.Equal(11, result); + } + + [Fact(Timeout = 60000)] + public async Task ResumeFrom_UnknownCheckpoint_Throws() + { + var graph = BuildGraph(); + var cp = new InMemoryGraphCheckpointer(); + await Assert.ThrowsAsync( + () => graph.ResumeFromAsync(cp, "t5", "does-not-exist")); + } + } +} From 82fb6f685f5ebe582a4103fe9adf55ec00c722a0 Mon Sep 17 00:00:00 2001 From: franklinic Date: Mon, 8 Jun 2026 23:24:16 -0400 Subject: [PATCH 3/8] =?UTF-8?q?feat(agentic):=20Phase=201=20=E2=80=94=20hu?= =?UTF-8?q?man-in-the-loop=20interrupts=20(#1544)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds interrupt-before HITL to the graph runtime, built on checkpointing: - StateGraph.AddInterruptBefore(node) marks pause points (validated in Compile). - GraphRunResult (IsComplete/IsInterrupted/State/InterruptedBefore). - CompiledStateGraph.RunAsync(state, checkpointer, threadId, applyOnResume?): fresh run pauses just before the first interrupt node and returns Interrupted; re-invoking the same thread resumes past the pause (optionally applying the human's state edit) until the next interrupt or completion. RunCoreAsync now returns GraphRunResult and honors interrupts; plain/checkpointed InvokeAsync run straight through (no pause). - 4 HITL tests (pause+resume, human edit on resume, interrupt-before-entry, InvokeAsync-ignores-interrupts); 16 graph tests total green on net10.0 + net471. No null-forgiving. NEXT: channels/reducers + dynamic fan-out, DB-backed checkpointers, subgraphs. Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 --- src/Agentic/Graph/CompiledStateGraph.cs | 98 +++++++++++++++++-- src/Agentic/Graph/GraphRunResult.cs | 55 +++++++++++ src/Agentic/Graph/StateGraph.cs | 28 ++++++ .../Agentic/Graph/GraphInterruptTests.cs | 81 +++++++++++++++ 4 files changed, 256 insertions(+), 6 deletions(-) create mode 100644 src/Agentic/Graph/GraphRunResult.cs create mode 100644 tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphInterruptTests.cs diff --git a/src/Agentic/Graph/CompiledStateGraph.cs b/src/Agentic/Graph/CompiledStateGraph.cs index a9b8b8c706..5af1c6a343 100644 --- a/src/Agentic/Graph/CompiledStateGraph.cs +++ b/src/Agentic/Graph/CompiledStateGraph.cs @@ -25,17 +25,20 @@ public sealed class CompiledStateGraph private readonly IReadOnlyDictionary>> _nodes; private readonly IReadOnlyDictionary _edges; private readonly IReadOnlyDictionary> _conditionalEdges; + private readonly HashSet _interruptBefore; private readonly string _entryPoint; internal CompiledStateGraph( IReadOnlyDictionary>> nodes, IReadOnlyDictionary edges, IReadOnlyDictionary> conditionalEdges, + HashSet interruptBefore, string entryPoint) { _nodes = nodes; _edges = edges; _conditionalEdges = conditionalEdges; + _interruptBefore = interruptBefore; _entryPoint = entryPoint; } @@ -47,11 +50,16 @@ internal CompiledStateGraph( /// Token used to cancel the run. /// The final state when flow reaches the end node. /// Thrown when the step budget is exceeded. - public Task InvokeAsync( + public async Task InvokeAsync( TState initialState, GraphRunOptions? options = null, CancellationToken cancellationToken = default) - => RunCoreAsync(initialState, _entryPoint, startStep: 0, ResolveMaxSteps(options), checkpointer: null, threadId: null, cancellationToken); + { + var result = await RunCoreAsync( + initialState, _entryPoint, startStep: 0, ResolveMaxSteps(options), + checkpointer: null, threadId: null, honorInterrupts: false, honorFirstInterrupt: false, cancellationToken).ConfigureAwait(false); + return result.State; + } /// /// Runs the graph with durable checkpointing under a thread id, resuming automatically from the latest @@ -100,7 +108,10 @@ await checkpointer.SaveAsync( return state; // thread already complete } - return await RunCoreAsync(state, startNode, startStep, maxSteps, checkpointer, threadId, cancellationToken).ConfigureAwait(false); + var result = await RunCoreAsync( + state, startNode, startStep, maxSteps, checkpointer, threadId, + honorInterrupts: false, honorFirstInterrupt: false, cancellationToken).ConfigureAwait(false); + return result.State; } /// @@ -137,25 +148,100 @@ public async Task ResumeFromAsync( return checkpoint.State; } + var result = await RunCoreAsync( + checkpoint.State, checkpoint.NextNode, checkpoint.Step, ResolveMaxSteps(options), checkpointer, threadId, + honorInterrupts: false, honorFirstInterrupt: false, cancellationToken).ConfigureAwait(false); + return result.State; + } + + /// + /// Runs the graph with human-in-the-loop interrupts under a thread id. On the first call it runs from + /// the entry node and pauses just before the first interrupt node (returning an interrupted result); + /// calling it again on the same thread resumes past that pause and runs until the next interrupt or + /// completion. Optionally applies the human's edit to the state when resuming. + /// + /// The starting state (used only when the thread has no prior checkpoint). + /// The checkpoint store. + /// The run/thread id. + /// When resuming, an optional transform applied to the saved state (the human's input). Ignored on a fresh run. + /// Run options (step budget). null uses defaults. + /// Token used to cancel the run. + /// A indicating completion or an interrupt point. + /// Thrown when or is null. + public async Task> RunAsync( + TState initialState, + IGraphCheckpointer checkpointer, + string threadId, + Func? applyOnResume = null, + GraphRunOptions? options = null, + CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpointer); + Guard.NotNullOrWhiteSpace(threadId); + var maxSteps = ResolveMaxSteps(options); + + var latest = await checkpointer.GetLatestAsync(threadId, cancellationToken).ConfigureAwait(false); + string startNode; + TState state; + int startStep; + bool honorFirstInterrupt; + if (latest is not null) + { + // Resuming: the caller chose to continue, so do not re-pause at the node we stopped before. + startNode = latest.NextNode; + state = applyOnResume is not null ? applyOnResume(latest.State) : latest.State; + startStep = latest.Step; + honorFirstInterrupt = false; + } + else + { + startNode = _entryPoint; + state = initialState; + startStep = 0; + honorFirstInterrupt = true; + await checkpointer.SaveAsync( + new GraphCheckpoint(threadId, $"{threadId}-0", 0, _entryPoint, initialState), cancellationToken).ConfigureAwait(false); + } + + if (startNode == GraphSpecialNodes.End) + { + return GraphRunResult.Complete(state); + } + return await RunCoreAsync( - checkpoint.State, checkpoint.NextNode, checkpoint.Step, ResolveMaxSteps(options), checkpointer, threadId, cancellationToken).ConfigureAwait(false); + state, startNode, startStep, maxSteps, checkpointer, threadId, + honorInterrupts: true, honorFirstInterrupt: honorFirstInterrupt, cancellationToken).ConfigureAwait(false); } - private async Task RunCoreAsync( + private async Task> RunCoreAsync( TState state, string startNode, int startStep, int maxSteps, IGraphCheckpointer? checkpointer, string? threadId, + bool honorInterrupts, + bool honorFirstInterrupt, CancellationToken cancellationToken) { var current = startNode; var step = startStep; + var firstNode = true; while (current != GraphSpecialNodes.End) { cancellationToken.ThrowIfCancellationRequested(); + + var considerInterrupt = honorInterrupts && (honorFirstInterrupt || !firstNode); + if (considerInterrupt && _interruptBefore.Contains(current)) + { + // Pause before this node. The latest checkpoint already records NextNode == current, + // so a subsequent resume picks up here. + return GraphRunResult.Interrupted(state, current); + } + + firstNode = false; + if (++step > maxSteps) { throw new GraphRecursionException(maxSteps); @@ -171,7 +257,7 @@ await checkpointer.SaveAsync( } } - return state; + return GraphRunResult.Complete(state); } /// diff --git a/src/Agentic/Graph/GraphRunResult.cs b/src/Agentic/Graph/GraphRunResult.cs new file mode 100644 index 0000000000..3304edce85 --- /dev/null +++ b/src/Agentic/Graph/GraphRunResult.cs @@ -0,0 +1,55 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// The outcome of a human-in-the-loop graph run: either the run completed, or it paused before an +/// interrupt node awaiting input. Carries the state either way. +/// +/// The graph's state type. +/// +/// +/// When a graph has interrupt points (see ), a run +/// stops just before such a node and returns an interrupted result with +/// set to that node. Call the run again on the same thread to resume past the pause (optionally editing +/// the state with the human's input first). +/// +/// For Beginners: Tells you whether the graph finished or stopped to ask a human. If it +/// stopped, says which step it's waiting to run, and +/// is the data so far so you can review/edit it before continuing. +/// +/// +public sealed class GraphRunResult +{ + private GraphRunResult(bool isComplete, TState state, string? interruptedBefore) + { + IsComplete = isComplete; + State = state; + InterruptedBefore = interruptedBefore; + } + + /// Gets a value indicating whether the run reached the end node. + public bool IsComplete { get; } + + /// Gets a value indicating whether the run paused at an interrupt point. + public bool IsInterrupted => InterruptedBefore is not null; + + /// Gets the state (final when complete, or as of the pause when interrupted). + public TState State { get; } + + /// Gets the node the run paused before, or null when the run completed. + public string? InterruptedBefore { get; } + + /// Creates a completed result. + /// The final state. + /// A completed . + public static GraphRunResult Complete(TState state) => new(true, state, null); + + /// Creates an interrupted result. + /// The state as of the pause. + /// The node the run paused before. + /// An interrupted . + public static GraphRunResult Interrupted(TState state, string node) + { + Guard.NotNullOrWhiteSpace(node); + return new GraphRunResult(false, state, node); + } +} diff --git a/src/Agentic/Graph/StateGraph.cs b/src/Agentic/Graph/StateGraph.cs index b69bb8fef8..7aebac9318 100644 --- a/src/Agentic/Graph/StateGraph.cs +++ b/src/Agentic/Graph/StateGraph.cs @@ -44,6 +44,7 @@ public sealed class StateGraph private readonly Dictionary>> _nodes = new(StringComparer.Ordinal); private readonly Dictionary _edges = new(StringComparer.Ordinal); private readonly Dictionary> _conditionalEdges = new(StringComparer.Ordinal); + private readonly HashSet _interruptBefore = new(StringComparer.Ordinal); private string? _entryPoint; /// @@ -127,6 +128,24 @@ public StateGraph AddConditionalEdges(string from, Func return this; } + /// + /// Marks a node as a human-in-the-loop interrupt point: a run pauses just before this node so a human + /// can review (and optionally edit) the state, then resume. + /// + /// The node to pause before. + /// This builder, for chaining. + /// Thrown when is empty/whitespace. + /// + /// Interrupts are honored by ; + /// the plain InvokeAsync overloads run straight through without pausing. + /// + public StateGraph AddInterruptBefore(string name) + { + Guard.NotNullOrWhiteSpace(name); + _interruptBefore.Add(name); + return this; + } + /// /// Sets the node where execution begins. /// @@ -185,11 +204,20 @@ public CompiledStateGraph Compile() } } + foreach (var node in _interruptBefore) + { + if (!_nodes.ContainsKey(node)) + { + throw new InvalidOperationException($"Interrupt node '{node}' is not a registered node."); + } + } + // Snapshot so post-compile builder mutations don't affect the compiled graph. return new CompiledStateGraph( new Dictionary>>(_nodes, StringComparer.Ordinal), new Dictionary(_edges, StringComparer.Ordinal), new Dictionary>(_conditionalEdges, StringComparer.Ordinal), + new HashSet(_interruptBefore, StringComparer.Ordinal), entry); } } diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphInterruptTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphInterruptTests.cs new file mode 100644 index 0000000000..3cd87206ca --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphInterruptTests.cs @@ -0,0 +1,81 @@ +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using AiDotNet.Agentic.Graph.Checkpointing; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class GraphInterruptTests + { + // draft: +1, approve: *10, draft -> approve -> End, pause before "approve". + private static CompiledStateGraph BuildApprovalGraph() => + new StateGraph() + .AddNode("draft", s => s + 1) + .AddNode("approve", s => s * 10) + .AddEdge("draft", "approve") + .AddEdge("approve", StateGraph.End) + .AddInterruptBefore("approve") + .SetEntryPoint("draft") + .Compile(); + + [Fact(Timeout = 60000)] + public async Task RunAsync_PausesBeforeInterruptNode_ThenResumesToCompletion() + { + var graph = BuildApprovalGraph(); + var cp = new InMemoryGraphCheckpointer(); + + var first = await graph.RunAsync(1, cp, "t1"); + Assert.True(first.IsInterrupted); + Assert.Equal("approve", first.InterruptedBefore); + Assert.Equal(2, first.State); // draft ran (1 -> 2); approve did not + + var second = await graph.RunAsync(0, cp, "t1"); // initialState ignored on resume + Assert.True(second.IsComplete); + Assert.Equal(20, second.State); // approve ran (2 -> 20) + } + + [Fact(Timeout = 60000)] + public async Task RunAsync_AppliesHumanEdit_OnResume() + { + var graph = BuildApprovalGraph(); + var cp = new InMemoryGraphCheckpointer(); + + var first = await graph.RunAsync(1, cp, "t2"); + Assert.True(first.IsInterrupted); + + // Human edits the state (adds 100) before approving. + var second = await graph.RunAsync(0, cp, "t2", applyOnResume: s => s + 100); + Assert.True(second.IsComplete); + Assert.Equal(1020, second.State); // (2 + 100) * 10 + } + + [Fact(Timeout = 60000)] + public async Task RunAsync_CanInterruptBeforeEntryNode() + { + var graph = new StateGraph() + .AddNode("start", s => s + 1) + .AddEdge("start", StateGraph.End) + .AddInterruptBefore("start") + .SetEntryPoint("start") + .Compile(); + var cp = new InMemoryGraphCheckpointer(); + + var first = await graph.RunAsync(5, cp, "t3"); + Assert.True(first.IsInterrupted); + Assert.Equal("start", first.InterruptedBefore); + Assert.Equal(5, first.State); // nothing ran yet + + var second = await graph.RunAsync(0, cp, "t3"); + Assert.True(second.IsComplete); + Assert.Equal(6, second.State); + } + + [Fact(Timeout = 60000)] + public async Task InvokeAsync_RunsStraightThrough_IgnoringInterrupts() + { + var graph = BuildApprovalGraph(); + var result = await graph.InvokeAsync(1); // plain run, no checkpointer + Assert.Equal(20, result); // (1 + 1) * 10 — did not pause + } + } +} From e86ece8e57c01c534f1c534cca079d104a9cd501 Mon Sep 17 00:00:00 2001 From: franklinic Date: Tue, 9 Jun 2026 07:27:15 -0400 Subject: [PATCH 4/8] =?UTF-8?q?feat(agentic):=20Phase=201=20=E2=80=94=20su?= =?UTF-8?q?bgraphs,=20reward-gated=20edges,=20deterministic=20replay=20(#1?= =?UTF-8?q?544)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - StateGraph.AddSubgraph(name, compiledGraph): run a whole graph as one step (composition). - StateGraph.AddRewardGatedEdges(from, reward, threshold, ifMeets, ifBelow): route by a critic/reward score (differentiator); sugar over conditional edges. - CompiledStateGraph.ReplayAsync + GetRecordedStateAsync: deterministically reproduce a recorded run's (node,state) trajectory from checkpoint history without executing nodes. - 3 tests; 19 graph tests total green on net10.0 + net471. No null-forgiving. Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 --- src/Agentic/Graph/CompiledStateGraph.cs | 57 ++++++++++++++++ src/Agentic/Graph/StateGraph.cs | 40 +++++++++++ .../Agentic/Graph/GraphCompositionTests.cs | 68 +++++++++++++++++++ 3 files changed, 165 insertions(+) create mode 100644 tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCompositionTests.cs diff --git a/src/Agentic/Graph/CompiledStateGraph.cs b/src/Agentic/Graph/CompiledStateGraph.cs index 5af1c6a343..09e0214c74 100644 --- a/src/Agentic/Graph/CompiledStateGraph.cs +++ b/src/Agentic/Graph/CompiledStateGraph.cs @@ -293,6 +293,63 @@ public async IAsyncEnumerable> StreamAsync( } } + /// + /// Deterministically replays a recorded run from its checkpoint history WITHOUT executing any nodes, + /// yielding the same (node, state) sequence the original run produced. + /// + /// The checkpoint store holding the recorded run. + /// The recorded run/thread id. + /// Token used to cancel the replay. + /// The recorded per-node updates, in order. + /// + /// For Beginners: Re-watch a past run exactly as it happened, reading from the saved + /// checkpoints instead of re-running the (possibly slow or non-deterministic) nodes. Great for + /// debugging and audit. + /// + /// + public async IAsyncEnumerable> ReplayAsync( + IGraphCheckpointer checkpointer, + string threadId, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpointer); + Guard.NotNullOrWhiteSpace(threadId); + + var history = await checkpointer.GetHistoryAsync(threadId, cancellationToken).ConfigureAwait(false); + for (var k = 1; k < history.Count; k++) + { + cancellationToken.ThrowIfCancellationRequested(); + // The node that ran to produce history[k] is the one recorded as "next" by history[k-1]. + var executedNode = history[k - 1].NextNode; + yield return new GraphStepUpdate(executedNode, history[k].State); + } + } + + /// + /// Returns the final recorded state for a thread (read from checkpoints, no execution). + /// + /// The checkpoint store. + /// The run/thread id. + /// Token used to cancel the operation. + /// The latest recorded state. + /// Thrown when the thread has no checkpoints. + public async Task GetRecordedStateAsync( + IGraphCheckpointer checkpointer, + string threadId, + CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpointer); + Guard.NotNullOrWhiteSpace(threadId); + + var latest = await checkpointer.GetLatestAsync(threadId, cancellationToken).ConfigureAwait(false); + if (latest is null) + { + throw new InvalidOperationException($"No checkpoints found for thread '{threadId}'."); + } + + return latest.State; + } + private static int ResolveMaxSteps(GraphRunOptions? options) { var max = options?.MaxSteps ?? new GraphRunOptions().MaxSteps; diff --git a/src/Agentic/Graph/StateGraph.cs b/src/Agentic/Graph/StateGraph.cs index 7aebac9318..405b3b7649 100644 --- a/src/Agentic/Graph/StateGraph.cs +++ b/src/Agentic/Graph/StateGraph.cs @@ -128,6 +128,46 @@ public StateGraph AddConditionalEdges(string from, Func return this; } + /// + /// Adds a node that runs an entire compiled graph as a single step (a subgraph), threading the same + /// state in and out. Enables composition: build small graphs and embed them in larger ones. + /// + /// The node name for the subgraph. + /// The compiled graph to run when this node executes. + /// This builder, for chaining. + /// Thrown when is null. + public StateGraph AddSubgraph(string name, CompiledStateGraph subgraph) + { + Guard.NotNull(subgraph); + return AddNode(name, (state, ct) => subgraph.InvokeAsync(state, null, ct)); + } + + /// + /// Adds reward-gated routing after a node: a scoring function rates the state, and flow goes to + /// when the score is at least , else to + /// . A differentiator for verifier/critic-driven control flow. + /// + /// The source node name. + /// Scores the current state (e.g., a critic/reward model). + /// The minimum score required to take the "meets" branch. + /// The next node when reward(state) >= threshold (may be ). + /// The next node otherwise (may be ). + /// This builder, for chaining. + /// Thrown when is null. + /// Thrown when a node name is empty/whitespace. + public StateGraph AddRewardGatedEdges( + string from, + Func reward, + double threshold, + string ifMeetsThreshold, + string ifBelowThreshold) + { + Guard.NotNull(reward); + Guard.NotNullOrWhiteSpace(ifMeetsThreshold); + Guard.NotNullOrWhiteSpace(ifBelowThreshold); + return AddConditionalEdges(from, state => reward(state) >= threshold ? ifMeetsThreshold : ifBelowThreshold); + } + /// /// Marks a node as a human-in-the-loop interrupt point: a run pauses just before this node so a human /// can review (and optionally edit) the state, then resume. diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCompositionTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCompositionTests.cs new file mode 100644 index 0000000000..c4aff98c9e --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCompositionTests.cs @@ -0,0 +1,68 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using AiDotNet.Agentic.Graph.Checkpointing; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class GraphCompositionTests + { + [Fact(Timeout = 60000)] + public async Task Subgraph_RunsAsSingleStep() + { + var inner = new StateGraph() + .AddNode("plus1", s => s + 1) + .AddEdge("plus1", StateGraph.End) + .SetEntryPoint("plus1") + .Compile(); + + var outer = new StateGraph() + .AddNode("pre", s => s * 2) + .AddSubgraph("sub", inner) + .AddEdge("pre", "sub") + .AddEdge("sub", StateGraph.End) + .SetEntryPoint("pre") + .Compile(); + + Assert.Equal(7, await outer.InvokeAsync(3)); // (3*2) then +1 + } + + [Fact(Timeout = 60000)] + public async Task RewardGatedEdges_LoopUntilThresholdMet() + { + var graph = new StateGraph() + .AddNode("work", s => s + 1) + .AddRewardGatedEdges("work", reward: s => s, threshold: 3, ifMeetsThreshold: StateGraph.End, ifBelowThreshold: "work") + .SetEntryPoint("work") + .Compile(); + + Assert.Equal(3, await graph.InvokeAsync(0)); // loops work until score >= 3 + } + + [Fact(Timeout = 60000)] + public async Task Replay_ReproducesRecordedTrajectory_WithoutExecuting() + { + var graph = new StateGraph() + .AddNode("a", s => s + 1) + .AddNode("b", s => s + 10) + .AddEdge("a", "b") + .AddEdge("b", StateGraph.End) + .SetEntryPoint("a") + .Compile(); + var cp = new InMemoryGraphCheckpointer(); + await graph.InvokeAsync(0, cp, "r1"); + + var replay = new List>(); + await foreach (var u in graph.ReplayAsync(cp, "r1")) + { + replay.Add(u); + } + + Assert.Equal(new[] { "a", "b" }, replay.Select(u => u.NodeName)); + Assert.Equal(new[] { 1, 11 }, replay.Select(u => u.State)); + Assert.Equal(11, await graph.GetRecordedStateAsync(cp, "r1")); + } + } +} From 577d5665bce40878740fc55a2923bad4160a237f Mon Sep 17 00:00:00 2001 From: franklinic Date: Tue, 9 Jun 2026 07:39:43 -0400 Subject: [PATCH 5/8] =?UTF-8?q?feat(agentic):=20Phase=201=20=E2=80=94=20dy?= =?UTF-8?q?namic=20fan-out=20/=20map-reduce=20nodes=20(#1544)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StateGraph.AddFanOutNode(map, branch, reduce, maxDegreeOfParallelism?): derives items from state, runs branches in parallel (optional SemaphoreSlim throttle), and reduces results back into the typed state — the typed equivalent of LangGraph Send/map-reduce, as a single composable node. 4 tests (map-reduce, throttled, empty, composed); 23 graph tests total green on net10.0 + net471. No null-forgiving. Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 --- src/Agentic/Graph/StateGraph.cs | 66 +++++++++++++++++++ .../Agentic/Graph/GraphFanOutTests.cs | 63 ++++++++++++++++++ 2 files changed, 129 insertions(+) create mode 100644 tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphFanOutTests.cs diff --git a/src/Agentic/Graph/StateGraph.cs b/src/Agentic/Graph/StateGraph.cs index 405b3b7649..984d0d61b4 100644 --- a/src/Agentic/Graph/StateGraph.cs +++ b/src/Agentic/Graph/StateGraph.cs @@ -168,6 +168,72 @@ public StateGraph AddRewardGatedEdges( return AddConditionalEdges(from, state => reward(state) >= threshold ? ifMeetsThreshold : ifBelowThreshold); } + /// + /// Adds a dynamic fan-out (map-reduce) node: it derives a set of items from the current state, runs a + /// branch over each item in parallel, then reduces the branch results back into the state. This is the + /// typed equivalent of LangGraph's Send/map-reduce. + /// + /// The per-branch input item type. + /// The per-branch result type. + /// The node name. + /// Derives the items to fan out over from the current state. + /// The work run for each item (in parallel). + /// Merges the ordered branch results back into the state. + /// Optional cap on concurrent branches; null runs all at once. + /// This builder, for chaining. + /// Thrown when , , or is null. + /// Thrown when is less than 1. + public StateGraph AddFanOutNode( + string name, + Func> map, + Func> branch, + Func, TState> reduce, + int? maxDegreeOfParallelism = null) + { + Guard.NotNull(map); + Guard.NotNull(branch); + Guard.NotNull(reduce); + if (maxDegreeOfParallelism is { } dop && dop < 1) + { + throw new ArgumentOutOfRangeException(nameof(maxDegreeOfParallelism), "Max degree of parallelism must be at least 1."); + } + + return AddNode(name, async (state, ct) => + { + var items = (map(state) ?? Enumerable.Empty()).ToList(); + IReadOnlyList results = await RunBranchesAsync(items, branch, maxDegreeOfParallelism, ct).ConfigureAwait(false); + return reduce(state, results); + }); + } + + private static async Task RunBranchesAsync( + IReadOnlyList items, + Func> branch, + int? maxDegreeOfParallelism, + CancellationToken cancellationToken) + { + if (maxDegreeOfParallelism is not { } dop) + { + return await Task.WhenAll(items.Select(item => branch(item, cancellationToken))).ConfigureAwait(false); + } + + using var throttle = new SemaphoreSlim(dop); + async Task RunThrottled(TItem item) + { + await throttle.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + return await branch(item, cancellationToken).ConfigureAwait(false); + } + finally + { + throttle.Release(); + } + } + + return await Task.WhenAll(items.Select(RunThrottled)).ConfigureAwait(false); + } + /// /// Marks a node as a human-in-the-loop interrupt point: a run pauses just before this node so a human /// can review (and optionally edit) the state, then resume. diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphFanOutTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphFanOutTests.cs new file mode 100644 index 0000000000..5c62abb9ef --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphFanOutTests.cs @@ -0,0 +1,63 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class GraphFanOutTests + { + // Fan out over 1..N, square each branch in parallel, reduce by summing into the state. + private static CompiledStateGraph BuildSquareSumGraph(int? maxParallelism = null) => + new StateGraph() + .AddFanOutNode( + "squareSum", + map: s => Enumerable.Range(1, s), + branch: async (i, ct) => { await Task.Yield(); return i * i; }, + reduce: (s, results) => results.Sum(), + maxDegreeOfParallelism: maxParallelism) + .AddEdge("squareSum", StateGraph.End) + .SetEntryPoint("squareSum") + .Compile(); + + [Fact(Timeout = 60000)] + public async Task FanOut_MapsBranchesAndReduces() + { + var graph = BuildSquareSumGraph(); + Assert.Equal(14, await graph.InvokeAsync(3)); // 1 + 4 + 9 + } + + [Fact(Timeout = 60000)] + public async Task FanOut_RespectsParallelismCap_StillCorrect() + { + var graph = BuildSquareSumGraph(maxParallelism: 2); + Assert.Equal(30, await graph.InvokeAsync(4)); // 1 + 4 + 9 + 16 + } + + [Fact(Timeout = 60000)] + public async Task FanOut_EmptyItemSet_ReducesOverNothing() + { + var graph = BuildSquareSumGraph(); + Assert.Equal(0, await graph.InvokeAsync(0)); // no items -> Sum() == 0 + } + + [Fact(Timeout = 60000)] + public async Task FanOut_ComposesWithOtherNodes() + { + var graph = new StateGraph() + .AddNode("seed", s => s + 2) // 1 -> 3 + .AddFanOutNode( + "squareSum", + map: s => Enumerable.Range(1, s), + branch: (i, ct) => Task.FromResult(i * i), + reduce: (s, results) => results.Sum()) + .AddEdge("seed", "squareSum") + .AddEdge("squareSum", StateGraph.End) + .SetEntryPoint("seed") + .Compile(); + + Assert.Equal(14, await graph.InvokeAsync(1)); // seed: 1->3, then 1+4+9 + } + } +} From f53e21f32c1f12b924f31c8912e9ebdcd95458ef Mon Sep 17 00:00:00 2001 From: franklinic Date: Tue, 9 Jun 2026 08:08:12 -0400 Subject: [PATCH 6/8] =?UTF-8?q?feat(agentic):=20Phase=201=20=E2=80=94=20du?= =?UTF-8?q?rable=20checkpointers=20(JSON-file=20+=20SQLite)=20(#1544)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - JsonFileGraphCheckpointer (core, zero new deps): persists all threads' checkpoints to a JSON file for single-process durability across restarts. - SqliteGraphCheckpointer (opt-in AiDotNet.Storage.Sqlite): durable DB-backed checkpoints (table + JSON state column) for cross-process/cross-machine durability; keeps the SQLite dep out of core. Uses explicit arg checks (core's Guard is internal). - Durability tests prove a fresh checkpointer instance over the same store reads history + resumes. JSON test runs on both TFMs; SQLite test is net10-gated because the e_sqlite3 native lib isn't deployed to the net471 shadow-copy test host (pre-existing, repo-wide; also affects SqliteSqlSyntaxValidatorTests). 25 graph tests green net10.0 / 24 net471. No null-forgiving. Postgres/Redis checkpointers are interface-ready follow-ups (need live servers; not unit-testable in CI). Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 --- .../JsonFileGraphCheckpointer.cs | 138 +++++++++++++++ .../SqliteGraphCheckpointer.cs | 167 ++++++++++++++++++ .../Graph/GraphDurableCheckpointerTests.cs | 94 ++++++++++ 3 files changed, 399 insertions(+) create mode 100644 src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs create mode 100644 src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs create mode 100644 tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphDurableCheckpointerTests.cs diff --git a/src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs b/src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs new file mode 100644 index 0000000000..c8f89da1dd --- /dev/null +++ b/src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs @@ -0,0 +1,138 @@ +using System.IO; +using Newtonsoft.Json; + +namespace AiDotNet.Agentic.Graph.Checkpointing; + +/// +/// A durable that persists all threads' checkpoints to a single +/// JSON file, so runs survive process restarts. Zero extra dependencies (uses Newtonsoft.Json). +/// +/// The graph's state type (must be JSON-serializable). +/// +/// +/// Simple and self-contained: every save reads, mutates, and rewrites the file under an in-process lock, +/// so it is correct for single-process durability but not tuned for high-throughput or multi-process +/// concurrent writers. For those, use a database-backed checkpointer (e.g., SQLite in +/// AiDotNet.Storage.Sqlite). +/// +/// For Beginners: Saves your graph's progress to a file on disk, so if the app restarts you +/// can resume right where it left off. +/// +/// +public sealed class JsonFileGraphCheckpointer : IGraphCheckpointer +{ + private readonly object _gate = new(); + private readonly string _path; + + /// + /// Initializes the checkpointer, creating the containing directory if needed. + /// + /// The JSON file used to persist checkpoints. + /// Thrown when is null. + /// Thrown when is empty/whitespace. + public JsonFileGraphCheckpointer(string filePath) + { + Guard.NotNullOrWhiteSpace(filePath); + _path = filePath; + var dir = Path.GetDirectoryName(filePath); + if (!string.IsNullOrEmpty(dir) && !Directory.Exists(dir)) + { + Directory.CreateDirectory(dir); + } + } + + /// + public Task SaveAsync(GraphCheckpoint checkpoint, CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpoint); + lock (_gate) + { + var data = Load(); + if (!data.TryGetValue(checkpoint.ThreadId, out var list)) + { + list = new List>(); + data[checkpoint.ThreadId] = list; + } + + list.Add(checkpoint); + Persist(data); + } + + return Task.CompletedTask; + } + + /// + public Task?> GetLatestAsync(string threadId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + lock (_gate) + { + var data = Load(); + if (data.TryGetValue(threadId, out var list) && list.Count > 0) + { + return Task.FromResult?>(list[list.Count - 1]); + } + } + + return Task.FromResult?>(null); + } + + /// + public Task?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + Guard.NotNull(checkpointId); + lock (_gate) + { + var data = Load(); + if (data.TryGetValue(threadId, out var list)) + { + foreach (var cp in list) + { + if (cp.CheckpointId == checkpointId) + { + return Task.FromResult?>(cp); + } + } + } + } + + return Task.FromResult?>(null); + } + + /// + public Task>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + lock (_gate) + { + var data = Load(); + IReadOnlyList> history = data.TryGetValue(threadId, out var list) + ? list.ToList() + : new List>(); + return Task.FromResult(history); + } + } + + private Dictionary>> Load() + { + if (!File.Exists(_path)) + { + return new Dictionary>>(StringComparer.Ordinal); + } + + var json = File.ReadAllText(_path); + if (string.IsNullOrWhiteSpace(json)) + { + return new Dictionary>>(StringComparer.Ordinal); + } + + var data = JsonConvert.DeserializeObject>>>(json); + return data ?? new Dictionary>>(StringComparer.Ordinal); + } + + private void Persist(Dictionary>> data) + { + File.WriteAllText(_path, JsonConvert.SerializeObject(data, Formatting.None)); + } +} diff --git a/src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs b/src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs new file mode 100644 index 0000000000..1631b907ec --- /dev/null +++ b/src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs @@ -0,0 +1,167 @@ +using System.Collections.Generic; +using System.Data.Common; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Data.Sqlite; +using Newtonsoft.Json; + +namespace AiDotNet.Agentic.Graph.Checkpointing; + +/// +/// A SQLite-backed that persists graph checkpoints in a database +/// table, giving durable resume and time-travel across process restarts and machines. Ships in the opt-in +/// AiDotNet.Storage.Sqlite package so the core stays free of the SQLite dependency. +/// +/// The graph's state type (serialized to JSON for storage). +/// +/// For Beginners: Saves your graph's progress into a SQLite database file. Point it at a +/// connection string and runs survive restarts — resume or rewind by thread id. +/// +/// +public sealed class SqliteGraphCheckpointer : IGraphCheckpointer +{ + private readonly string _connectionString; + + /// + /// Initializes the checkpointer with a SQLite connection string (e.g. Data Source=graph.db). + /// + /// The SQLite connection string. + /// Thrown when is null. + /// Thrown when is empty/whitespace. + public SqliteGraphCheckpointer(string connectionString) + { + if (string.IsNullOrWhiteSpace(connectionString)) + { + throw new ArgumentException("Connection string cannot be empty.", nameof(connectionString)); + } + + _connectionString = connectionString; + } + + /// + public async Task SaveAsync(GraphCheckpoint checkpoint, CancellationToken cancellationToken = default) + { + if (checkpoint is null) + { + throw new ArgumentNullException(nameof(checkpoint)); + } + + using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false); + using var command = connection.CreateCommand(); + command.CommandText = + "INSERT INTO GraphCheckpoints (ThreadId, CheckpointId, Step, NextNode, StateJson) " + + "VALUES ($threadId, $checkpointId, $step, $nextNode, $stateJson);"; + command.Parameters.AddWithValue("$threadId", checkpoint.ThreadId); + command.Parameters.AddWithValue("$checkpointId", checkpoint.CheckpointId); + command.Parameters.AddWithValue("$step", checkpoint.Step); + command.Parameters.AddWithValue("$nextNode", checkpoint.NextNode); + command.Parameters.AddWithValue("$stateJson", JsonConvert.SerializeObject(checkpoint.State)); + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + + /// + public async Task?> GetLatestAsync(string threadId, CancellationToken cancellationToken = default) + { + if (threadId is null) + { + throw new ArgumentNullException(nameof(threadId)); + } + + using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false); + using var command = connection.CreateCommand(); + command.CommandText = + "SELECT CheckpointId, Step, NextNode, StateJson FROM GraphCheckpoints " + + "WHERE ThreadId = $threadId ORDER BY Seq DESC LIMIT 1;"; + command.Parameters.AddWithValue("$threadId", threadId); + using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + return ReadRow(threadId, reader); + } + + return null; + } + + /// + public async Task?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default) + { + if (threadId is null) + { + throw new ArgumentNullException(nameof(threadId)); + } + + if (checkpointId is null) + { + throw new ArgumentNullException(nameof(checkpointId)); + } + + using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false); + using var command = connection.CreateCommand(); + command.CommandText = + "SELECT CheckpointId, Step, NextNode, StateJson FROM GraphCheckpoints " + + "WHERE ThreadId = $threadId AND CheckpointId = $checkpointId ORDER BY Seq DESC LIMIT 1;"; + command.Parameters.AddWithValue("$threadId", threadId); + command.Parameters.AddWithValue("$checkpointId", checkpointId); + using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + return ReadRow(threadId, reader); + } + + return null; + } + + /// + public async Task>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default) + { + if (threadId is null) + { + throw new ArgumentNullException(nameof(threadId)); + } + + var history = new List>(); + using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false); + using var command = connection.CreateCommand(); + command.CommandText = + "SELECT CheckpointId, Step, NextNode, StateJson FROM GraphCheckpoints " + + "WHERE ThreadId = $threadId ORDER BY Seq ASC;"; + command.Parameters.AddWithValue("$threadId", threadId); + using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + history.Add(ReadRow(threadId, reader)); + } + + return history; + } + + private static GraphCheckpoint ReadRow(string threadId, DbDataReader reader) + { + var checkpointId = reader.GetString(0); + var step = reader.GetInt32(1); + var nextNode = reader.GetString(2); + var stateJson = reader.GetString(3); + var state = JsonConvert.DeserializeObject(stateJson); + return new GraphCheckpoint(threadId, checkpointId, step, nextNode, state); + } + + private async Task OpenAsync(CancellationToken cancellationToken) + { + var connection = new SqliteConnection(_connectionString); + await connection.OpenAsync(cancellationToken).ConfigureAwait(false); + + using var command = connection.CreateCommand(); + command.CommandText = + "CREATE TABLE IF NOT EXISTS GraphCheckpoints (" + + "Seq INTEGER PRIMARY KEY AUTOINCREMENT, " + + "ThreadId TEXT NOT NULL, " + + "CheckpointId TEXT NOT NULL, " + + "Step INTEGER NOT NULL, " + + "NextNode TEXT NOT NULL, " + + "StateJson TEXT NOT NULL);" + + "CREATE INDEX IF NOT EXISTS IX_GraphCheckpoints_Thread ON GraphCheckpoints (ThreadId, Seq);"; + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + + return connection; + } +} diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphDurableCheckpointerTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphDurableCheckpointerTests.cs new file mode 100644 index 0000000000..dd7fb52215 --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphDurableCheckpointerTests.cs @@ -0,0 +1,94 @@ +using System.IO; +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using AiDotNet.Agentic.Graph.Checkpointing; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class GraphDurableCheckpointerTests + { + // a: +1, b: +10, a -> b -> End + private static CompiledStateGraph BuildGraph() => + new StateGraph() + .AddNode("a", s => s + 1) + .AddNode("b", s => s + 10) + .AddEdge("a", "b") + .AddEdge("b", StateGraph.End) + .SetEntryPoint("a") + .Compile(); + + // Runs a checkpointed graph via 'writer', then verifies a *fresh* checkpointer instance over the + // same backing store can read the history and resume — proving durability across instances/restarts. + private static async Task AssertDurableAsync(IGraphCheckpointer writer, IGraphCheckpointer freshReader) + { + var graph = BuildGraph(); + + var result = await graph.InvokeAsync(0, writer, "d1"); + Assert.Equal(11, result); + + var history = await freshReader.GetHistoryAsync("d1"); + Assert.Equal(3, history.Count); + Assert.True(history[history.Count - 1].IsComplete); + Assert.Equal(11, history[history.Count - 1].State); + + // Resuming a completed thread through the fresh instance returns the persisted final state. + Assert.Equal(11, await graph.InvokeAsync(999, freshReader, "d1")); + } + + [Fact(Timeout = 60000)] + public async Task JsonFileCheckpointer_PersistsAcrossInstances() + { + var path = Path.GetTempFileName(); + try + { + await AssertDurableAsync( + new JsonFileGraphCheckpointer(path), + new JsonFileGraphCheckpointer(path)); + } + finally + { + TryDelete(path); + } + } + +#if NET5_0_OR_GREATER + // SqliteGraphCheckpointer works on both target frameworks, but the native SQLite library + // (e_sqlite3) is not deployed to the .NET Framework (net471) shadow-copy test host in this + // environment — a pre-existing, repo-wide condition that also affects SqliteSqlSyntaxValidatorTests. + // The cross-TFM durability contract is covered by the JSON-file test above; this adds SQLite coverage + // on the framework where the native dependency is available. + [Fact(Timeout = 60000)] + public async Task SqliteCheckpointer_PersistsAcrossInstances() + { + var path = Path.GetTempFileName(); + var connectionString = $"Data Source={path};Pooling=False"; + try + { + await AssertDurableAsync( + new SqliteGraphCheckpointer(connectionString), + new SqliteGraphCheckpointer(connectionString)); + } + finally + { + TryDelete(path); + } + } +#endif + + private static void TryDelete(string path) + { + try + { + if (File.Exists(path)) + { + File.Delete(path); + } + } + catch (IOException) + { + // Best-effort cleanup; the OS reclaims temp files eventually. + } + } + } +} From a6550165f9b28b9ce45d23577cc3bfb8002a5b23 Mon Sep 17 00:00:00 2001 From: franklinic Date: Wed, 10 Jun 2026 22:38:23 -0400 Subject: [PATCH 7/8] =?UTF-8?q?fix(agentic):=20address=20PR=20#1548=20revi?= =?UTF-8?q?ew=20=E2=80=94=20checkpointer=20ordering/atomicity/corruption?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - IGraphCheckpointer.GetHistoryAsync: clarify the contract to chronological/append order (persistent sequence, not logical step) so replay/time-travel is consistent across backends. - InMemoryGraphCheckpointer.GetAsync: return the NEWEST matching checkpoint (LastOrDefault), aligning with the durable backends. - JsonFileGraphCheckpointer.Load: throw a descriptive InvalidOperationException on a corrupt file (with the path) instead of silently returning empty (which would let Save overwrite and lose recoverable data). - JsonFileGraphCheckpointer.Persist: atomic write — temp file + flush-to-disk + atomic replace (File.Move overwrite on net5+, File.Replace/Move fallback on net471), with temp cleanup. Co-Authored-By: Claude Opus 4.8 --- .../Graph/Checkpointing/IGraphCheckpointer.cs | 11 +++- .../InMemoryGraphCheckpointer.cs | 11 ++-- .../JsonFileGraphCheckpointer.cs | 64 ++++++++++++++++++- 3 files changed, 74 insertions(+), 12 deletions(-) diff --git a/src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs b/src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs index c258bdd92c..8521f40743 100644 --- a/src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs +++ b/src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs @@ -37,10 +37,17 @@ public interface IGraphCheckpointer Task?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default); /// - /// Gets the full checkpoint history for a thread, ordered by step ascending. + /// Gets the full history for in + /// append (chronological) order — i.e. the order checkpoints were saved, by a persistent monotonic + /// sequence (e.g. the SQLite backend orders by Seq ASC), NOT by logical step number. /// + /// + /// Replay/time-travel relies on this chronological ordering: logical step values can repeat or branch + /// (loops, re-runs from an earlier checkpoint), so ordering by step would diverge across backends. + /// Implementations MUST return checkpoints oldest-first by save order. + /// /// The run/thread id. /// Token used to cancel the operation. - /// The ordered history (possibly empty). + /// The chronologically-ordered history (possibly empty). Task>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default); } diff --git a/src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs b/src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs index f4fc8a8c5d..81c17cdaf5 100644 --- a/src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs +++ b/src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs @@ -63,13 +63,10 @@ public Task SaveAsync(GraphCheckpoint checkpoint, CancellationToken canc { if (_byThread.TryGetValue(threadId, out var list)) { - foreach (var cp in list) - { - if (cp.CheckpointId == checkpointId) - { - return Task.FromResult?>(cp); - } - } + // Return the NEWEST checkpoint with this id (history is append-ordered), matching the durable + // backends which order by descending sequence — so re-saving an id resolves to the latest. + var match = list.LastOrDefault(cp => cp.CheckpointId == checkpointId); + return Task.FromResult?>(match); } } diff --git a/src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs b/src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs index c8f89da1dd..07905b357a 100644 --- a/src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs +++ b/src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs @@ -127,12 +127,70 @@ private Dictionary>> Load() return new Dictionary>>(StringComparer.Ordinal); } - var data = JsonConvert.DeserializeObject>>>(json); - return data ?? new Dictionary>>(StringComparer.Ordinal); + try + { + var data = JsonConvert.DeserializeObject>>>(json); + return data ?? new Dictionary>>(StringComparer.Ordinal); + } + catch (JsonException ex) + { + // Surface corruption explicitly rather than silently returning empty — silently dropping the + // history would let the next Save overwrite (and permanently lose) recoverable checkpoint data. + throw new InvalidOperationException( + $"The checkpoint file at '{_path}' is corrupt and could not be parsed. " + + "Move or delete it to start a fresh checkpoint store.", ex); + } } private void Persist(Dictionary>> data) { - File.WriteAllText(_path, JsonConvert.SerializeObject(data, Formatting.None)); + var json = JsonConvert.SerializeObject(data, Formatting.None); + + // Atomic write: serialize to a temp file in the same directory, flush it to disk, then atomically + // replace the target. A crash mid-write therefore leaves the previous (valid) file intact rather than + // a half-written, unparseable one. + var directory = Path.GetDirectoryName(_path); + var tempPath = Path.Combine( + string.IsNullOrEmpty(directory) ? "." : directory, + $".{Path.GetFileName(_path)}.{Guid.NewGuid():N}.tmp"); + + try + { + using (var stream = new FileStream(tempPath, FileMode.Create, FileAccess.Write, FileShare.None)) + using (var writer = new StreamWriter(stream)) + { + writer.Write(json); + writer.Flush(); + stream.Flush(flushToDisk: true); + } + +#if NET5_0_OR_GREATER + File.Move(tempPath, _path, overwrite: true); +#else + if (File.Exists(_path)) + { + // File.Replace is atomic when the destination exists; it also clears the temp file. + File.Replace(tempPath, _path, destinationBackupFileName: null); + } + else + { + File.Move(tempPath, _path); + } +#endif + } + finally + { + if (File.Exists(tempPath)) + { + try + { + File.Delete(tempPath); + } + catch (IOException) + { + // Best-effort cleanup of the temp file; a leftover .tmp is harmless. + } + } + } } } From 256f62ccdb0fe0bd9d88782847a2ab3927492d7b Mon Sep 17 00:00:00 2001 From: franklinic Date: Wed, 10 Jun 2026 22:50:40 -0400 Subject: [PATCH 8/8] =?UTF-8?q?fix(agentic):=20address=20PR=20#1548=20revi?= =?UTF-8?q?ew=20=E2=80=94=20MaxSteps=20validation,=20Sqlite=20null-state,?= =?UTF-8?q?=20test=20cleanup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - GraphRunOptions.MaxSteps: validate > 0 in the setter (ArgumentOutOfRangeException) instead of allowing zero/negative. - SqliteGraphCheckpointer.ReadRow: throw a descriptive InvalidOperationException (with checkpointId/threadId) when the persisted state deserializes to null, rather than passing null into GraphCheckpoint. (Constructor null-checks kept as-is: Guard is internal to core and not accessible from the opt-in AiDotNet.Storage.Sqlite project.) - StateGraphTests: drop the no-op `await Task.CompletedTask;` from two synchronous tests by making them non-async Task methods returning Task.CompletedTask (preserves [Fact(Timeout)], avoids the CS1998 that deleting the await alone would introduce). Co-Authored-By: Claude Opus 4.8 --- src/Agentic/Graph/GraphRunOptions.cs | 20 ++++++++++++++++++- .../SqliteGraphCheckpointer.cs | 7 +++++++ .../Agentic/Graph/StateGraphTests.cs | 8 ++++---- 3 files changed, 30 insertions(+), 5 deletions(-) diff --git a/src/Agentic/Graph/GraphRunOptions.cs b/src/Agentic/Graph/GraphRunOptions.cs index 7dea5db4e0..009c8e787d 100644 --- a/src/Agentic/Graph/GraphRunOptions.cs +++ b/src/Agentic/Graph/GraphRunOptions.cs @@ -1,3 +1,5 @@ +using System; + namespace AiDotNet.Agentic.Graph; /// @@ -11,9 +13,25 @@ namespace AiDotNet.Agentic.Graph; /// public sealed class GraphRunOptions { + private int _maxSteps = 25; + /// /// Gets or sets the maximum number of node executions allowed in a single run before a /// is thrown. Default: 25. Must be positive. /// - public int MaxSteps { get; set; } = 25; + /// Thrown when set to a value <= 0. + public int MaxSteps + { + get => _maxSteps; + set + { + if (value <= 0) + { + throw new ArgumentOutOfRangeException( + nameof(value), value, "MaxSteps must be positive (a run needs at least one step)."); + } + + _maxSteps = value; + } + } } diff --git a/src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs b/src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs index 1631b907ec..93ef9484a0 100644 --- a/src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs +++ b/src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs @@ -142,6 +142,13 @@ private static GraphCheckpoint ReadRow(string threadId, DbDataReader rea var nextNode = reader.GetString(2); var stateJson = reader.GetString(3); var state = JsonConvert.DeserializeObject(stateJson); + if (state is null) + { + throw new InvalidOperationException( + $"Checkpoint '{checkpointId}' for thread '{threadId}' has a null or unparseable state payload " + + "and cannot be reconstructed."); + } + return new GraphCheckpoint(threadId, checkpointId, step, nextNode, state); } diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs index 704c81c40a..b14dbb783e 100644 --- a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs @@ -87,7 +87,7 @@ public async Task AsyncNode_IsAwaited() } [Fact(Timeout = 60000)] - public async Task Compile_Validates_EntryPointEdgesAndExclusiveRouting() + public Task Compile_Validates_EntryPointEdgesAndExclusiveRouting() { // No entry point. Assert.Throws(() => @@ -106,16 +106,16 @@ public async Task Compile_Validates_EntryPointEdgesAndExclusiveRouting() .SetEntryPoint("a") .Compile()); - await Task.CompletedTask; + return Task.CompletedTask; } [Fact(Timeout = 60000)] - public async Task AddNode_RejectsDuplicateAndReservedNames() + public Task AddNode_RejectsDuplicateAndReservedNames() { var graph = new StateGraph().AddNode("a", s => s); Assert.Throws(() => graph.AddNode("a", s => s)); // duplicate Assert.Throws(() => new StateGraph().AddNode(StateGraph.End, s => s)); // reserved - await Task.CompletedTask; + return Task.CompletedTask; } } }